#!/usr/bin/env python3
import argparse
import ast
import datetime as dt
import json
import re
import subprocess
import sys
from collections import defaultdict
from typing import Dict, List, Tuple

BASE_BIN = "/usr/local/bin/customer-service-analysis.base"

GREETING_KEYWORDS = ("欢迎", "您好", "亲,", "亲，", "亲~")
EMPATHY_KEYWORDS = ("抱歉", "不好意思", "理解您", "理解你的", "辛苦了")
COMMITMENT_KEYWORDS = ("马上", "尽快", "给您处理", "为您处理", "帮您处理", "我来帮您")
CLOSING_KEYWORDS = ("感谢", "祝您", "辛苦", "生活愉快", "欢迎下次")
RISK_KEYWORDS = ("不知道", "没办法", "无法处理", "不清楚", "等不了", "太慢")

AFTERSALE_ROLE_KEYWORDS = (
    "退货",
    "退款",
    "售后",
    "安装",
    "工单",
    "物流",
    "补发",
    "换货",
    "维修",
    "催退款",
    "催取件",
    "拒收",
    "赔付",
)


def _parse_date(value: str) -> dt.date:
    try:
        return dt.datetime.strptime(value, "%Y-%m-%d").date()
    except ValueError as exc:
        raise argparse.ArgumentTypeError(f"invalid date '{value}', expected YYYY-MM-DD") from exc


def _sql_escape(value: str) -> str:
    return value.replace("'", "''")


def _fmt(v):
    if v is None:
        return ""
    if isinstance(v, float):
        return round(v, 2)
    return v


def _safe_pct(num: float, den: float):
    if den <= 0:
        return None
    return round((num / den) * 100.0, 2)


def _score_grade(score: float) -> str:
    if score >= 90:
        return "S"
    if score >= 80:
        return "A"
    if score >= 70:
        return "B"
    if score >= 60:
        return "C"
    return "D"


def _contains_any(text: str, keywords: Tuple[str, ...]) -> bool:
    return any(k in text for k in keywords)


def _split_result_sets(output_text: str) -> Tuple[Dict[int, List[str]], int]:
    blocks: Dict[int, List[str]] = {}
    current_idx = None
    current_lines: List[str] = []
    total_sets = 0

    for line in (output_text or "").splitlines():
        if line.startswith("TOTAL_RESULT_SETS:"):
            try:
                total_sets = int(line.split(":", 1)[1].strip())
            except Exception:
                total_sets = 0
            continue
        if line.startswith("RESULT_SET "):
            if current_idx is not None:
                blocks[current_idx] = current_lines
            try:
                current_idx = int(line.split()[1])
            except Exception:
                current_idx = None
            current_lines = [line] if current_idx is not None else []
            continue
        if current_idx is not None:
            current_lines.append(line)

    if current_idx is not None:
        blocks[current_idx] = current_lines

    return blocks, total_sets


def _extract_json_rows(output: str) -> list:
    rows = []
    for line in output.splitlines():
        line = line.strip()
        if not line.startswith("{") or not line.endswith("}"):
            continue
        try:
            rows.append(json.loads(line))
        except json.JSONDecodeError:
            continue
    return rows


def _run_company_query(sql_text: str, limit: int = 100000) -> List[Dict]:
    proc = subprocess.run(
        ["company-db", "query", "--limit", str(limit), "--sql", sql_text],
        capture_output=True,
        text=True,
    )
    if proc.returncode != 0:
        err = (proc.stderr or proc.stdout or "company-db failed").strip()
        raise RuntimeError(err)
    return _extract_json_rows(proc.stdout or "")


def _print_result_set(index: int, columns: List[str], rows: List[Dict]) -> None:
    print(f"RESULT_SET {index}")
    print("COLUMNS: " + ", ".join(columns))
    print(f"ROWS: {len(rows)}")
    for row in rows:
        obj = {c: row.get(c) for c in columns}
        print(json.dumps(obj, ensure_ascii=False))


def _clean_user_label(label: str) -> str:
    return (label or "").replace("\ue7f6", "").strip()


def _extract_agent_name(user_label: str) -> str | None:
    label = _clean_user_label(user_label)
    if "-->" in label:
        return None
    if ":" not in label:
        return None

    prefix, name = label.rsplit(":", 1)
    prefix = prefix.strip()
    name = name.strip()
    if not prefix or not name:
        return None
    if len(name) > 40:
        return None
    if "店" not in prefix and "客服" not in prefix:
        return None
    return name


def _parse_dialog_entries(content_text: str) -> list[tuple[str, str]]:
    text = (content_text or "").strip()
    if not text or text == "[]":
        return []

    entries: list[tuple[str, str]] = []
    try:
        data = ast.literal_eval(text)
        if isinstance(data, list):
            for item in data:
                if isinstance(item, dict):
                    entries.append((str(item.get("user", "")), str(item.get("message", ""))))
            if entries:
                return entries
    except Exception:
        pass

    pattern = re.compile(r"'user':\s*'([^']*)'.*?'message':\s*'([^']*)'", re.S)
    for user, message in pattern.findall(text):
        entries.append((user, message))
    return entries


def _classify_role(main_reason: str, second_reason: str, third_reason: str) -> str:
    text = f"{main_reason or ''}|{second_reason or ''}|{third_reason or ''}"
    if _contains_any(text, AFTERSALE_ROLE_KEYWORDS):
        return "售后"
    return "售前"


def _timeliness_score(avg_submit_to_kf_hours):
    if avg_submit_to_kf_hours is None:
        return 60.0
    h = float(avg_submit_to_kf_hours)
    if h <= 4:
        return 100.0
    if h <= 12:
        return 90.0
    if h <= 24:
        return 80.0
    if h <= 48:
        return 70.0
    if h <= 72:
        return 60.0
    return 40.0


def _to_float(value):
    if value is None:
        return None
    if isinstance(value, (int, float)):
        return float(value)
    text = str(value).strip().replace(',', '')
    if text == '' or text.lower() in ('none', 'null', 'nan'):
        return None
    if text.endswith('%'):
        text = text[:-1]
    try:
        return float(text)
    except Exception:
        return None


def _to_int(value):
    v = _to_float(value)
    if v is None:
        return 0
    return int(round(v))


def _to_pct(value):
    v = _to_float(value)
    if v is None:
        return None
    if v <= 1.2:
        v = v * 100.0
    return max(0.0, min(100.0, v))


