diff options
Diffstat (limited to 'comm/python/l10n/tbxchannel/quarantine_to_strings.py')
| -rw-r--r-- | comm/python/l10n/tbxchannel/quarantine_to_strings.py | 194 |
1 files changed, 194 insertions, 0 deletions
diff --git a/comm/python/l10n/tbxchannel/quarantine_to_strings.py b/comm/python/l10n/tbxchannel/quarantine_to_strings.py new file mode 100644 index 0000000000..def3f59e5e --- /dev/null +++ b/comm/python/l10n/tbxchannel/quarantine_to_strings.py @@ -0,0 +1,194 @@ +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at https://mozilla.org/MPL/2.0/. + +import logging +import os +import shutil +import subprocess +import tempfile +from pathlib import Path + +from typing_extensions import Literal + +from mozversioncontrol import HgRepository +from mozversioncontrol.repoupdate import update_mercurial_repo + +from .l10n_merge import COMM_L10N, COMM_L10N_PUSH, COMM_STRINGS_QUARANTINE + +ACTIONS = Literal["clean", "prep", "migrate", "push"] + + +class HgL10nRepository(HgRepository): + log_trans_table = str.maketrans({"{": "{{", "}": "}}"}) + + def __init__(self, path: Path, check_url=None, logger=print): + super(HgL10nRepository, self).__init__(path, hg="hg") + self._logger = logger + if check_url is not None: + self._check_hg_url(check_url) + + def logger(self, *args): + # Escape python-style format string substitutions because Sentry is annoying + self._logger(*args[:-1], args[-1].translate(self.log_trans_table)) + + def _check_hg_url(self, repo_url): + configured_url = self._run("config", "paths.default").strip() + if configured_url != repo_url: + raise Exception(f"Repository does not match {repo_url}.") + + def check_status(self): + if not self.working_directory_clean() or self.get_outgoing_files(): + raise Exception(f"Repository at {self.path} is not clean, run with 'clean'.") + + def last_convert_rev(self): + args = ( + "log", + "-r", + "last(extra('convert_source', 'comm-strings-quarantine'))", + "--template", + "{get(extras,'convert_revision')}\n", + ) + self.logger(logging.INFO, "last_convert_rev", {}, " ".join(args)) + rv = self._run(*args).strip() + self.logger(logging.INFO, "last_convert_rev", {}, rv) + return rv + + def next_convert_rev(self, last_converted): + args = ("log", "-r", f"first(children({last_converted}))", "--template", "{node}\n") + self.logger(logging.INFO, "next_convert_rev", {}, " ".join(args)) + rv = self._run(*args).strip() + self.logger(logging.INFO, "next_convert_rev", {}, rv) + return rv + + def convert_quarantine(self, strings_path, filemap_path, splicemap_path, next_converted_rev): + args = ( + "convert", + "--config", + "convert.hg.saverev=True", + "--config", + "convert.hg.sourcename=comm-strings-quarantine", + "--config", + f"convert.hg.revs={next_converted_rev}:tip", + "--filemap", + filemap_path, + "--splicemap", + splicemap_path, + "--datesort", + str(self.path), + str(strings_path.absolute()), + ) + self.logger(logging.INFO, "convert_quarantine", {}, " ".join(args)) + rv = self._run(*args) + self.logger(logging.INFO, "convert_quarantine", {}, rv) + return rv + + def push(self, push_url): + popen_kwargs = { + "stdout": subprocess.PIPE, + "stderr": subprocess.PIPE, + "cwd": self.path, + "env": self._env, + "universal_newlines": True, + "bufsize": 1, + } + cmd = ("hg", "push", "-r", ".", push_url) + self.logger(logging.INFO, "push", {}, " ".join(cmd)) + # This function doesn't really push to try... + self._push_to_try_with_log_capture(cmd, popen_kwargs) + + +def _nuke_hg_repos(*paths: Path): + failed = {} + for path in paths: + try: + if path.exists(): + shutil.rmtree(str(path)) + except Exception as e: + failed[str(path)] = e + + if failed: + for f in failed: + print(f"Unable to nuke '{f}': {failed[f]}") + raise Exception() + + +def publish_strings( + command_context, + quarantine_path: Path, + comm_l10n_path: Path, + actions: ACTIONS, + **kwargs, +): + if "clean" in actions: + command_context.log(logging.INFO, "clean", {}, "Removing old repository clones.") + _nuke_hg_repos(quarantine_path, comm_l10n_path) + + if "prep" in actions: + # update_mercurial_repo also will clone if a repo is not already there + command_context.log( + logging.INFO, "prep", {}, f"Updating comm-strings-quarantine at {quarantine_path}." + ) + update_mercurial_repo("hg", COMM_STRINGS_QUARANTINE, quarantine_path) + command_context.log(logging.INFO, "prep", {}, f"Updating comm-l10n at {comm_l10n_path}.") + update_mercurial_repo("hg", COMM_L10N, comm_l10n_path) + + local_quarantine = HgL10nRepository( + quarantine_path, COMM_STRINGS_QUARANTINE, command_context.log + ) + local_comm_l10n = HgL10nRepository(comm_l10n_path, COMM_L10N, command_context.log) + + if "prep" not in actions: + local_quarantine.update("tip") + local_comm_l10n.update("tip") + + if "migrate" in actions: + local_quarantine.check_status() + local_comm_l10n.check_status() + + command_context.log( + logging.INFO, "migrate", {}, "Starting string migration from quarantine." + ) + head_rev = local_comm_l10n.head_ref + last_convert_rev = local_comm_l10n.last_convert_rev() + first_convert_rev = local_quarantine.next_convert_rev(last_convert_rev) + command_context.log( + logging.INFO, "migrate", {}, f" Last converted rev: {last_convert_rev}" + ) + command_context.log( + logging.INFO, "migrate", {}, f" First converted rev: {first_convert_rev}" + ) + + with tempfile.NamedTemporaryFile( + prefix="splicemap", suffix=".txt", delete=False + ) as splice_fp: + splicemap = splice_fp.name + command_context.log( + logging.INFO, "migrate", {}, f" Writing splicemap to: {splicemap}" + ) + splice_fp.write(f"{first_convert_rev} {head_rev}\n".encode("utf-8")) + + with tempfile.NamedTemporaryFile(prefix="filemap", suffix=".txt", delete=False) as file_fp: + filemap = file_fp.name + command_context.log(logging.INFO, "migrate", {}, f" Writing filemap to: {filemap}") + file_fp.writelines( + ["exclude _configs\n".encode("utf-8"), "rename . en-US\n".encode("utf-8")] + ) + + command_context.log(logging.INFO, "migrate", {}, " Running hg convert...") + local_quarantine.convert_quarantine(comm_l10n_path, filemap, splicemap, first_convert_rev) + try: + os.unlink(splicemap) + os.unlink(filemap) + except Exception: + pass + + local_comm_l10n.update("tip") + command_context.log(logging.INFO, "migrate", {}, " Finished!") + + if "push" in actions: + if local_comm_l10n.get_outgoing_files(): + command_context.log(logging.INFO, "push", {}, " Pushing to comm-l10n.") + local_comm_l10n.push(COMM_L10N_PUSH) + else: + command_context.log(logging.INFO, "push", {}, "Skipping empty push.") |