Coverage for app/venv/lib/python3.14/site-packages/weblate/trans/views/create.py: 19%
415 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 07:15 +0000
1# Copyright © Michal Čihař <michal@weblate.org>
2#
3# SPDX-License-Identifier: GPL-3.0-or-later
4from __future__ import annotations
6import json
7import os
8import subprocess
9from contextlib import suppress
10from typing import TYPE_CHECKING
11from zipfile import BadZipfile
13from django.conf import settings
14from django.contrib.auth.decorators import login_required
15from django.forms import HiddenInput
16from django.shortcuts import redirect
17from django.urls import reverse
18from django.utils.decorators import method_decorator
19from django.utils.functional import cached_property
20from django.utils.http import urlencode
21from django.utils.translation import gettext
22from django.views.generic.edit import CreateView
24from weblate.trans.backups import ProjectBackup
25from weblate.trans.forms import (
26 ComponentBranchForm,
27 ComponentCreateForm,
28 ComponentDiscoverForm,
29 ComponentDocCreateForm,
30 ComponentInitCreateForm,
31 ComponentScratchCreateForm,
32 ComponentSelectForm,
33 ComponentZipCreateForm,
34 ProjectCreateForm,
35 ProjectImportCreateForm,
36 ProjectImportForm,
37)
38from weblate.trans.models import Category, Component, Project
39from weblate.trans.tasks import perform_update
40from weblate.trans.util import get_clean_env
41from weblate.utils import messages
42from weblate.utils.errors import report_error
43from weblate.utils.licenses import LICENSE_URLS
44from weblate.utils.ratelimit import session_ratelimit_post
45from weblate.utils.views import create_component_from_doc, create_component_from_zip
46from weblate.vcs.models import VCS_REGISTRY
48if TYPE_CHECKING: 48 ↛ 49line 48 didn't jump to line 49 because the condition on line 48 was never true
49 from django.forms import Form
51 from weblate.auth.models import AuthenticatedHttpRequest
52 from weblate.trans.forms import (
53 ComponentProjectForm,
54 )
55 from weblate.trans.models.component import ComponentQuerySet
57SESSION_CREATE_KEY = "session_component"
60class BaseCreateView(CreateView):
61 request: AuthenticatedHttpRequest
63 def __init__(self, **kwargs) -> None:
64 super().__init__(**kwargs)
65 self.has_billing = "weblate.billing" in settings.INSTALLED_APPS
67 def get_form_kwargs(self):
68 kwargs = super().get_form_kwargs()
69 kwargs["request"] = self.request
70 return kwargs
72 def form_invalid(self, form):
73 messages.error(
74 self.request,
75 gettext(
76 "The supplied configuration is incorrect. Please check the errors below.",
77 ),
78 )
79 return super().form_invalid(form)
82@method_decorator(login_required, name="dispatch")
83@method_decorator(session_ratelimit_post("project"), name="dispatch")
84class CreateProject(BaseCreateView):
85 model = Project
86 object: Project
87 form_class: type[Form] = ProjectCreateForm
88 billings = None
90 def get_form(self, form_class=None):
91 form = super().get_form(form_class)
92 if "billing" in form.fields:
93 billing_field = form.fields["billing"]
94 if self.has_billing:
95 billing_field.queryset = self.billings
96 with suppress(ValueError, KeyError):
97 billing_field.initial = int(self.request.GET["billing"])
98 billing_field.required = not self.request.user.is_superuser
99 if self.request.user.is_superuser:
100 billing_field.empty_label = "-- without billing --"
101 else:
102 billing_field.required = False
103 billing_field.widget = HiddenInput()
104 return form
106 def form_valid(self, form):
107 result = super().form_valid(form)
108 if self.has_billing and form.cleaned_data["billing"]:
109 billing = form.cleaned_data["billing"]
110 else:
111 billing = None
112 self.object.post_create(self.request.user, billing)
113 return result
115 def can_create(self):
116 return (self.has_billing and self.billings) or self.request.user.has_perm(
117 "project.add"
118 )
120 def post(self, request: AuthenticatedHttpRequest, *args, **kwargs): # type: ignore[override]
121 if not self.can_create():
122 return redirect("create-project")
123 return super().post(request, *args, **kwargs)
125 def get_context_data(self, **kwargs):
126 kwargs = super().get_context_data(**kwargs)
127 kwargs["can_create"] = self.can_create()
128 kwargs["import_form"] = self.get_form(ProjectImportForm)
129 if self.has_billing:
130 from weblate.billing.models import Billing
132 kwargs["user_billings"] = Billing.objects.for_user(
133 self.request.user
134 ).exists()
135 return kwargs
137 def dispatch(self, request: AuthenticatedHttpRequest, *args, **kwargs): # type: ignore[override]
138 if self.has_billing:
139 from weblate.billing.models import Billing
141 self.billings = Billing.objects.for_user_within_limits(request.user)
142 return super().dispatch(request, *args, **kwargs)
144 def get_success_url(self) -> str:
145 return f"{super().get_success_url()}#components"
148class ImportProject(CreateProject):
149 form_class = ProjectImportForm
150 template_name = "trans/project_import.html"
152 def setup(self, request: AuthenticatedHttpRequest, *args, **kwargs) -> None: # type: ignore[override]
153 if "import_project" in request.session and os.path.exists(
154 request.session["import_project"]
155 ):
156 if "zipfile" in request.FILES:
157 # Delete previous (stale) import data
158 del request.session["import_project"]
159 request.session.pop("import_billing", None)
160 self.projectbackup = None
161 else:
162 self.projectbackup = ProjectBackup(request.session["import_project"])
163 # The backup is already validated at this point,
164 # but we need to load the info.
165 self.projectbackup.validate()
166 else:
167 request.session.pop("import_project", None)
168 request.session.pop("import_billing", None)
169 self.projectbackup = None
170 super().setup(request, *args, **kwargs)
172 def get_form(self, form_class=None):
173 form = super().get_form(form_class)
174 if "billing" in form.fields and self.has_billing:
175 from weblate.billing.models import Billing
177 billing = self.request.session.get("import_billing")
178 if billing:
179 form.fields["billing"].initial = Billing.objects.get(pk=billing)
180 return form
182 def get_form_class(self):
183 """Return the form class to use."""
184 if self.projectbackup:
185 return ProjectImportCreateForm
186 return self.form_class
188 def get_form_kwargs(self):
189 kwargs = super().get_form_kwargs()
190 if self.projectbackup:
191 kwargs["projectbackup"] = self.projectbackup
192 return kwargs
194 def post(self, request: AuthenticatedHttpRequest, *args, **kwargs): # type: ignore[override]
195 if "zipfile" in request.FILES and self.projectbackup:
196 # Delete previous (stale) import data
197 os.unlink(self.projectbackup.filename)
198 del self.request.session["import_project"]
199 self.request.session.pop("import_billing", None)
200 self.projectbackup = None
201 return super().post(request, *args, **kwargs)
203 def form_valid(self, form):
204 if isinstance(form, ProjectImportForm):
205 # Save current zip to the import dir
206 self.request.session["import_project"] = form.cleaned_data[
207 "projectbackup"
208 ].store_for_import()
209 if form.cleaned_data["billing"]:
210 self.request.session["import_billing"] = form.cleaned_data["billing"].pk
211 return redirect("create-project-import")
212 # Perform actual import
213 project = self.projectbackup.restore(
214 project_name=form.cleaned_data["name"],
215 project_slug=form.cleaned_data["slug"],
216 user=self.request.user,
217 billing=form.cleaned_data["billing"],
218 )
219 del self.request.session["import_project"]
220 return redirect(project)
223@method_decorator(login_required, name="dispatch")
224class CreateComponent(BaseCreateView):
225 model = Component
226 projects = None
227 stage = None
228 selected_project = None
229 selected_category = None
230 basic_fields = ("repo", "name", "slug", "vcs", "source_language")
231 empty_form = False
232 form_class: type[ComponentProjectForm] = ComponentInitCreateForm
233 origin = "vcs"
234 object: Component
235 duplicate_existing_component: int | None = None
237 def get_form_class(self):
238 """Return the form class to use."""
239 if self.stage == "create":
240 return ComponentCreateForm
241 if self.stage == "discover":
242 return ComponentDiscoverForm
243 return self.form_class
245 def get_form_kwargs(self):
246 result = super().get_form_kwargs()
247 if self.request.method != "POST":
248 if self.initial:
249 # When going from other form (for example ZIP import)
250 result.pop("data", None)
251 result.pop("files", None)
252 if self.has_all_fields() and not self.empty_form:
253 if SESSION_CREATE_KEY in self.request.session:
254 result["data"] = self.request.session[SESSION_CREATE_KEY]
255 else:
256 result["data"] = self.request.GET
257 return result
259 def get_success_url(self):
260 return reverse("show_progress", kwargs={"path": self.object.get_url_path()})
262 def warn_outdated(self, form) -> None:
263 linked = form.instance.linked_component
264 if linked:
265 perform_update.delay("Component", linked.pk, auto=True)
266 if linked.repo_needs_merge():
267 messages.warning(
268 self.request,
269 gettext(
270 "The repository is outdated, you might not get "
271 "expected results until you update it."
272 ),
273 )
275 def detect_license(self, form) -> None:
276 """Automatic license detection based on licensee."""
277 try:
278 process_result = subprocess.run(
279 ["licensee", "detect", "--json", form.instance.full_path],
280 text=True,
281 capture_output=True,
282 env=get_clean_env(),
283 check=True,
284 )
285 except FileNotFoundError:
286 return
287 except (OSError, subprocess.CalledProcessError) as error:
288 if getattr(error, "returncode", 0) != 1:
289 report_error("Failed licensee invocation")
290 return
291 result = json.loads(process_result.stdout)
292 for license_data in result["licenses"]:
293 spdx_id = license_data["spdx_id"]
294 for license_id in (f"{spdx_id}-or-later", f"{spdx_id}-only", spdx_id):
295 if license_id in LICENSE_URLS:
296 self.initial["license"] = license_id
297 messages.info(
298 self.request,
299 gettext(
300 "Detected license as %s, please check whether it is correct."
301 )
302 % license_id,
303 )
304 return
306 def form_valid(self, form):
307 if self.stage == "create":
308 form.instance.manage_units = (
309 bool(form.instance.template) or form.instance.file_format == "tbx"
310 )
311 if self.duplicate_existing_component and (
312 source_component := form.cleaned_data["source_component"]
313 ):
314 fields_to_duplicate = [
315 "agreement",
316 "merge_style",
317 "commit_message",
318 "add_message",
319 "delete_message",
320 "merge_message",
321 "addon_message",
322 "pull_message",
323 ]
324 for field in fields_to_duplicate:
325 setattr(form.instance, field, getattr(source_component, field))
327 result = super().form_valid(form)
328 self.object.post_create(self.request.user, origin=self.origin)
329 return result
330 if self.stage == "discover":
331 # Move to create
332 self.initial = form.cleaned_data
333 self.stage = "create"
334 self.request.method = "GET"
335 self.warn_outdated(form)
336 self.detect_license(form)
337 return self.get(self.request)
338 # Move to discover
339 self.stage = "discover"
340 self.request.method = "GET"
341 self.initial = form.cleaned_data
342 self.warn_outdated(form)
343 return self.get(self.request)
345 def get_form(self, form_class=None, empty=False):
346 self.empty_form = empty
347 form = super().get_form(form_class)
348 if "project" in form.fields:
349 project_field = form.fields["project"]
350 category_field = form.fields["category"]
351 project_field.queryset = self.projects
352 category_field.queryset = Category.objects.filter(project__in=self.projects)
353 project_field.empty_label = None
354 if self.selected_project:
355 project_field.initial = self.selected_project
356 with suppress(IndexError):
357 form.fields["source_language"].initial = Component.objects.filter(
358 project=self.selected_project
359 )[0].source_language_id
360 if self.selected_category:
361 category_field.initial = self.selected_category
362 self.empty_form = False
363 if "source_component" in form.fields and self.duplicate_existing_component:
364 self.components = Component.objects.filter(
365 pk=self.duplicate_existing_component
366 )
367 form.fields["source_component"].queryset = self.components
368 form.initial["source_component"] = self.duplicate_existing_component
369 return form
371 def get_context_data(self, **kwargs):
372 kwargs = super().get_context_data(**kwargs)
373 kwargs["projects"] = self.projects
374 kwargs["stage"] = self.stage
375 return kwargs
377 def fetch_params(self, request: AuthenticatedHttpRequest) -> None:
378 try:
379 self.selected_project = int(
380 request.POST.get("project", request.GET.get("project", ""))
381 )
382 except ValueError:
383 self.selected_project = None
384 try:
385 self.selected_category = int(
386 request.POST.get("category", request.GET.get("category", ""))
387 )
388 except ValueError:
389 self.selected_category = None
390 if request.user.is_superuser:
391 self.projects = Project.objects.order()
392 elif self.has_billing:
393 from weblate.billing.models import Billing
395 self.projects = request.user.managed_projects.filter(
396 billing__in=Billing.objects.get_valid()
397 ).order()
398 else:
399 self.projects = request.user.managed_projects
400 self.initial = {}
401 session_data = {}
402 if SESSION_CREATE_KEY in request.GET and SESSION_CREATE_KEY in request.session:
403 session_data = request.session[SESSION_CREATE_KEY]
404 for field in self.basic_fields:
405 if field in session_data:
406 self.initial[field] = session_data[field]
407 elif field in request.GET:
408 self.initial[field] = request.GET[field]
410 try:
411 self.duplicate_existing_component = int(request.GET.get("source_component"))
412 except (ValueError, TypeError):
413 self.duplicate_existing_component = None
415 def has_all_fields(self):
416 session_data = {}
417 if (
418 SESSION_CREATE_KEY in self.request.GET
419 and SESSION_CREATE_KEY in self.request.session
420 ):
421 session_data = self.request.session[SESSION_CREATE_KEY]
422 return self.stage == "init" and all(
423 field in session_data or field in self.request.GET
424 for field in self.basic_fields
425 )
427 def dispatch(self, request: AuthenticatedHttpRequest, *args, **kwargs): # type: ignore[override]
428 if "new_base" in request.POST:
429 self.stage = "create"
430 elif "discovery" in request.POST:
431 self.stage = "discover"
432 else:
433 self.stage = "init"
435 self.fetch_params(request)
437 # Proceed to post if all params are present
438 if self.has_all_fields():
439 return self.post(request, *args, **kwargs)
441 return super().dispatch(request, *args, **kwargs)
444class CreateFromZip(CreateComponent):
445 form_class = ComponentZipCreateForm
446 origin = "zip"
448 def form_valid(self, form):
449 if self.stage != "init":
450 return super().form_valid(form)
452 try:
453 create_component_from_zip(form.cleaned_data)
454 except (BadZipfile, OSError):
455 form.add_error("zipfile", gettext("Could not parse uploaded ZIP file."))
456 return self.form_invalid(form)
458 # Move to discover phase
459 self.stage = "discover"
460 self.initial = form.cleaned_data
461 self.initial["vcs"] = "local"
462 self.initial["repo"] = "local:"
463 self.initial["branch"] = "main"
464 self.initial.pop("zipfile")
465 self.request.method = "GET"
466 return self.get(self.request)
469class CreateFromDoc(CreateComponent):
470 form_class = ComponentDocCreateForm
471 origin = "document"
473 def form_valid(self, form):
474 if self.stage != "init":
475 return super().form_valid(form)
477 fake = create_component_from_doc(
478 form.cleaned_data,
479 form.cleaned_data.pop("docfile"),
480 form.cleaned_data.pop("target_language", None),
481 )
482 # Move to discover phase
483 self.stage = "discover"
484 self.initial = form.cleaned_data
485 self.initial["vcs"] = "local"
486 self.initial["repo"] = "local:"
487 self.initial["branch"] = "main"
488 self.initial["template"] = fake.template
489 self.initial["filemask"] = fake.filemask
491 self.request.method = "GET"
492 return self.get(self.request)
495def component_branches(repo: str) -> set[str]:
496 return set(Component.objects.filter(repo=repo).values_list("branch", flat=True))
499class CreateComponentSelection(CreateComponent):
500 template_name = "trans/component_create.html"
502 components: ComponentQuerySet
503 origin: str | None = None
504 duplicate_existing_component: int | None = None
506 @cached_property
507 def branch_data(self):
508 result = {}
509 existing_branches: dict[str, set[str]] = {}
510 for component in self.components:
511 repo = component.repo
512 if repo not in existing_branches:
513 existing_branches[repo] = component_branches(repo)
514 branches = [
515 branch
516 for branch in component.repository.list_remote_branches()
517 if branch != component.branch and branch not in existing_branches[repo]
518 ]
519 if branches:
520 result[component.pk] = branches
521 return result
523 def fetch_params(self, request: AuthenticatedHttpRequest) -> None:
524 super().fetch_params(request)
525 self.components = (
526 Component.objects.filter_access(request.user)
527 .with_repo()
528 .prefetch()
529 .filter(project__in=self.projects)
530 .order_project()
531 )
532 if self.selected_project:
533 self.components = self.components.filter(project__pk=self.selected_project)
534 self.origin = request.POST.get("origin")
536 try:
537 self.duplicate_existing_component = int(request.GET.get("component"))
538 except (ValueError, TypeError):
539 self.duplicate_existing_component = None
540 self.initial = {}
541 if self.duplicate_existing_component:
542 source_component = Component.objects.get(
543 pk=self.duplicate_existing_component
544 )
545 self.initial |= {
546 "component": source_component,
547 "is_glossary": source_component.is_glossary,
548 }
550 def get_context_data(self, **kwargs):
551 kwargs = super().get_context_data(**kwargs)
552 kwargs["components"] = self.components
553 kwargs["selected_project"] = self.selected_project
554 kwargs["existing_form"] = self.get_form(ComponentSelectForm, empty=True)
555 kwargs["branch_form"] = self.get_form(ComponentBranchForm, empty=True)
556 kwargs["branch_data"] = json.dumps(self.branch_data)
557 kwargs["full_form"] = self.get_form(ComponentInitCreateForm, empty=True)
558 if "local" in VCS_REGISTRY:
559 kwargs["zip_form"] = self.get_form(ComponentZipCreateForm, empty=True)
560 kwargs["scratch_form"] = self.get_form(
561 ComponentScratchCreateForm, empty=True
562 )
563 kwargs["doc_form"] = self.get_form(ComponentDocCreateForm, empty=True)
564 if self.origin == "branch":
565 kwargs["branch_form"] = kwargs["form"]
566 elif self.origin == "scratch":
567 kwargs["scratch_form"] = kwargs["form"]
568 else:
569 kwargs["existing_form"] = kwargs["form"]
570 return kwargs
572 def get_form(self, form_class=None, empty=False):
573 form = super().get_form(form_class, empty=empty)
574 if isinstance(form, ComponentBranchForm):
575 form.fields["component"].queryset = Component.objects.filter(
576 pk__in=self.branch_data.keys()
577 ).order_project()
578 form.branch_data = self.branch_data
579 elif isinstance(form, ComponentSelectForm):
580 if self.duplicate_existing_component:
581 self.components |= Component.objects.filter_access(
582 self.request.user
583 ).filter(pk=self.duplicate_existing_component)
584 form.fields["component"].queryset = self.components
585 return form
587 def get_form_class(self):
588 if self.origin == "branch":
589 return ComponentBranchForm
590 if self.origin == "scratch":
591 return ComponentScratchCreateForm
592 return ComponentSelectForm
594 def redirect_create(self, **kwargs):
595 # Store params in session
596 self.request.session[SESSION_CREATE_KEY] = kwargs
598 return redirect(
599 "{}?{}".format(
600 reverse("create-component-vcs"), urlencode({SESSION_CREATE_KEY: 1})
601 )
602 )
604 def form_valid(self, form):
605 if self.origin == "scratch":
606 project = form.cleaned_data["project"]
607 component = project.scratch_create_component(**form.cleaned_data)
608 component.post_create(self.request.user, origin="scratch")
609 return redirect(
610 reverse("show_progress", kwargs={"path": component.get_url_path()})
611 )
612 component = form.cleaned_data["component"]
613 if self.origin == "existing":
614 return self.redirect_create(
615 repo=component.repo or component.get_repo_link_url(),
616 project=component.project.pk,
617 category=component.category.pk if component.category else "",
618 name=form.cleaned_data["name"],
619 slug=form.cleaned_data["slug"],
620 is_glossary=form.cleaned_data["is_glossary"],
621 vcs=component.vcs,
622 source_language=component.source_language.pk,
623 license=component.license,
624 source_component=component.pk,
625 )
626 if self.origin == "branch":
627 form.instance.save()
628 form.instance.post_create(self.request.user, origin="branch")
629 return redirect(
630 reverse("show_progress", kwargs={"path": form.instance.get_url_path()})
631 )
633 return redirect("create-component")
635 def post(self, request: AuthenticatedHttpRequest, *args, **kwargs): # type: ignore[override]
636 if self.origin == "vcs":
637 kwargs = {}
638 if self.selected_project:
639 kwargs["project"] = self.selected_project
640 return self.redirect_create(**kwargs)
641 return super().post(request, *args, **kwargs)