UmbraTrans/umbratrans_web/project_status.py

# Copyright 2026 Umbra Tools (umbratools.com)
# SPDX-License-Identifier: Apache-2.0

"""Project-status adapter backed by the ``productions/`` directory.

The master store does not yet hold translation/audio/format state, so this
adapter reads ``productions/<book>-<lang>/`` directly. It is a seam: when the
store gains those rows, this module can be replaced without changing the API.

For each program, the adapter returns every configured target language plus the
English source row. Languages without a production directory are reported as
``not_started``; languages with a production directory report real chapter
progress from the runs directory.

The chapter count is a claim about the *files*, not about the *text*. A
directory full of Chinese can sit in a directory named for Spanish and be
reported as a complete Spanish translation - measured on oldbox, that is
exactly what ``deal-hunter-es`` was. So the text is measured too, and a
production whose text is not in the language it claims is reported as
``mismatched`` rather than ``complete``.
"""

from __future__ import annotations

import importlib.util
import json
import sys
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any

# slug -> productions directory name stem
_SLUG_TO_PROD = {
    "dealhunter-trans": "deal-hunter-zh",
}

# Configured target languages per program slug.
# Each entry: language_tag -> (label, role)
_TARGET_LANGUAGES = {
    "dealhunter-trans": {
        "en": ("English (source)", "source"),
        "zh-Hans": ("Simplified Chinese (zh-CN)", "translation"),
        "es-ES": ("Spanish (es-ES)", "translation"),
        "fr-FR": ("French (fr-FR)", "translation"),
        "de-DE": ("German (de-DE)", "translation"),
        "it-IT": ("Italian (it-IT)", "translation"),
        "pt-BR": ("Portuguese (pt-BR)", "translation"),
        "ru-RU": ("Russian (ru-RU)", "translation"),
        "ja-JP": ("Japanese (ja-JP)", "translation"),
        "ko-KR": ("Korean (ko-KR)", "translation"),
        "ar-SA": ("Arabic (ar-SA)", "translation"),
    },
}


@dataclass(slots=True)
class LanguageStatus:
    language_tag: str
    language_label: str
    source_present: bool
    translation: dict[str, Any]
    audio: dict[str, Any]
    outputs: dict[str, Any]
    review: dict[str, Any]
    updated_at: str | None


def _load_json(path: Path) -> dict[str, Any] | None:
    if not path.is_file():
        return None
    try:
        with path.open("r", encoding="utf-8") as fh:
            data = json.load(fh)
        return data if isinstance(data, dict) else None
    except Exception:
        return None


def _max_mtime(paths: list[Path]) -> str | None:
    mtimes = [p.stat().st_mtime for p in paths if p.exists()]
    if not mtimes:
        return None
    return datetime.fromtimestamp(max(mtimes)).isoformat()


# The translation language registry, loaded once. It lives in
# ``tools/translation/languages.py``, which is not a distributed package, so
# it is loaded from the repository by path. It is the same file the pipeline
# uses, so the dashboard cannot disagree with it about which script a
# language is written in.
_LANGUAGE_REGISTRY: Any | None = None
_LANGUAGE_REGISTRY_TRIED = False


def _language_registry() -> Any | None:
    """The translation language registry, or None if it cannot be loaded."""
    global _LANGUAGE_REGISTRY, _LANGUAGE_REGISTRY_TRIED
    if _LANGUAGE_REGISTRY_TRIED:
        return _LANGUAGE_REGISTRY
    _LANGUAGE_REGISTRY_TRIED = True
    path = Path(__file__).resolve().parents[1] / "tools" / "translation" / "languages.py"
    try:
        if path.is_file():
            spec = importlib.util.spec_from_file_location("umbratrans_languages", path)
            if spec and spec.loader:
                module = importlib.util.module_from_spec(spec)
                # The module has to be in sys.modules before it is executed:
                # @dataclass resolves its own module through sys.modules.
                sys.modules[spec.name] = module
                spec.loader.exec_module(module)
                _LANGUAGE_REGISTRY = module
                return module
        from tools.translation import languages  # type: ignore[import-not-found]

        _LANGUAGE_REGISTRY = languages
        return languages
    except Exception:
        return None


def _language_problem(runs_dir: Path | None, tag: str) -> str | None:
    """Why the text in a production is not in the language it claims.

    Returns None when the text is in the declared language. When the check
    cannot be performed at all it says so, rather than passing silently: a
    production that cannot be verified is not one that can be reported
    complete.
    """
    if runs_dir is None or not runs_dir.is_dir():
        return None
    registry = _language_registry()
    if registry is None:
        return "the language registry could not be loaded, so the text could not be checked"
    try:
        return registry.check_runs_language(runs_dir, tag)
    except Exception as exc:  # pragma: no cover - defensive
        return f"the language could not be checked: {exc}"


