#!/usr/bin/env python3 """ Translate en -> vi fields in data/questions.json (offline batch). Uses deep-translator (Google). Saves progress every N questions. """ from __future__ import annotations import argparse import json import sys import time from datetime import datetime, timezone from pathlib import Path ROOT = Path(__file__).resolve().parents[1] SCRIPTS = Path(__file__).resolve().parent if str(SCRIPTS) not in __import__("sys").path: __import__("sys").path.insert(0, str(SCRIPTS)) from azure_vi_glossary import improve_vi_text, protect_terms, restore_terms from exam_config import get_profile QUESTIONS_PATH = ROOT / "data" / "questions.json" CHECKPOINT_PATH = ROOT / "data" / ".translate_checkpoint.json" try: from deep_translator import GoogleTranslator except ImportError: GoogleTranslator = None # type: ignore def translate_text(text: str | None, translator: GoogleTranslator) -> str | None: if not text or not str(text).strip(): return None t = str(text).strip() protected, placeholders = protect_terms(t) if len(protected) > 4500: parts = [] for i in range(0, len(protected), 4000): chunk = protected[i : i + 4000] parts.append(translator.translate(chunk)) time.sleep(0.2) out = " ".join(parts) else: out = translator.translate(protected) out = restore_terms(out, placeholders) return improve_vi_text(out) def translate_bilingual(obj: dict | None, translator: GoogleTranslator, dry_run: bool) -> bool: if not obj or not obj.get("en") or obj.get("vi"): return False if dry_run: obj["vi"] = "[dry-run]" return True try: obj["vi"] = translate_text(obj["en"], translator) time.sleep(0.05) return True except Exception as e: print(f" translate error: {e}", file=sys.stderr) return False def question_needs_translation(q: dict) -> bool: stem = q.get("stem") or {} if stem.get("en") and not stem.get("vi"): return True ex = q.get("explanation") or {} if ex.get("en") and not ex.get("vi"): return True ans_text = (q.get("answer") or {}).get("text") or {} if ans_text.get("en") and not ans_text.get("vi"): return True meta = q.get("meta") or {} if isinstance(meta.get("scenario"), dict): sc = meta["scenario"] if sc.get("en") and not sc.get("vi"): return True for opt in q.get("options") or []: t = opt.get("text") or {} if t.get("en") and not t.get("vi"): return True for box in q.get("boxes") or []: t = box.get("answer") or {} if t.get("en") and not t.get("vi"): return True return False def translate_question(q: dict, translator: GoogleTranslator, dry_run: bool) -> bool: changed = False if translate_bilingual(q.get("stem", {}), translator, dry_run): changed = True if translate_bilingual(q.get("explanation", {}), translator, dry_run): changed = True if translate_bilingual(q.get("answer", {}).get("text", {}), translator, dry_run): changed = True meta = q.get("meta") or {} if isinstance(meta.get("scenario"), dict): if translate_bilingual(meta["scenario"], translator, dry_run): changed = True for opt in q.get("options", []): if translate_bilingual(opt.get("text", {}), translator, dry_run): changed = True for box in q.get("boxes", []): if translate_bilingual(box.get("answer", {}), translator, dry_run): changed = True if changed: q.setdefault("meta", {})["translatedAt"] = datetime.now(timezone.utc).isoformat() return changed def save_all(questions: list, done_ids: set) -> None: with QUESTIONS_PATH.open("w", encoding="utf-8") as f: json.dump(questions, f, ensure_ascii=False, indent=2) CHECKPOINT_PATH.write_text(json.dumps(sorted(done_ids, key=int)), encoding="utf-8") def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--dry-run", action="store_true") parser.add_argument("--limit", type=int, default=0, help="Max questions to touch (0=all)") parser.add_argument( "--only-untranslated", action="store_true", default=True, help="Skip questions with all vi fields filled", ) parser.add_argument( "--no-only-untranslated", action="store_false", dest="only_untranslated", help="Re-check every question for missing vi fields", ) parser.add_argument("--ids", type=str, help="Comma-separated question ids") parser.add_argument("--reset-checkpoint", action="store_true") parser.add_argument( "--skip-unknown", action="store_true", default=True, help="Skip questions with type unknown (default: on)", ) parser.add_argument( "--no-skip-unknown", action="store_false", dest="skip_unknown", help="Include unknown-type questions in translation", ) parser.add_argument( "--exam", default="az-104", choices=sorted(__import__("exam_config").EXAMS.keys()), ) args = parser.parse_args() profile = get_profile(args.exam) global QUESTIONS_PATH, CHECKPOINT_PATH QUESTIONS_PATH = profile.questions_json CHECKPOINT_PATH = profile.translate_checkpoint if not GoogleTranslator and not args.dry_run: print("Install: pip install deep-translator", file=sys.stderr) return 1 with QUESTIONS_PATH.open(encoding="utf-8") as f: questions = json.load(f) translator = GoogleTranslator(source="en", target="vi") if not args.dry_run else None id_filter = None if args.ids: id_filter = {x.strip() for x in args.ids.split(",")} done_ids: set[str] = set() if CHECKPOINT_PATH.exists() and not args.reset_checkpoint: done_ids = set(json.loads(CHECKPOINT_PATH.read_text(encoding="utf-8"))) count = 0 for q in questions: qid = str(q["id"]) if id_filter and qid not in id_filter: continue meta = q.get("meta") or {} if meta.get("parseStatus") in ("deferred", "pending_review", "missing"): continue if args.skip_unknown and q.get("type") == "unknown": continue if args.only_untranslated and not question_needs_translation(q): done_ids.add(qid) continue if args.only_untranslated and qid in done_ids and not question_needs_translation(q): continue print(f"Translating Q{qid} ({q.get('type')})...", flush=True) if translate_question(q, translator, args.dry_run): done_ids.add(qid) count += 1 if count % 5 == 0: save_all(questions, done_ids) print(f" saved ({count} this run)") if args.limit and count >= args.limit: break save_all(questions, done_ids) print(f"Done. Updated {count} questions this run. Fully translated: {sum(1 for q in questions if not question_needs_translation(q))}/{len(questions)}") return 0 if __name__ == "__main__": sys.exit(main())