from __future__ import annotations import importlib.util import sys import unittest from pathlib import Path from openpyxl import Workbook ROOT = Path(__file__).resolve().parents[1] COMMON = ROOT / 'scripts' / 'common' sys.path.insert(0, str(COMMON)) def load_module(name, relative_path): spec = importlib.util.spec_from_file_location(name, ROOT / relative_path) module = importlib.util.module_from_spec(spec) sys.modules[name] = module spec.loader.exec_module(module) return module writer = load_module('facebook_conversation_writer', 'scripts/social/write_facebook_conversations.py') dashboard = load_module('customer_dashboard', 'scripts/dashboard/build_dashboard.py') class FacebookConversationSyncTests(unittest.TestCase): def test_five_intents_and_translation_guard(self): expected = { '明确有意向': '已回复,有合作意向', '潜在意向': '已回复,待跟进', '需澄清': '已回复,待澄清', '暂不考虑': '已回复,暂不考虑', '明确拒绝': '已回复,明确拒绝', } for intent, status in expected.items(): self.assertEqual(writer.merge_status('已发送邮件', intent), f'已发送邮件,{status}') self.assertEqual(writer.default_followup('明确有意向', '2026-08-07'), '2026-08-10') self.assertEqual(writer.default_followup('明确拒绝', '2026-08-07'), '') with self.assertRaises(ValueError): writer.chinese_or_same('Bonjour', '') def test_conversation_records_deduplicate(self): wb = Workbook() ws = writer.ensure_conversation_sheet(wb, writer.CONVERSATION_SHEET) row = {header: '' for header in writer.CONVERSATION_HEADERS} row.update({'记录ID': 'message-1', '客户序号': '1', '对话原文': 'Bonjour', '中文翻译': '您好'}) self.assertEqual(writer.append_or_update_conversations(ws, [row]), (1, 0)) self.assertEqual(writer.append_or_update_conversations(ws, [row]), (0, 0)) self.assertEqual(ws.max_row, 2) def test_dashboard_counts_unique_customers(self): rows = [ {'customer_index': '1', 'direction': '我方发送', 'effective': '否'}, {'customer_index': '1', 'direction': '客户回复', 'effective': '是', 'intent': '明确有意向', 'message_time': '2026-08-05', 'company': 'Atlas'}, {'customer_index': '1', 'direction': '客户回复', 'effective': '是', 'intent': '明确有意向', 'message_time': '2026-08-06', 'company': 'Atlas'}, ] result = dashboard.build_conversation_insights(rows) self.assertEqual(len(result['sent_customers']), 1) self.assertEqual(len(result['replied_customers']), 1) self.assertEqual(result['intent_counter']['明确有意向'], 1)