#!/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())