Implement Po mode

checkmk_weblate_syncer/__main__.py

@@ -1,9 +1,11 @@
+import sys
 from pathlib import Path
 from typing import TypeVar
 
 from .cli import Mode, parse_arguments
-from .config import PotModeConfig
+from .config import PoModeConfig, PotModeConfig
 from .logging import LOGGER, configure_logger
+from .po import run as run_po_mode
 from .pot import run as run_pot_mode
 
 
@@ -13,16 +15,12 @@ def _main() -> None:
 
     match args.mode:
         case Mode.POT:
-            run_pot_mode(_load_config(args.config_path, PotModeConfig))
-        # TODO (apparently does not work with enums with only one variant):  # pylint: disable=fixme
-        # case _:
-        #     assert_never(args.mode)
+            sys.exit(run_pot_mode(_load_config(args.config_path, PotModeConfig)))
+        case Mode.PO:
+            sys.exit(run_po_mode(_load_config(args.config_path, PoModeConfig)))
 
 
-# TODO:  # pylint: disable=fixme
-# _ConfigTypeT = TypeVar("_ConfigTypeT", PotModeConfig, PoModeConfig)
-# apparently, type vars cannot have a single constraint, so we have to use bound for now
-_ConfigTypeT = TypeVar("_ConfigTypeT", bound=PotModeConfig)
+_ConfigTypeT = TypeVar("_ConfigTypeT", PotModeConfig, PoModeConfig)
 
 
 def _load_config(config_path: Path, config_type: type[_ConfigTypeT]) -> _ConfigTypeT:

checkmk_weblate_syncer/cli.py

@@ -9,6 +9,7 @@
 
 class Mode(Enum):
     POT = "pot"
+    PO = "po"
 
 
 def _parse_log_level(raw: int) -> int:
@@ -33,7 +34,7 @@ def parse_arguments() -> Arguments:
         type=str,
         choices=[mode.value for mode in Mode],
         metavar="MODE",
-        help="Operation mode. pot: Update pot file.",
+        help="Operation mode. pot: Update pot file. po: Update po files.",
     )
     parser.add_argument(
         "config_path",

checkmk_weblate_syncer/config.py

@@ -1,3 +1,4 @@
+from collections.abc import Sequence
 from pathlib import Path
 from typing import Annotated
 
@@ -29,3 +30,12 @@ class PotModeConfig(BaseConfig, frozen=True):
         Path, AfterValidator(_validate_path_is_relative)
     ]
     locale_pot_path: Annotated[Path, AfterValidator(_validate_path_is_relative)]
+
+
+class PoFilePair(BaseModel, frozen=True):
+    checkmk: Annotated[Path, AfterValidator(_validate_path_is_relative)]
+    locale: Annotated[Path, AfterValidator(_validate_path_is_relative)]
+
+
+class PoModeConfig(BaseConfig, frozen=True):
+    po_file_pairs: Sequence[PoFilePair]

checkmk_weblate_syncer/git.py

@@ -1,12 +1,53 @@
+from collections.abc import Sequence
 from pathlib import Path
+from subprocess import CalledProcessError
 
 from git import Repo
 
+from .config import RepositoryConfig
+from .logging import LOGGER
 
-def repository_in_clean_state(path: Path, branch: str) -> Repo:
+
+def repository_in_clean_state(
+    repo_config: RepositoryConfig,
+) -> Repo:
+    LOGGER.info("Cleaning up and updating %s repository", repo_config.path)
+    try:
+        return _repository_in_clean_state(
+            repo_config.path,
+            repo_config.branch,
+        )
+    except Exception as e:
+        LOGGER.error(
+            "Error while cleaning up and updating %s repository", repo_config.path
+        )
+        LOGGER.exception(e)
+        raise e
+
+
+def _repository_in_clean_state(path: Path, branch: str) -> Repo:
     repo = Repo(path)
     repo.git.reset("--hard")
     repo.remotes.origin.fetch()
     repo.git.checkout(branch)
     repo.git.reset("--hard", f"origin/{branch}")
     return repo
