"""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"))] 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 = Query(0, ge=0), limit: int = Query(50, ge=1, le=200), locate: str = "") -> dict[str, Any]: v = await ready_version(db, user, version_id) return await run_in_threadpool( lambda: tr.entries_page(v.manifest, lang, kind=kind, state=state, module=module, q=q, offset=offset, limit=limit, locate=locate)) @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)}