import_sentences.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""句库导入脚本：CSV → SQLite（data/app.db）。

用法：
    python import_sentences.py sentences.csv [manifest.csv]

- sentences.csv：必需列 id,text；可选列 scene,wugniu,mandarin。缺列/缺值
  均容错（注音数据后续补齐后重跑本脚本即可，不会清空已有值）；
- manifest.csv：M0 批量合成产物（默认位置 audio_cache/manifest.csv），提供
  audio_file / model / voice / status。参数省略时自动找默认位置；找不到则
  仅导入文本，audio_file 置空，页面显示"无音频"；
- 可重复执行：按 id UPSERT，只用"本次提供了非空值"的字段覆盖旧值；
- 本脚本对 audio_cache/ 只读，绝不写入。
"""
import csv
import sys
from pathlib import Path

from backend import config
from backend.db import connect, init_db

REQUIRED_COLS = {"id", "text"}
OPTIONAL_COLS = ("scene", "wugniu", "mandarin")


def read_sentences(path: Path) -> list[dict]:
    with path.open(encoding="utf-8-sig", newline="") as f:
        reader = csv.DictReader(f)
        cols = set(reader.fieldnames or [])
        missing = REQUIRED_COLS - cols
        if missing:
            sys.exit(f"错误：{path} 缺少必需列 {sorted(missing)}（现有列：{sorted(cols)}）")
        rows = []
        for lineno, raw in enumerate(reader, start=2):
            sid = (raw.get("id") or "").strip()
            text = (raw.get("text") or "").strip()
            if not sid and not text:
                continue
            if not sid or not text:
                print(f"[跳过] 第 {lineno} 行 id/text 不完整：{raw}")
                continue
            row = {"id": sid, "text": text}
            for col in OPTIONAL_COLS:
                val = (raw.get(col) or "").strip()
                row[col] = val or None
            rows.append(row)
    return rows


def read_manifest(path: Path) -> dict[str, dict]:
    info: dict[str, dict] = {}
    with path.open(encoding="utf-8-sig", newline="") as f:
        for raw in csv.DictReader(f):
            sid = (raw.get("id") or "").strip()
            if not sid:
                continue
            model = (raw.get("model") or "").strip()
            voice = (raw.get("voice") or "").strip()
            info[sid] = {
                "text": (raw.get("text") or "").strip(),
                "audio_file": (raw.get("audio_file") or "").strip(),
                "status": (raw.get("status") or "").strip().lower(),
                "source": "/".join(x for x in (model, voice) if x) or None,
            }
    return info


def resolve_manifest_path(arg: str | None) -> Path | None:
    default = config.AUDIO_CACHE_DIR / "manifest.csv"
    if arg:
        p = Path(arg)
        if p.is_file():
            return p
        if default.is_file():
            print(f"[提示] {p} 不存在，改用默认位置 {default}")
            return default
        print(f"[警告] manifest 不存在（{p}），本次仅导入文本，音频字段置空")
        return None
    if default.is_file():
        print(f"[提示] 未指定 manifest，自动使用 {default}")
        return default
    print("[警告] 未找到 manifest.csv（M0 尚未合成？），本次仅导入文本，音频字段置空")
    return None


def main() -> None:
    if len(sys.argv) < 2 or len(sys.argv) > 3:
        sys.exit(__doc__)
    sentences_path = Path(sys.argv[1])
    if not sentences_path.is_file():
        sys.exit(f"错误：句库文件不存在：{sentences_path}")

    manifest_path = resolve_manifest_path(sys.argv[2] if len(sys.argv) == 3 else None)
    sentences = read_sentences(sentences_path)
    manifest = read_manifest(manifest_path) if manifest_path else {}

    init_db()
    conn = connect()
    inserted = updated = 0
    drift: list[str] = []
    try:
        for row in sentences:
            audio_file = audio_source = None
            m = manifest.get(row["id"])
            if m:
                if m["text"] and m["text"] != row["text"]:
                    drift.append(
                        f"  {row['id']}: 句库「{row['text']}」≠ 合成时「{m['text']}」"
                    )
                wav = config.AUDIO_CACHE_DIR / m["audio_file"] if m["audio_file"] else None
                if m["status"] == "ok" and wav is not None and wav.is_file():
                    audio_file, audio_source = m["audio_file"], m["source"]
                elif m["status"] == "ok":
                    print(f"[警告] manifest 标记 ok 但音频文件缺失：{wav}")

            old = conn.execute(
                "SELECT * FROM sentences WHERE id = ?", (row["id"],)
            ).fetchone()
            if old is None:
                conn.execute(
                    "INSERT INTO sentences"
                    " (id, text, wugniu, mandarin, scene, audio_file, audio_source)"
                    " VALUES (?, ?, ?, ?, ?, ?, ?)",
                    (
                        row["id"],
                        row["text"],
                        row["wugniu"],
                        row["mandarin"],
                        row["scene"] or "未分类",
                        audio_file,
                        audio_source or "qwen3-tts-flash/Jada",
                    ),
                )
                inserted += 1
            else:
                conn.execute(
                    "UPDATE sentences SET text=?, wugniu=?, mandarin=?, scene=?,"
                    " audio_file=?, audio_source=? WHERE id=?",
                    (
                        row["text"],
                        row["wugniu"] or old["wugniu"],
                        row["mandarin"] or old["mandarin"],
                        row["scene"] or old["scene"],
                        audio_file or old["audio_file"],
                        audio_source or old["audio_source"],
                        row["id"],
                    ),
                )
                updated += 1
        conn.commit()

        extra = sorted(set(manifest) - {r["id"] for r in sentences})
        if extra:
            print(f"[提示] manifest 中有 {len(extra)} 个 id 不在句库 CSV：{extra[:10]}")
        if drift:
            print("[警告] 以下句子文本与合成时不一致（音频可能对应旧文本）：")
            print("\n".join(drift))

        total = conn.execute("SELECT COUNT(*) FROM sentences").fetchone()[0]
        have = conn.execute(
            "SELECT COUNT(*) FROM sentences WHERE audio_file IS NOT NULL"
        ).fetchone()[0]
        print(f"\n导入完成：新增 {inserted} 句，更新 {updated} 句。")
        print(f"库中共 {total} 句，其中 {have} 句有标准音，{total - have} 句缺音频。")
        if total - have:
            rows = conn.execute(
                "SELECT id, text FROM sentences WHERE audio_file IS NULL ORDER BY id LIMIT 20"
            ).fetchall()
            for r in rows:
                print(f"  [缺音频] {r['id']}: {r['text']}")
    finally:
        conn.close()


if __name__ == "__main__":
    main()