"""
VOC 评论结构化服务（命令行 / 直接调用，无 MCP）。

----------------------------------------------------------------------
虚拟环境（推荐 Python 3.10+）

  cd "/Users/onesvmwhoops/Cursor_Project/VOC_LLM结构化"
  source 310py/bin/activate
  pip install -r requirements.txt

----------------------------------------------------------------------
运行前提供密钥（任选其一；勿把密钥写进代码仓库）::

  export DASHSCOPE_API_KEY="sk-xxx"
  # 或项目根单行文件 .dashscope_key
  # 或 export DASHSCOPE_API_KEY_FILE="/path/to/key.txt"

----------------------------------------------------------------------
用法::

  python3 结构化_server.py --industry "Pet supplements" --product "Turkey tail mushroom for dogs" --file merged_reviews_cleaned.csv

  python3 结构化_server.py --smoke

模型：默认 ``qwen3.6-flash``（OpenAI 兼容 Chat API；可通过环境变量 ``DASHSCOPE_MODEL`` 覆盖）。

地域：固定中国大陆华北2（北京）``https://dashscope.aliyuncs.com/compatible-mode/v1``，不使用新加坡/国际节点。

数据库：项目根目录 ``voc_structured.sqlite``；每次写入前会清理该库及下游 ``voc_embeddings.sqlite``、``voc_clustering.sqlite``（冒烟测试用临时库时不清理项目根文件）。
"""
from __future__ import annotations

import argparse
import importlib.util
import json
import logging
import os
import re
import sqlite3
import sys
from csv import DictReader
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Tuple

from prompts.loader import product_feedback_categories

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    stream=sys.stderr,
)
logger = logging.getLogger("voc_structured")

PROJECT_ROOT = Path(__file__).resolve().parent
DB_PATH = PROJECT_ROOT / "voc_structured.sqlite"
EMBED_DB = PROJECT_ROOT / "voc_embeddings.sqlite"
CLUSTER_DB = PROJECT_ROOT / "voc_clustering.sqlite"
# 中国大陆华北2（北京）OpenAI 兼容模式
DASHSCOPE_BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
MODEL_NAME = os.environ.get("DASHSCOPE_MODEL", "qwen3.6-flash").strip()
REVIEW_COLUMN_NAMES = ("content", "review", "评论")
BATCH_FETCH_MAX_ATTEMPTS = 3
BATCH_CORRECTION_MAX_ATTEMPTS = 3

_VALID_CATEGORIES = product_feedback_categories()  # 模块加载时快照；校验时用 _get_valid_categories()


def _get_valid_categories() -> frozenset[str]:
    return product_feedback_categories()

# 模型上下文上限（保守估）；单批仍用下方 DEFAULT_MAX_* 控制，保证 JSON 稳定
MODEL_MAX_INPUT_TOKENS = 991_800
MODEL_MAX_OUTPUT_TOKENS = 65_530
CHARS_PER_TOKEN_EST = 3.2
DEFAULT_MAX_BATCH_INPUT_TOKENS = 32_000
DEFAULT_MAX_BATCH_OUTPUT_TOKENS = 12_000
DEFAULT_OUTPUT_TOKENS_PER_REVIEW = 450
BATCH_COUNT_MIN = 1
BATCH_COUNT_MAX = 50
# 批量结构化单次请求输出 token（含 JSON 开销，上限不超过模型）
BATCH_OUTPUT_TOKEN_BUFFER = 1024
BATCH_OUTPUT_TOKEN_FLOOR = 4096


def _batch_max_output_tokens(review_count: int) -> int:
    est = review_count * DEFAULT_OUTPUT_TOKENS_PER_REVIEW + BATCH_OUTPUT_TOKEN_BUFFER
    return min(MODEL_MAX_OUTPUT_TOKENS, max(BATCH_OUTPUT_TOKEN_FLOOR, est))


def _resolve_dashscope_api_key() -> str:
    """环境变量 > DASHSCOPE_API_KEY_FILE > 项目根 .dashscope_key（均仅一行密钥，无引号）。"""
    v = os.environ.get("DASHSCOPE_API_KEY", "").strip()
    if v:
        return v
    fp = os.environ.get("DASHSCOPE_API_KEY_FILE", "").strip()
    if fp:
        p = Path(fp).expanduser()
        if p.is_file():
            return p.read_text(encoding="utf-8").strip().strip('"').strip("'")
    local = PROJECT_ROOT / ".dashscope_key"
    if local.is_file():
        return local.read_text(encoding="utf-8").strip().strip('"').strip("'")
    return ""


def _load_prompt_module():
    path = PROJECT_ROOT / "结构化_Prompt.py"
    spec = importlib.util.spec_from_file_location("voc_structured_prompts", path)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"Cannot load prompt module from {path}")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