def _normalize_shop_name(name: str) -> str:
    t = (name or '').strip().lower()
    t = t.replace(' ', '').replace('　', '')
    for p in ('天猫-', '淘宝-', '京东-', '京喜-', '拼多多-', '抖音-', '阿里巴巴-', '唯品会-', '得物-', '快手-'):
        if t.startswith(p):
            t = t[len(p):]
    t = t.replace('-', '').replace('_', '')
    return t


def _normalize_agent_name(name: str) -> str:
    t = (name or '').strip().lower()
    t = t.replace(' ', '').replace('　', '')
    t = re.sub(r'^\d+号', '', t)
    t = re.sub(r'(师傅|客服|专员|老师)$', '', t)
    return t


def _normalize_role_text(value: str) -> str:
    t = (value or '').strip()
    if '售后' in t:
        return '售后'
    if '售前' in t:
        return '售前'
    return ''


def _expand_agent_aliases(name: str) -> List[str]:
    raw = (name or '').strip()
    if not raw:
        return []

    raw_candidates = {raw}
    for sep in (':', '：'):
        if sep in raw:
            parts = [x.strip() for x in raw.split(sep) if x and x.strip()]
            if parts:
                raw_candidates.add(parts[-1])

    norm_set = set()
    for item in raw_candidates:
        cleaned = re.sub(r'[（(].*?[）)]', '', item).strip()
        if not cleaned:
            continue

        normalized = _normalize_agent_name(cleaned)
        if not normalized:
            continue

        norm_set.add(normalized)
        norm_set.add(re.sub(r'\d+$', '', normalized))

        no_prefix = re.sub(r'^搭配师', '', normalized)
        norm_set.add(no_prefix)
        norm_set.add(re.sub(r'\d+$', '', no_prefix))

    norm_set.discard('')
    return sorted(norm_set)


def _parse_month_key(month_text: str) -> int:
    if not month_text:
        return 0
    s = str(month_text)
    m = re.search(r'(20\d{2})-(\d{1,2})', s)
    if m:
        y = int(m.group(1))
        mm = int(m.group(2))
        if 1 <= mm <= 12:
            return y * 100 + mm
    m = re.search(r'(\d{2,4})年(\d{1,2})月', s)
    if m:
        y = int(m.group(1))
        if y < 100:
            y += 2000
        mm = int(m.group(2))
        if 1 <= mm <= 12:
            return y * 100 + mm
    return 0


def _build_external_conversion_sql(shop: str) -> str:
    shop_escaped = _sql_escape(shop.strip())
    return f"""
USE [JSDataMiddlePlatform];
SET NOCOUNT ON;
DECLARE @shop nvarchar(100) = N'{shop_escaped}';
SELECT
    [月份] AS month_text,
    [店铺名称] AS shop_name,
    [客服昵称] AS agent_name,
    [询单人数] AS inquiry_count_text,
    [当日或次日付款人数] AS paid_next_count_text,
    [最终付款人数] AS paid_final_count_text,
    [询单->次日付款转化率] AS conv_next_text,
    [询单->最终付款转化率] AS conv_final_text
FROM dbo.cs_conversion_rate WITH (NOLOCK)
WHERE (@shop = N'' OR [店铺名称] LIKE N'%' + @shop + N'%');
"""


def _build_external_satisfaction_sql(shop: str) -> str:
    shop_escaped = _sql_escape(shop.strip())
    return f"""
USE [JSDataMiddlePlatform];
SET NOCOUNT ON;
DECLARE @shop nvarchar(100) = N'{shop_escaped}';
SELECT
    [月份] AS month_text,
    [店铺名称] AS shop_name,
    [客服昵称] AS agent_name,
    [聊天人次(原咨询人次)] AS chat_count_text,
    [评价展示量] AS eval_show_text,
    [评价参与量] AS eval_join_text,
    [客户满意率] AS customer_satisfaction_text,
    [服务满意度评分] AS service_score_text
FROM dbo.cs_satisfaction WITH (NOLOCK)
WHERE (@shop = N'' OR [店铺名称] LIKE N'%' + @shop + N'%');
"""


def _build_external_roster_sql(shop: str) -> str:
    shop_escaped = _sql_escape(shop.strip())
    return f"""
USE [JSDataMiddlePlatform];
SET NOCOUNT ON;
DECLARE @shop nvarchar(100) = N'{shop_escaped}';
SELECT
    [店铺] AS shop_name,
    [岗位] AS position_role,
    [账号名] AS account_name,
    [赤兔展示名] AS display_name,
    [花名] AS nick_name,
    [姓名] AS real_name
FROM dbo.cs_roster WITH (NOLOCK)
WHERE (@shop = N'' OR [店铺] LIKE N'%' + @shop + N'%');
"""


def _roster_entry_score(entry: Dict) -> Tuple[int, int]:
    return (
        1 if entry.get('position_role') else 0,
        len(entry.get('alias_norms') or []),
    )


def _load_cs_roster(shop: str) -> Dict:
    rows = _run_company_query(_build_external_roster_sql(shop), limit=100000)
    by_shop_alias: Dict[Tuple[str, str], Dict] = {}
    by_alias: Dict[str, Dict] = {}

    for r in rows:
        shop_name = str(r.get('shop_name') or '')
        position_role = _normalize_role_text(str(r.get('position_role') or ''))
        account_name = str(r.get('account_name') or '').strip()
        display_name = str(r.get('display_name') or '').strip()
        nick_name = str(r.get('nick_name') or '').strip()
        real_name = str(r.get('real_name') or '').strip()

        alias_norms = set()
        for raw in (account_name, display_name, nick_name, real_name):
            for alias in _expand_agent_aliases(raw):
                alias_norms.add(alias)

        if not alias_norms:
            continue

        entry = {
            'shop_name': shop_name,
            'shop_norm': _normalize_shop_name(shop_name),
            'position_role': position_role,
            'account_name': account_name,
            'display_name': display_name,
            'nick_name': nick_name,
            'real_name': real_name,
            'alias_norms': sorted(alias_norms),
        }

        for alias in alias_norms:
            k = (entry['shop_norm'], alias)
            cur = by_shop_alias.get(k)
            if cur is None or _roster_entry_score(entry) > _roster_entry_score(cur):
                by_shop_alias[k] = entry

            cur_alias = by_alias.get(alias)
            if cur_alias is None or _roster_entry_score(entry) > _roster_entry_score(cur_alias):
                by_alias[alias] = entry

    return {
        'by_shop_alias': by_shop_alias,
        'by_alias': by_alias,
    }


