amz_review_analyse/结构化_server.py

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

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

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

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

  export DEEPSEEK_API_KEY="sk-xxx"
  # 或项目根单行文件 .deepseek_key
  # 或 export DEEPSEEK_API_KEY_FILE="/path/to/key.txt"

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

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

  ./310py/bin/python 结构化_server.py --smoke

模型：默认 ``deepseek-v4-pro``（思考关闭；DeepSeek OpenAI 兼容 Chat API；可通过 ``DEEPSEEK_MODEL`` 覆盖）。

数据库：项目根目录 ``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 concurrent.futures import ThreadPoolExecutor, as_completed
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
from voc_llm import CHAT_MODEL, chat_extra_body, create_chat_client, require_chat_api_key

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"
MODEL_NAME = CHAT_MODEL
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_BATCH_INPUT_TOKENS 与 DEFAULT_MAX_BATCH_REVIEWS 约束
MODEL_MAX_INPUT_TOKENS = 1000_800
MODEL_MAX_OUTPUT_TOKENS = 200_530
CHARS_PER_TOKEN_EST = 3.2
DEFAULT_MAX_BATCH_INPUT_TOKENS = 200_000
# 仅用于 batch_plan 日志中的输出 token 粗估，不参与分批与 API max_tokens
DEFAULT_OUTPUT_TOKENS_PER_REVIEW = 450
BATCH_COUNT_MIN = 1
# 动态分批时单批评论条数上限（避免单请求过大导致输出截断）
DEFAULT_MAX_BATCH_REVIEWS = 100
# Chat 批间并行；与 embedding 共用账号时不宜过高，避免连带 429
STRUCT_DEFAULT_WORKERS = 20


def _resolve_max_batch_reviews(explicit: int | None = None) -> int:
    if explicit is not None and explicit > 0:
        return explicit
    env = os.environ.get("VOC_STRUCT_BATCH_MAX_REVIEWS", "").strip()
    if env.isdigit() and int(env) > 0:
        return int(env)
    return DEFAULT_MAX_BATCH_REVIEWS


def _resolve_struct_workers(explicit: int | None = None) -> int:
    if explicit is not None and explicit > 0:
        return explicit
    env = os.environ.get("VOC_STRUCT_WORKERS", "").strip()
    if env.isdigit() and int(env) > 0:
        return int(env)
    return STRUCT_DEFAULT_WORKERS