def _count_translated_chapters(runs_dir: Path) -> tuple[int, int]:
    """Return (done, total) chapter counts from the runs directory."""
    done = 0
    chapter_dirs: list[Path] = []
    if runs_dir.is_dir():
        for ch in sorted(runs_dir.iterdir()):
            if ch.is_dir() and ch.name.startswith("ch-"):
                chapter_dirs.append(ch)
                if (ch / "translation.json").is_file():
                    done += 1
    return done, len(chapter_dirs)


def _read_outputs(out_progress: Path, out_final: Path) -> dict[str, Any]:
    """Read available output files from progress and final directories."""
    outputs: dict[str, Any] = {}
    candidates: list[tuple[Path, str]] = []
    if out_progress.is_dir():
        candidates.append((out_progress, "output-progress"))
    if out_final.is_dir():
        candidates.append((out_final, "output"))
    for subdir, key in candidates:
        for p in subdir.iterdir():
            if not p.is_file():
                continue
            name = p.name
            if name.endswith(".pdf"):
                outputs.setdefault("pdf-paperback", {"available": True, "path": str(p), "source": key})
            elif name.endswith(".odt") or name.endswith(".asciidoc"):
                outputs.setdefault("epub", {"available": True, "path": str(p), "source": key})
    return outputs


class ProjectStatusAdapter:
    def __init__(self, productions_root: Path | None = None) -> None:
        if productions_root is not None:
            self.productions_root = Path(productions_root)
        else:
            self.productions_root = Path(__file__).resolve().parents[1] / "productions"

    def status_for(self, project) -> dict[str, Any]:
        prod_name = _SLUG_TO_PROD.get(getattr(project, "slug", ""))
        prod_dir = self.productions_root / prod_name if prod_name else None
        runs_dir = prod_dir / "runs" if prod_dir else None

        # Chapter totals come from the source book when available; fall back
        # to the number of chapter directories found in runs.
        done, total = _count_translated_chapters(runs_dir) if runs_dir else (0, 0)

        languages: list[dict[str, Any]] = []

        # English source row is always present when the project exists.
        languages.append(
            {
                "language_tag": "en",
                "language_label": "English (source)",
                "role": "source",
                "source_present": True,
                "translation": {
                    "state": "complete",
                    "chapters_done": total,
                    "chapters_total": total,
                    "language_problem": None,
                },
                "audio": {"state": "unknown"},
                "outputs": {},
                "review": {"state": "ok"},
            }
        )

        target_langs = _TARGET_LANGUAGES.get(getattr(project, "slug", ""), {})

        for tag in sorted(target_langs.keys()):
            if tag == "en":
                continue  # source row already added above
            label, role = target_langs[tag]

            # A language-specific production directory would be named
            # deal-hunter-<tag> (e.g. deal-hunter-es for Spanish). It is
            # looked up on its own: a Spanish production exists or it does
            # not, regardless of whether the Chinese one is there.
            lang_prod_name = f"deal-hunter-{tag.split('-')[0]}"
            lang_prod_dir = self.productions_root / lang_prod_name
            lang_runs = lang_prod_dir / "runs" if lang_prod_dir.is_dir() else None
            lang_out_progress = lang_prod_dir / "output-progress" if lang_prod_dir.is_dir() else None
            lang_out_final = lang_prod_dir / "output" if lang_prod_dir.is_dir() else None

            if lang_runs and lang_runs.is_dir():
                lang_done, lang_total = _count_translated_chapters(lang_runs)
                state = (
                    "complete"
                    if lang_total and lang_done >= lang_total
                    else ("in_progress" if lang_done else "not_started")
                )
                outputs = _read_outputs(lang_out_progress or Path(), lang_out_final or Path())
                updated_at = _max_mtime(
                    [lang_runs / f"ch-{i}" / "translation.json" for i in range(1, 40)] if lang_runs.is_dir() else []
                )
            else:
                state = "not_started"
                lang_done, lang_total = 0, 0
                outputs = {}
                updated_at = None

            # The chapter count says the files are there. The script says
            # which language they are in. A production that declares one
            # language and holds another is not complete in the language
            # it declares.
            language_problem = _language_problem(lang_runs, tag)
            if language_problem:
                state = "mismatched"

            languages.append(
                {
                    "language_tag": tag,
                    "language_label": label,
                    "role": role,
                    "source_present": tag == "en",
                    "translation": {
                        "state": state,
                        "chapters_done": lang_done,
                        "chapters_total": lang_total,
                        "language_problem": language_problem,
                    },
                    "audio": {"state": "not_started"},
                    "outputs": outputs,
                    "review": {"state": "needs_review" if state in {"complete", "mismatched"} else "pending"},
                    "updated_at": updated_at,
                }
            )

        return {"project_id": str(getattr(project, "id")), "languages": languages}