def _pick_roster_entry(roster_index: Dict, shop_name: str, agent_name: str) -> Dict:
    if not roster_index:
        return {}

    ns = _normalize_shop_name(shop_name)
    alias_list = _expand_agent_aliases(agent_name)
    by_shop_alias = roster_index.get('by_shop_alias') or {}
    by_alias = roster_index.get('by_alias') or {}

    for alias in alias_list:
        hit = by_shop_alias.get((ns, alias))
        if hit:
            return hit

    fuzzy_candidates = []
    for alias in alias_list:
        for (shop_norm, alias_norm), hit in by_shop_alias.items():
            if alias_norm != alias:
                continue
            if shop_norm == ns or (shop_norm and ns and (shop_norm in ns or ns in shop_norm)):
                fuzzy_candidates.append(hit)

    if fuzzy_candidates:
        fuzzy_candidates.sort(key=_roster_entry_score, reverse=True)
        return fuzzy_candidates[0]

    for alias in alias_list:
        hit = by_alias.get(alias)
        if hit:
            return hit

    return {}


def _load_external_cs_metrics(shop: str) -> Dict[Tuple[str, str], Dict]:
    conv_rows = _run_company_query(_build_external_conversion_sql(shop), limit=200000)
    sat_rows = _run_company_query(_build_external_satisfaction_sql(shop), limit=200000)

    conv_best: Dict[Tuple[str, str], Dict] = {}
    sat_best: Dict[Tuple[str, str], Dict] = {}

    for r in conv_rows:
        key = (_normalize_shop_name(str(r.get('shop_name') or '')), _normalize_agent_name(str(r.get('agent_name') or '')))
        if not key[0] or not key[1]:
            continue
        mk = _parse_month_key(r.get('month_text'))
        cur = conv_best.get(key)
        inquiry_count = _to_int(r.get('inquiry_count_text'))
        if cur is None or mk > cur['month_key'] or (mk == cur['month_key'] and inquiry_count > cur.get('inquiry_count', 0)):
            conv_best[key] = {
                'month_key': mk,
                'month_text': str(r.get('month_text') or ''),
                'shop_name': str(r.get('shop_name') or ''),
                'agent_name': str(r.get('agent_name') or ''),
                'inquiry_count': inquiry_count,
                'paid_next_count': _to_int(r.get('paid_next_count_text')),
                'paid_final_count': _to_int(r.get('paid_final_count_text')),
                'conv_next_pct': _to_pct(r.get('conv_next_text')),
                'conv_final_pct': _to_pct(r.get('conv_final_text')),
            }

    for r in sat_rows:
        key = (_normalize_shop_name(str(r.get('shop_name') or '')), _normalize_agent_name(str(r.get('agent_name') or '')))
        if not key[0] or not key[1]:
            continue
        mk = _parse_month_key(r.get('month_text'))
        cur = sat_best.get(key)
        eval_join = _to_int(r.get('eval_join_text'))
        if cur is None or mk > cur['month_key'] or (mk == cur['month_key'] and eval_join > cur.get('eval_join_count', 0)):
            sat_best[key] = {
                'month_key': mk,
                'month_text': str(r.get('month_text') or ''),
                'shop_name': str(r.get('shop_name') or ''),
                'agent_name': str(r.get('agent_name') or ''),
                'chat_count': _to_int(r.get('chat_count_text')),
                'eval_show_count': _to_int(r.get('eval_show_text')),
                'eval_join_count': eval_join,
                'customer_satisfaction_pct': _to_pct(r.get('customer_satisfaction_text')),
                'service_score_5': _to_float(r.get('service_score_text')),
            }

    out: Dict[Tuple[str, str], Dict] = {}
    all_keys = set(conv_best.keys()) | set(sat_best.keys())
    for k in all_keys:
        c = conv_best.get(k, {})
        s = sat_best.get(k, {})
        out[k] = {
            'shop_norm': k[0],
            'agent_norm': k[1],
            'month_key': max(c.get('month_key', 0), s.get('month_key', 0)),
            'month_text': c.get('month_text') or s.get('month_text') or '',
            'shop_name': c.get('shop_name') or s.get('shop_name') or '',
            'agent_name': c.get('agent_name') or s.get('agent_name') or '',
            'inquiry_count': c.get('inquiry_count', 0),
            'paid_next_count': c.get('paid_next_count', 0),
            'paid_final_count': c.get('paid_final_count', 0),
            'conv_next_pct': c.get('conv_next_pct'),
            'conv_final_pct': c.get('conv_final_pct'),
            'chat_count': s.get('chat_count', 0),
            'eval_show_count': s.get('eval_show_count', 0),
            'eval_join_count': s.get('eval_join_count', 0),
            'customer_satisfaction_pct': s.get('customer_satisfaction_pct'),
            'service_score_5': s.get('service_score_5'),
        }
    return out


def _pick_external_metric(
    external: Dict[Tuple[str, str], Dict],
    shop_name: str,
    agent_name: str,
    agent_alias_norms=None,
) -> Dict:
    ns = _normalize_shop_name(shop_name)
    alias_list = []
    if agent_alias_norms:
        alias_list.extend([str(x) for x in agent_alias_norms if x])

    na = _normalize_agent_name(agent_name)
    if na:
        alias_list.append(na)

    seen = set()
    dedup_aliases = []
    for alias in alias_list:
        if alias in seen:
            continue
        seen.add(alias)
        dedup_aliases.append(alias)

    for alias in dedup_aliases:
        direct = external.get((ns, alias))
        if direct:
            return direct

    candidates = [
        v for (s, a), v in external.items()
        if a in seen and (s == ns or (s and ns and (s in ns or ns in s)))
    ]
    if not candidates:
        candidates = [v for (s, a), v in external.items() if a in seen]

    if not candidates:
        return {}

    candidates.sort(key=lambda x: (x.get('month_key', 0), x.get('inquiry_count', 0), x.get('eval_join_count', 0)), reverse=True)
    return candidates[0]