def _api_max_output_tokens() -> int:
    """单次 Chat 请求 max_tokens：使用模型上限，不按条数折算。"""
    return MODEL_MAX_OUTPUT_TOKENS


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_reviews_per_batch: int = DEFAULT_MAX_BATCH_REVIEWS,
) -> List[List[Tuple[int, str]]]:
    """
    按预估输入 token 将评论动态打包。
    单批条数不超过 max_reviews_per_batch；输入 token 超过 max_input_tokens 时拆批。
    单条过长则单独成批。
    """
    max_input_tokens = min(max_input_tokens, MODEL_MAX_INPUT_TOKENS)
    max_reviews_per_batch = max(BATCH_COUNT_MIN, 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]]
            inp = _estimate_batch_input_tokens(industry, product_name, candidate)

            if batch and inp > max_input_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 len(batch) >= max_reviews_per_batch:
                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:
    require_chat_api_key()
    client = create_chat_client()
    extra_body = chat_extra_body(MODEL_NAME)

    out_tokens = (
        _api_max_output_tokens()
        if max_tokens is None
        else min(MODEL_MAX_OUTPUT_TOKENS, max_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",
}

# 模型常见误写/自创词 -> 合法 sentiment 键（校验前自动映射，减少无效重试）
_SENTIMENT_ALIASES: Dict[str, str] = {
    "mixed": "neutral",
    "ambivalent": "neutral",
    "ambiguous": "neutral",
    "both": "neutral",
    "balanced": "neutral",
    "unclear": "neutral",
    "unsure": "neutral",
    "unknown": "neutral",
    "pos": "positive",
    "positve": "positive",
    "positiv": "positive",
    "neg": "negative",
    "neu": "neutral",
    "somewhat positive": "positive",
    "slightly positive": "positive",
    "mildly positive": "positive",
    "somewhat negative": "negative",
    "slightly negative": "negative",
    "mildly negative": "negative",
}

# 模型常见误写 -> 8 个标准 category（校验前自动映射，减少无效重试）
_CATEGORY_ALIASES: Dict[str, str] = {
    # Price（价格）
    "value": "Price",
    "values": "Price",
    "cost": "Price",
    "pricing": "Price",
    "value for money": "Price",
    "worth": "Price",
    # Logistics（物流）
    "packaging": "Logistics",
    "shipping": "Logistics",
    "delivery": "Logistics",
    # Customer Service（客服）
    "service": "Customer Service",
    "customer support": "Customer Service",
    "support": "Customer Service",
    # Appearance（外观）
    "design": "Appearance",
    "look": "Appearance",
    # Function（功效/功能）
    "efficacy": "Function",
    "performance": "Function",
    "effectiveness": "Function",
    "effect": "Function",
    # Ingredient（原料）
    "ingredients": "Ingredient",
    "material": "Ingredient",
    "materials": "Ingredient",
    "formula": "Ingredient",
    # Trust（信任）
    "reliability": "Trust",
    "reliable": "Trust",
    "brand": "Trust",
    "authenticity": "Trust",
    "genuine": "Trust",
    "trustworthy": "Trust",
}


def _canonical_sentiment_key(raw: Any) -> str | None:
    key = str(raw or "").strip().lower()
    if not key:
        return None
    if key in _SENTIMENT_CANONICAL:
        return key
    return _SENTIMENT_ALIASES.get(key)


def _normalize_sentiment(raw: Any) -> str:
    key = _canonical_sentiment_key(raw)
    if key is None:
        return "Neutral"
    return _SENTIMENT_CANONICAL[key]


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
    # 不属于 8 个标准类别：允许模型自行补充新类别，规范为首字母大写后保留
    return key.title()


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}")
        if _canonical_sentiment_key(item.get("sentiment")) is None:
            raise ValueError(
                f"{prefix}product_feedback[{i}].sentiment 必须为 "
                "Positive、Negative 或 Neutral（禁止 Mixed 等自创词）"
            )
        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，禁止 Mixed/Ambiguous；褒贬交织选主倾向或拆条）、"
        "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=_api_max_output_tokens()
            )
            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 _apply_batch_extraction_result(
    *,
    b_idx: int,
    batch: List[Tuple[int, str]],
    raw_obj: Dict[str, Any],
    extractions_by_row: Dict[str, Dict[str, Any]],
    sqlite_rows: List[Tuple[int, int, str, Dict[str, Any]]],
    batch_details: List[Dict[str, Any]],
    total_batches: int,
) -> None:
    _, keys, key_to_row = _format_tagged_batch(batch)
    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,
        }
    )


