call-center/scripts/seed_kb_localizations.py

from __future__ import annotations

from argparse import ArgumentParser
from pathlib import Path
import json
import sys

from sqlalchemy import select

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

from services.shared.core import new_id, utc_now_iso
from services.shared.db import get_session
from services.shared.kb_localization import normalize_kb_language, resolve_article_group_id
from services.shared.sql_models import KBArticleRow, KBCategoryRow


def _parser() -> ArgumentParser:
    parser = ArgumentParser(description="Seed KB article localizations without creating duplicates.")
    parser.add_argument(
        "--seed-file",
        default=str(ROOT / "scripts" / "kb_voice_basic_kz_localizations.json"),
        help="Path to the localization seed JSON file.",
    )
    parser.add_argument(
        "--apply",
        action="store_true",
        help="Persist changes. Without this flag the script runs in dry-run mode.",
    )
    parser.add_argument(
        "--update-existing",
        action="store_true",
        help="Update existing localized articles instead of leaving them untouched.",
    )
    return parser


def _load_seed(path: Path) -> dict:
    return json.loads(path.read_text(encoding="utf-8"))


def _find_category(session, category_name: str) -> KBCategoryRow:
    row = session.execute(
        select(KBCategoryRow).where(KBCategoryRow.name == category_name)
    ).scalar_one_or_none()
    if row is None:
        raise RuntimeError(f"KB category not found: {category_name}")
    return row


def _find_source_article(
    session,
    *,
    category_id: str,
    source_title: str,
    source_language: str,
) -> KBArticleRow:
    rows = session.execute(
        select(KBArticleRow)
        .where(KBArticleRow.category_id == category_id)
        .where(KBArticleRow.title == source_title)
        .where(KBArticleRow.language == source_language)
        .order_by(KBArticleRow.id.asc())
    ).scalars().all()
    if not rows:
        raise RuntimeError(f"Source article not found: {source_title}")
    if len(rows) > 1:
        raise RuntimeError(f"Multiple source articles found for title: {source_title}")
    return rows[0]


def _find_localized_article(
    session,
    *,
    article_group_id: str,
    target_language: str,
) -> KBArticleRow | None:
    return session.execute(
        select(KBArticleRow)
        .where(KBArticleRow.article_group_id == article_group_id)
        .where(KBArticleRow.language == target_language)
    ).scalar_one_or_none()


def main() -> None:
    args = _parser().parse_args()
    seed_file = Path(args.seed_file).resolve()
    seed = _load_seed(seed_file)
    source_language = normalize_kb_language(seed.get("source_language"))
    target_language = normalize_kb_language(seed.get("target_language"))
    category_name = str(seed.get("category_name") or "").strip()
    if not category_name:
        raise RuntimeError("Seed file must define category_name")

    session = get_session()
    try:
        category = _find_category(session, category_name)
        summary = {
            "category_id": category.category_id,
            "category_name": category.name,
            "seed_file": str(seed_file),
            "source_language": source_language,
            "target_language": target_language,
            "apply": bool(args.apply),
            "update_existing": bool(args.update_existing),
            "created": [],
            "updated": [],
            "skipped": [],
        }

        for item in seed.get("localizations", []):
            source_title = str(item.get("source_title") or "").strip()
            title = str(item.get("title") or "").strip()
            body = str(item.get("body") or "").strip()
            tags = [str(tag).strip() for tag in item.get("tags", []) if str(tag).strip()]
            if not source_title or not title or not body:
                raise RuntimeError(f"Incomplete localization entry: {item!r}")

            source_article = _find_source_article(
                session,
                category_id=category.category_id,
                source_title=source_title,
                source_language=source_language,
            )
            article_group_id = resolve_article_group_id(
                source_article.article_id,
                source_article.article_group_id,
            )
            existing = _find_localized_article(
                session,
                article_group_id=article_group_id,
                target_language=target_language,
            )
            if existing is None:
                row = KBArticleRow(
                    article_id=new_id("kba"),
                    category_id=category.category_id,
                    article_group_id=article_group_id,
                    language=target_language,
                    title=title,
                    body=body,
                    tags_json=json.dumps(tags, ensure_ascii=False),
                    created_at=utc_now_iso(),
                    updated_at=utc_now_iso(),
                )
                session.add(row)
                summary["created"].append(
                    {
                        "source_title": source_title,
                        "article_group_id": article_group_id,
                        "target_title": title,
                    }
                )
                continue

            if args.update_existing:
                existing.title = title
                existing.body = body
                existing.tags_json = json.dumps(tags, ensure_ascii=False)
                existing.updated_at = utc_now_iso()
                summary["updated"].append(
                    {
                        "source_title": source_title,
                        "article_id": existing.article_id,
                        "article_group_id": article_group_id,
                        "target_title": title,
                    }
                )
            else:
                summary["skipped"].append(
                    {
                        "source_title": source_title,
                        "article_id": existing.article_id,
                        "article_group_id": article_group_id,
                        "target_title": existing.title,
                    }
                )

        if args.apply:
            session.commit()
        else:
            session.rollback()

        print(json.dumps(summary, ensure_ascii=True, indent=2))
    finally:
        session.close()


if __name__ == "__main__":
    main()