def _build_agent_raw_sql(start_date: dt.date, end_date: dt.date, shop: str, max_rows: int) -> str:
    shop_escaped = _sql_escape(shop.strip())
    return f"""
USE [JSDataMiddlePlatform];
SET NOCOUNT ON;

DECLARE @start_time datetime2 = CAST('{start_date.isoformat()}' AS datetime2);
DECLARE @end_time datetime2 = DATEADD(day, 1, CAST('{end_date.isoformat()}' AS datetime2));
DECLARE @shop nvarchar(100) = N'{shop_escaped}';

;WITH base AS (
    SELECT
        c.id,
        NULLIF(c.orderNo, '') AS order_no,
        ISNULL(NULLIF(c.shopName, ''), N'未知店铺') AS shop_name,
        ISNULL(NULLIF(LTRIM(RTRIM(c.mainReason)), ''), N'未标注') AS main_reason,
        ISNULL(NULLIF(LTRIM(RTRIM(c.secondReason)), ''), N'未标注') AS second_reason,
        ISNULL(NULLIF(LTRIM(RTRIM(c.thirdReason)), ''), N'未标注') AS third_reason,
        CAST(c.content AS nvarchar(max)) AS content_text,
        c.createdAt,
        ROW_NUMBER() OVER (ORDER BY c.createdAt DESC, c.id DESC) AS rn,
        COUNT(1) OVER () AS total_rows
    FROM dbo.pc_kfChatLog AS c WITH (NOLOCK)
    WHERE c.createdAt >= @start_time
      AND c.createdAt < @end_time
      AND (@shop = N'' OR c.shopName LIKE N'%' + @shop + N'%')
)
SELECT TOP ({max_rows})
    id,
    order_no,
    shop_name,
    main_reason,
    second_reason,
    third_reason,
    content_text,
    total_rows
FROM base
ORDER BY rn ASC;
"""


def _build_order_snapshot_sql(start_date: dt.date, end_date: dt.date, shop: str) -> str:
    shop_escaped = _sql_escape(shop.strip())
    return f"""
USE [JSDataMiddlePlatform];
SET NOCOUNT ON;

DECLARE @start_time datetime2 = CAST('{start_date.isoformat()}' AS datetime2);
DECLARE @end_time datetime2 = DATEADD(day, 1, CAST('{end_date.isoformat()}' AS datetime2));
DECLARE @shop nvarchar(100) = N'{shop_escaped}';

;WITH chat_orders AS (
    SELECT DISTINCT NULLIF(c.orderNo, '') AS order_no
    FROM dbo.pc_kfChatLog AS c WITH (NOLOCK)
    WHERE c.createdAt >= @start_time
      AND c.createdAt < @end_time
      AND NULLIF(c.orderNo, '') IS NOT NULL
      AND (@shop = N'' OR c.shopName LIKE N'%' + @shop + N'%')
)
SELECT
    co.order_no,
    CASE WHEN o.pay_time IS NOT NULL THEN 1 ELSE 0 END AS is_paid,
    CASE WHEN o.send_time IS NOT NULL THEN 1 ELSE 0 END AS is_shipped,
    CASE WHEN t.OrderNum IS NOT NULL THEN 1 ELSE 0 END AS has_ticket,
    t.StatusId AS ticket_status,
    DATEDIFF(minute, t.SubmitTime, t.kfshTime) AS submit_to_kf_minutes,
    DATEDIFF(minute, t.kfshTime, t.cwshTime) AS kf_to_cw_minutes
FROM chat_orders AS co
OUTER APPLY (
    SELECT TOP 1 x.pay_time, x.send_time
    FROM dbo.jst_order_express_details AS x WITH (NOLOCK)
    WHERE x.so_id = co.order_no
    ORDER BY x.updatedAt DESC, x.id DESC
) AS o
OUTER APPLY (
    SELECT TOP 1 i.OrderNum, i.StatusId, i.SubmitTime, i.kfshTime, i.cwshTime
    FROM dbo.fys_Ticketing_install AS i WITH (NOLOCK)
    WHERE i.OrderNum = co.order_no
    ORDER BY i.SubmitTime DESC, i.Id DESC
) AS t;
"""