def _run_struct_batches_parallel(
    industry: str,
    product_name: str,
    batches: List[List[Tuple[int, str]]],
    workers: int,
    *,
    extractions_by_row: Dict[str, Dict[str, Any]],
    sqlite_rows: List[Tuple[int, int, str, Dict[str, Any]]],
    batch_details: List[Dict[str, Any]],
) -> None:
    total_batches = len(batches)
    workers = max(1, min(workers, total_batches))
    logger.info(
        "结构化批间并行：%s 批，workers=%s（模型 %s）",
        total_batches,
        workers,
        MODEL_NAME,
    )

    if workers == 1:
        for b_idx, batch in enumerate(batches):
            logger.info("批次 %s/%s：处理 %s 条评论…", b_idx + 1, total_batches, len(batch))
            raw_obj = _fetch_batch_extractions(industry, product_name, batch, b_idx)
            _apply_batch_extraction_result(
                b_idx=b_idx,
                batch=batch,
                raw_obj=raw_obj,
                extractions_by_row=extractions_by_row,
                sqlite_rows=sqlite_rows,
                batch_details=batch_details,
                total_batches=total_batches,
            )
        return

    done = 0
    with ThreadPoolExecutor(max_workers=workers) as pool:
        future_map = {
            pool.submit(
                _fetch_batch_extractions, industry, product_name, batch, b_idx
            ): (b_idx, batch)
            for b_idx, batch in enumerate(batches)
        }
        for fut in as_completed(future_map):
            b_idx, batch = future_map[fut]
            raw_obj = fut.result()
            done += 1
            if done == total_batches or done % 5 == 0:
                logger.info("结构化进度 %s/%s 批", done, total_batches)
            _apply_batch_extraction_result(
                b_idx=b_idx,
                batch=batch,
                raw_obj=raw_obj,
                extractions_by_row=extractions_by_row,
                sqlite_rows=sqlite_rows,
                batch_details=batch_details,
                total_batches=total_batches,
            )


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_reviews: int | None = None,
    workers: int | None = None,
) -> 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:
        cap_reviews = _resolve_max_batch_reviews(max_batch_reviews)
        batches = chunk_reviews_by_token_budget(
            reviews,
            industry,
            product_name,
            max_input_tokens=max_batch_input_tokens,
            max_reviews_per_batch=cap_reviews,
        )
        batching_mode = "dynamic"
        batch_size_record = 0

    batch_plan = _summarize_batch_plan(batches, industry, product_name)
    total_batches = len(batches)
    counts = [p["review_count"] for p in batch_plan]
    if batching_mode == "dynamic" and counts:
        batch_size_record = max(counts)
    cap_note = ""
    if batching_mode == "dynamic":
        cap_note = f"，单批≤{_resolve_max_batch_reviews(max_batch_reviews)}条"
    logger.info(
        "开始结构化：共 %s 条评论，模式=%s%s，共 %s 批，每批条数 min/med/max=%s/%s/%s",
        len(reviews),
        batching_mode,
        cap_note,
        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]]] = []
    struct_workers = _resolve_struct_workers(workers)

    _run_struct_batches_parallel(
        industry,
        product_name,
        batches,
        struct_workers,
        extractions_by_row=extractions_by_row,
        sqlite_rows=sqlite_rows,
        batch_details=batch_details,
    )
    batch_details.sort(key=lambda x: x["batch_index"])

    if not sqlite_rows:
        require_chat_api_key()  # 无结果时若缺 key 则给出明确错误
        raise RuntimeError(
            f"结构化 0/{len(reviews)} 条成功。"
            "请检查 DEEPSEEK_API_KEY / .deepseek_key 与网络；"
            "查看上方 WARNING 中的具体原因。"
        )

    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,
            "api_max_output_tokens": MODEL_MAX_OUTPUT_TOKENS,
            "struct_workers": struct_workers,
            "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 估算上限（默认 {DEFAULT_MAX_BATCH_INPUT_TOKENS}；"
            f"模型上限约 {MODEL_MAX_INPUT_TOKENS}）；"
            f"单批最多 {DEFAULT_MAX_BATCH_REVIEWS} 条（环境变量 VOC_STRUCT_BATCH_MAX_REVIEWS）。"
        ),
    )
    parser.add_argument(
        "--max-batch-reviews",
        type=int,
        default=None,
        help=(
            f"动态分批单批最多评论条数（默认 {DEFAULT_MAX_BATCH_REVIEWS}；"
            "环境变量 VOC_STRUCT_BATCH_MAX_REVIEWS）"
        ),
    )
    parser.add_argument(
        "--workers",
        type=int,
        default=None,
        help=(
            f"批间并行请求数（默认 {STRUCT_DEFAULT_WORKERS}；"
            "1=串行；可用环境变量 VOC_STRUCT_WORKERS）"
        ),
    )
    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_reviews=args.max_batch_reviews,
        workers=args.workers,
    )
    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()