+
+
+def commit_and_push_files(
+    repo: Repo,
+    files_to_push_to_repo: Sequence[Path],
+) -> None:
+    try:
+        repo.index.add(files_to_push_to_repo)
+        repo.index.commit("Updating files")
+        repo.remotes.origin.push()
+    except CalledProcessError as e:
+        LOGGER.error(
+            "Committing and pushing files for repository %s failed", repo.working_dir
+        )
+        LOGGER.exception(e)
+        raise e
+    LOGGER.info(
+        "Committing and pushing files for repository %s succeeded", repo.working_dir
+    )

checkmk_weblate_syncer/po.py

@@ -0,0 +1,141 @@
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from subprocess import CalledProcessError
+from subprocess import run as run_subprocess
+from typing import assert_never
+
+from git import Repo
+
+from .config import PoFilePair, PoModeConfig, RepositoryConfig
+from .git import commit_and_push_files, repository_in_clean_state
+from .logging import LOGGER
+
+
+@dataclass(frozen=True)
+class _Success:
+    path: Path
+
+
+@dataclass(frozen=True)
+class _Failure:
+    error_message: str
+    path: Path
+
+
+def run(config: PoModeConfig) -> int:
+    checkmk_repo = repository_in_clean_state(config.checkmk_repository)
+    repository_in_clean_state(config.locale_repository)
+
+    failures: list[_Failure] = []
+    successes: list[_Success] = []
+
+    for file_pair in config.po_file_pairs:
+        match (
+            result := _process_po_file_pair(
+                file_pair=file_pair,
+                checkmk_repo=config.checkmk_repository,
+                locale_repo=config.locale_repository,
+            )
+        ):
+            case _Success():
+                successes.append(result)
+            case _Failure():
+                LOGGER.error(
+                    "We encountered an error while processing the .po file. "
+                    "See the logging output at the end for more information."
+                )
+                failures.append(result)
+            case _:
+                assert_never(result)
+    LOGGER.info("Checking if any .po files changed in the checkmk repository")
+
+    if _check_if_repo_is_dirty(checkmk_repo):
+        LOGGER.info("Committing and pushing .po file to checkmk repository")
+        commit_and_push_files(
+            repo=checkmk_repo,
+            files_to_push_to_repo=[success.path for success in successes],
+        )
+    else:
+        LOGGER.info("No changes in checkmk repository.")
+
+    if not failures:
+        return 0
+
+    for failure in failures:
+        LOGGER.error(
+            "Encountered the following error while processing %s:\n%s",
+            failure.path,
+            failure.error_message,
+        )
+    return 1
+
+
+def _process_po_file_pair(
+    file_pair: PoFilePair,
+    checkmk_repo: RepositoryConfig,
+    locale_repo: RepositoryConfig,
+) -> _Success | _Failure:
+    checkmk_po_file = checkmk_repo.path / file_pair.checkmk
+    locale_po_file = locale_repo.path / file_pair.locale
+    LOGGER.info("Checking formatting errors in %s", locale_po_file)
+    try:
+        run_subprocess(
+            ["msgfmt", "--check-format", "-o", "-", locale_po_file],
+            check=True,
+            capture_output=True,
+            encoding="UTF-8",
+        )
+    except CalledProcessError as e:
+        return _Failure(
+            error_message=f"Found formatting errors: {e.stderr}", path=locale_po_file
+        )
+    except Exception as e:  # pylint: disable=broad-except
+        return _Failure(error_message=str(e), path=locale_po_file)
+
+    LOGGER.info("Removing unwanted lines from %s", locale_po_file)
+    if isinstance(po_file_content := _remove_unwanted_lines(locale_po_file), _Failure):
+        return po_file_content
+
+    LOGGER.info("Writing stripped .po file to checkmk repository: %s", checkmk_po_file)
+    try:
+        checkmk_po_file.write_text(po_file_content)
+    except Exception as e:  # pylint: disable=broad-except
+        return _Failure(
+            f"Encountered error while writing po file to checkmk repository: {str(e)}",
+            checkmk_po_file,
+        )
+    return _Success(checkmk_po_file)
+
+
+def _check_if_repo_is_dirty(repo: Repo) -> bool:
+    try:
+        if not repo.is_dirty(untracked_files=True):
+            LOGGER.info("No changes, exiting")
+            return False
+    except Exception as e:
+        LOGGER.error(
+            "Checking if any .po files changed in the checkmk repository failed"
+        )
+        LOGGER.exception(e)
+        raise e
+    return True
+
+
+def _remove_unwanted_lines(file_path: Path) -> str | _Failure:
+    LOGGER.info("Reading %s", file_path)
+    try:
+        po_file_content = file_path.read_text()
+    except Exception as e:  # pylint: disable=broad-except
+        return _Failure(
+            error_message=f"Encountered error while reading file: {str(e)}",
+            path=file_path,
+        )
+    LOGGER.info("Removing code comments from %s", file_path)
+    po_file_content = re.sub(r"^#.+\d\n", "", po_file_content, flags=re.DOTALL)
+
+    LOGGER.info("Removing last translator information from %s", file_path)
+    po_file_content = re.sub(
+        r"\"Last-Translator:.+?\n", "", po_file_content, flags=re.DOTALL
+    )
+    return po_file_content