def _compute_role_score_rows(
    start_date: dt.date,
    end_date: dt.date,
    shop: str,
    limit: int,
    agent_max_rows: int,
) -> List[Dict]:
    chat_rows = _run_company_query(_build_agent_raw_sql(start_date, end_date, shop, agent_max_rows), limit=agent_max_rows)
    if not chat_rows:
        return []

    total_rows = int(chat_rows[0].get("total_rows") or len(chat_rows))
    sampled_rows = len(chat_rows)
    sample_scope = f"full_{sampled_rows}/{total_rows}"
    if total_rows > sampled_rows:
        sample_scope = f"sampled_recent_{sampled_rows}/{total_rows}"
    roster_index = _load_cs_roster(shop)

    stats = defaultdict(
        lambda: {
            "session_set": set(),
            "order_set": set(),
            "primary_session_count": 0,
            "message_count": 0,
            "greeting_hits": 0,
            "empathy_hits": 0,
            "commitment_hits": 0,
            "closing_hits": 0,
            "risk_hits": 0,
            "no_match_session_count": 0,
            "missing_session_count": 0,
            "role_source_roster_count": 0,
            "role_source_reason_count": 0,
            "roster_position_role": "",
            "roster_shop_name": "",
            "roster_display_name": "",
            "alias_norms": set(),
        }
    )

    for row in chat_rows:
        dialogs = _parse_dialog_entries(row.get("content_text") or "")
        if not dialogs:
            continue

        main_reason = str(row.get("main_reason") or "")
        second_reason = str(row.get("second_reason") or "")
        third_reason = str(row.get("third_reason") or "")
        role_reason = _classify_role(main_reason, second_reason, third_reason)
        shop_name = str(row.get("shop_name") or "未知店铺")
        order_no = (row.get("order_no") or "").strip()
        session_key = order_no or f"id:{row.get('id')}"

        agent_turns = defaultdict(int)
        agent_features = defaultdict(lambda: {"g": 0, "e": 0, "c": 0, "cl": 0, "r": 0, "m": 0})

        for user_label, message in dialogs:
            agent_name = _extract_agent_name(user_label)
            if not agent_name:
                continue
            agent_turns[agent_name] += 1

            msg = (message or "").strip()
            f = agent_features[agent_name]
            f["m"] += 1
            if _contains_any(msg, GREETING_KEYWORDS):
                f["g"] += 1
            if _contains_any(msg, EMPATHY_KEYWORDS):
                f["e"] += 1
            if _contains_any(msg, COMMITMENT_KEYWORDS):
                f["c"] += 1
            if _contains_any(msg, CLOSING_KEYWORDS):
                f["cl"] += 1
            if _contains_any(msg, RISK_KEYWORDS):
                f["r"] += 1

        if not agent_turns:
            continue

        primary_agent = max(agent_turns.items(), key=lambda x: (x[1], x[0]))[0]
        roster_entry = _pick_roster_entry(roster_index, shop_name, primary_agent)
        roster_role = _normalize_role_text(str(roster_entry.get("position_role") or "")) if roster_entry else ""
        role = roster_role or role_reason
        key = (shop_name, primary_agent, role)
        m = stats[key]
        m["session_set"].add(session_key)
        if order_no:
            m["order_set"].add(order_no)
        m["primary_session_count"] += 1
        if roster_role:
            m["role_source_roster_count"] += 1
            m["roster_position_role"] = roster_role
        else:
            m["role_source_reason_count"] += 1
        if roster_entry:
            if roster_entry.get("shop_name"):
                m["roster_shop_name"] = str(roster_entry.get("shop_name"))
            if roster_entry.get("display_name"):
                m["roster_display_name"] = str(roster_entry.get("display_name"))
            for alias in roster_entry.get("alias_norms") or []:
                m["alias_norms"].add(alias)

        ff = agent_features[primary_agent]
        m["message_count"] += ff["m"]
        m["greeting_hits"] += ff["g"]
        m["empathy_hits"] += ff["e"]
        m["commitment_hits"] += ff["c"]
        m["closing_hits"] += ff["cl"]
        m["risk_hits"] += ff["r"]

        if main_reason == "无对应分类" or second_reason == "无对应分类":
            m["no_match_session_count"] += 1
        if main_reason == "未标注" or second_reason == "未标注":
            m["missing_session_count"] += 1

    order_snapshot_rows = _run_company_query(_build_order_snapshot_sql(start_date, end_date, shop), limit=200000)
    order_map: Dict[str, Dict] = {}
    for r in order_snapshot_rows:
        order_no = (r.get("order_no") or "").strip()
        if not order_no:
            continue
        order_map[order_no] = {
            "is_paid": int(r.get("is_paid") or 0),
            "is_shipped": int(r.get("is_shipped") or 0),
            "has_ticket": int(r.get("has_ticket") or 0),
            "ticket_status": r.get("ticket_status"),
            "submit_to_kf_minutes": r.get("submit_to_kf_minutes"),
        }

    external_metrics = _load_external_cs_metrics(shop)

    result_rows: List[Dict] = []

    for (shop_name, agent_name, role), m in stats.items():
        session_count = len(m["session_set"])
        message_count = m["message_count"]
        if session_count <= 0 or message_count <= 0:
            continue

        greeting_rate = m["greeting_hits"] / message_count
        empathy_rate = m["empathy_hits"] / message_count
        commitment_rate = m["commitment_hits"] / message_count
        closing_rate = m["closing_hits"] / message_count
        risk_rate = m["risk_hits"] / message_count

        no_match_rate = m["no_match_session_count"] / session_count
        missing_rate = m["missing_session_count"] / session_count

        resp_score = max(0.0, min(100.0, 100.0 * (0.70 * commitment_rate + 0.30 * (1.0 - risk_rate))))
        sat_chat_score = max(0.0, min(100.0, 100.0 * (0.40 * empathy_rate + 0.35 * closing_rate + 0.25 * greeting_rate)))
        risk_ctrl_score = max(0.0, min(100.0, 100.0 * (1.0 - (0.75 * no_match_rate + 0.25 * missing_rate))))

        order_set = m["order_set"]
        inquiry_order_count = len(order_set)
        converted_order_count = 0
        ticket_linked_order_count = 0
        ticket_resolved_order_count = 0
        submit_to_kf_hours_list: List[float] = []

        for order_no in order_set:
            snap = order_map.get(order_no)
            if not snap:
                continue
            if snap.get("is_paid") == 1 or snap.get("is_shipped") == 1:
                converted_order_count += 1
            if snap.get("has_ticket") == 1:
                ticket_linked_order_count += 1
                if str(snap.get("ticket_status")) == "2":
                    ticket_resolved_order_count += 1
                mins = snap.get("submit_to_kf_minutes")
                try:
                    mins = float(mins)
                except Exception:
                    mins = None
                if mins is not None and 0 <= mins <= 43200:
                    submit_to_kf_hours_list.append(mins / 60.0)

        inquiry_conversion_rate_pct = _safe_pct(converted_order_count, inquiry_order_count)
        ticket_resolve_rate_pct = _safe_pct(ticket_resolved_order_count, ticket_linked_order_count)
        avg_submit_to_kf_hours = (
            round(sum(submit_to_kf_hours_list) / len(submit_to_kf_hours_list), 2)
            if submit_to_kf_hours_list
            else None
        )

        alias_norms = set(m.get("alias_norms") or set())
        for alias in _expand_agent_aliases(agent_name):
            alias_norms.add(alias)
        ext = _pick_external_metric(external_metrics, shop_name, agent_name, alias_norms)
        ext_month_text = ext.get("month_text") if ext else ""
        ext_shop_name = ext.get("shop_name") if ext else ""
        ext_inquiry_count = ext.get("inquiry_count", 0) if ext else 0
        ext_paid_next_count = ext.get("paid_next_count", 0) if ext else 0
        ext_paid_final_count = ext.get("paid_final_count", 0) if ext else 0
        ext_conv_next_pct = ext.get("conv_next_pct") if ext else None
        ext_conv_final_pct = ext.get("conv_final_pct") if ext else None
        ext_chat_count = ext.get("chat_count", 0) if ext else 0
        ext_eval_show_count = ext.get("eval_show_count", 0) if ext else 0
        ext_eval_join_count = ext.get("eval_join_count", 0) if ext else 0
        ext_customer_satisfaction_pct = ext.get("customer_satisfaction_pct") if ext else None
        ext_service_score_5 = ext.get("service_score_5") if ext else None

        sat_external_score = None
        if ext_customer_satisfaction_pct is not None:
            sat_external_score = max(0.0, min(100.0, float(ext_customer_satisfaction_pct)))
        elif ext_service_score_5 is not None:
            sat_external_score = max(0.0, min(100.0, float(ext_service_score_5) * 20.0))

        if sat_external_score is None:
            sat_final_score = sat_chat_score
            satisfaction_source = "chat_lexicon"
        else:
            sat_final_score = 0.6 * sat_external_score + 0.4 * sat_chat_score
            satisfaction_source = "cs_satisfaction+chat_lexicon"

        conversion_score = inquiry_conversion_rate_pct if inquiry_conversion_rate_pct is not None else 0.0
        conversion_source = "order_snapshot"
        if ext_conv_final_pct is not None:
            conversion_score = float(ext_conv_final_pct)
            conversion_source = "cs_conversion_rate"

        resolution_score = ticket_resolve_rate_pct if ticket_resolve_rate_pct is not None else 60.0
        timeliness_score = _timeliness_score(avg_submit_to_kf_hours)
        role_source = "cs_roster" if m["role_source_roster_count"] > 0 else "reason_inference"
        roster_position_role = m["roster_position_role"] or ""
        roster_shop_name = m["roster_shop_name"] or ""
        roster_display_name = m["roster_display_name"] or ""

        if role == "售前":
            role_score = (
                0.20 * resp_score
                + 0.25 * sat_final_score
                + 0.40 * conversion_score
                + 0.15 * risk_ctrl_score
            )
            rule_note = "售前评分=响应20%+满意25%+询单转化40%+风险控制15%"
        else:
            role_score = (
                0.15 * resp_score
                + 0.20 * sat_final_score
                + 0.35 * resolution_score
                + 0.20 * timeliness_score
                + 0.10 * risk_ctrl_score
            )
            rule_note = "售后评分=响应15%+满意20%+工单闭环35%+处理时效20%+风险控制10%"

        reliable = session_count >= 10 and message_count >= 20

        result_rows.append(
            {
                "shop_name": shop_name,
                "agent_name": agent_name,
                "position_role": role,
                "role_source": role_source,
                "roster_position_role": roster_position_role,
                "roster_shop_name": roster_shop_name,
                "roster_display_name": roster_display_name,
                "session_count": session_count,
                "primary_session_count": m["primary_session_count"],
                "message_count": message_count,
                "inquiry_order_count": inquiry_order_count,
                "converted_order_count": converted_order_count,
                "inquiry_conversion_rate_pct": _fmt(inquiry_conversion_rate_pct),
                "ticket_linked_order_count": ticket_linked_order_count,
                "ticket_resolved_order_count": ticket_resolved_order_count,
                "ticket_resolve_rate_pct": _fmt(ticket_resolve_rate_pct),
                "avg_submit_to_kf_hours": _fmt(avg_submit_to_kf_hours),
                "external_month": ext_month_text,
                "external_shop_name": ext_shop_name,
                "external_inquiry_count": ext_inquiry_count,
                "external_paid_next_count": ext_paid_next_count,
                "external_paid_final_count": ext_paid_final_count,
                "external_conv_next_pct": _fmt(ext_conv_next_pct),
                "external_conv_final_pct": _fmt(ext_conv_final_pct),
                "external_chat_count": ext_chat_count,
                "external_eval_show_count": ext_eval_show_count,
                "external_eval_join_count": ext_eval_join_count,
                "external_customer_satisfaction_pct": _fmt(ext_customer_satisfaction_pct),
                "external_service_score_5": _fmt(ext_service_score_5),
                "resp_ability_score": round(resp_score, 2),
                "satisfaction_chat_score": round(sat_chat_score, 2),
                "satisfaction_external_score": _fmt(sat_external_score),
                "satisfaction_score": round(sat_final_score, 2),
                "conversion_score": round(conversion_score, 2),
                "conversion_source": conversion_source,
                "resolution_score": round(resolution_score, 2),
                "timeliness_score": round(timeliness_score, 2),
                "risk_control_score": round(risk_ctrl_score, 2),
                "role_capability_score": round(role_score, 2),
                "role_grade": _score_grade(role_score),
                "is_reliable_sample": "是" if reliable else "否",
                "sample_scope": sample_scope,
                "satisfaction_source": satisfaction_source,
                "score_rule_note": rule_note,
            }
        )

    reliable_rows = [r for r in result_rows if r["is_reliable_sample"] == "是"]
    ranked_rows = reliable_rows if reliable_rows else result_rows
    ranked_rows.sort(
        key=lambda x: (
            x["role_capability_score"],
            x["session_count"],
            x["message_count"],
            x["agent_name"],
        ),
        reverse=True,
    )
    return ranked_rows[:limit]


