mirror of https://github.com/stenzek/duckstation
You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1015 lines
42 KiB
Python
1015 lines
42 KiB
Python
#!/usr/bin/env python3
|
|
"""Create, resume, validate, and atomically apply Qt translation sessions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import collections
|
|
import datetime
|
|
import difflib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import uuid
|
|
import xml.etree.ElementTree as ET
|
|
from pathlib import Path
|
|
from typing import Sequence
|
|
|
|
if __package__ in (None, ""):
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from translation.ts_utils import ( # noqa: E402
|
|
SKIPPED_TYPES,
|
|
CatalogMessage,
|
|
atomic_write_text,
|
|
catalog_fingerprint,
|
|
extra_rich_tags,
|
|
iter_catalog_messages,
|
|
jsonl_text,
|
|
message_to_task_record,
|
|
missing_rich_tags,
|
|
parse_catalog,
|
|
placeholder_counts_match,
|
|
replace_translation_node,
|
|
scan_raw_messages,
|
|
text_sha256,
|
|
unbalanced_rich_tags,
|
|
validate_translation,
|
|
)
|
|
|
|
|
|
SESSION_SCHEMA = "duckstation-qt-translation-session-v2"
|
|
TASK_SCHEMA = "duckstation-qt-translation-task-v2"
|
|
RESPONSE_SCHEMA = "duckstation-qt-translation-response-v2"
|
|
REVIEW_SCHEMA = "duckstation-qt-translation-reviews-v2"
|
|
DECISION_FIELDS = ("translation", "plural_translations", "accept_current", "accept_source")
|
|
DEFAULT_MAX_RECORDS = 300
|
|
DEFAULT_MAX_BYTES = 192 * 1024
|
|
TASK_METADATA_RESERVE = 256
|
|
|
|
|
|
def json_line(value: dict[str, object]) -> str:
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True)
|
|
|
|
|
|
def read_jsonl(path: Path) -> list[dict[str, object]]:
|
|
output: list[dict[str, object]] = []
|
|
with path.open("r", encoding="utf-8") as stream:
|
|
for line_number, line in enumerate(stream, 1):
|
|
if not line.strip():
|
|
continue
|
|
try:
|
|
item = json.loads(line)
|
|
except json.JSONDecodeError as error:
|
|
raise ValueError(f"{path}:{line_number}: invalid JSON: {error}") from error
|
|
if not isinstance(item, dict):
|
|
raise ValueError(f"{path}:{line_number}: JSONL records must be objects")
|
|
output.append(item)
|
|
return output
|
|
|
|
|
|
def read_json(path: Path) -> dict[str, object]:
|
|
try:
|
|
item = json.loads(path.read_text(encoding="utf-8"))
|
|
except json.JSONDecodeError as error:
|
|
raise ValueError(f"{path}: invalid JSON: {error}") from error
|
|
if not isinstance(item, dict):
|
|
raise ValueError(f"{path}: expected a JSON object")
|
|
return item
|
|
|
|
|
|
def translated_text(message: CatalogMessage) -> str | list[str] | None:
|
|
if message.plural_translations:
|
|
return message.plural_translations if any(value.strip() for value in message.plural_translations) else None
|
|
return message.translation if message.translation and message.translation.strip() else None
|
|
|
|
|
|
def build_suggestion_indexes(
|
|
messages: Sequence[CatalogMessage],
|
|
) -> tuple[dict[str, list[CatalogMessage]], dict[str, list[CatalogMessage]]]:
|
|
exact: dict[str, list[CatalogMessage]] = collections.defaultdict(list)
|
|
contexts: dict[str, list[CatalogMessage]] = collections.defaultdict(list)
|
|
for message in messages:
|
|
if translated_text(message) is None:
|
|
continue
|
|
exact[message.identity.source].append(message)
|
|
contexts[message.identity.context].append(message)
|
|
return exact, contexts
|
|
|
|
|
|
def build_suggestions(
|
|
target: CatalogMessage,
|
|
exact_index: dict[str, list[CatalogMessage]],
|
|
context_index: dict[str, list[CatalogMessage]],
|
|
limit: int,
|
|
minimum_similarity: float,
|
|
) -> list[dict[str, object]]:
|
|
if limit <= 0:
|
|
return []
|
|
candidates: list[tuple[float, CatalogMessage]] = []
|
|
visited: set[str] = set()
|
|
for candidate in exact_index.get(target.identity.source, []):
|
|
if candidate.identifier != target.identifier:
|
|
candidates.append((1.0, candidate))
|
|
visited.add(candidate.identifier)
|
|
for candidate in context_index.get(target.identity.context, []):
|
|
if candidate.identifier == target.identifier or candidate.identifier in visited:
|
|
continue
|
|
similarity = difflib.SequenceMatcher(None, target.identity.source, candidate.identity.source).ratio()
|
|
if similarity >= minimum_similarity:
|
|
candidates.append((similarity, candidate))
|
|
candidates.sort(
|
|
key=lambda pair: (
|
|
pair[0],
|
|
pair[1].translation_type == "finished",
|
|
pair[1].identity.source == target.identity.source,
|
|
),
|
|
reverse=True,
|
|
)
|
|
suggestions: list[dict[str, object]] = []
|
|
seen: set[tuple[str, str]] = set()
|
|
for similarity, candidate in candidates:
|
|
text = translated_text(candidate)
|
|
signature = (candidate.identity.source, repr(text))
|
|
if signature in seen:
|
|
continue
|
|
seen.add(signature)
|
|
suggestions.append(
|
|
{
|
|
"source": candidate.identity.source,
|
|
"translation": text,
|
|
"context": candidate.identity.context,
|
|
"type": candidate.translation_type,
|
|
"similarity": round(similarity, 4),
|
|
}
|
|
)
|
|
if len(suggestions) == limit:
|
|
break
|
|
return suggestions
|
|
|
|
|
|
def task_record(message: CatalogMessage, suggestions: list[dict[str, object]]) -> dict[str, object]:
|
|
record = message_to_task_record(message)
|
|
record["suggestions"] = suggestions
|
|
return record
|
|
|
|
|
|
def split_tasks(
|
|
records: Sequence[dict[str, object]], max_records: int, max_bytes: int
|
|
) -> list[list[dict[str, object]]]:
|
|
tasks: list[list[dict[str, object]]] = []
|
|
current: list[dict[str, object]] = []
|
|
current_bytes = TASK_METADATA_RESERVE
|
|
for record in records:
|
|
record_bytes = len((json_line(record) + "\n").encode("utf-8"))
|
|
if TASK_METADATA_RESERVE + record_bytes > max_bytes:
|
|
raise ValueError(
|
|
f"{record.get('id')}: serialized record exceeds --max-bytes; "
|
|
"increase the limit for this catalog"
|
|
)
|
|
if current and (len(current) >= max_records or current_bytes + record_bytes > max_bytes):
|
|
tasks.append(current)
|
|
current = []
|
|
current_bytes = TASK_METADATA_RESERVE
|
|
current.append(record)
|
|
current_bytes += record_bytes
|
|
if current:
|
|
tasks.append(current)
|
|
return tasks
|
|
|
|
|
|
def find_repo_root(path: Path) -> Path:
|
|
for candidate in (path, *path.parents):
|
|
if (candidate / ".git").exists():
|
|
return candidate
|
|
return Path.cwd().resolve()
|
|
|
|
|
|
def session_file(session: Path, relative: object) -> Path:
|
|
path = (session / str(relative)).resolve()
|
|
try:
|
|
path.relative_to(session.resolve())
|
|
except ValueError as error:
|
|
raise ValueError(f"session artifact escapes session directory: {relative}") from error
|
|
return path
|
|
|
|
|
|
def state_for_message(message: CatalogMessage) -> dict[str, object]:
|
|
return {
|
|
"type": message.translation_type,
|
|
"translation": message.translation,
|
|
"plural_translations": message.plural_translations,
|
|
}
|
|
|
|
|
|
def state_for_record(record: dict[str, object]) -> dict[str, object]:
|
|
return {
|
|
"type": record.get("current_type"),
|
|
"translation": record.get("current_translation"),
|
|
"plural_translations": record.get("current_plural_translations") or [],
|
|
}
|
|
|
|
|
|
def load_session(
|
|
session_value: Path,
|
|
) -> tuple[Path, dict[str, object], list[dict[str, object]], dict[str, dict[str, object]]]:
|
|
session = session_value.resolve()
|
|
manifest = read_json(session / "manifest.json")
|
|
if manifest.get("schema") != SESSION_SCHEMA:
|
|
raise ValueError(f"{session}: unsupported session schema: {manifest.get('schema')!r}")
|
|
task_records: list[dict[str, object]] = []
|
|
tasks_by_id: dict[str, dict[str, object]] = {}
|
|
seen_ids: set[str] = set()
|
|
tasks = manifest.get("tasks")
|
|
if not isinstance(tasks, list):
|
|
raise ValueError(f"{session}: manifest tasks must be a list")
|
|
for task_value in tasks:
|
|
if not isinstance(task_value, dict):
|
|
raise ValueError(f"{session}: invalid task manifest record")
|
|
task_id = str(task_value.get("id", ""))
|
|
if not task_id or task_id in tasks_by_id:
|
|
raise ValueError(f"{session}: missing or duplicate task id: {task_id!r}")
|
|
task_path = session_file(session, task_value.get("path"))
|
|
task_text = task_path.read_text(encoding="utf-8")
|
|
if text_sha256(task_text) != task_value.get("sha256"):
|
|
raise ValueError(f"{task_path}: immutable task checksum mismatch")
|
|
lines = read_jsonl(task_path)
|
|
if not lines or lines[0] != {
|
|
"record_type": "metadata",
|
|
"schema": TASK_SCHEMA,
|
|
"session_id": manifest.get("session_id"),
|
|
"task_id": task_id,
|
|
}:
|
|
raise ValueError(f"{task_path}: invalid task metadata")
|
|
records = lines[1:]
|
|
expected_ids = task_value.get("record_ids")
|
|
actual_ids = [record.get("id") for record in records]
|
|
if actual_ids != expected_ids:
|
|
raise ValueError(f"{task_path}: task record list does not match manifest")
|
|
for record in records:
|
|
identifier = str(record.get("id", ""))
|
|
if record.get("record_type") != "message" or not identifier or identifier in seen_ids:
|
|
raise ValueError(f"{task_path}: invalid or duplicate message id: {identifier!r}")
|
|
seen_ids.add(identifier)
|
|
task_value = dict(task_value)
|
|
task_value["records"] = records
|
|
tasks_by_id[task_id] = task_value
|
|
task_records.extend(records)
|
|
if len(task_records) != manifest.get("target_count"):
|
|
raise ValueError(f"{session}: manifest target count does not match tasks")
|
|
return session, manifest, task_records, tasks_by_id
|
|
|
|
|
|
def load_reviews(
|
|
session: Path, manifest: dict[str, object], records_by_id: dict[str, dict[str, object]]
|
|
) -> dict[str, dict[str, object]]:
|
|
path = session / "reviews.jsonl"
|
|
lines = read_jsonl(path)
|
|
expected_metadata = {
|
|
"record_type": "metadata",
|
|
"schema": REVIEW_SCHEMA,
|
|
"session_id": manifest.get("session_id"),
|
|
}
|
|
if not lines or lines[0] != expected_metadata:
|
|
raise ValueError(f"{path}: invalid review journal metadata")
|
|
reviews: dict[str, dict[str, object]] = {}
|
|
for decision in lines[1:]:
|
|
identifier = str(decision.get("id", ""))
|
|
if identifier not in records_by_id:
|
|
raise ValueError(f"{path}: review has unknown id: {identifier or '<missing>'}")
|
|
normalized, _ = validate_decision(records_by_id[identifier], decision)
|
|
if identifier in reviews:
|
|
raise ValueError(f"{path}: duplicate review id: {identifier}")
|
|
reviews[identifier] = normalized
|
|
return reviews
|
|
|
|
|
|
def accelerator_count(text: str) -> int:
|
|
count = 0
|
|
index = 0
|
|
while index < len(text):
|
|
if text[index] != "&":
|
|
index += 1
|
|
continue
|
|
if index + 1 < len(text) and text[index + 1] == "&":
|
|
index += 2
|
|
continue
|
|
if index + 1 < len(text) and not text[index + 1].isspace():
|
|
count += 1
|
|
index += 1
|
|
return count
|
|
|
|
|
|
def resolved_values(
|
|
record: dict[str, object], decision: dict[str, object]
|
|
) -> tuple[str | None, list[str] | None]:
|
|
if decision.get("accept_current") is True:
|
|
plurals = record.get("current_plural_translations")
|
|
if plurals:
|
|
return None, list(plurals)
|
|
return str(record.get("current_translation") or ""), None
|
|
if decision.get("accept_source") is True:
|
|
return str(record.get("source", "")), None
|
|
if "translation" in decision:
|
|
return str(decision["translation"]), None
|
|
return None, list(decision["plural_translations"])
|
|
|
|
|
|
def validate_decision(
|
|
record: dict[str, object], decision: dict[str, object]
|
|
) -> tuple[dict[str, object], list[str]]:
|
|
identifier = str(decision.get("id", ""))
|
|
unexpected = set(decision) - {"id", *DECISION_FIELDS}
|
|
if unexpected:
|
|
raise ValueError(f"{identifier}: unsupported response fields: {sorted(unexpected)}")
|
|
if identifier != record.get("id"):
|
|
raise ValueError(f"{identifier or '<missing>'}: response id does not match its task record")
|
|
selected: list[str] = []
|
|
for field in DECISION_FIELDS:
|
|
value = decision.get(field)
|
|
if (field.startswith("accept_") and value is True) or (
|
|
not field.startswith("accept_") and value is not None
|
|
):
|
|
selected.append(field)
|
|
if len(selected) != 1:
|
|
raise ValueError(f"{identifier}: set exactly one reviewed decision field")
|
|
field = selected[0]
|
|
normalized: dict[str, object] = {"id": identifier, field: decision[field]}
|
|
numerus = record.get("numerus") is True
|
|
source = str(record.get("source", ""))
|
|
warnings: list[str] = []
|
|
if field == "accept_current":
|
|
current_values = record.get("current_plural_translations") or [record.get("current_translation")]
|
|
if not current_values or any(not isinstance(value, str) or not value.strip() for value in current_values):
|
|
raise ValueError(f"{identifier}: cannot accept an empty or incomplete current translation")
|
|
if not numerus and current_values == [source]:
|
|
normalized = {"id": identifier, "accept_source": True}
|
|
warnings.append(
|
|
f"{identifier}: normalized source-equal accept_current to accept_source=true"
|
|
)
|
|
elif field == "accept_source":
|
|
if numerus:
|
|
raise ValueError(f"{identifier}: accept_source is supported only for singular messages")
|
|
elif field == "translation":
|
|
value = decision[field]
|
|
if numerus or not isinstance(value, str) or not value.strip():
|
|
raise ValueError(f"{identifier}: translation must be a nonempty singular string")
|
|
if value == source:
|
|
normalized = {"id": identifier, "accept_source": True}
|
|
warnings.append(
|
|
f"{identifier}: normalized source-equal translation to accept_source=true"
|
|
)
|
|
else:
|
|
values = decision[field]
|
|
arity = int(record.get("plural_arity", 0))
|
|
if (
|
|
not numerus
|
|
or not isinstance(values, list)
|
|
or len(values) != arity
|
|
or any(not isinstance(value, str) or not value.strip() for value in values)
|
|
):
|
|
raise ValueError(
|
|
f"{identifier}: plural_translations must contain exactly {arity} nonempty strings"
|
|
)
|
|
if any(value == source for value in values):
|
|
raise ValueError(f"{identifier}: plural translations must not copy the source")
|
|
|
|
singular, plurals = resolved_values(record, normalized)
|
|
values = plurals if plurals is not None else [singular or ""]
|
|
for value in values:
|
|
problems = validate_translation(source, value, allow_missing_placeholders=numerus)
|
|
if problems:
|
|
raise ValueError(f"{identifier}: {'; '.join(problems)}")
|
|
if not numerus and not placeholder_counts_match(source, value):
|
|
raise ValueError(f"{identifier}: placeholder counts do not match the source")
|
|
missing = missing_rich_tags(source, value)
|
|
if missing:
|
|
raise ValueError(f"{identifier}: missing required rich-text tags: {dict(missing)}")
|
|
unbalanced = unbalanced_rich_tags(value)
|
|
if unbalanced:
|
|
raise ValueError(f"{identifier}: unbalanced rich-text tags: {unbalanced}")
|
|
source_accelerators = accelerator_count(source)
|
|
if source_accelerators and accelerator_count(value) != source_accelerators:
|
|
raise ValueError(
|
|
f"{identifier}: accelerator count mismatch: "
|
|
f"source={source_accelerators} translation={accelerator_count(value)}"
|
|
)
|
|
extras = extra_rich_tags(source, value)
|
|
if extras:
|
|
warnings.append(f"{identifier}: additional rich-text tags: {dict(extras)}")
|
|
if source.count("\n") != value.count("\n"):
|
|
warnings.append(
|
|
f"{identifier}: newline count differs: "
|
|
f"source={source.count(chr(10))} translation={value.count(chr(10))}"
|
|
)
|
|
if numerus and not any(placeholder_counts_match(source, value) for value in values):
|
|
raise ValueError(f"{identifier}: no plural form preserves the complete source placeholder set")
|
|
return normalized, warnings
|
|
|
|
|
|
def load_response(
|
|
path: Path,
|
|
manifest: dict[str, object],
|
|
tasks_by_id: dict[str, dict[str, object]],
|
|
records_by_id: dict[str, dict[str, object]],
|
|
) -> tuple[str, dict[str, dict[str, object]], list[str]]:
|
|
lines = read_jsonl(path)
|
|
if not lines:
|
|
raise ValueError(f"{path}: response file is empty")
|
|
metadata = lines[0]
|
|
task_id = str(metadata.get("task_id", ""))
|
|
task = tasks_by_id.get(task_id)
|
|
if task is None:
|
|
raise ValueError(f"{path}: response refers to unknown task: {task_id or '<missing>'}")
|
|
expected_metadata = response_metadata(manifest, task_id, task)
|
|
if metadata != expected_metadata:
|
|
raise ValueError(f"{path}: stale or invalid response metadata")
|
|
task_ids = set(task.get("record_ids", []))
|
|
decisions: dict[str, dict[str, object]] = {}
|
|
warnings: list[str] = []
|
|
for decision in lines[1:]:
|
|
identifier = str(decision.get("id", ""))
|
|
if identifier not in task_ids:
|
|
raise ValueError(f"{path}: response id is not owned by {task_id}: {identifier or '<missing>'}")
|
|
if identifier in decisions:
|
|
raise ValueError(f"{path}: duplicate response id: {identifier}")
|
|
normalized, decision_warnings = validate_decision(records_by_id[identifier], decision)
|
|
decisions[identifier] = normalized
|
|
warnings.extend(decision_warnings)
|
|
return task_id, decisions, warnings
|
|
|
|
|
|
def response_metadata(
|
|
manifest: dict[str, object], task_id: str, task: dict[str, object]
|
|
) -> dict[str, object]:
|
|
return {
|
|
"record_type": "metadata",
|
|
"schema": RESPONSE_SCHEMA,
|
|
"session_id": manifest.get("session_id"),
|
|
"task_id": task_id,
|
|
"task_sha256": task.get("sha256"),
|
|
}
|
|
|
|
|
|
def write_review_journal(
|
|
path: Path,
|
|
manifest: dict[str, object],
|
|
record_order: Sequence[dict[str, object]],
|
|
reviews: dict[str, dict[str, object]],
|
|
) -> None:
|
|
metadata = {
|
|
"record_type": "metadata",
|
|
"schema": REVIEW_SCHEMA,
|
|
"session_id": manifest.get("session_id"),
|
|
}
|
|
ordered = [reviews[str(record["id"])] for record in record_order if str(record["id"]) in reviews]
|
|
atomic_write_text(path, jsonl_text(metadata, ordered))
|
|
|
|
|
|
def catalog_path(manifest: dict[str, object]) -> Path:
|
|
return Path(str(manifest.get("catalog", ""))).resolve()
|
|
|
|
|
|
def command_start(args: argparse.Namespace) -> int:
|
|
catalog = args.catalog.resolve()
|
|
_, messages = parse_catalog(catalog)
|
|
contexts = set(args.context)
|
|
targets: list[CatalogMessage] = []
|
|
for message in messages:
|
|
if message.translation_type in SKIPPED_TYPES or (contexts and message.identity.context not in contexts):
|
|
continue
|
|
values = message.plural_translations if message.plural_translations else [message.translation or ""]
|
|
is_empty = not any(value.strip() for value in values)
|
|
is_source_equal = (
|
|
not message.identity.numerus
|
|
and message.translation is not None
|
|
and message.translation == message.identity.source
|
|
)
|
|
if (
|
|
message.translation_type == "unfinished"
|
|
or (args.include_empty_active and is_empty)
|
|
or (args.include_source_equal and is_source_equal)
|
|
):
|
|
targets.append(message)
|
|
|
|
exact_index, context_index = build_suggestion_indexes(messages)
|
|
records = [
|
|
task_record(
|
|
message,
|
|
build_suggestions(message, exact_index, context_index, args.suggestions, args.similarity),
|
|
)
|
|
for message in targets
|
|
]
|
|
task_groups = split_tasks(records, args.max_records, args.max_bytes)
|
|
fingerprint = catalog_fingerprint(messages)
|
|
session_id = uuid.uuid4().hex
|
|
repo_root = find_repo_root(catalog.parent)
|
|
timestamp = datetime.datetime.now(datetime.UTC).strftime("%Y%m%dT%H%M%SZ")
|
|
session = (
|
|
args.session_dir.resolve()
|
|
if args.session_dir
|
|
else repo_root / ".translation-work" / f"{catalog.stem}-{timestamp}-{fingerprint[:8]}-{session_id[:8]}"
|
|
)
|
|
if session.exists():
|
|
raise ValueError(f"{session}: session directory already exists")
|
|
session.parent.mkdir(parents=True, exist_ok=True)
|
|
staging = session.with_name(f".{session.name}.tmp-{session_id[:8]}")
|
|
if staging.exists():
|
|
raise ValueError(f"{staging}: staging directory already exists")
|
|
|
|
manifest: dict[str, object] = {
|
|
"schema": SESSION_SCHEMA,
|
|
"session_id": session_id,
|
|
"catalog": str(catalog),
|
|
"catalog_fingerprint": fingerprint,
|
|
"target_count": len(records),
|
|
"selection": {
|
|
"contexts": args.context,
|
|
"include_empty_active": args.include_empty_active,
|
|
"include_source_equal": args.include_source_equal,
|
|
"suggestions": args.suggestions,
|
|
"similarity": args.similarity,
|
|
"max_records": args.max_records,
|
|
"max_bytes": args.max_bytes,
|
|
},
|
|
"tasks": [],
|
|
}
|
|
try:
|
|
(staging / "tasks").mkdir(parents=True)
|
|
(staging / "responses").mkdir()
|
|
for index, group in enumerate(task_groups, 1):
|
|
task_id = f"task-{index:04d}"
|
|
task_relative = f"tasks/{task_id}.jsonl"
|
|
response_relative = f"responses/{task_id}.jsonl"
|
|
task_metadata = {
|
|
"record_type": "metadata",
|
|
"schema": TASK_SCHEMA,
|
|
"session_id": session_id,
|
|
"task_id": task_id,
|
|
}
|
|
task_text = jsonl_text(task_metadata, group)
|
|
task_hash = text_sha256(task_text)
|
|
atomic_write_text(staging / task_relative, task_text)
|
|
initial_response_metadata = {
|
|
"record_type": "metadata",
|
|
"schema": RESPONSE_SCHEMA,
|
|
"session_id": session_id,
|
|
"task_id": task_id,
|
|
"task_sha256": task_hash,
|
|
}
|
|
atomic_write_text(
|
|
staging / response_relative, jsonl_text(initial_response_metadata, [])
|
|
)
|
|
manifest["tasks"].append(
|
|
{
|
|
"id": task_id,
|
|
"path": task_relative,
|
|
"response_path": response_relative,
|
|
"sha256": task_hash,
|
|
"record_ids": [record["id"] for record in group],
|
|
}
|
|
)
|
|
review_metadata = {
|
|
"record_type": "metadata",
|
|
"schema": REVIEW_SCHEMA,
|
|
"session_id": session_id,
|
|
}
|
|
atomic_write_text(staging / "reviews.jsonl", jsonl_text(review_metadata, []))
|
|
atomic_write_text(
|
|
staging / "manifest.json",
|
|
json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
|
|
)
|
|
os.replace(staging, session)
|
|
finally:
|
|
if staging.exists():
|
|
shutil.rmtree(staging)
|
|
|
|
print(f"Created translation session with {len(records)} message(s) in {len(task_groups)} task(s).")
|
|
print(session)
|
|
return 0
|
|
|
|
|
|
def command_check(args: argparse.Namespace) -> int:
|
|
session, manifest, records, tasks_by_id = load_session(args.session)
|
|
records_by_id = {str(record["id"]): record for record in records}
|
|
task_id, decisions, warnings = load_response(args.response.resolve(), manifest, tasks_by_id, records_by_id)
|
|
expected = len(tasks_by_id[task_id]["record_ids"])
|
|
if args.require_complete_task and len(decisions) != expected:
|
|
raise ValueError(f"{args.response}: {task_id} has {len(decisions)}/{expected} reviewed record(s)")
|
|
for warning in warnings:
|
|
print(f"WARNING: {warning}")
|
|
print(f"Valid response for {task_id}: {len(decisions)}/{expected} reviewed record(s).")
|
|
return 0
|
|
|
|
|
|
def command_salvage(args: argparse.Namespace) -> int:
|
|
session, manifest, records, tasks_by_id = load_session(args.session)
|
|
records_by_id = {str(record["id"]): record for record in records}
|
|
response = args.response.resolve()
|
|
task_id = ""
|
|
task: dict[str, object] | None = None
|
|
for candidate_id, candidate in tasks_by_id.items():
|
|
if session_file(session, candidate["response_path"]) == response:
|
|
task_id = candidate_id
|
|
task = candidate
|
|
break
|
|
if task is None:
|
|
raise ValueError(f"{response}: response is not owned by this session")
|
|
|
|
task_ids = set(task["record_ids"])
|
|
decisions: dict[str, dict[str, object]] = {}
|
|
rejected: list[str] = []
|
|
warnings: list[str] = []
|
|
with response.open("r", encoding="utf-8") as stream:
|
|
for line_number, line in enumerate(stream, 1):
|
|
if not line.strip():
|
|
continue
|
|
try:
|
|
item = json.loads(line)
|
|
except json.JSONDecodeError as error:
|
|
rejected.append(f"line {line_number}: invalid JSON: {error}")
|
|
continue
|
|
if not isinstance(item, dict):
|
|
rejected.append(f"line {line_number}: JSONL record is not an object")
|
|
continue
|
|
if item.get("record_type") == "metadata":
|
|
continue
|
|
identifier = str(item.get("id", ""))
|
|
if identifier not in task_ids:
|
|
rejected.append(
|
|
f"line {line_number}: response id is not owned by {task_id}: "
|
|
f"{identifier or '<missing>'}"
|
|
)
|
|
continue
|
|
try:
|
|
normalized, decision_warnings = validate_decision(records_by_id[identifier], item)
|
|
except ValueError as error:
|
|
rejected.append(f"line {line_number}: {error}")
|
|
continue
|
|
existing = decisions.get(identifier)
|
|
if existing is not None and existing != normalized:
|
|
rejected.append(f"line {line_number}: conflicting duplicate response id: {identifier}")
|
|
continue
|
|
decisions[identifier] = normalized
|
|
warnings.extend(decision_warnings)
|
|
|
|
ordered = [
|
|
decisions[str(record["id"])]
|
|
for record in task["records"]
|
|
if str(record["id"]) in decisions
|
|
]
|
|
if args.write:
|
|
atomic_write_text(response, jsonl_text(response_metadata(manifest, task_id, task), ordered))
|
|
for warning in warnings:
|
|
print(f"WARNING: {warning}")
|
|
for problem in rejected:
|
|
print(f"SKIPPED: {problem}")
|
|
action = "Salvaged" if args.write else "Would salvage"
|
|
print(
|
|
f"{action} {len(ordered)}/{len(task_ids)} valid decision(s) for {task_id}; "
|
|
f"skipped {len(rejected)} invalid line(s)."
|
|
)
|
|
if not args.write:
|
|
print("Dry-run only; pass --write to rebuild the response atomically.")
|
|
return 0
|
|
|
|
|
|
def response_paths(
|
|
session: Path,
|
|
manifest: dict[str, object],
|
|
provided: Sequence[Path],
|
|
) -> list[Path]:
|
|
if provided:
|
|
return [path.resolve() for path in provided]
|
|
return [session_file(session, task["response_path"]) for task in manifest.get("tasks", [])]
|
|
|
|
|
|
def command_merge(args: argparse.Namespace) -> int:
|
|
session, manifest, records, tasks_by_id = load_session(args.session)
|
|
records_by_id = {str(record["id"]): record for record in records}
|
|
reviews = load_reviews(session, manifest, records_by_id)
|
|
merged = dict(reviews)
|
|
incoming_count = 0
|
|
warnings: list[str] = []
|
|
for path in response_paths(session, manifest, args.responses):
|
|
_, decisions, response_warnings = load_response(path, manifest, tasks_by_id, records_by_id)
|
|
warnings.extend(response_warnings)
|
|
for identifier, decision in decisions.items():
|
|
existing = merged.get(identifier)
|
|
if existing is not None and existing != decision and not args.replace:
|
|
raise ValueError(
|
|
f"{identifier}: response conflicts with the merged review; use --replace to correct it"
|
|
)
|
|
if existing != decision:
|
|
incoming_count += 1
|
|
merged[identifier] = decision
|
|
if args.write:
|
|
write_review_journal(session / "reviews.jsonl", manifest, records, merged)
|
|
for warning in warnings:
|
|
print(f"WARNING: {warning}")
|
|
action = "Merged" if args.write else "Would merge"
|
|
print(f"{action} {incoming_count} new or changed review(s); total={len(merged)}/{len(records)}.")
|
|
if not args.write:
|
|
print("Dry-run only; pass --write to update the review journal.")
|
|
return 0
|
|
|
|
|
|
def catalog_conflicts(
|
|
manifest: dict[str, object],
|
|
records: Sequence[dict[str, object]],
|
|
reviews: dict[str, dict[str, object]],
|
|
) -> tuple[bool, list[str]]:
|
|
catalog = catalog_path(manifest)
|
|
_, messages = parse_catalog(catalog)
|
|
if catalog_fingerprint(messages) != manifest.get("catalog_fingerprint"):
|
|
return True, ["catalog source identity has changed since the session started"]
|
|
messages_by_id = {message.identifier: message for message in messages}
|
|
conflicts: list[str] = []
|
|
for record in records:
|
|
identifier = str(record["id"])
|
|
message = messages_by_id.get(identifier)
|
|
if message is None:
|
|
conflicts.append(f"{identifier}: target message is missing")
|
|
continue
|
|
current = state_for_message(message)
|
|
if current == state_for_record(record):
|
|
continue
|
|
decision = reviews.get(identifier)
|
|
if decision is not None:
|
|
singular, plurals = resolved_values(record, decision)
|
|
desired_values = plurals if plurals is not None else [singular or ""]
|
|
current_values = (
|
|
message.plural_translations
|
|
if message.plural_translations
|
|
else [message.translation or ""]
|
|
)
|
|
if message.translation_type == "finished" and current_values == desired_values:
|
|
continue
|
|
conflicts.append(f"{identifier}: targeted translation changed outside this session")
|
|
return bool(conflicts), conflicts
|
|
|
|
|
|
def command_status(args: argparse.Namespace) -> int:
|
|
session, manifest, records, tasks_by_id = load_session(args.session)
|
|
records_by_id = {str(record["id"]): record for record in records}
|
|
reviews = load_reviews(session, manifest, records_by_id)
|
|
invalid_responses: list[str] = []
|
|
invalid_response_paths: list[str] = []
|
|
unmerged_ids: set[str] = set()
|
|
response_decisions_by_task: dict[str, dict[str, dict[str, object]]] = {}
|
|
for path in response_paths(session, manifest, []):
|
|
try:
|
|
task_id, decisions, _ = load_response(path, manifest, tasks_by_id, records_by_id)
|
|
except (OSError, ValueError) as error:
|
|
invalid_responses.append(str(error))
|
|
invalid_response_paths.append(str(path))
|
|
continue
|
|
response_decisions_by_task[task_id] = decisions
|
|
unmerged_ids.update(
|
|
identifier for identifier, decision in decisions.items() if reviews.get(identifier) != decision
|
|
)
|
|
stale, conflicts = catalog_conflicts(manifest, records, reviews)
|
|
task_status: list[dict[str, object]] = []
|
|
for task_id, task in tasks_by_id.items():
|
|
ids = set(task["record_ids"])
|
|
merged_count = len(ids & reviews.keys())
|
|
checkpointed = response_decisions_by_task.get(task_id, {})
|
|
task_unmerged = {
|
|
identifier
|
|
for identifier, decision in checkpointed.items()
|
|
if reviews.get(identifier) != decision
|
|
}
|
|
task_status.append(
|
|
{
|
|
"id": task_id,
|
|
"merged": merged_count,
|
|
"checkpointed": len(checkpointed),
|
|
"unmerged": len(task_unmerged),
|
|
"total": len(ids),
|
|
"remaining": len(ids) - merged_count,
|
|
"task": str(session_file(session, task["path"])),
|
|
"response": str(session_file(session, task["response_path"])),
|
|
}
|
|
)
|
|
result = {
|
|
"session": str(session),
|
|
"catalog": str(catalog_path(manifest)),
|
|
"reviewed": len(reviews),
|
|
"total": len(records),
|
|
"remaining": len(records) - len(reviews),
|
|
"unmerged": len(unmerged_ids),
|
|
"catalog_stale_or_conflicted": stale,
|
|
"conflicts": conflicts,
|
|
"invalid_responses": invalid_responses,
|
|
"invalid_response_paths": invalid_response_paths,
|
|
"tasks": task_status,
|
|
}
|
|
if stale:
|
|
result["queue_complete"] = False
|
|
result["recommended_action"] = "resolve_errors"
|
|
elif invalid_responses:
|
|
result["queue_complete"] = False
|
|
result["recommended_action"] = "salvage_responses"
|
|
elif unmerged_ids:
|
|
result["queue_complete"] = False
|
|
result["recommended_action"] = "merge_responses"
|
|
elif result["remaining"]:
|
|
result["queue_complete"] = False
|
|
result["recommended_action"] = "dispatch_workers"
|
|
else:
|
|
result["queue_complete"] = True
|
|
result["recommended_action"] = "dry_run_apply"
|
|
if args.json:
|
|
print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True))
|
|
else:
|
|
print(
|
|
f"Session status: reviewed={result['reviewed']}, remaining={result['remaining']}, "
|
|
f"unmerged={result['unmerged']}, invalid_responses={len(invalid_responses)}."
|
|
)
|
|
for item in task_status:
|
|
if item["remaining"]:
|
|
print(
|
|
f"{item['id']}: {item['merged']}/{item['total']} merged, "
|
|
f"{item['checkpointed']} checkpointed, {item['unmerged']} unmerged; "
|
|
f"task={item['task']} response={item['response']}"
|
|
)
|
|
for problem in conflicts:
|
|
print(f"ERROR: {problem}")
|
|
for problem in invalid_responses:
|
|
print(f"ERROR: {problem}")
|
|
for path in invalid_response_paths:
|
|
print(
|
|
"RECOVERY: "
|
|
f"python3 scripts/translation/translate_ts.py salvage {session} {path} --write"
|
|
)
|
|
if not stale and not invalid_responses:
|
|
if unmerged_ids:
|
|
print(
|
|
"INCOMPLETE: merge the checkpointed responses, then continue the rolling worker queue. "
|
|
"Do not stop at a partial session."
|
|
)
|
|
elif result["remaining"]:
|
|
print(
|
|
"INCOMPLETE: dispatch the next bounded tasks and continue across worker waves until "
|
|
"remaining=0. Concurrent agent slots are not a total-capacity limit."
|
|
)
|
|
else:
|
|
print("REVIEW QUEUE COMPLETE: proceed with dry-run apply, write, validation, and diff inspection.")
|
|
return 1 if stale or invalid_responses else 0
|
|
|
|
|
|
def command_apply(args: argparse.Namespace) -> int:
|
|
session, manifest, records, _ = load_session(args.session)
|
|
records_by_id = {str(record["id"]): record for record in records}
|
|
reviews = load_reviews(session, manifest, records_by_id)
|
|
missing = [str(record["id"]) for record in records if str(record["id"]) not in reviews]
|
|
if missing and not args.reviewed_only:
|
|
raise ValueError(f"session is incomplete: {len(missing)} record(s) remain unreviewed")
|
|
selected_records = (
|
|
[record for record in records if str(record["id"]) in reviews]
|
|
if args.reviewed_only
|
|
else records
|
|
)
|
|
if not selected_records:
|
|
raise ValueError("session has no reviewed translations to apply")
|
|
for record in selected_records:
|
|
validate_decision(record, reviews[str(record["id"])])
|
|
|
|
catalog = catalog_path(manifest)
|
|
_, messages = parse_catalog(catalog)
|
|
if catalog_fingerprint(messages) != manifest.get("catalog_fingerprint"):
|
|
raise ValueError("catalog source identity has changed since the session started")
|
|
messages_by_id = {message.identifier: message for message in messages}
|
|
with catalog.open("r", encoding="utf-8", newline="") as stream:
|
|
original = stream.read()
|
|
spans = scan_raw_messages(original)
|
|
replacements: list[tuple[int, int, str]] = []
|
|
already_applied = 0
|
|
for record in selected_records:
|
|
identifier = str(record["id"])
|
|
message = messages_by_id.get(identifier)
|
|
span = spans.get(identifier)
|
|
if message is None or span is None:
|
|
raise ValueError(f"{identifier}: target message no longer exists")
|
|
singular, plurals = resolved_values(record, reviews[identifier])
|
|
desired_values = plurals if plurals is not None else [singular or ""]
|
|
current_values = (
|
|
message.plural_translations if message.plural_translations else [message.translation or ""]
|
|
)
|
|
if message.translation_type == "finished" and current_values == desired_values:
|
|
already_applied += 1
|
|
continue
|
|
if state_for_message(message) != state_for_record(record):
|
|
raise ValueError(f"{identifier}: targeted translation changed outside this session")
|
|
replacement = replace_translation_node(span.block, singular, plurals)
|
|
replacements.append((span.start, span.end, replacement))
|
|
|
|
updated = original
|
|
for start, end, replacement in sorted(replacements, reverse=True):
|
|
updated = updated[:start] + replacement + updated[end:]
|
|
try:
|
|
staged_messages = list(iter_catalog_messages(ET.fromstring(updated)))
|
|
except ET.ParseError as error:
|
|
raise ValueError(f"staged catalog is not well-formed XML: {error}") from error
|
|
staged_by_id = {message.identifier: message for message in staged_messages}
|
|
for record in selected_records:
|
|
identifier = str(record["id"])
|
|
message = staged_by_id.get(identifier)
|
|
if message is None or message.translation_type != "finished":
|
|
raise ValueError(f"{identifier}: staged translation is missing or unfinished")
|
|
singular, plurals = resolved_values(record, reviews[identifier])
|
|
expected = plurals if plurals is not None else [singular or ""]
|
|
actual = message.plural_translations if message.plural_translations else [message.translation or ""]
|
|
if actual != expected:
|
|
raise ValueError(f"{identifier}: staged translation differs from its reviewed value")
|
|
if args.write and replacements:
|
|
atomic_write_text(catalog, updated, mode=catalog.stat().st_mode)
|
|
action = "Applied" if args.write else "Would apply"
|
|
print(
|
|
f"{action} {len(replacements)} translation(s); "
|
|
f"{already_applied} already matched the reviewed session"
|
|
f"{f'; {len(missing)} remain unreviewed' if args.reviewed_only else ''}."
|
|
)
|
|
if not args.write:
|
|
print("Dry-run only; pass --write to update the catalog.")
|
|
return 0
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
start = subparsers.add_parser("start", help="create a new resumable translation session")
|
|
start.add_argument("catalog", type=Path)
|
|
start.add_argument("--session-dir", type=Path)
|
|
start.add_argument("--context", action="append", default=[])
|
|
start.add_argument("--include-empty-active", action="store_true")
|
|
start.add_argument("--include-source-equal", action="store_true")
|
|
start.add_argument("--suggestions", type=int, default=2)
|
|
start.add_argument("--similarity", type=float, default=0.72)
|
|
start.add_argument("--max-records", type=int, default=DEFAULT_MAX_RECORDS)
|
|
start.add_argument("--max-bytes", type=int, default=DEFAULT_MAX_BYTES)
|
|
start.set_defaults(function=command_start)
|
|
|
|
status = subparsers.add_parser("status", help="show durable and unmerged session progress")
|
|
status.add_argument("session", type=Path)
|
|
status.add_argument("--json", action="store_true")
|
|
status.set_defaults(function=command_status)
|
|
|
|
check = subparsers.add_parser("check", help="validate one agent response file")
|
|
check.add_argument("session", type=Path)
|
|
check.add_argument("response", type=Path)
|
|
check.add_argument("--require-complete-task", action="store_true")
|
|
check.set_defaults(function=command_check)
|
|
|
|
salvage = subparsers.add_parser(
|
|
"salvage", help="atomically rebuild a damaged response from its valid decisions"
|
|
)
|
|
salvage.add_argument("session", type=Path)
|
|
salvage.add_argument("response", type=Path)
|
|
salvage.add_argument("--write", action="store_true")
|
|
salvage.set_defaults(function=command_salvage)
|
|
|
|
merge = subparsers.add_parser("merge", help="merge response checkpoints into the review journal")
|
|
merge.add_argument("session", type=Path)
|
|
merge.add_argument("responses", nargs="*", type=Path)
|
|
merge.add_argument("--write", action="store_true")
|
|
merge.add_argument("--replace", action="store_true")
|
|
merge.set_defaults(function=command_merge)
|
|
|
|
apply = subparsers.add_parser(
|
|
"apply", help="atomically apply complete or explicitly selected reviewed translations"
|
|
)
|
|
apply.add_argument("session", type=Path)
|
|
apply.add_argument(
|
|
"--reviewed-only",
|
|
action="store_true",
|
|
help="apply only checkpointed reviews when explicitly abandoning or restarting a session",
|
|
)
|
|
apply.add_argument("--write", action="store_true")
|
|
apply.set_defaults(function=command_apply)
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
if args.command == "start":
|
|
if args.max_records < 1:
|
|
raise SystemExit("--max-records must be at least 1")
|
|
if args.max_bytes < 1:
|
|
raise SystemExit("--max-bytes must be at least 1")
|
|
if args.suggestions < 0:
|
|
raise SystemExit("--suggestions cannot be negative")
|
|
if not 0.0 <= args.similarity <= 1.0:
|
|
raise SystemExit("--similarity must be between 0 and 1")
|
|
try:
|
|
return int(args.function(args))
|
|
except (OSError, ValueError) as error:
|
|
print(f"ERROR: {error}")
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|