_PROMPTS = _load_prompt_module()


def _clean_voc_sqlite_databases(structured_db: Path) -> None:
    """写入前删除 SQLite，避免与旧 job / 下游向量、聚类结果混用。"""
    paths = [structured_db.resolve()]
    if structured_db.resolve() == DB_PATH.resolve():
        paths.extend([EMBED_DB.resolve(), CLUSTER_DB.resolve()])
    for p in paths:
        if p.is_file():
            p.unlink()
            logger.info("已清理 SQLite: %s", p.name)


def init_sqlite_schema(conn: sqlite3.Connection) -> None:
    conn.executescript(
        """
        CREATE TABLE IF NOT EXISTS analysis_jobs (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            created_at TEXT NOT NULL,
            industry TEXT NOT NULL,
            product_name TEXT NOT NULL,
            source_file TEXT NOT NULL,
            batch_size INTEGER NOT NULL,
            model TEXT NOT NULL,
            full_result_json TEXT NOT NULL
        );
        CREATE TABLE IF NOT EXISTS comment_extractions (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            job_id INTEGER NOT NULL,
            source_row INTEGER NOT NULL,
            batch_index INTEGER NOT NULL,
            comment_key TEXT NOT NULL,
            extraction_json TEXT NOT NULL,
            UNIQUE(job_id, source_row),
            FOREIGN KEY (job_id) REFERENCES analysis_jobs(id)
        );
        """
    )


def load_reviews_from_file(file_path: str) -> List[Tuple[int, str]]:
    """
    返回 (source_row, text) 列表。
    CSV：source_row 为表头下的数据行序号（第 1 行数据=1），空单元格跳过。
    TXT：source_row 为文件物理行号（从 1 起），空行跳过。
    """
    p = Path(file_path).expanduser().resolve()
    if not p.is_file():
        raise FileNotFoundError(f"File not found: {p}")

    suffix = p.suffix.lower()
    if suffix == ".csv":
        return _load_csv_reviews(p)
    if suffix == ".txt" or suffix == ".text":
        return _load_txt_reviews(p)
    raise ValueError(f"Unsupported file type {suffix!r}; use .csv or .txt")


def _load_txt_reviews(p: Path) -> List[Tuple[int, str]]:
    out: List[Tuple[int, str]] = []
    with p.open(encoding="utf-8-sig", newline="") as f:
        for line_no, line in enumerate(f, start=1):
            t = line.strip("\r\n").strip()
            if not t:
                continue
            out.append((line_no, t))
    return out


def _load_csv_reviews(p: Path) -> List[Tuple[int, str]]:
    with p.open(encoding="utf-8-sig", newline="") as f:
        reader = DictReader(f)
        if not reader.fieldnames:
            raise ValueError("CSV has no header row")
        headers = [h.strip() for h in reader.fieldnames]
        lower_map = {h.lower(): h for h in headers}
        col: str | None = None
        for name in REVIEW_COLUMN_NAMES:
            if name.lower() in lower_map:
                col = lower_map[name.lower()]
                break
        if col is None:
            if len(headers) == 1:
                col = headers[0]
            else:
                raise ValueError(
                    "CSV must contain column 'content', 'review', or '评论', "
                    f"or be single-column. Found columns: {headers}"
                )
        rows = list(reader)
    out: List[Tuple[int, str]] = []
    for idx, row in enumerate(rows, start=1):
        cell = row.get(col)
        if cell is None:
            cell = ""
        t = str(cell).strip()
        if not t:
            continue
        out.append((idx, t))
    return out


def _chunked(items: List[Tuple[int, str]], size: int) -> List[List[Tuple[int, str]]]:
    return [items[i : i + size] for i in range(0, len(items), size)]


def _estimate_tokens(text: str) -> int:
    """英文评论粗略估 token（偏保守，避免顶满上下文）。"""
    return max(1, int(len(text) / CHARS_PER_TOKEN_EST))


def _estimate_batch_input_tokens(
    industry: str,
    product_name: str,
    batch: List[Tuple[int, str]],
) -> int:
    tagged, keys, _ = _format_tagged_batch(batch)
    system, user = _PROMPTS.build_batch_review_analysis_prompts(
        industry=industry,
        product_name=product_name,
        keys=keys,
        tagged_input=tagged,
    )
    return _estimate_tokens(system) + _estimate_tokens(user)


def _estimate_batch_output_tokens(review_count: int) -> int:
    return review_count * DEFAULT_OUTPUT_TOKENS_PER_REVIEW