def _build_ticket_summary_sql(start_date: dt.date, end_date: dt.date, shop: str) -> str:
    shop_escaped = _sql_escape(shop.strip())
    return f"""
USE [JSDataMiddlePlatform];
SET NOCOUNT ON;
DECLARE @start_time datetime2 = CAST('{start_date.isoformat()}' AS datetime2);
DECLARE @end_time datetime2 = DATEADD(day, 1, CAST('{end_date.isoformat()}' AS datetime2));
DECLARE @shop nvarchar(100) = N'{shop_escaped}';

;WITH chat_orders AS (
    SELECT DISTINCT
        NULLIF(c.orderNo, '') AS order_no,
        ISNULL(NULLIF(c.shopName, ''), N'未知店铺') AS shop_name
    FROM dbo.pc_kfChatLog AS c WITH (NOLOCK)
    WHERE c.createdAt >= @start_time
      AND c.createdAt < @end_time
      AND NULLIF(c.orderNo, '') IS NOT NULL
      AND (@shop = N'' OR c.shopName LIKE N'%' + @shop + N'%')
), joined_tickets AS (
    SELECT
        o.order_no,
        o.shop_name,
        i.StatusId,
        i.SubmitTime,
        i.kfshTime,
        i.cwshTime,
        DATEDIFF(minute, i.SubmitTime, i.kfshTime) AS submit_to_kf_minutes,
        DATEDIFF(minute, i.kfshTime, i.cwshTime) AS kf_to_cw_minutes
    FROM chat_orders AS o
    LEFT JOIN dbo.fys_Ticketing_install AS i WITH (NOLOCK)
      ON i.OrderNum = o.order_no
)
SELECT
    CAST(@start_time AS date) AS start_date,
    CAST(DATEADD(day, -1, @end_time) AS date) AS end_date,
    (SELECT COUNT(1) FROM chat_orders) AS chat_order_count,
    COUNT(DISTINCT CASE WHEN SubmitTime IS NOT NULL THEN order_no END) AS matched_ticket_orders,
    COUNT(CASE WHEN SubmitTime IS NOT NULL THEN 1 END) AS matched_ticket_rows,
    SUM(CASE WHEN StatusId = 2 THEN 1 ELSE 0 END) AS status_done_count,
    SUM(CASE WHEN StatusId <> 2 AND StatusId IS NOT NULL THEN 1 ELSE 0 END) AS status_pending_count,
    SUM(CASE WHEN SubmitTime IS NOT NULL AND kfshTime IS NULL THEN 1 ELSE 0 END) AS kf_audit_pending_count,
    SUM(CASE WHEN SubmitTime IS NOT NULL AND kfshTime IS NOT NULL AND cwshTime IS NULL THEN 1 ELSE 0 END) AS cw_audit_pending_count,
    CAST(AVG(CASE WHEN submit_to_kf_minutes BETWEEN 0 AND 43200 THEN submit_to_kf_minutes / 60.0 END) AS decimal(10,2)) AS avg_submit_to_kf_hours,
    CAST(AVG(CASE WHEN kf_to_cw_minutes BETWEEN 0 AND 43200 THEN kf_to_cw_minutes / 60.0 END) AS decimal(10,2)) AS avg_kf_to_cw_hours
FROM joined_tickets;
"""


