Keep skipped imports out of translation PRs

README.md

@@ -406,6 +406,13 @@ Server guide: [docs/new-project-deployment.md](docs/new-project-deployment.md).
   `suffix`, `locale_directory`, or `locale_filename`.
 - **`Permission denied (publickey)` on push:** the deploy key is missing or does
   not have write access to the fork repository.
+- **Skipped translation inputs:** files explicitly skipped by validation or
+  processing are excluded before PR batching and semantic review. The publisher
+  preserves their imported contents and validation summary under
+  `logs/skipped-inputs-*/`. Other valid files can still be published, but the run
+  exits unsuccessfully, retains the git-source baseline, and withholds its
+  success heartbeat for that run. Fix the reported input
+  problem and rerun; the pipeline does not restore old translations automatically.
 - **Quality gate failed:** inspect the PR report. The pipeline reports skipped
   files, placeholder errors, semantic findings, and suspicious source-identical
   values.

localize/connectors.py

@@ -42,7 +42,7 @@ def archive_original_files(self, changed_files: List[str], input_folder: str, ar
     def copy_files_to_translation_queue(self, changed_files: List[str], input_folder: str, queue_folder: str) -> None:
         """Copy changed files into the processing queue."""
 
-    def copy_translated_files_back(self, translated_queue: str, input_folder: str) -> None:
+    def copy_translated_files_back(self, translated_queue: str, input_folder: str, *, skipped_files: Optional[Dict[str, List[str]]] = None) -> None:
         """Copy translated output back to the project localization folder."""
 
     def cleanup_queue_folders(self, translation_queue: str, translated_queue: str) -> None:
@@ -169,7 +169,8 @@ def archive_original_files(self, changed_files: List[str], input_folder: str, ar
     def copy_files_to_translation_queue(self, changed_files: List[str], input_folder: str, queue_folder: str) -> None:
         self._copy_relative_files(changed_files, input_folder, queue_folder)
 
-    def copy_translated_files_back(self, translated_queue: str, input_folder: str) -> None:
+    def copy_translated_files_back(self, translated_queue: str, input_folder: str, *, skipped_files: Optional[Dict[str, List[str]]] = None) -> None:
+        """Copy successful outputs without overwriting explicitly skipped inputs."""
         translated_queue_path = Path(translated_queue).resolve()
         input_folder_path = Path(input_folder).resolve()
         if not translated_queue_path.exists():
@@ -186,6 +187,8 @@ def copy_translated_files_back(self, translated_queue: str, input_folder: str) -
                     source_path,
                 )
                 continue
+            if safe_relative_path.as_posix() in (skipped_files or {}):
+                continue
             target_path = (input_folder_path / safe_relative_path).resolve()
             try:
                 target_path.relative_to(input_folder_path)
@@ -320,11 +323,12 @@ def write_translation_validation_summary(
         validation_files: Dict[str, Dict[str, object]],
         skipped_files: Dict[str, List[str]],
     ) -> None:
+        """Record validated files and explicit skips for the publication gate."""
         self._write_json(
             summary_path,
             {
                 "files": validation_files,
-                "skipped_files": skipped_files,
+                "skipped_files": sorted(skipped_files),
                 "pipeline_warnings": [
                     {"file": filename, "errors": list(errors)}
                     for filename, errors in sorted(skipped_files.items())

localize/pipeline_core.py

@@ -137,6 +137,9 @@ async def run_translation_pipeline(
     )
     if not changed_files:
         logger.info("No translation files to process. Exiting.")
+        steps.write_translation_validation_summary(
+            paths.validation_summary_path, validation_files={}, skipped_files={}
+        )
         return TranslationPipelineResult(changed_files=[])
 
     logger.info("Detected %d translation file(s) to process.", len(changed_files))
@@ -215,7 +218,12 @@ async def run_translation_pipeline(
 
     steps.write_token_usage_summary(paths.token_usage_summary_path)
 
-    steps.copy_translated_files_back(paths.translated_queue_folder, paths.input_folder)
+    copy_kwargs = {}
+    if supports_keyword_argument(steps.copy_translated_files_back, "skipped_files"):
+        copy_kwargs["skipped_files"] = skipped_files
+    steps.copy_translated_files_back(
+        paths.translated_queue_folder, paths.input_folder, **copy_kwargs
+    )
     if processed_files_count > 0:
         logger.info("Copied translated files back to the input folder.")
 

localize/translate_localization_files.py

@@ -2175,18 +2175,23 @@ def is_source_localization_file(
 
 def copy_translated_files_back(
         translated_queue_folder: str,
-        input_folder_path: str
+        input_folder_path: str,
+        *,
+        skipped_files: Optional[Dict[str, List[str]]] = None,
 ):
     """
     Copy translated translation files back to the input folder, overwriting existing ones and preserving subdirectories.
 
     Args:
         translated_queue_folder (str): The folder containing translated files.
         input_folder_path (str): The input folder path.
+        skipped_files: Inputs that must not be overwritten by stale queue output.
     """
     for root, _dirs, files in os.walk(translated_queue_folder):
         for name in files:
             rel_path = os.path.relpath(os.path.join(root, name), translated_queue_folder)
+            if rel_path.replace('\\', '/') in (skipped_files or {}):
+                continue
             if is_target_localization_file(rel_path):
                 translated_file_path = os.path.join(translated_queue_folder, rel_path)
                 dest_path = os.path.join(input_folder_path, rel_path)
@@ -2512,6 +2517,7 @@ async def process_translation_queue(
         Run metrics are attached to the result as ``result.run_metrics``.
     """
     localization_files: List[Tuple[str, LocalizationProfile]] = []
+    skipped_files: Dict[str, List[str]] = {}
     for root, dirs, files in os.walk(translation_queue_folder):
         dirs.sort()
         for name in files:
@@ -2522,6 +2528,11 @@ async def process_translation_queue(
             profile = find_target_localization_profile(relative_path)
             if profile:
                 localization_files.append((relative_path, profile))
+            elif not is_source_localization_file(relative_path) and any(
+                item.localization_format.is_supported_file(relative_path)
+                for item in _iter_localization_profiles()
+            ):
+                skipped_files[relative_path] = ["No supported target locale profile."]
     localization_files.sort(key=lambda item: item[0])
 
     # Load the glossary from the JSON file
@@ -2543,8 +2554,6 @@ async def process_translation_queue(
     processed_filenames: List[str] = []
     total_keys_translated = 0
     run_metrics = new_run_metrics()
-    skipped_files: Dict[str, List[str]] = {}
-
     for translation_file, localization_profile in localization_files:
         try:
             localization_format = localization_profile.localization_format
@@ -2558,11 +2567,13 @@ async def process_translation_queue(
             )
             if not language_code:
                 logger.warning(f"Skipping file {translation_file}: unable to extract language code.")
+                skipped_files[translation_file] = ["Unable to extract language code."]
                 continue
             # 4) Now we find the "friendly name" from the dictionary
             target_language = language_code_to_name(language_code)
             if not target_language:
                 logger.warning(f"Skipping file {translation_file}: unsupported language code '{language_code}'.")
+                skipped_files[translation_file] = [f"Unsupported language code: {language_code}."]
                 continue
             style_rules_text_for_review = PRECOMPUTED_STYLE_RULES_TEXT.get(language_code, "")
             raw_language_glossary = glossary.get(language_code, {})
@@ -2594,6 +2605,7 @@ async def process_translation_queue(
 
             if not os.path.exists(source_file_path):
                 logger.warning(f"Source file '{source_file_name}' not found in '{INPUT_FOLDER}'. Skipping.")
+                skipped_files[translation_file] = [f"Source file not found: {source_file_name}."]
                 continue
 
             logger.info(f"Processing file '{translation_file}' for language '{target_language}'...")
@@ -3290,6 +3302,7 @@ def write_translation_validation_summary(
     """Write structured validation data consumed by the PR quality gate."""
     summary = {
         "files": validation_files,
+        "skipped_files": sorted(skipped_files),
         "pipeline_warnings": [
             {"file": filename, "errors": errors}
             for filename, errors in sorted(skipped_files.items())

localize/translation_publication.py

@@ -0,0 +1,63 @@
+"""Exclude skipped imports from publication and retain their diagnostic inputs."""
+
+import argparse
+import json
+from pathlib import Path, PurePosixPath
+import shutil
+import tempfile
+
+
+def plan_publication(summary_path: Path, repo: Path, input_folder: Path, candidates: list[str]) -> dict:
+    """Validate explicit skip records before filtering any publication candidates."""
+    summary = json.loads(summary_path.read_text(encoding="utf-8"))
+    skipped = summary.get("skipped_files")
+    if not isinstance(skipped, list):
+        raise ValueError("Validation summary must contain an explicit skipped_files list")
+    repo = repo.resolve(strict=True)
+    input_folder = input_folder.resolve(strict=True)
+    relative_input = input_folder.relative_to(repo)
+    excluded = set()
+    sources = []
+    for name in skipped:
+        if (
+            not isinstance(name, str)
+            or not name
+            or "\\" in name
+            or any(ord(char) < 32 or ord(char) == 127 for char in name)
+            or PurePosixPath(name).is_absolute()
+            or any(part in ("", ".", "..") for part in name.split("/"))
+        ):
+            raise ValueError("Invalid skipped translation path")
+        source = input_folder / name
+        if source.resolve(strict=True) != source or not source.is_file():
+            raise ValueError("Skipped translation must be a regular file without symlinks")
+        excluded.add((relative_input / name).as_posix())
+        sources.append((name, source))
+    evidence = None
+    if sources:
+        evidence = Path(tempfile.mkdtemp(prefix="skipped-inputs-", dir=summary_path.parent))
+        for name, source in sources:
+            destination = evidence / name
+            destination.parent.mkdir(parents=True, exist_ok=True)
+            shutil.copyfile(source, destination)
+        shutil.copyfile(summary_path, evidence / "validation-summary.json")
+    return {
+        "files": [path for path in candidates if path not in excluded],
+        "skipped_count": len(excluded),
+        "evidence_directory": str(evidence) if evidence else None,
+    }
+
+
+def main() -> None:
+    """Print the publication plan or fail before any Git mutation."""
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--summary", type=Path, required=True)
+    parser.add_argument("--repo", type=Path, required=True)
+    parser.add_argument("--input", type=Path, required=True)
+    parser.add_argument("files", nargs="*")
+    args = parser.parse_args()
+    print(json.dumps(plan_publication(args.summary, args.repo, args.input, args.files)))
+
+
+if __name__ == "__main__":
+    main()

tests/integration/test_skipped_translation_publication.py

@@ -0,0 +1,99 @@
+"""Exercise the shell publisher against imported, explicitly skipped files."""
+
+import json
+import os
+from pathlib import Path
+from shlex import quote
+import subprocess
+import sys
+
+import pytest
+
+from tests.integration.test_update_translations_batches import _run
+
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+@pytest.mark.parametrize("dry_run,valid_file,new_file", [(False, True, False), (False, False, False), (True, True, False), (False, True, True)])
+def test_skipped_import_never_reaches_publication(tmp_path, dry_run, valid_file, new_file):
+    """Keep the imported English regression out of commits and preserve evidence."""
+    repo = tmp_path / "target"
+    repo.mkdir()
+    folder = repo / "l10n"
+    folder.mkdir()
+    logs = tmp_path / "logs"
+    logs.mkdir()
+    (tmp_path / "localize").symlink_to(ROOT / "localize", target_is_directory=True)
+    skipped = folder / "settings_pt_PT.properties"
+    valid = folder / "settings_ta.properties"
+    original = "label=Opções\nhelp=Escolha as opções\nunrelated=Valor\n"
+    imported = "label=Options\nhelp=Choose options\nunrelated=Valor\x7f\n"
+    skipped.write_text(original)
+    valid.write_text("label=Original\n")
+    _run(["git", "init", "-q"], cwd=repo)
+    for key, value in [("user.name", "Test"), ("user.email", "test@example.invalid"), ("commit.gpgSign", "false")]:
+        _run(["git", "config", key, value], cwd=repo)
+    _run(["git", "add", "."], cwd=repo)
+    _run(["git", "commit", "-qm", "Original"], cwd=repo)
+    if new_file:
+        skipped = folder / "new_pt_PT.properties"
+    skipped.write_text(imported)
+    if valid_file:
+        valid.write_text("label=மொழிபெயர்ப்பு\n")
+    (logs / "translation_validation_summary.json").write_text(json.dumps({
+        "files": {"settings_ta.properties": {}} if valid_file else {},
+        "skipped_files": [skipped.name],
+        "pipeline_warnings": [{"file": skipped.name, "errors": ["U+007F"]}],
+    }))
+    script = (ROOT / "update-translations.sh").read_text()
+    collector = script[script.index("collect_changed_translation_files() {"):script.index("# Send a heartbeat")]
+    publish = script[script.index("publish_translation_changes() {"):script.index("# Go back to original branch")]
+    harness = rf"""
+set -euo pipefail
+mapfile() {{
+  shift
+  local array_name="$1" line
+  eval "$array_name=()"
+  while IFS= read -r line; do eval "$array_name+=(\"\$line\")"; done
+}}
+log() {{ printf '%s\n' "$1"; }}
+record_pipeline_event() {{ :; }}
+command_exists() {{ command -v "$1" >/dev/null; }}
+translation_file_extension_regex() {{ printf properties; }}
+update_git_source_baseline_if_safe() {{ touch {quote(str(tmp_path / 'baseline-advanced'))}; }}
+stage_and_submit_batch() {{
+  printf '%s\n' "${{BATCH_FILES[@]}}" > {quote(str(tmp_path / 'reviewed-files'))}
+  git add -- "${{BATCH_FILES[@]}}"
+  git -c commit.gpgSign=false commit -qm Translations
+}}
+{collector}
+APP_ROOT={quote(str(tmp_path))}
+TARGET_PROJECT_ROOT={quote(str(repo))}
+ABSOLUTE_INPUT_FOLDER={quote(str(folder))}
+INPUT_FOLDER=l10n
+DRY_RUN={str(dry_run).lower()}
+MAX_FILES_PER_PR=90
+TRANSLATION_BRANCH_PREFIX=translation-updates
+FORK_REPO_NAME=owner/repo
+TRANSLATION_SOURCE=transifex
+git remote add origin git@github.com:owner/repo.git
+{publish}
+touch {quote(str(tmp_path / 'heartbeat'))}
+"""
+    env = dict(os.environ, PATH=f"{Path(sys.executable).parent}:{os.environ['PATH']}")
+    env.pop("PYTHONPATH", None)
+    result = subprocess.run(["bash", "-c", harness], cwd=repo, env=env, text=True, capture_output=True)
+    assert result.returncode != 0, result.stdout + result.stderr
+    assert not (tmp_path / "baseline-advanced").exists()
+    assert not (tmp_path / "heartbeat").exists()
+    reviewed = tmp_path / "reviewed-files"
+    if valid_file and not dry_run:
+        assert reviewed.read_text().splitlines() == ["l10n/settings_ta.properties"]
+        assert _run(["git", "show", "HEAD:l10n/settings_pt_PT.properties"], cwd=repo).stdout == original
+    else:
+        assert not reviewed.exists()
+    evidence = list(logs.glob("skipped-inputs-*/" + skipped.name))
+    assert len(evidence) == 1
+    assert evidence[0].read_text() == imported
+    assert skipped.read_text() == imported

tests/integration/test_translate_localization_files.py

@@ -24,6 +24,21 @@
 
 # All fixtures are now defined in conftest.py and are auto-discovered by pytest.
 
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("name", ["app_xx.properties", "missing_de.properties"])
+async def test_unprocessable_import_has_explicit_skip_record(integration_test_environment, name):
+    """Unknown locales and absent sources remain excluded from publication."""
+    env = integration_test_environment
+    with open(os.path.join(env["translation_queue_folder"], name), "w", encoding="utf-8") as handle:
+        handle.write("label=English import\n")
+    result = await localize.translate_localization_files.process_translation_queue(
+        translation_queue_folder=env["translation_queue_folder"],
+        translated_queue_folder=env["translated_queue_folder"],
+        glossary_file_path=env["mock_glossary_path_resolved"],
+    )
+    assert result[2][name]
+
 @pytest.mark.asyncio
 @patch('localize.translate_localization_files.get_changed_translation_files')
 @patch('localize.translate_localization_files.copy_files_to_translation_queue')

tests/unit/test_pipeline_core.py

@@ -177,6 +177,7 @@ def pipeline_options(**overrides) -> TranslationPipelineOptions:
 
 @pytest.mark.asyncio
 async def test_pipeline_stops_after_detection_when_no_files_changed(pipeline_paths):
+    """An empty run replaces stale skip evidence with an explicit empty list."""
     fake = FakePipelineSteps(changed_files=[])
 
     result = await run_translation_pipeline(
@@ -191,6 +192,7 @@ async def test_pipeline_stops_after_detection_when_no_files_changed(pipeline_pat
     assert fake.calls == [
         "validate:/repo/i18n:/app/translation_queue:/app/translated_queue:/repo",
         "detect:/repo/i18n:/repo:True",
+        "validation_summary:/app/logs/translation_validation_summary.json:[]:[]",
     ]