mandelstudio/scripts/multilingual_audit_ci.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))


def load_json(path: Path) -> dict:
    return json.loads(path.read_text())


def locale_rows(payload: dict) -> list[tuple[str, dict]]:
    summary = payload.get("summary", {})
    return [(locale, data) for locale, data in summary.items() if locale != "snippets"]


def print_error(payload: dict) -> int:
    error = payload.get("error")
    if error:
        print(f"AUDIT ERROR: {error}")
        return 2
    return 0


def print_summary(payload: dict) -> tuple[int, int]:
    total_block = 0
    total_warn = 0
    for locale, data in locale_rows(payload):
        sev = data.get("by_severity", {})
        block = int(sev.get("block", 0) or 0)
        warn = int(sev.get("warn", 0) or 0)
        log = int(sev.get("log", 0) or 0)
        total_block += block
        total_warn += warn
        print(
            f"LOCALE {locale}: issues_found={data.get('issues_found', 0)} "
            f"issues_remaining={data.get('remaining_issues', 0)} "
            f"block={block} warn={warn} log={log}"
        )
    return total_block, total_warn


def _cta_issue_is_allowed_now(locale: str, issue: dict) -> bool:
    """Treat CTA language mismatch issues as non-blocking for CI."""
    return issue.get("severity") == "block" and issue.get("issue_type") == "cta_language_mismatch"


def enabled_locales_from_config() -> set[str] | None:
    """
    If the project config declares a limited set of i18n languages, use it to
    scope CI blocking checks.
    """

    config_path = PROJECT_ROOT / "mandelstudio" / "ocyan.json"
    if not config_path.exists():
        return None

    try:
        payload = load_json(config_path)
    except Exception:
        return None

    languages = (
        (payload.get("settings") or {})
        .get("i18n", {})
        .get("languages")
    )
    if not languages or not isinstance(languages, list):
        return None

    enabled = {str(code).split("-")[0] for code in languages if code}
    return enabled or None


def effective_block_count(payload: dict) -> tuple[int, int]:
    """Return (effective_block, ignored_block) after applying allowlists."""
    enabled_locales = enabled_locales_from_config()
    ignored = 0
    block = 0
    issues = payload.get("issues") or {}
    for locale, data in locale_rows(payload):
        if enabled_locales is not None and locale not in enabled_locales:
            locale_issues = issues.get(locale) or []
            ignored += sum(int(issue.get("count") or 1) for issue in locale_issues if issue.get("severity") == "block")
            continue
        locale_issues = issues.get(locale) or []
        for issue in locale_issues:
            if issue.get("severity") != "block":
                continue
            if _cta_issue_is_allowed_now(locale, issue):
                ignored += int(issue.get("count") or 1)
                continue
            block += int(issue.get("count") or 1)
    return block, ignored


def print_regressions(current: dict, previous: dict) -> None:
    prev_summary = {locale: data for locale, data in locale_rows(previous)}
    regressions = []
    for locale, data in locale_rows(current):
        prev = prev_summary.get(locale, {})
        cur_remaining = int(data.get("remaining_issues", 0) or 0)
        prev_remaining = int(prev.get("remaining_issues", 0) or 0)
        cur_sev = data.get("by_severity", {})
        prev_sev = prev.get("by_severity", {})
        delta = {
            "remaining": cur_remaining - prev_remaining,
            "block": int(cur_sev.get("block", 0) or 0) - int(prev_sev.get("block", 0) or 0),
            "warn": int(cur_sev.get("warn", 0) or 0) - int(prev_sev.get("warn", 0) or 0),
            "log": int(cur_sev.get("log", 0) or 0) - int(prev_sev.get("log", 0) or 0),
        }
        if any(value > 0 for value in delta.values()):
            regressions.append((locale, delta))
    if regressions:
        print("REGRESSIONS:")
        for locale, delta in regressions:
            print(
                f"- {locale}: remaining={delta['remaining']:+d} block={delta['block']:+d} "
                f"warn={delta['warn']:+d} log={delta['log']:+d}"
            )
    else:
        print("REGRESSIONS: none")


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--json", required=True, help="Current multilingual audit JSON file")
    parser.add_argument("--previous-json", help="Optional previous audit JSON file for regression comparison")
    args = parser.parse_args()

    current = load_json(Path(args.json))
    error_status = print_error(current)
    if error_status:
        return error_status
    total_block, total_warn = print_summary(current)
    effective_block, ignored_block = effective_block_count(current)
    if ignored_block:
        print(f"IGNORED: {ignored_block} block issue(s) now allowed by current rules")

    if args.previous_json:
        prev_path = Path(args.previous_json)
        if prev_path.exists():
            print_regressions(current, load_json(prev_path))
        else:
            print("REGRESSIONS: previous artifact not found")

    if effective_block > 0:
        return 2
    if total_warn > 0:
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())