def _build_ticket_reason_sql(start_date: dt.date, end_date: dt.date, shop: str, limit: int) -> str:
    shop_escaped = _sql_escape(shop.strip())
    return f"""
USE [JSDataMiddlePlatform];
SET NOCOUNT ON;
DECLARE @start_time datetime2 = CAST('{start_date.isoformat()}' AS datetime2);
DECLARE @end_time datetime2 = DATEADD(day, 1, CAST('{end_date.isoformat()}' AS datetime2));
DECLARE @shop nvarchar(100) = N'{shop_escaped}';

;WITH chat_orders AS (
    SELECT DISTINCT
        NULLIF(c.orderNo, '') AS order_no,
        ISNULL(NULLIF(c.shopName, ''), N'未知店铺') AS shop_name
    FROM dbo.pc_kfChatLog AS c WITH (NOLOCK)
    WHERE c.createdAt >= @start_time
      AND c.createdAt < @end_time
      AND NULLIF(c.orderNo, '') IS NOT NULL
      AND (@shop = N'' OR c.shopName LIKE N'%' + @shop + N'%')
)
SELECT TOP ({limit})
    o.shop_name,
    ISNULL(NULLIF(i.ErrorReason, ''), N'未填写') AS ticket_error_reason,
    ISNULL(NULLIF(i.Type, ''), N'未填写') AS ticket_type,
    ISNULL(CAST(i.StatusId AS varchar(20)), 'NULL') AS status_id,
    COUNT(1) AS ticket_count
FROM chat_orders AS o
JOIN dbo.fys_Ticketing_install AS i WITH (NOLOCK)
  ON i.OrderNum = o.order_no
GROUP BY
    o.shop_name,
    ISNULL(NULLIF(i.ErrorReason, ''), N'未填写'),
    ISNULL(NULLIF(i.Type, ''), N'未填写'),
    ISNULL(CAST(i.StatusId AS varchar(20)), 'NULL')
ORDER BY ticket_count DESC, o.shop_name ASC;
"""


def _build_ticket_item_sql(start_date: dt.date, end_date: dt.date, shop: str, limit: int) -> str:
    shop_escaped = _sql_escape(shop.strip())
    return f"""
USE [JSDataMiddlePlatform];
SET NOCOUNT ON;
DECLARE @start_time datetime2 = CAST('{start_date.isoformat()}' AS datetime2);
DECLARE @end_time datetime2 = DATEADD(day, 1, CAST('{end_date.isoformat()}' AS datetime2));
DECLARE @shop nvarchar(100) = N'{shop_escaped}';

;WITH chat_orders AS (
    SELECT DISTINCT
        NULLIF(c.orderNo, '') AS order_no,
        ISNULL(NULLIF(c.shopName, ''), N'未知店铺') AS shop_name
    FROM dbo.pc_kfChatLog AS c WITH (NOLOCK)
    WHERE c.createdAt >= @start_time
      AND c.createdAt < @end_time
      AND NULLIF(c.orderNo, '') IS NOT NULL
      AND (@shop = N'' OR c.shopName LIKE N'%' + @shop + N'%')
)
SELECT TOP ({limit})
    o.shop_name,
    ISNULL(NULLIF(ii.remark, ''), N'未填写') AS install_item_remark,
    ISNULL(NULLIF(ii.scmname, ''), N'未填写') AS install_provider,
    COUNT(1) AS item_count,
    COUNT(DISTINCT ii.Sourceno) AS order_count
FROM chat_orders AS o
JOIN dbo.fys_Ticketing_install_item AS ii WITH (NOLOCK)
  ON ii.Sourceno = o.order_no
GROUP BY
    o.shop_name,
    ISNULL(NULLIF(ii.remark, ''), N'未填写'),
    ISNULL(NULLIF(ii.scmname, ''), N'未填写')
ORDER BY item_count DESC, order_count DESC;
"""


def _build_platform_snapshot_sql(start_date: dt.date, end_date: dt.date, shop: str) -> str:
    shop_escaped = _sql_escape(shop.strip())
    return f"""
USE [JSDataMiddlePlatform];
SET NOCOUNT ON;
DECLARE @start_date date = '{start_date.isoformat()}';
DECLARE @end_date date = '{end_date.isoformat()}';
DECLARE @shop nvarchar(100) = N'{shop_escaped}';

WITH src AS (
    SELECT
        'jd_customerServiceScore' AS metric_source,
        TRY_CONVERT(date, REPLACE(NULLIF([date], ''), '/', '-')) AS metric_date,
        ISNULL(NULLIF(shopName, ''), N'未知店铺') AS shop_name
    FROM dbo.jd_customerServiceScore WITH (NOLOCK)
    WHERE (@shop = N'' OR shopName LIKE N'%' + @shop + N'%')

    UNION ALL

    SELECT
        'jd_customerServiceDuration',
        TRY_CONVERT(date, REPLACE(NULLIF([date], ''), '/', '-')),
        ISNULL(NULLIF(shopName, ''), N'未知店铺')
    FROM dbo.jd_customerServiceDuration WITH (NOLOCK)
    WHERE (@shop = N'' OR shopName LIKE N'%' + @shop + N'%')

    UNION ALL

    SELECT
        'pdd_DataCustomerService',
        TRY_CONVERT(date, REPLACE(NULLIF(createTime, ''), '/', '-')),
        ISNULL(NULLIF(shopName, ''), N'未知店铺')
    FROM dbo.pdd_DataCustomerService WITH (NOLOCK)
    WHERE (@shop = N'' OR shopName LIKE N'%' + @shop + N'%')

    UNION ALL

    SELECT
        'pdd_MallServiceScore',
        COALESCE(
            TRY_CONVERT(date, REPLACE(NULLIF(createTime, ''), '/', '-')),
            TRY_CONVERT(date, createdAt)
        ),
        ISNULL(NULLIF(shopName, ''), N'未知店铺')
    FROM dbo.pdd_MallServiceScore WITH (NOLOCK)
    WHERE (@shop = N'' OR shopName LIKE N'%' + @shop + N'%')
)
SELECT
    metric_source,
    COUNT(1) AS rows_total,
    MIN(metric_date) AS min_metric_date,
    MAX(metric_date) AS max_metric_date,
    SUM(CASE WHEN metric_date >= @start_date AND metric_date <= @end_date THEN 1 ELSE 0 END) AS rows_in_query_range,
    CASE
        WHEN MAX(metric_date) IS NULL THEN '无数据'
        WHEN DATEDIFF(day, MAX(metric_date), @end_date) <= 30 THEN '新鲜(<=30天)'
        WHEN DATEDIFF(day, MAX(metric_date), @end_date) <= 180 THEN '可参考(31-180天)'
        ELSE '过期(>180天)'
    END AS freshness_flag
FROM src
GROUP BY metric_source
ORDER BY metric_source ASC;
"""