def chunk_reviews_by_token_budget(
    reviews: List[Tuple[int, str]],
    industry: str,
    product_name: str,
    *,
    max_input_tokens: int = DEFAULT_MAX_BATCH_INPUT_TOKENS,
    max_output_tokens: int = DEFAULT_MAX_BATCH_OUTPUT_TOKENS,
    max_reviews_per_batch: int = BATCH_COUNT_MAX,
    min_reviews_per_batch: int = BATCH_COUNT_MIN,
) -> List[List[Tuple[int, str]]]:
    """
    按预估输入/输出 token 与单批条数上限，将评论动态打包。
    短评可多条约一批，长评自动减少条数（单条超长则单独成批）。
    """
    max_input_tokens = min(max_input_tokens, MODEL_MAX_INPUT_TOKENS)
    max_output_tokens = min(max_output_tokens, MODEL_MAX_OUTPUT_TOKENS)
    max_reviews_per_batch = max(min_reviews_per_batch, max_reviews_per_batch)

    batches: List[List[Tuple[int, str]]] = []
    i = 0
    while i < len(reviews):
        batch: List[Tuple[int, str]] = []
        while i < len(reviews):
            candidate = batch + [reviews[i]]
            if len(candidate) > max_reviews_per_batch:
                break

            inp = _estimate_batch_input_tokens(industry, product_name, candidate)
            out = _estimate_batch_output_tokens(len(candidate))

            if batch and (inp > max_input_tokens or out > max_output_tokens):
                break

            batch = candidate
            i += 1

            if inp > max_input_tokens:
                logger.warning(
                    "单条评论过长（约 %s input tokens），单独成批: source_row=%s",
                    inp,
                    batch[0][0],
                )
                break

        if not batch:
            break
        batches.append(batch)

    return batches


def _summarize_batch_plan(
    batches: List[List[Tuple[int, str]]],
    industry: str,
    product_name: str,
) -> List[Dict[str, Any]]:
    plan: List[Dict[str, Any]] = []
    for idx, batch in enumerate(batches):
        chars = sum(len(t) for _, t in batch)
        plan.append(
            {
                "batch_index": idx,
                "review_count": len(batch),
                "total_chars": chars,
                "est_input_tokens": _estimate_batch_input_tokens(
                    industry, product_name, batch
                ),
                "est_output_tokens": _estimate_batch_output_tokens(len(batch)),
            }
        )
    return plan


def _format_tagged_batch(batch: List[Tuple[int, str]]) -> Tuple[str, List[str], Dict[str, int]]:
    """返回 tagged 文本、键列表 C1..Cn、Ci -> source_row。"""
    lines: List[str] = []
    keys: List[str] = []
    key_to_row: Dict[str, int] = {}
    for i, (source_row, text) in enumerate(batch, start=1):
        key = f"C{i}"
        keys.append(key)
        key_to_row[key] = source_row
        safe = text.replace("\r\n", "\n").replace("\r", "\n")
        lines.append(f"[{key}] {safe}")
    return "\n".join(lines), keys, key_to_row


def _strip_code_fence(text: str) -> str:
    t = text.strip()
    if t.startswith("```"):
        lines = t.split("\n")
        if len(lines) >= 2 and lines[0].startswith("```"):
            lines = lines[1:]
        if lines and lines[-1].strip() == "```":
            lines = lines[:-1]
        t = "\n".join(lines).strip()
    return t


def parse_model_json(raw: str) -> Any:
    t = _strip_code_fence(raw)
    try:
        return json.loads(t)
    except json.JSONDecodeError:
        m = re.search(r"[\{\[][\s\S]*[\}\]]\s*$", t)
        if not m:
            raise
        return json.loads(m.group(0))


def parse_model_json_object(raw: str) -> Dict[str, Any]:
    parsed = parse_model_json(raw)
    if isinstance(parsed, dict):
        return parsed
    raise ValueError(
        f"模型 JSON 应为对象，实际为 {type(parsed).__name__}；请使用批量对象格式 {{\"C1\": ...}}"
    )


