#!/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 = "Facebook" CONVERSATION_SHEET = "Facebook对话记录" CONVERSATION_HEADERS = [ "记录ID", "客户序号", "客户姓名/公司", "Facebook主页链接", "Messenger线程ID", "消息时间", "消息方向", "发件人", "原文语言", "对话原文", "中文翻译", "消息类型", "是否有效客户回复", "合作意向", "意向判断依据", "下一步建议", "同步时间", "来源账号/Profile ID", "风险标记", ] INTENTS = ["明确有意向", "潜在意向", "需澄清", "暂不考虑", "明确拒绝"] STATUS_BY_INTENT = { "明确有意向": "已回复,有合作意向", "潜在意向": "已回复,待跟进", "需澄清": "已回复,待澄清", "暂不考虑": "已回复,暂不考虑", "明确拒绝": "已回复,明确拒绝", } FOLLOWUP_DAYS = { "明确有意向": ("business", 1), "潜在意向": ("business", 3), "需澄清": ("business", 2), "暂不考虑": ("calendar", 30), "明确拒绝": ("none", 0), } REPLY_STATUS_PHRASES = list(STATUS_BY_INTENT.values()) NOTE_START = "【Facebook回复分析】" NOTE_END = "【/Facebook回复分析】" 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: value = clean(value) if value not in INTENTS: raise ValueError(f"Invalid cooperation intent: {value!r}. Expected one of {INTENTS}") return 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 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) if not athread: raise ValueError(f"Missing analysis thread for {raw_thread_key(raw_thread)}") amap = message_analysis_map(athread) customer = raw_thread.get("customer") or {} effective_reply_ids: List[str] = [] 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 "文本" original = clean(raw_message.get("original_text")) translation = chinese_or_same(original, clean(analyzed.get("chinese_translation"))) original_language = clean(analyzed.get("original_language")) or ("zh" if CJK_RE.search(original) else "") non_reply = direction != "客户回复" or kind in {"系统消息", "自动回复", "已读提示"} effective = bool(analyzed.get("is_effective_customer_reply")) and not non_reply intent = reason = action = "" if effective: intent = validate_intent(clean(analyzed.get("intent"))) reason, action = clean(analyzed.get("intent_reason")), clean(analyzed.get("next_action")) if not reason or not action: raise ValueError(f"Effective reply {record_id} requires intent_reason and next_action") effective_reply_ids.append(record_id) risks = list_values(raw_message.get("risk_flags")) + list_values(analyzed.get("risk_flags")) output_rows.append({ "记录ID": record_id, "客户序号": clean(customer.get("index")), "客户姓名/公司": clean(customer.get("company")), "Facebook主页链接": clean(customer.get("facebook_link")), "Messenger线程ID": clean(raw_thread.get("thread_id")), "消息时间": clean(analyzed.get("message_time")) or clean(raw_message.get("message_time_raw")), "消息方向": direction, "发件人": clean(raw_message.get("sender")), "原文语言": original_language, "对话原文": original, "中文翻译": translation, "消息类型": kind, "是否有效客户回复": "是" if effective else "否", "合作意向": intent, "意向判断依据": reason, "下一步建议": action, "同步时间": sync_time, "来源账号/Profile ID": clean(raw.get("profile_id")), "风险标记": "；".join(dict.fromkeys(risks)), }) latest = athread.get("latest_analysis") or {} if effective_reply_ids: latest_record_id = clean(latest.get("latest_reply_record_id")) if latest_record_id not in effective_reply_ids: raise ValueError(f"latest_reply_record_id must reference an effective reply for {clean(customer.get('company'))}") latest_intent = validate_intent(clean(latest.get("intent"))) summary = clean(latest.get("chinese_summary")) reason, next_action = clean(latest.get("intent_reason")), clean(latest.get("next_action")) latest_reply_at = clean(latest.get("latest_reply_at")) if not all([summary, reason, next_action, latest_reply_at]): raise ValueError(f"latest_analysis is incomplete for {clean(customer.get('company'))}") customer_updates.append({ "index": clean(customer.get("index")), "company": clean(customer.get("company")), "facebook_link": clean(customer.get("facebook_link")), "latest_reply_at": latest_reply_at, "summary": summary, "intent": latest_intent, "intent_reason": reason, "next_action": next_action, "next_followup": default_followup(latest_intent, latest_reply_at, clean(latest.get("next_followup"))), "latest_reply_record_id": latest_record_id, }) 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, intent: str) -> str: result = clean(existing) for phrase in REPLY_STATUS_PHRASES: result = result.replace(phrase, "") result = re.sub(r"[,，;；|]+", ",", result).strip(", ") latest = STATUS_BY_INTENT[intent] return ",".join(dict.fromkeys([clean(part) for part in [*result.split(","), *latest.split(",")] if clean(part)])) def reply_note(update: Dict[str, Any]) -> str: return ( f"{NOTE_START}最新回复时间：{update['latest_reply_at']}；中文摘要：{update['summary']}；" f"合作意向：{update['intent']}；判断依据：{update['intent_reason']}；" f"下一步建议：{update['next_action']}{NOTE_END}" ) def replace_reply_note(existing: str, block: str) -> str: existing = clean(existing) pattern = re.compile(re.escape(NOTE_START) + r".*?" + re.escape(NOTE_END)) return clean(pattern.sub(block, existing)) if pattern.search(existing) else 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), update["intent"]) 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 intent analysis 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.26", "run_id": run_id, "workbook": str(workbook_path), "dry_run": not args.write_workbook, "conversation_rows_ready": len(rows), "customer_updates_ready": len(updates), "intent_counts": {intent: sum(1 for item in updates if item.get("intent") == intent) for intent in INTENTS}, "high_intent_customers": [ { "customer": item["company"], "intent": item["intent"], "summary": item["summary"], "next_action": item["next_action"], "next_followup": item["next_followup"], } for item in updates if item.get("intent") in {"明确有意向", "潜在意向"} ], "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())