def main() -> int:
    today = dt.date.today()
    default_start = today - dt.timedelta(days=6)

    parser = argparse.ArgumentParser(description="客服用户反馈体验/话术分析（增强版，含售前售后分岗评分）")
    parser.add_argument("--start", type=_parse_date, default=default_start, help="开始日期 YYYY-MM-DD（默认近7天）")
    parser.add_argument("--end", type=_parse_date, default=today, help="结束日期 YYYY-MM-DD（默认今天）")
    parser.add_argument("--shop", default="", help="店铺关键词，可选")
    parser.add_argument("--limit", type=int, default=20, help="每个结果集TOP条数")
    parser.add_argument("--agent-max-rows", type=int, default=15000, help="客服评分阶段最多读取会话条数")
    args = parser.parse_args()

    if args.start > args.end:
        print("start date must be <= end date", file=sys.stderr)
        return 2

    if (args.end - args.start).days > 366:
        print("date range too large, please keep within 367 days", file=sys.stderr)
        return 2

    limit = max(1, min(args.limit, 200))
    agent_max_rows = max(1000, min(args.agent_max_rows, 60000))

    base_cmd = [
        BASE_BIN,
        "--start",
        args.start.isoformat(),
        "--end",
        args.end.isoformat(),
        "--limit",
        str(limit),
        "--agent-max-rows",
        str(agent_max_rows),
    ]
    if args.shop:
        base_cmd.extend(["--shop", args.shop])

    base_proc = subprocess.run(base_cmd, capture_output=True, text=True)
    if base_proc.returncode != 0:
        err = (base_proc.stderr or base_proc.stdout or "base analysis failed").strip()
        print(err, file=sys.stderr)
        return base_proc.returncode

    base_blocks, _ = _split_result_sets(base_proc.stdout or "")

    # Keep base result sets 1..8, replace legacy set9 with role-split score set
    for idx in sorted(base_blocks.keys()):
        if idx <= 8:
            print("\n".join(base_blocks[idx]))

    next_idx = 9

    try:
        role_rows = _compute_role_score_rows(
            start_date=args.start,
            end_date=args.end,
            shop=args.shop,
            limit=limit,
            agent_max_rows=agent_max_rows,
        )
        _print_result_set(
            next_idx,
            [
                "shop_name",
                "agent_name",
                "position_role",
                "role_source",
                "roster_position_role",
                "roster_shop_name",
                "roster_display_name",
                "session_count",
                "primary_session_count",
                "message_count",
                "inquiry_order_count",
                "converted_order_count",
                "inquiry_conversion_rate_pct",
                "ticket_linked_order_count",
                "ticket_resolved_order_count",
                "ticket_resolve_rate_pct",
                "avg_submit_to_kf_hours",
                "external_month",
                "external_shop_name",
                "external_inquiry_count",
                "external_paid_next_count",
                "external_paid_final_count",
                "external_conv_next_pct",
                "external_conv_final_pct",
                "external_chat_count",
                "external_eval_show_count",
                "external_eval_join_count",
                "external_customer_satisfaction_pct",
                "external_service_score_5",
                "resp_ability_score",
                "satisfaction_chat_score",
                "satisfaction_external_score",
                "satisfaction_score",
                "satisfaction_source",
                "conversion_score",
                "conversion_source",
                "resolution_score",
                "timeliness_score",
                "risk_control_score",
                "role_capability_score",
                "role_grade",
                "is_reliable_sample",
                "sample_scope",
                "score_rule_note",
            ],
            role_rows,
        )
    except Exception as e:
        _print_result_set(next_idx, ["error"], [{"error": f"分岗评分计算失败: {e}"}])
    next_idx += 1

    try:
        summary_rows = _run_company_query(_build_ticket_summary_sql(args.start, args.end, args.shop))
        reason_rows = _run_company_query(_build_ticket_reason_sql(args.start, args.end, args.shop, limit))
        item_rows = _run_company_query(_build_ticket_item_sql(args.start, args.end, args.shop, limit))
        platform_rows = _run_company_query(_build_platform_snapshot_sql(args.start, args.end, args.shop))

        _print_result_set(
            next_idx,
            [
                "start_date",
                "end_date",
                "chat_order_count",
                "matched_ticket_orders",
                "matched_ticket_rows",
                "status_done_count",
                "status_pending_count",
                "kf_audit_pending_count",
                "cw_audit_pending_count",
                "avg_submit_to_kf_hours",
                "avg_kf_to_cw_hours",
            ],
            summary_rows,
        )
        next_idx += 1

        _print_result_set(
            next_idx,
            ["shop_name", "ticket_error_reason", "ticket_type", "status_id", "ticket_count"],
            reason_rows,
        )
        next_idx += 1

        _print_result_set(
            next_idx,
            ["shop_name", "install_item_remark", "install_provider", "item_count", "order_count"],
            item_rows,
        )
        next_idx += 1

        _print_result_set(
            next_idx,
            [
                "metric_source",
                "rows_total",
                "min_metric_date",
                "max_metric_date",
                "rows_in_query_range",
                "freshness_flag",
            ],
            platform_rows,
        )
        next_idx += 1

    except Exception as e:
        _print_result_set(next_idx, ["error"], [{"error": f"增强结果集计算失败: {e}"}])
        next_idx += 1

    print(f"TOTAL_RESULT_SETS: {next_idx - 1}")
    return 0


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