Refactor sync

pontoon/base/models/changed_entity_locale.py

@@ -1,15 +1,21 @@
+from typing import TYPE_CHECKING
+
 from django.db import models
 from django.utils import timezone
 
 
+if TYPE_CHECKING:
+    from pontoon.base.models import Entity, Locale
+
+
 class ChangedEntityLocale(models.Model):
     """
     ManyToMany model for storing what locales have changed translations for a
     specific entity since the last sync.
     """
 
-    entity = models.ForeignKey("Entity", models.CASCADE)
-    locale = models.ForeignKey("Locale", models.CASCADE)
+    entity: models.ForeignKey["Entity"] = models.ForeignKey("Entity", models.CASCADE)
+    locale: models.ForeignKey["Locale"] = models.ForeignKey("Locale", models.CASCADE)
     when = models.DateTimeField(default=timezone.now)
 
     class Meta:

pontoon/base/models/project.py

@@ -1,11 +1,13 @@
 from collections import defaultdict
 from os.path import basename, join, normpath
+from typing import TYPE_CHECKING
 from urllib.parse import urlparse
 
 from django.conf import settings
 from django.contrib.auth.models import User
 from django.db import models
 from django.db.models import Prefetch
+from django.db.models.manager import BaseManager
 from django.utils import timezone
 from django.utils.functional import cached_property
 
@@ -15,6 +17,10 @@
 from pontoon.base.models.locale import Locale
 
 
+if TYPE_CHECKING:
+    from pontoon.base.models import Resource
+
+
 class Priority(models.IntegerChoices):
     LOWEST = 1, "Lowest"
     LOW = 2, "Low"
@@ -103,6 +109,8 @@ class Project(AggregatedStats):
     slug = models.SlugField(unique=True)
     locales = models.ManyToManyField(Locale, through="ProjectLocale")
 
+    resources: BaseManager["Resource"]
+
     class DataSource(models.TextChoices):
         REPOSITORY = "repository", "Repository"
         DATABASE = "database", "Database"

pontoon/base/tests/__init__.py

@@ -90,7 +90,9 @@ def locales(self, create, extracted, **kwargs):
 
         if extracted:
             for locale in extracted:
-                ProjectLocaleFactory.create(project=self, locale=locale)
+                ProjectLocaleFactory.create(
+                    project=self, locale=locale, total_strings=self.total_strings
+                )
 
     @factory.post_generation
     def repositories(self, create, extracted, **kwargs):

pontoon/sync/checkouts.py

@@ -0,0 +1,92 @@
+import logging
+
+from os import walk
+from os.path import join, normpath, relpath
+from typing import NamedTuple, cast
+
+from django.db.models.manager import BaseManager
+
+from pontoon.base.models import Project, Repository
+from pontoon.sync.repositories import get_repo
+
+
+log = logging.getLogger(__name__)
+
+
+class Checkout:
+    repo: Repository
+    is_source: bool
+    url: str
+    path: str
+    prev_commit: str | None
+    commit: str | None
+    changed: list[str]
+    """Relative paths from the checkout base"""
+    removed: list[str]
+    """Relative paths from the checkout base"""
+
+    def __init__(self, slug: str, db_repo: Repository, pull: bool) -> None:
+        self.repo = db_repo
+        self.is_source = db_repo.source_repo
+        self.url = db_repo.url
+        self.path = normpath(db_repo.checkout_path)
+
+        if db_repo.last_synced_revisions is None:
+            self.prev_commit = None
+        else:
+            pc = db_repo.last_synced_revisions.get("single_locale", None)
+            self.prev_commit = pc if isinstance(pc, str) else None
+
+        versioncontrol = get_repo(db_repo.type)
+        if pull:
+            log.info(f"[{slug}] Pulling updates from {self.url}")
+            versioncontrol.update(self.url, self.path, db_repo.branch)
+        else:
+            log.info(f"[{slug}] Skipping pull")
+        self.commit = versioncontrol.revision(self.path)
+        log.info(f"[{slug}] Repo {self.url} now at {self.commit}")
+
+        delta = (
+            versioncontrol.changed_files(self.path, self.prev_commit)
+            if isinstance(self.prev_commit, str)
+            else None
+        )
+        if delta is not None:
+            self.changed, self.removed = delta
+        else:
+            # Initially and on error, consider all files changed
+            self.changed = []
+            for root, dirnames, filenames in walk(self.path):
+                dirnames[:] = (dn for dn in dirnames if not dn.startswith("."))
+                rel_root = relpath(root, self.path) if root != self.path else ""
+                self.changed.extend(
+                    join(rel_root, fn) for fn in filenames if not fn.startswith(".")
+                )
+            self.removed = []
+
+
+class Checkouts(NamedTuple):
+    source: Checkout
+    target: Checkout
+
+
+def get_checkouts(project: Project, pull: bool = True) -> Checkouts:
+    """
+    For each project repository including all multi-locale repositories,
+    update its local checkout (unless `pull` is false),
+    and provide a `Checkout` representing their current state.
+    """
+    source: Checkout | None = None
+    target: Checkout | None = None
+    for repo in cast(BaseManager[Repository], project.repositories).all():
+        if repo.source_repo:
+            if source:
+                raise Exception("Multiple source repositories")
+            source = Checkout(project.slug, repo, pull)
+        elif target:
+            raise Exception("Multiple target repositories")
+        else:
+            target = Checkout(project.slug, repo, pull)
+    if source is None and target is None:
+        raise Exception("No repository found")
+    return Checkouts(source or target, target or source)

