"""Translation editing API for a version.""" from typing import Annotated, Any from fastapi import APIRouter, Depends, HTTPException, Query, Request, status from fastapi.responses import Response from pydantic import BaseModel, Field from starlette.concurrency import run_in_threadpool from app import translation as tr from app.blobs import put_bytes from app.converter.locales import LANG_NAMES from app.deps import DB, has_perm, require from app.models import Game, GameVersion, User from app.routers.versions import can_manage, get_version, lock_version router = APIRouter(prefix="/api/versions/{version_id}/i18n", tags=["i18n"]) Player = Annotated[User, Depends(require("games.play"))] Translator = Annotated[User, Depends(require("games.play", "translations.edit"))] TODO_STATES = {"untranslated", "outdated", "fuzzy", "machine"} LANG_RE = r"^[a-z]{2,3}(-[A-Za-z0-9]{2,8})*$" FONT_EXTS = (".ttf", ".otf", ".woff", ".woff2") MAX_FONT_BYTES = 30 * 1024 * 1024 MAX_IMPORT_BYTES = 50 * 1024 * 1024 def can_edit(user: User, v: GameVersion) -> bool: """Drafts are edited by their owner; publishers may also fix published versions.""" if not has_perm(user, "translations.edit"): return False if has_perm(user, "versions.publish"): return True return v.visibility == "draft" and can_manage(user, v) async def ready_version(db: DB, user: User, version_id: int) -> GameVersion: v = await get_version(db, user, version_id) if v.status != "ready" or not v.manifest: raise HTTPException(status.HTTP_409_CONFLICT, "Версия ещё не готова") return v async def editable_version(db: DB, user: User, version_id: int) -> GameVersion: await ready_version(db, user, version_id) v = await lock_version(db, version_id) if not can_edit(user, v): raise HTTPException(status.HTTP_403_FORBIDDEN, "Переводить можно в своём черновике (форке «перевод»)") return v def check_lang(lang: str) -> str: import re if not re.match(LANG_RE, lang): raise HTTPException(status.HTTP_400_BAD_REQUEST, "Код языка в формате BCP-47, например ru") return lang @router.get("") async def overview(version_id: int, user: Player, db: DB) -> dict[str, Any]: v = await ready_version(db, user, version_id) def build() -> dict[str, Any]: game = tr.file_json(v.manifest, "game.json") idx = tr.text_index(v.manifest) langs = [] for code, info in sorted((game.get("locales") or {}).items()): langs.append({"code": code, "name": info.get("name", code), "font": info.get("font"), **tr.stats(v.manifest, code)}) return { "sourceLang": game.get("sourceLang"), "editable": can_edit(user, v), "canApprove": has_perm(user, "versions.publish"), "lines": len(idx.lines), "strings": len(idx.strings), "modules": [{"path": p, "lines": n} for p, n in idx.modules if n], "languages": langs, "knownLanguages": LANG_NAMES, } return await run_in_threadpool(build) @router.get("/{lang}/entries") async def entries(version_id: int, lang: str, user: Player, db: DB, kind: str = "lines", state: str = "all", module: str = "", q: str = "", offset: int = 0, limit: int = Query(50, ge=1, le=200)) -> dict[str, Any]: v = await ready_version(db, user, version_id) needle = q.strip().lower() def matches(st: str, *texts: str) -> bool: if state == "todo" and st not in TODO_STATES: return False if state not in ("all", "todo") and st != state: return False return not needle or any(needle in t.lower() for t in texts if t) def build() -> dict[str, Any]: idx = tr.text_index(v.manifest) lines, strings, _ = tr.locale_tables(v.manifest, lang) files = v.manifest["files"] items: list[dict[str, Any]] = [] total = 0 if kind == "strings": for src, usage in idx.strings.items(): e = strings.get(src) or {} st = e.get("state", "translated") if e.get("t") else "untranslated" if not matches(st, src, e.get("t", "")): continue total += 1 if offset < total <= offset + limit: items.append({"key": src, "usage": usage, "source": src, "translation": e.get("t", ""), "state": st}) else: for info in idx.lines: if module and info.module != module: continue e = lines.get(info.tid) st = tr.line_state(e, info) if not matches(st, info.source, (e or {}).get("t", ""), info.who): continue total += 1 if offset < total <= offset + limit: scene = files.get(info.scene) if info.scene else None items.append({ "key": info.tid, "nodeId": info.node_id, "module": info.module, "label": info.label, "who": info.who, "source": info.source, "resource": info.resource, "prev": info.prev, "next": info.next, "scene": f"/blobs/{scene['hash']}" if scene else None, "translation": (e or {}).get("t", ""), "state": st, }) return {"total": total, "items": items} return await run_in_threadpool(build) @router.get("/{lang}/meta") async def get_meta(version_id: int, lang: str, user: Player, db: DB) -> dict[str, Any]: v = await ready_version(db, user, version_id) _, _, meta = await run_in_threadpool(tr.locale_tables, v.manifest, lang) return {"glossary": meta.get("glossary", []), "font": meta.get("font")} class Changes(BaseModel): name: str | None = Field(default=None, max_length=64) lines: dict[str, dict[str, Any] | None] = {} strings: dict[str, dict[str, Any] | None] = {} glossary: list[dict[str, str]] | None = None @router.put("/{lang}") async def save(version_id: int, lang: str, body: Changes, user: Translator, db: DB ) -> dict[str, Any]: """Create a language or change entries. `null` removes an entry.""" check_lang(lang) v = await editable_version(db, user, version_id) meta = None if body.glossary is not None: meta = {"glossary": [{"src": g.get("src", "").strip(), "t": g.get("t", "").strip(), "note": g.get("note", "").strip()} for g in body.glossary if g.get("src", "").strip()]} try: manifest = await run_in_threadpool( lambda: tr.apply_changes(v.manifest, lang, lines=body.lines, strings=body.strings, meta=meta, name=body.name, can_approve=has_perm(user, "versions.publish"))) except ValueError as e: raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e)) from None v.manifest = manifest await db.commit() return await run_in_threadpool(tr.stats, manifest, lang) @router.delete("/{lang}", status_code=204) async def remove_language(version_id: int, lang: str, user: Translator, db: DB) -> None: v = await editable_version(db, user, version_id) files = {p: f for p, f in v.manifest["files"].items() if not p.startswith(f"locales/{lang}/")} game = dict(tr.file_json(v.manifest, "game.json")) game["locales"] = {k: i for k, i in (game.get("locales") or {}).items() if k != lang} tr.put_json(files, "game.json", game) v.manifest = {**v.manifest, "files": files} await db.commit() @router.put("/{lang}/font") async def upload_font(version_id: int, lang: str, filename: str, request: Request, user: Translator, db: DB) -> dict[str, Any]: """A font for this language, for games whose fonts lack its glyphs.""" check_lang(lang) ext = "." + filename.rsplit(".", 1)[-1].lower() if "." in filename else "" if ext not in FONT_EXTS: raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Шрифт: {', '.join(FONT_EXTS)}") data = await request.body() if not data or len(data) > MAX_FONT_BYTES: raise HTTPException(status.HTTP_400_BAD_REQUEST, "Пустой или слишком большой файл") v = await editable_version(db, user, version_id) path = f"locales/{lang}/font{ext}" def store() -> dict[str, Any]: blob_id, size = put_bytes(data, ext) manifest = {**v.manifest, "files": {**v.manifest["files"], path: {"hash": blob_id, "size": size}}} return tr.apply_changes(manifest, lang, meta={"font": path}) v.manifest = await run_in_threadpool(store) await db.commit() return {"font": path} @router.get("/{lang}/export") async def export(version_id: int, lang: str, user: Player, db: DB, format: str = "po") -> Response: v = await ready_version(db, user, version_id) game = await db.get(Game, v.game_id) if format == "csv": body = await run_in_threadpool(tr.export_csv, v.manifest, lang) media, ext = "text/csv", "csv" else: body = await run_in_threadpool(tr.export_po, v.manifest, lang, game.title) media, ext = "text/x-gettext-translation", "po" name = f"{game.id}-{v.id}-{lang}.{ext}" return Response("" + body if ext == "csv" else body, media_type=f"{media}; charset=utf-8", headers={"Content-Disposition": f'attachment; filename="{name}"'}) @router.post("/{lang}/import") async def import_file(version_id: int, lang: str, request: Request, user: Translator, db: DB, format: str = "po", overwrite: bool = True) -> dict[str, Any]: check_lang(lang) data = await request.body() if len(data) > MAX_IMPORT_BYTES: raise HTTPException(status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, "Слишком большой файл") text = data.decode("utf-8-sig", errors="replace") lines, strings = tr.parse_csv(text) if format == "csv" else tr.parse_po(text) v = await editable_version(db, user, version_id) if not overwrite: have_lines, have_strings, _ = tr.locale_tables(v.manifest, lang) lines = {k: x for k, x in lines.items() if not (have_lines.get(k) or {}).get("t")} strings = {k: x for k, x in strings.items() if not (have_strings.get(k) or {}).get("t")} try: v.manifest = await run_in_threadpool( lambda: tr.apply_changes(v.manifest, lang, lines=lines, strings=strings, can_approve=has_perm(user, "versions.publish"))) except ValueError as e: raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e)) from None await db.commit() return {"lines": len(lines), "strings": len(strings), **tr.stats(v.manifest, lang)}