def _call_dashscope_chat(
    messages: List[Dict[str, str]],
    *,
    max_tokens: int | None = None,
) -> Any:
    api_key = _resolve_dashscope_api_key()
    if not api_key:
        raise RuntimeError(
            "Missing DashScope API key: set DASHSCOPE_API_KEY in the process environment, "
            "or DASHSCOPE_API_KEY_FILE to a one-line key file, "
            "or create a one-line file at project root: .dashscope_key"
        )

    from openai import OpenAI

    client = OpenAI(api_key=api_key, base_url=DASHSCOPE_BASE_URL)
    extra_body: Dict[str, Any] = {}
    model_lower = MODEL_NAME.lower()
    if model_lower.startswith(("qwen3.6", "qwen3.5", "qwen3")) or model_lower.startswith(
        "deepseek-v4"
    ):
        extra_body["enable_thinking"] = False

    out_tokens = max_tokens if max_tokens is not None else DEFAULT_MAX_BATCH_OUTPUT_TOKENS
    out_tokens = min(MODEL_MAX_OUTPUT_TOKENS, max(BATCH_OUTPUT_TOKEN_FLOOR, out_tokens))

    resp = client.chat.completions.create(
        model=MODEL_NAME,
        messages=messages,
        response_format={"type": "json_object"},
        temperature=0.2,
        max_tokens=out_tokens,
        **({"extra_body": extra_body} if extra_body else {}),
    )
    msg = resp.choices[0].message
    choice = msg.content
    if not choice and getattr(msg, "reasoning_content", None):
        choice = msg.reasoning_content
    if not choice:
        raise RuntimeError("Empty model response")
    return parse_model_json(choice)


def _call_dashscope_model(system: str, user: str) -> Dict[str, Any]:
    return _call_dashscope_chat(
        [
            {"role": "system", "content": system},
            {"role": "user", "content": user},
        ]
    )


def _coalesce_batch_response(raw: Dict[str, Any], keys: List[str]) -> Dict[str, Any]:
    """若模型把 C1..Cn 包在嵌套对象里，尝试展开到顶层。"""
    if any(k in raw for k in keys):
        return raw
    for value in raw.values():
        if isinstance(value, dict) and any(k in value for k in keys):
            return value
    return raw


def _normalize_batch_response(raw: Any, keys: List[str]) -> Dict[str, Any]:
    """
    统一为 {C1: {...}, C2: {...}}。
    兼容模型误返回 JSON 数组、嵌套对象或按序号的对象。
    """
    if isinstance(raw, list):
        out: Dict[str, Any] = {}
        for i, key in enumerate(keys):
            if i >= len(raw):
                break
            item = raw[i]
            if not isinstance(item, dict):
                continue
            if len(item) == 1:
                sole_k, sole_v = next(iter(item.items()))
                if sole_k in keys and isinstance(sole_v, dict):
                    out[sole_k] = sole_v
                    continue
            if any(k in item for k in keys):
                for k in keys:
                    if k in item and isinstance(item[k], dict):
                        out[k] = item[k]
                continue
            out[key] = item
        if out:
            return out
        raise ValueError(
            "模型返回 JSON 数组，但无法按顺序映射为 "
            + ", ".join(keys[: min(len(raw), len(keys))])
        )

    if not isinstance(raw, dict):
        raise TypeError(f"批量 JSON 应为 object 或 array，实际为 {type(raw).__name__}")

    coalesced = _coalesce_batch_response(raw, keys)
    if any(k in coalesced for k in keys):
        return coalesced

    # 如 {"1": {...}, "2": {...}} 映射到 C1, C2
    remapped: Dict[str, Any] = {}
    for i, key in enumerate(keys, start=1):
        for alias in (str(i), f"C{i}"):
            if alias in coalesced and isinstance(coalesced[alias], dict):
                remapped[key] = coalesced[alias]
                break
    if remapped:
        return remapped

    return coalesced


_SENTIMENT_CANONICAL = {
    "positive": "Positive",
    "negative": "Negative",
    "neutral": "Neutral",
}

# 模型常见误写 -> 合法 category（校验前自动映射，减少无效重试）
_CATEGORY_ALIASES: Dict[str, str] = {
    "value": "Price",
    "values": "Price",
    "cost": "Price",
    "pricing": "Price",
    "value for money": "Price",
    "worth": "Price",
    "packaging": "Logistics",
    "shipping": "Logistics",
    "delivery": "Logistics",
    "service": "Customer Service",
    "customer support": "Customer Service",
    "support": "Customer Service",
    "design": "Appearance",
    "look": "Appearance",
    "performance": "Function",
    "efficacy": "Function",
    "effectiveness": "Function",
}


def _normalize_sentiment(raw: Any) -> str:
    key = str(raw or "").strip().lower()
    return _SENTIMENT_CANONICAL.get(key, "Neutral")


def _normalize_category(raw: Any) -> str | None:
    valid = _get_valid_categories()
    key = str(raw or "").strip()
    if not key:
        return None
    if key in valid:
        return key
    mapped = _CATEGORY_ALIASES.get(key.lower())
    if mapped:
        return mapped
    lower = key.lower()
    for v in valid:
        if v.lower() == lower:
            return v
    return None


