wuling-overseas-dealer-expansion-skill/scripts/social/write_facebook_conversations.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Validate translated Facebook conversations and write them to the workbook."""
from __future__ import annotations

import argparse
import json
import re
import subprocess
import sys
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple

from openpyxl import load_workbook
from openpyxl.styles import Font, PatternFill
from openpyxl.utils import get_column_letter

SCRIPT_DIR = Path(__file__).resolve().parent
SKILL_ROOT = SCRIPT_DIR.parents[1]
COMMON_DIR = SKILL_ROOT / "scripts" / "common"
if str(COMMON_DIR) not in sys.path:
    sys.path.insert(0, str(COMMON_DIR))

from artifact_manager import create_backup_once, new_run_id, project_root, resolve_artifact_path  # type: ignore  # noqa: E402
from workbook_resolver import resolve_workbook_path  # type: ignore  # noqa: E402

FACEBOOK_SHEET = "\u5ba2\u6237\u4fe1\u606f\u6c47\u603b\u8868"
CONVERSATION_SHEET = "Facebook对话记录"
CONVERSATION_HEADERS = [
    "记录ID", "客户序号", "客户姓名/公司", "Facebook主页链接", "Messenger线程ID",
    "消息时间", "消息方向", "发件人", "原文语言", "对话原文", "中文翻译", "消息类型",
    "是否有效客户回复", "合作意向", "意向判断依据", "下一步建议", "同步时间",
    "来源账号/Profile ID", "风险标记",
]
FACEBOOK_REPLY_STATUS = "Facebook\u5df2\u56de\u590d"
OLD_REPLY_STATUS_PHRASES = [
    "\u5df2\u56de\u590d,\u6709\u5408\u4f5c\u610f\u5411", "\u5df2\u56de\u590d,\u5f85\u8ddf\u8fdb", "\u5df2\u56de\u590d,\u5f85\u6f84\u6e05", "\u5df2\u56de\u590d,\u6682\u4e0d\u8003\u8651", "\u5df2\u56de\u590d,\u660e\u786e\u62d2\u7edd",
    "\u5df2\u56de\u590d\uff0c\u6709\u5408\u4f5c\u610f\u5411", "\u5df2\u56de\u590d\uff0c\u5f85\u8ddf\u8fdb", "\u5df2\u56de\u590d\uff0c\u5f85\u6f84\u6e05", "\u5df2\u56de\u590d\uff0c\u6682\u4e0d\u8003\u8651", "\u5df2\u56de\u590d\uff0c\u660e\u786e\u62d2\u7edd",
]
NOTE_START = "\u3010Facebook\u56de\u590d\u8bb0\u5f55\u3011"
NOTE_END = "\u3010/Facebook\u56de\u590d\u8bb0\u5f55\u3011"
QUESTION_MARK_RE = re.compile(r"\?{3,}")
CJK_RE = re.compile(r"[\u3400-\u9fff]")
FACEBOOK_ALIASES = {
    "index": ["序号", "编号", "ID"],
    "company": ["客户姓名/公司", "公司名称", "公司姓名", "客户名称"],
    "link": ["主页/链接", "Facebook主页链接", "Facebook链接", "facebook链接"],
    "status": ["建联状态", "建联情况"],
    "followup": ["下次跟进", "下次跟进时间"],
    "note": ["备注", "说明"],
}


def clean(value: Any) -> str:
    if value is None:
        return ""
    return re.sub(r"\s+", " ", str(value).strip())


def now_text() -> str:
    return datetime.now().astimezone().isoformat(timespec="seconds")


def read_json(path: Path) -> Dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8-sig"))


def write_json(path: Path, data: Dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


def normalized_link(value: str) -> str:
    value = clean(value).casefold().split("?", 1)[0].rstrip("/")
    return value.removeprefix("https://").removeprefix("http://").removeprefix("www.")


def list_values(value: Any) -> List[str]:
    if isinstance(value, list):
        return [clean(item) for item in value if clean(item)]
    return [part for part in re.split(r"[；;|]+", clean(value)) if part] if clean(value) else []


def analysis_key(thread: Dict[str, Any]) -> Tuple[str, str, str]:
    return (
        clean(thread.get("customer_index")),
        clean(thread.get("thread_id")),
        normalized_link(clean(thread.get("facebook_link"))),
    )


def raw_thread_key(thread: Dict[str, Any]) -> Tuple[str, str, str]:
    customer = thread.get("customer") or {}
    return (
        clean(customer.get("index")),
        clean(thread.get("thread_id") or customer.get("expected_identity")),
        normalized_link(clean(customer.get("facebook_link"))),
    )


def fallback_analysis_match(raw_thread: Dict[str, Any], candidates: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
    raw_key = raw_thread_key(raw_thread)
    for candidate in candidates:
        candidate_key = analysis_key(candidate)
        checks = [left and right and left == right for left, right in zip(raw_key, candidate_key)]
        if sum(checks) >= 2:
            return candidate
    return None


def chinese_or_same(original: str, translated: str) -> str:
    original, translated = clean(original), clean(translated)
    if translated:
        return translated
    if not original:
        return ""
    if CJK_RE.search(original):
        return original
    raise ValueError(f"Missing Chinese translation for message: {original[:80]}")


def validate_intent(value: str) -> str:
    return clean(value)

def parse_date(value: str) -> Optional[date]:
    match = re.search(r"(20\d{2})[-/](\d{1,2})[-/](\d{1,2})", clean(value))
    if not match:
        return None
    try:
        return date(int(match.group(1)), int(match.group(2)), int(match.group(3)))
    except ValueError:
        return None


def add_business_days(start: date, days: int) -> date:
    result = start
    while days > 0:
        result += timedelta(days=1)
        if result.weekday() < 5:
            days -= 1
    return result


def default_followup(intent: str, latest_reply_at: str, explicit: str = "") -> str:
    if clean(explicit):
        parsed = parse_date(explicit)
        if not parsed:
            raise ValueError(f"Invalid next_followup date: {explicit!r}")
        return parsed.isoformat()
    kind, amount = FOLLOWUP_DAYS[intent]
    if kind == "none":
        return ""
    start = parse_date(latest_reply_at) or date.today()
    return (add_business_days(start, amount) if kind == "business" else start + timedelta(days=amount)).isoformat()


def header_map(ws, aliases: Dict[str, List[str]]) -> Dict[str, int]:
    raw = {clean(cell.value): idx for idx, cell in enumerate(ws[1], start=1) if clean(cell.value)}
    mapped: Dict[str, int] = {}
    for key, names in aliases.items():
        for name in names:
            if name in raw:
                mapped[key] = raw[name]
                break
    return mapped


def ensure_conversation_sheet(wb, sheet_name: str):
    if sheet_name not in wb.sheetnames:
        ws = wb.create_sheet(sheet_name)
        ws.append(CONVERSATION_HEADERS)
    else:
        ws = wb[sheet_name]
        existing = {clean(cell.value) for cell in ws[1] if clean(cell.value)}
        for header in CONVERSATION_HEADERS:
            if header not in existing:
                ws.cell(row=1, column=ws.max_column + 1).value = header
    header_fill = PatternFill(fill_type="solid", fgColor="D9EAF7")
    for cell in ws[1]:
        cell.font, cell.fill = Font(bold=True), header_fill
    headers = {clean(cell.value): idx for idx, cell in enumerate(ws[1], start=1)}
    widths = [38, 12, 28, 38, 24, 22, 12, 20, 12, 56, 56, 14, 18, 16, 52, 52, 22, 22, 32]
    for header, width in zip(CONVERSATION_HEADERS, widths):
        ws.column_dimensions[get_column_letter(headers[header])].width = width
    ws.freeze_panes = "A2"
    ws.auto_filter.ref = ws.dimensions
    return ws


def message_analysis_map(thread: Dict[str, Any]) -> Dict[str, Dict[str, Any]]:
    return {clean(item.get("record_id")): item for item in thread.get("messages", []) if clean(item.get("record_id"))}


def is_real_customer_reply(direction: str, kind: str, original: str, translation: str) -> bool:
    if clean(direction) != "\u5ba2\u6237\u56de\u590d":
        return False
    blocked_types = {
        "\u7cfb\u7edf\u6d88\u606f", "\u81ea\u52a8\u56de\u590d", "\u5df2\u8bfb\u63d0\u793a",
        "\u70b9\u8d5e", "\u8868\u60c5", "reaction", "read_receipt",
    }
    if clean(kind).casefold() in {item.casefold() for item in blocked_types}:
        return False
    # 翻译以“（自动回复）”开头说明分析判定为自动回复，不视为人工有效回复
    if clean(translation).startswith("\uff08\u81ea\u52a8\u56de\u590d\uff09"):
        return False
    # 纯时间戳/加载提示不是有效回复
    _orig = clean(original).strip().lower()
    _trans = clean(translation).strip().lower()
    if _orig in ("loading...",) or "loading..." in _trans:
        return False
    import re as _re
    if _re.match(r"^((mon|tue|wed|thu|fri|sat|sun)\s+)?\d{1,2}:\d{2}(\s*(am|pm))?$", _orig):
        return False
    # 主页简介（Dealership/Dealer/Showroom 结尾的页面介绍）不是有效回复
    if _re.search(r"(automotive dealership|car dealership|car dealer|car rental|showroom|cars)$", _orig):
        return False
    emoji_only = clean(original or translation) in {"??", "??", "??", "??", "?", "??", "?"}
    return not emoji_only


def merge_threads(raw: Dict[str, Any], analysis: Dict[str, Any]) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], List[str]]:
    analysis_threads = analysis.get("threads") or []
    exact = {analysis_key(thread): thread for thread in analysis_threads}
    output_rows: List[Dict[str, Any]] = []
    customer_updates: List[Dict[str, Any]] = []
    warnings: List[str] = []
    sync_time = now_text()
    for raw_thread in raw.get("threads") or []:
        if raw_thread.get("status") != "collected":
            warnings.append(f"{clean((raw_thread.get('customer') or {}).get('company'))}: {clean(raw_thread.get('status'))}")
            continue
        messages = raw_thread.get("messages") or []
        if not messages:
            continue
        athread = exact.get(raw_thread_key(raw_thread)) or fallback_analysis_match(raw_thread, analysis_threads) or {}
        amap = message_analysis_map(athread) if athread else {}
        customer = raw_thread.get("customer") or {}
        real_reply_rows: List[Dict[str, Any]] = []
        for raw_message in messages:
            record_id = clean(raw_message.get("record_id"))
            if not record_id:
                raise ValueError("Raw message is missing record_id")
            analyzed = amap.get(record_id) or {}
            direction = clean(raw_message.get("direction"))
            kind = clean(raw_message.get("message_type")) or "\u6587\u672c"
            original = clean(raw_message.get("original_text"))
            translated_text = clean(analyzed.get("chinese_translation"))
            if direction == "\u5ba2\u6237\u56de\u590d":
                translation = chinese_or_same(original, translated_text)
            else:
                translation = translated_text or original
            original_language = clean(analyzed.get("original_language")) or ("zh" if CJK_RE.search(original) else "")
            is_reply = is_real_customer_reply(direction, kind, original, translation)
            next_action = clean(analyzed.get("next_action")) if is_reply else ""
            if is_reply and not next_action:
                next_action = "\u8bf7\u4eba\u5de5\u67e5\u770b\u5ba2\u6237\u56de\u590d\uff0c\u5e76\u5224\u65ad\u662f\u5426\u9700\u8981\u7ee7\u7eed\u8ddf\u8fdb\u3002"
            risks = list_values(raw_message.get("risk_flags")) + list_values(analyzed.get("risk_flags"))
            row = {
                "\u8bb0\u5f55ID": record_id,
                "\u5ba2\u6237\u5e8f\u53f7": clean(customer.get("index")),
                "\u5ba2\u6237\u59d3\u540d/\u516c\u53f8": clean(customer.get("company")),
                "Facebook\u4e3b\u9875\u94fe\u63a5": clean(customer.get("facebook_link")),
                "Messenger\u7ebf\u7a0bID": clean(raw_thread.get("thread_id")),
                "\u6d88\u606f\u65f6\u95f4": clean(analyzed.get("message_time")) or clean(raw_message.get("message_time_raw")),
                "\u6d88\u606f\u65b9\u5411": direction,
                "\u53d1\u4ef6\u4eba": clean(raw_message.get("sender")),
                "\u539f\u6587\u8bed\u8a00": original_language,
                "\u5bf9\u8bdd\u539f\u6587": original,
                "\u4e2d\u6587\u7ffb\u8bd1": translation,
                "\u6d88\u606f\u7c7b\u578b": kind,
                "\u662f\u5426\u6709\u6548\u5ba2\u6237\u56de\u590d": "\u662f" if is_reply else "\u5426",
                "\u5408\u4f5c\u610f\u5411": "",
                "\u610f\u5411\u5224\u65ad\u4f9d\u636e": "",
                "\u4e0b\u4e00\u6b65\u5efa\u8bae": next_action,
                "\u540c\u6b65\u65f6\u95f4": sync_time,
                "\u6765\u6e90\u8d26\u53f7/Profile ID": clean(raw.get("profile_id")),
                "\u98ce\u9669\u6807\u8bb0": "\uff1b".join(dict.fromkeys(risks)),
            }
            output_rows.append(row)
            if is_reply:
                real_reply_rows.append(row)
        if real_reply_rows:
            latest = athread.get("latest_analysis") or {}
            latest_row = real_reply_rows[-1]
            latest_record_id = clean(latest.get("latest_reply_record_id"))
            if latest_record_id:
                latest_row = next((row for row in real_reply_rows if row.get("\u8bb0\u5f55ID") == latest_record_id), latest_row)
            summary = (
                clean(latest.get("chinese_summary"))
                or clean(latest_row.get("\u4e2d\u6587\u7ffb\u8bd1"))
                or clean(latest_row.get("\u5bf9\u8bdd\u539f\u6587"))
                or "\u5ba2\u6237\u5df2\u5728 Facebook Messenger \u56de\u590d\uff0c\u9700\u4eba\u5de5\u67e5\u770b\u5177\u4f53\u5185\u5bb9\u3002"
            )
            next_action = (
                clean(latest.get("next_action"))
                or clean(latest_row.get("\u4e0b\u4e00\u6b65\u5efa\u8bae"))
                or "\u8bf7\u4eba\u5de5\u67e5\u770b\u5ba2\u6237\u56de\u590d\uff0c\u5e76\u5224\u65ad\u662f\u5426\u9700\u8981\u7ee7\u7eed\u8ddf\u8fdb\u3002"
            )
            customer_updates.append({
                "index": clean(customer.get("index")),
                "company": clean(customer.get("company")),
                "facebook_link": clean(customer.get("facebook_link")),
                "latest_reply_at": clean(latest.get("latest_reply_at")) or clean(latest_row.get("\u6d88\u606f\u65f6\u95f4")),
                "summary": summary,
                "next_action": next_action,
                "next_followup": clean(latest.get("next_followup")),
                "latest_reply_record_id": clean(latest_row.get("\u8bb0\u5f55ID")),
            })
    return output_rows, customer_updates, warnings

def append_or_update_conversations(ws, rows: Iterable[Dict[str, Any]]) -> Tuple[int, int]:
    headers = {clean(cell.value): idx for idx, cell in enumerate(ws[1], start=1)}
    existing = {
        clean(ws.cell(row=row, column=headers["记录ID"]).value): row
        for row in range(2, ws.max_row + 1)
        if clean(ws.cell(row=row, column=headers["记录ID"]).value)
    }
    immutable = {"记录ID", "客户序号", "客户姓名/公司", "Facebook主页链接", "Messenger线程ID", "对话原文"}
    added = updated = 0
    for item in rows:
        record_id = clean(item.get("记录ID"))
        if record_id in existing:
            target, changed = existing[record_id], False
            for header in CONVERSATION_HEADERS:
                if header in immutable:
                    continue
                value = item.get(header, "")
                if value not in (None, "") and ws.cell(target, headers[header]).value != value:
                    ws.cell(target, headers[header]).value = value
                    changed = True
            updated += int(changed)
        else:
            target = ws.max_row + 1
            for header in CONVERSATION_HEADERS:
                ws.cell(target, headers[header]).value = item.get(header, "")
            existing[record_id] = target
            added += 1
    ws.auto_filter.ref = ws.dimensions
    return added, updated


def merge_status(existing: str) -> str:
    result = clean(existing)
    for phrase in OLD_REPLY_STATUS_PHRASES:
        result = result.replace(phrase, "")
    result = re.sub(r"[,\uFF0C;\uFF1B|]+", ",", result).strip(", ")
    parts = [clean(part) for part in result.split(",") if clean(part)]
    parts.append(FACEBOOK_REPLY_STATUS)
    return ",".join(dict.fromkeys(parts))


def reply_note(update: Dict[str, Any]) -> str:
    return (
        f"{NOTE_START}\u6700\u65b0\u56de\u590d\u65f6\u95f4\uff1a{update['latest_reply_at']}\uff1b\u4e2d\u6587\u6458\u8981\uff1a{update['summary']}\uff1b"
        f"\u4e0b\u4e00\u6b65\u5efa\u8bae\uff1a{update['next_action']}{NOTE_END}"
    )


def replace_reply_note(existing: str, block: str) -> str:
    existing = clean(existing)
    patterns = [
        re.compile(re.escape(NOTE_START) + r".*?" + re.escape(NOTE_END)),
        re.compile(re.escape("\u3010Facebook\u56de\u590d\u5206\u6790\u3011") + r".*?" + re.escape("\u3010/Facebook\u56de\u590d\u5206\u6790\u3011")),
    ]
    for pattern in patterns:
        if pattern.search(existing):
            return clean(pattern.sub(block, existing))
    return clean(existing + (" | " if existing else "") + block)


def update_facebook_rows(wb, sheet_name: str, updates: Sequence[Dict[str, Any]]) -> Tuple[int, List[str]]:
    if sheet_name not in wb.sheetnames:
        raise KeyError(f"Sheet not found: {sheet_name}")
    ws = wb[sheet_name]
    columns = header_map(ws, FACEBOOK_ALIASES)
    required = {"index", "company", "link", "status", "followup", "note"}
    if missing := sorted(required - set(columns)):
        raise RuntimeError(f"Facebook sheet is missing columns: {', '.join(missing)}")
    by_index = {clean(item.get("index")): item for item in updates if clean(item.get("index"))}
    written, missing_customers, matched = 0, [], set()
    for row in range(2, ws.max_row + 1):
        row_index = clean(ws.cell(row, columns["index"]).value)
        update = by_index.get(row_index)
        if not update:
            continue
        if normalized_link(clean(ws.cell(row, columns["link"]).value)) != normalized_link(clean(update.get("facebook_link"))):
            missing_customers.append(f"{update.get('company')}: Facebook link mismatch")
            continue
        ws.cell(row, columns["status"]).value = merge_status(clean(ws.cell(row, columns["status"]).value))
        if clean(update.get("next_followup")):
            ws.cell(row, columns["followup"]).value = update["next_followup"]
        existing_note = clean(ws.cell(row, columns["note"]).value)
        ws.cell(row, columns["note"]).value = replace_reply_note(existing_note, reply_note(update))
        matched.add(row_index)
        written += 1
    for update in updates:
        if clean(update.get("index")) not in matched:
            missing_customers.append(f"{update.get('company')}: customer row not matched")
    return written, list(dict.fromkeys(missing_customers))

def scan_question_marks(wb) -> Tuple[int, int]:
    cells = note_rows = 0
    for ws in wb.worksheets:
        note_columns = {idx for idx, cell in enumerate(ws[1], start=1) if clean(cell.value) in {"备注", "说明"}}
        bad_note_rows = set()
        for row in ws.iter_rows():
            for cell in row:
                if isinstance(cell.value, str) and QUESTION_MARK_RE.search(cell.value):
                    cells += 1
                    if cell.column in note_columns:
                        bad_note_rows.add(cell.row)
        note_rows += len(bad_note_rows)
    return cells, note_rows


def ensure_generated_text_clean(rows: Sequence[Dict[str, Any]], updates: Sequence[Dict[str, Any]]) -> None:
    for item in [*rows, *updates]:
        for key, value in item.items():
            if isinstance(value, str) and QUESTION_MARK_RE.search(value):
                raise ValueError(f"Repeated question marks detected before write: field={key}")


def run_child(command: List[str]) -> Dict[str, Any]:
    completed = subprocess.run(command, text=True, encoding="utf-8", capture_output=True, check=True)
    try:
        return json.loads(completed.stdout)
    except json.JSONDecodeError:
        return {"stdout": completed.stdout.strip(), "stderr": completed.stderr.strip()}


def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Write translated Facebook conversations and unified reply records to Excel.")
    parser.add_argument("--excel", default="")
    parser.add_argument("--transcript", default="")
    parser.add_argument("--analysis", default="")
    parser.add_argument("--facebook-sheet", default=FACEBOOK_SHEET)
    parser.add_argument("--conversation-sheet", default=CONVERSATION_SHEET)
    parser.add_argument("--write-workbook", action="store_true")
    parser.add_argument("--ensure-sheet-only", action="store_true")
    parser.add_argument("--refresh-summary", action="store_true")
    parser.add_argument("--refresh-dashboard", action="store_true")
    parser.add_argument("--latest-dashboard-dir", default="dashboards/latest")
    parser.add_argument("--run-id", default="")
    parser.add_argument("--output", default="")
    parser.add_argument("--no-backup", action="store_true")
    return parser.parse_args(argv)


def main(argv: Optional[Sequence[str]] = None) -> int:
    args = parse_args(argv)
    run_id = args.run_id or new_run_id("facebook_conversation_write")
    resolved = resolve_workbook_path(args.excel, create_from_template=bool(args.write_workbook))
    workbook = resolved.get("path")
    if not workbook:
        raise FileNotFoundError("No outreach workbook found. Pass --excel.")
    workbook_path = Path(workbook)
    lock_path = workbook_path.with_name("~$" + workbook_path.name)
    if args.write_workbook and lock_path.exists():
        raise PermissionError(f"Workbook appears to be open: {lock_path}")

    rows, updates, warnings = [], [], []
    if not args.ensure_sheet_only:
        if not args.transcript or not args.analysis:
            raise ValueError("--transcript and --analysis are required unless --ensure-sheet-only is used")
        rows, updates, warnings = merge_threads(read_json(Path(args.transcript)), read_json(Path(args.analysis)))
        ensure_generated_text_clean(rows, updates)

    report_path = resolve_artifact_path(
        args.output, kind="facebook_conversation_write",
        default_name="facebook_conversation_write_report.json", run_id=run_id,
    )
    report: Dict[str, Any] = {
        "schema_version": "4.29",
        "run_id": run_id,
        "workbook": str(workbook_path),
        "dry_run": not args.write_workbook,
        "conversation_rows_ready": len(rows),
        "customer_updates_ready": len(updates),
        "facebook_reply_customers_ready": len(updates),
        "warnings": warnings,
    }
    if not args.write_workbook:
        write_json(report_path, report)
        print(json.dumps({"report": str(report_path), **report}, ensure_ascii=False, indent=2))
        return 0

    backup_path = None if args.no_backup else create_backup_once(
        workbook_path, purpose="facebook_conversations", run_id=run_id,
    )
    wb = load_workbook(workbook_path)
    before_question_cells, before_question_notes = scan_question_marks(wb)
    if before_question_cells:
        raise ValueError(
            f"Workbook already contains repeated-question-mark corruption: cells={before_question_cells}, "
            f"note_rows={before_question_notes}"
        )
    conversation_ws = ensure_conversation_sheet(wb, args.conversation_sheet)
    added, conversation_updated = append_or_update_conversations(conversation_ws, rows)
    customer_rows_updated, missing_customers = (
        update_facebook_rows(wb, args.facebook_sheet, updates) if updates else (0, [])
    )
    if scan_question_marks(wb)[0]:
        raise ValueError("Repeated question marks detected before save")
    wb.save(workbook_path)

    summary_report, dashboard_report = {}, {}
    refresh_summary = bool(args.refresh_summary or args.refresh_dashboard)
    if refresh_summary:
        summary_report = run_child([
            sys.executable, str(SKILL_ROOT / "scripts" / "common" / "build_customer_summary.py"),
            "--excel", str(workbook_path), "--write-summary", "--no-backup", "--run-id", run_id,
        ])
    if args.refresh_dashboard:
        dashboard_report = run_child([
            sys.executable, str(SKILL_ROOT / "scripts" / "dashboard" / "build_dashboard.py"),
            "--excel", str(workbook_path), "--run-id", run_id,
            "--latest-dir", args.latest_dashboard_dir,
        ])

    reopened = load_workbook(workbook_path, data_only=True, read_only=True)
    question_mark_cells, question_mark_note_rows = scan_question_marks(reopened)
    reopened.close()
    report.update({
        "dry_run": False,
        "backup": str(backup_path) if backup_path else "",
        "conversation_rows_added": added,
        "conversation_rows_updated": conversation_updated,
        "customer_rows_updated": customer_rows_updated,
        "missing_customers": missing_customers,
        "question_mark_cells": question_mark_cells,
        "question_mark_note_rows": question_mark_note_rows,
        "summary_refreshed": refresh_summary,
        "summary_report": summary_report,
        "dashboard_refreshed": bool(args.refresh_dashboard),
        "dashboard_report": dashboard_report,
    })
    config_path = project_root(workbook_path.parent) / "feishu_sync_config.json"
    if config_path.exists():
        try:
            config = read_json(config_path)
        except Exception as exc:
            config = {"enabled": False, "config_error": str(exc)}
        report["feishu_sync"] = {
            "config_found": True,
            "enabled": bool(config.get("enabled")),
            "sync_required": bool(config.get("enabled")),
            "conversation_sheet_name": config.get("conversation_sheet_name", args.conversation_sheet),
            "sync_conversation_sheet": config.get("sync_conversation_sheet", True),
            "agent_action": "Use lark-sheets after local write; Python never stores Feishu credentials.",
        }
    else:
        report["feishu_sync"] = {
            "config_found": False,
            "enabled": False,
            "sync_required": False,
            "message": "本地建联表已更新；未发现 feishu_sync_config.json，飞书同步未执行。",
        }
    write_json(report_path, report)
    print(json.dumps({"report": str(report_path), **report}, ensure_ascii=False, indent=2))
    return 3 if question_mark_cells or missing_customers else 0


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