pontoon/sync/formats/__init__.py

@@ -16,6 +16,7 @@
     xliff,
     xml,
 )
+from pontoon.sync.formats.base import ParsedResource
 
 
 # To add support for a new resource format, add an entry to this dict
@@ -56,7 +57,7 @@ def are_compatible_files(file_a, file_b):
     return False
 
 
-def parse(path, source_path=None, locale=None):
+def parse(path, source_path=None, locale=None) -> ParsedResource:
     """
     Parse the resource file at the given path and return a
     ParsedResource with its translations.

pontoon/sync/formats/base.py

@@ -1,3 +1,6 @@
+from pontoon.sync.vcs.translation import VCSTranslation
+
+
 class ParsedResource:
     """
     Parent class for parsed resources as returned by parse.
@@ -6,8 +9,10 @@ class ParsedResource:
     that inherits from this class.
     """
 
+    entities: dict[str, VCSTranslation]
+
     @property
-    def translations(self):
+    def translations(self) -> list[VCSTranslation]:
         """
         Return a list of VCSTranslation instances or subclasses that
         represent the translations in the resource.

pontoon/sync/formats/po.py

@@ -64,6 +64,8 @@ def __repr__(self):
 
 
 class POResource(ParsedResource):
+    entities: list[POEntity]
+
     def __init__(self, pofile):
         self.pofile = pofile
         self.entities = [

pontoon/sync/paths.py

@@ -0,0 +1,39 @@
+import logging
+
+from os.path import join
+
+from moz.l10n.paths import L10nConfigPaths, L10nDiscoverPaths, get_android_locale
+
+from pontoon.base.models import Project
+from pontoon.sync.checkouts import Checkouts
+from pontoon.sync.vcs.project import MissingLocaleDirectoryError
+
+
+log = logging.getLogger(__name__)
+
+
+def get_paths(
+    project: Project, checkouts: Checkouts
+) -> L10nConfigPaths | L10nDiscoverPaths:
+    force_paths = [
+        join(checkouts.source.path, path) for path in checkouts.source.removed
+    ]
+    if project.configuration_file:
+        paths = L10nConfigPaths(
+            join(checkouts.source.path, project.configuration_file),
+            locale_map={"android_locale": get_android_locale},
+            force_paths=force_paths,
+        )
+        if checkouts.target != checkouts.source:
+            paths.base = checkouts.target.repo.checkout_path
+        return paths
+    else:
+        paths = L10nDiscoverPaths(
+            project.checkout_path,
+            ref_root=checkouts.source.path,
+            force_paths=force_paths,
+            source_locale=["templates", "en-US", "en"],
+        )
+        if paths.base is None:
+            raise MissingLocaleDirectoryError("Base localization directory not found")
+        return paths