def _normalize_product_feedback(items: Any) -> List[Dict[str, str]]:
    if not isinstance(items, list):
        return []
    out: List[Dict[str, str]] = []
    for item in items:
        if not isinstance(item, dict):
            continue
        aspect = str(item.get("aspect", "")).strip()
        opinion = str(item.get("opinion", "")).strip()
        category = _normalize_category(item.get("category"))
        if aspect and opinion and category:
            out.append(
                {
                    "aspect": aspect,
                    "opinion": opinion,
                    "sentiment": _normalize_sentiment(item.get("sentiment")),
                    "category": category,
                }
            )
    return out


def _validate_extraction_strict(obj: Any, key_label: str = "") -> Dict[str, Any]:
    """校验结构化结果；失败时抛出带键名的 ValueError，供回传模型修正。"""
    prefix = f"{key_label}: " if key_label else ""
    if not isinstance(obj, dict):
        raise ValueError(f"{prefix}必须是 JSON 对象")

    for field in ("audience", "pain_points", "product_feedback"):
        if field not in obj:
            raise ValueError(f"{prefix}缺少必填字段 {field}")

    audience = obj.get("audience")
    if not isinstance(audience, str) or not audience.strip():
        raise ValueError(f"{prefix}audience 必须为非空字符串")

    pain_points = obj.get("pain_points")
    if not isinstance(pain_points, list):
        raise ValueError(f"{prefix}pain_points 必须为数组")
    for i, p in enumerate(pain_points):
        if not isinstance(p, str) or not str(p).strip():
            raise ValueError(f"{prefix}pain_points[{i}] 必须为非空字符串")

    pf = obj.get("product_feedback")
    if not isinstance(pf, list):
        raise ValueError(f"{prefix}product_feedback 必须为数组")
    for i, item in enumerate(pf):
        if not isinstance(item, dict):
            raise ValueError(f"{prefix}product_feedback[{i}] 必须是对象")
        for sub in ("aspect", "opinion", "sentiment", "category"):
            if not str(item.get(sub, "")).strip():
                raise ValueError(f"{prefix}product_feedback[{i}] 缺少或空的 {sub}")
        sent_key = str(item.get("sentiment", "")).strip().lower()
        if sent_key not in _SENTIMENT_CANONICAL:
            raise ValueError(
                f"{prefix}product_feedback[{i}].sentiment 必须为 "
                "Positive、Negative 或 Neutral"
            )
        raw_cat = str(item.get("category", "")).strip()
        cat = _normalize_category(raw_cat)
        if cat is None:
            raise ValueError(
                f"{prefix}product_feedback[{i}].category 非法: {raw_cat!r}，"
                f"允许: {', '.join(sorted(_get_valid_categories()))}"
            )

    return _normalize_extraction(obj)


def _normalize_extraction(obj: Any) -> Dict[str, Any]:
    """补齐缺省字段，避免模型漏写 pain_points / product_feedback 导致整条丢弃。"""
    if not isinstance(obj, dict):
        raise ValueError("Each extraction must be a JSON object")

    audience = obj.get("audience", "unknown")
    if not isinstance(audience, str) or not str(audience).strip():
        audience = "unknown"
    else:
        audience = str(audience).strip()

    pain_points = obj.get("pain_points", [])
    if pain_points is None:
        pain_points = []
    if not isinstance(pain_points, list):
        pain_points = [str(pain_points)] if str(pain_points).strip() else []
    pain_points = [str(p).strip() for p in pain_points if str(p).strip()]

    product_feedback = _normalize_product_feedback(obj.get("product_feedback", []))

    missing = [
        f
        for f in ("audience", "pain_points", "product_feedback")
        if f not in obj
    ]
    if missing:
        logger.info("Normalized missing fields %s in extraction", missing)

    return {
        "audience": audience,
        "pain_points": pain_points,
        "product_feedback": product_feedback,
    }


def _build_batch_correction_message(keys: List[str], errors: List[str]) -> str:
    keys_literal = ", ".join(json.dumps(k) for k in keys)
    err_block = "\n".join(f"- {e}" for e in errors)
    return (
        "你上一次返回的 JSON 未通过服务端校验，请根据下列错误修正后重新输出。\n\n"
        f"校验错误：\n{err_block}\n\n"
        "要求：\n"
        f"- 输出一个 JSON 对象，顶层键必须且仅能是：{keys_literal}\n"
        "- 每个键的值必须包含 audience、pain_points、product_feedback\n"
        "- product_feedback 每条须含 aspect、opinion、sentiment"
        "（Positive/Negative/Neutral）、category（禁止 Value，性价比用 Price）\n"
        "- 只输出一个 JSON 对象（不要用数组），不要 markdown 代码围栏或解释文字"
    )