checkmk_weblate_syncer/pot.py

@@ -1,22 +1,14 @@
 from subprocess import CalledProcessError
 from subprocess import run as run_subprocess
 
-from git import Repo
-
-from .config import PotModeConfig, RepositoryConfig
-from .git import repository_in_clean_state
+from .config import PotModeConfig
+from .git import commit_and_push_files, repository_in_clean_state
 from .logging import LOGGER
 
 
-def run(config: PotModeConfig) -> None:
-    _get_repository_in_clean_state_with_logging(
-        config.checkmk_repository,
-        "checkmk",
-    )
-    locale_repo = _get_repository_in_clean_state_with_logging(
-        config.locale_repository,
-        "locale",
-    )
+def run(config: PotModeConfig) -> int:
+    repository_in_clean_state(config.checkmk_repository)
+    locale_repo = repository_in_clean_state(config.locale_repository)
 
     LOGGER.info("Calling pot generation script")
     try:
@@ -52,33 +44,15 @@ def run(config: PotModeConfig) -> None:
     try:
         if not locale_repo.is_dirty(untracked_files=True):
             LOGGER.info("No changes, exiting")
-            return
+            return 0
     except Exception as e:
         LOGGER.error("Checking if pot file has changed failed")
         LOGGER.exception(e)
         raise e
 
     LOGGER.info("Committing and pushing pot file to locale repository")
-    try:
-        locale_repo.index.add([path_pot_file])
-        locale_repo.index.commit("Update pot file")
-        locale_repo.remotes.origin.push()
-    except CalledProcessError as e:
-        LOGGER.error("Committing and pushing pot file failed")
-        LOGGER.exception(e)
-        raise e
-
-
-def _get_repository_in_clean_state_with_logging(
-    repo_config: RepositoryConfig, repo_name: str
-) -> Repo:
-    LOGGER.info("Cleaning up and updating %s repository", repo_name)
-    try:
-        return repository_in_clean_state(
-            repo_config.path,
-            repo_config.branch,
-        )
-    except Exception as e:
-        LOGGER.error("Error while cleaning up and updating %s repository", repo_name)
-        LOGGER.exception(e)
-        raise e
+    commit_and_push_files(
+        repo=locale_repo,
+        files_to_push_to_repo=[path_pot_file],
+    )
+    return 0