def _parse_batch_with_model_correction(
    system: str,
    user: str,
    sub_keys: List[str],
    *,
    batch_index: int,
    attempt_label: str,
) -> Tuple[Dict[str, Any], Dict[str, Dict[str, Any]], List[str]]:
    """
    调用模型并在校验失败时将错误信息附在对话中重试。
    返回 (原始 JSON, 已通过校验的条目, 仍失败的错误列表)。
    """
    messages: List[Dict[str, str]] = [
        {"role": "system", "content": system},
        {"role": "user", "content": user},
    ]
    raw_obj: Dict[str, Any] = {}
    ok: Dict[str, Dict[str, Any]] = {}
    errors: List[str] = []

    for correction in range(BATCH_CORRECTION_MAX_ATTEMPTS):
        logger.info(
            "请求模型 batch %s（%s），校验轮次 %s/%s…",
            batch_index,
            attempt_label,
            correction + 1,
            BATCH_CORRECTION_MAX_ATTEMPTS,
        )
        try:
            raw_parsed = _call_dashscope_chat(
                messages, max_tokens=_batch_max_output_tokens(len(sub_keys))
            )
            raw_obj = _normalize_batch_response(raw_parsed, sub_keys)
        except (RuntimeError, json.JSONDecodeError, ValueError, TypeError) as e:
            errors = [f"模型响应无法解析为 JSON: {e}"]
            ok = {}
            if correction + 1 >= BATCH_CORRECTION_MAX_ATTEMPTS:
                break
            logger.warning(
                "Batch %s parse error (correction %s): %s",
                batch_index,
                correction + 1,
                e,
            )
            messages.append(
                {
                    "role": "assistant",
                    "content": json.dumps(
                        {"error": "invalid_json", "detail": str(e)},
                        ensure_ascii=False,
                    ),
                }
            )
            messages.append(
                {
                    "role": "user",
                    "content": _build_batch_correction_message(sub_keys, errors),
                }
            )
            continue

        ok = {}
        errors = []
        for sub_ck in sub_keys:
            if sub_ck not in raw_obj:
                errors.append(f"键 {sub_ck}：响应 JSON 缺少该顶层键")
                continue
            try:
                ok[sub_ck] = _validate_extraction_strict(raw_obj[sub_ck], sub_ck)
            except ValueError as e:
                errors.append(str(e))

        if not errors:
            return raw_obj, ok, []

        if correction + 1 >= BATCH_CORRECTION_MAX_ATTEMPTS:
            break

        logger.warning(
            "Batch %s validation failed (correction %s/%s): %s",
            batch_index,
            correction + 1,
            BATCH_CORRECTION_MAX_ATTEMPTS,
            errors,
        )
        messages.append(
            {"role": "assistant", "content": json.dumps(raw_obj, ensure_ascii=False)}
        )
        messages.append(
            {
                "role": "user",
                "content": _build_batch_correction_message(sub_keys, errors),
            }
        )

    return raw_obj, ok, errors


def _fetch_batch_extractions(
    industry: str,
    product_name: str,
    batch: List[Tuple[int, str]],
    b_idx: int,
) -> Dict[str, Any]:
    """请求一批评论的结构化结果；对缺失/无效键自动缩小批次重试。"""
    _, all_keys, key_to_row = _format_tagged_batch(batch)
    row_to_text = {row: text for row, text in batch}
    pending_keys: List[str] = list(all_keys)
    merged: Dict[str, Any] = {}

    for attempt in range(BATCH_FETCH_MAX_ATTEMPTS):
        if not pending_keys:
            break

        sub_batch = [
            (key_to_row[ck], row_to_text[key_to_row[ck]]) for ck in pending_keys
        ]
        tagged, sub_keys, sub_key_to_row = _format_tagged_batch(sub_batch)
        # 重试时子批次会重新编号为 C1..Cn，需映射回原始键名
        sub_to_orig = {sub_keys[i]: pending_keys[i] for i in range(len(sub_keys))}

        system, user = _PROMPTS.build_batch_review_analysis_prompts(
            industry=industry,
            product_name=product_name,
            keys=sub_keys,
            tagged_input=tagged,
        )
        logger.info(
            "请求模型 batch %s，第 %s 次尝试，本批 %s 条…",
            b_idx,
            attempt + 1,
            len(sub_keys),
        )
        _, ok_map, val_errors = _parse_batch_with_model_correction(
            system,
            user,
            sub_keys,
            batch_index=b_idx,
            attempt_label=f"attempt {attempt + 1}",
        )

        next_pending: List[str] = []
        for sub_ck in sub_keys:
            orig_ck = sub_to_orig[sub_ck]
            if sub_ck in ok_map:
                merged[orig_ck] = ok_map[sub_ck]
            else:
                next_pending.append(orig_ck)

        if val_errors:
            logger.warning(
                "Batch %s still has validation issues after correction (attempt %s): %s",
                b_idx,
                attempt + 1,
                val_errors,
            )

        pending_keys = next_pending
        if pending_keys and attempt + 1 < BATCH_FETCH_MAX_ATTEMPTS:
            logger.info(
                "Retrying %s review(s) in batch %s (attempt %s)",
                len(pending_keys),
                b_idx,
                attempt + 2,
            )

    if pending_keys:
        logger.warning(
            "Batch %s: gave up on %s review(s) after %s attempts: %s",
            b_idx,
            len(pending_keys),
            BATCH_FETCH_MAX_ATTEMPTS,
            pending_keys,
        )
    return merged


def run_analysis(
    industry: str,
    product_name: str,
    file_path: str,
    batch_size: int | None = None,
    *,
    clean_databases: bool = True,
    max_batch_input_tokens: int = DEFAULT_MAX_BATCH_INPUT_TOKENS,
    max_batch_output_tokens: int = DEFAULT_MAX_BATCH_OUTPUT_TOKENS,
    max_reviews_per_batch: int = BATCH_COUNT_MAX,
) -> Dict[str, Any]:
    src = str(Path(file_path).expanduser().resolve())
    reviews = load_reviews_from_file(src)
    if not reviews:
        raise ValueError("No non-empty reviews found in file")

    if batch_size is not None and batch_size > 0:
        if batch_size < BATCH_COUNT_MIN:
            raise ValueError(f"batch_size must be >= {BATCH_COUNT_MIN}, got {batch_size}")
        batches = _chunked(reviews, batch_size)
        batching_mode = "fixed"
        batch_size_record = batch_size
    else:
        batches = chunk_reviews_by_token_budget(
            reviews,
            industry,
            product_name,
            max_input_tokens=max_batch_input_tokens,
            max_output_tokens=max_batch_output_tokens,
            max_reviews_per_batch=max_reviews_per_batch,
        )
        batching_mode = "dynamic"
        batch_size_record = max_reviews_per_batch

    batch_plan = _summarize_batch_plan(batches, industry, product_name)
    total_batches = len(batches)
    counts = [p["review_count"] for p in batch_plan]
    logger.info(
        "开始结构化：共 %s 条评论，模式=%s，共 %s 批，每批条数 min/med/max=%s/%s/%s",
        len(reviews),
        batching_mode,
        total_batches,
        min(counts) if counts else 0,
        sorted(counts)[len(counts) // 2] if counts else 0,
        max(counts) if counts else 0,
    )
    for p in batch_plan:
        logger.info(
            "  批次 %s: %s 条, %s 字, 约 %s in / %s out tokens",
            p["batch_index"] + 1,
            p["review_count"],
            p["total_chars"],
            p["est_input_tokens"],
            p["est_output_tokens"],
        )
    extractions_by_row: Dict[str, Dict[str, Any]] = {}
    batch_details: List[Dict[str, Any]] = []
    sqlite_rows: List[Tuple[int, int, str, Dict[str, Any]]] = []

    for b_idx, batch in enumerate(batches):
        logger.info("批次 %s/%s：处理 %s 条评论…", b_idx + 1, total_batches, len(batch))
        _, keys, key_to_row = _format_tagged_batch(batch)
        raw_obj = _fetch_batch_extractions(industry, product_name, batch, b_idx)
        logger.info(
            "批次 %s/%s 完成：成功 %s/%s 条",
            b_idx + 1,
            total_batches,
            len(raw_obj),
            len(batch),
        )

        for ck in keys:
            if ck not in raw_obj:
                continue
            validated = raw_obj[ck]
            src_row = key_to_row[ck]
            extractions_by_row[str(src_row)] = validated
            sqlite_rows.append((src_row, b_idx, ck, validated))

        batch_details.append(
            {
                "batch_index": b_idx,
                "keys": keys,
                "source_rows": [key_to_row[k] for k in keys],
                "model_output": raw_obj,
            }
        )

    created = datetime.now(timezone.utc).isoformat()
    db_path = DB_PATH.resolve()
    if clean_databases:
        _clean_voc_sqlite_databases(db_path)
    conn = sqlite3.connect(db_path)
    try:
        init_sqlite_schema(conn)
        cur = conn.execute(
            """
            INSERT INTO analysis_jobs
            (created_at, industry, product_name, source_file, batch_size, model, full_result_json)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            (
                created,
                industry,
                product_name,
                src,
                batch_size_record,
                MODEL_NAME,
                "{}",
            ),
        )
        job_id = int(cur.lastrowid)

        for src_row, b_idx, ck, payload in sqlite_rows:
            conn.execute(
                """
                INSERT INTO comment_extractions
                (job_id, source_row, batch_index, comment_key, extraction_json)
                VALUES (?, ?, ?, ?, ?)
                """,
                (
                    job_id,
                    src_row,
                    b_idx,
                    ck,
                    json.dumps(payload, ensure_ascii=False),
                ),
            )

        result: Dict[str, Any] = {
            "job_id": job_id,
            "sqlite_path": str(db_path),
            "industry": industry,
            "product_name": product_name,
            "source_file": src,
            "batching_mode": batching_mode,
            "batch_size": batch_size_record,
            "batch_plan": batch_plan,
            "max_batch_input_tokens": max_batch_input_tokens,
            "max_batch_output_tokens": max_batch_output_tokens,
            "model": MODEL_NAME,
            "extractions": extractions_by_row,
            "batches": batch_details,
        }
        full_json = json.dumps(result, ensure_ascii=False)
        conn.execute(
            "UPDATE analysis_jobs SET full_result_json = ? WHERE id = ?",
            (full_json, job_id),
        )
        conn.commit()
    finally:
        conn.close()

    return result


def _run_smoke_test() -> None:
    """
    真实调用 DashScope 的冒烟：5 条英文评论、batch_size=5、一次模型请求。
    使用临时目录下的 SQLite，不写项目根 voc_structured.sqlite。
    """
    import tempfile

    global DB_PATH
    td = Path(tempfile.mkdtemp(prefix="voc_smoke_"))
    csv_p = td / "reviews.csv"
    csv_p.write_text(
        "review\n"
        "Soft ramp; zipper broke after one week.\n"
        "Shipping box was crushed but product fine.\n"
        "My senior dog uses it daily; worth the price.\n"
        "Smells a bit chemical at first, smell fades.\n"
        "Great for cats too, very stable.\n",
        encoding="utf-8",
    )
    saved_db = DB_PATH
    DB_PATH = td / "smoke.sqlite"
    try:
        out = run_analysis(
            industry="Pet supplies",
            product_name="Pet bed ramp",
            file_path=str(csv_p),
        )
        slim = {
            "ok": True,
            "job_id": out.get("job_id"),
            "model": out.get("model"),
            "sqlite_path": out.get("sqlite_path"),
            "extraction_row_keys": sorted(out.get("extractions", {}).keys()),
            "extractions": out.get("extractions"),
        }
        print(json.dumps(slim, ensure_ascii=False, indent=2))
    finally:
        DB_PATH = saved_db


def main() -> None:
    parser = argparse.ArgumentParser(
        description=f"VOC 评论结构化（{MODEL_NAME} + SQLite）"
    )
    parser.add_argument("--industry", required=True, help="行业，如 Pet supplements")
    parser.add_argument("--product", required=True, help="产品名")
    parser.add_argument("--file", required=True, type=Path, help="评论 CSV/TXT 路径")
    parser.add_argument(
        "--batch-size",
        type=int,
        default=None,
        help="固定每批条数；不指定则按评论长度与 token 预算动态分批",
    )
    parser.add_argument(
        "--max-batch-input-tokens",
        type=int,
        default=DEFAULT_MAX_BATCH_INPUT_TOKENS,
        help=f"单批最大输入 token 估算上限（模型上限约 {MODEL_MAX_INPUT_TOKENS}）",
    )
    parser.add_argument(
        "--max-batch-output-tokens",
        type=int,
        default=DEFAULT_MAX_BATCH_OUTPUT_TOKENS,
        help=f"单批最大输出 token 估算上限（模型上限约 {MODEL_MAX_OUTPUT_TOKENS}）",
    )
    parser.add_argument(
        "--max-reviews-per-batch",
        type=int,
        default=BATCH_COUNT_MAX,
        help=f"动态模式下每批最多条数（短评可接近此值，默认 {BATCH_COUNT_MAX}）",
    )
    parser.add_argument(
        "-o",
        "--output-json",
        type=Path,
        default=None,
        help="可选：将完整结果 JSON 写入该文件",
    )
    args = parser.parse_args()

    result = run_analysis(
        industry=args.industry,
        product_name=args.product,
        file_path=str(args.file),
        batch_size=args.batch_size,
        max_batch_input_tokens=args.max_batch_input_tokens,
        max_batch_output_tokens=args.max_batch_output_tokens,
        max_reviews_per_batch=args.max_reviews_per_batch,
    )
    text = json.dumps(result, ensure_ascii=False, indent=2)
    if args.output_json:
        args.output_json.write_text(text, encoding="utf-8")
        logger.info("结果已写入 %s", args.output_json)
    print(text)


if __name__ == "__main__":
    if len(sys.argv) >= 2 and sys.argv[1] == "--smoke":
        _run_smoke_test()
    else:
        main()