test_facebook_conversation_sync.py 2.8 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061
  1. from __future__ import annotations
  2. import importlib.util
  3. import sys
  4. import unittest
  5. from pathlib import Path
  6. from openpyxl import Workbook
  7. ROOT = Path(__file__).resolve().parents[1]
  8. COMMON = ROOT / 'scripts' / 'common'
  9. sys.path.insert(0, str(COMMON))
  10. def load_module(name, relative_path):
  11. spec = importlib.util.spec_from_file_location(name, ROOT / relative_path)
  12. module = importlib.util.module_from_spec(spec)
  13. sys.modules[name] = module
  14. spec.loader.exec_module(module)
  15. return module
  16. writer = load_module('facebook_conversation_writer', 'scripts/social/write_facebook_conversations.py')
  17. dashboard = load_module('customer_dashboard', 'scripts/dashboard/build_dashboard.py')
  18. class FacebookConversationSyncTests(unittest.TestCase):
  19. def test_five_intents_and_translation_guard(self):
  20. expected = {
  21. '明确有意向': '已回复,有合作意向',
  22. '潜在意向': '已回复,待跟进',
  23. '需澄清': '已回复,待澄清',
  24. '暂不考虑': '已回复,暂不考虑',
  25. '明确拒绝': '已回复,明确拒绝',
  26. }
  27. for intent, status in expected.items():
  28. self.assertEqual(writer.merge_status('已发送邮件', intent), f'已发送邮件,{status}')
  29. self.assertEqual(writer.default_followup('明确有意向', '2026-08-07'), '2026-08-10')
  30. self.assertEqual(writer.default_followup('明确拒绝', '2026-08-07'), '')
  31. with self.assertRaises(ValueError):
  32. writer.chinese_or_same('Bonjour', '')
  33. def test_conversation_records_deduplicate(self):
  34. wb = Workbook()
  35. ws = writer.ensure_conversation_sheet(wb, writer.CONVERSATION_SHEET)
  36. row = {header: '' for header in writer.CONVERSATION_HEADERS}
  37. row.update({'记录ID': 'message-1', '客户序号': '1', '对话原文': 'Bonjour', '中文翻译': '您好'})
  38. self.assertEqual(writer.append_or_update_conversations(ws, [row]), (1, 0))
  39. self.assertEqual(writer.append_or_update_conversations(ws, [row]), (0, 0))
  40. self.assertEqual(ws.max_row, 2)
  41. def test_dashboard_counts_unique_customers(self):
  42. rows = [
  43. {'customer_index': '1', 'direction': '我方发送', 'effective': '否'},
  44. {'customer_index': '1', 'direction': '客户回复', 'effective': '是', 'intent': '明确有意向', 'message_time': '2026-08-05', 'company': 'Atlas'},
  45. {'customer_index': '1', 'direction': '客户回复', 'effective': '是', 'intent': '明确有意向', 'message_time': '2026-08-06', 'company': 'Atlas'},
  46. ]
  47. result = dashboard.build_conversation_insights(rows)
  48. self.assertEqual(len(result['sent_customers']), 1)
  49. self.assertEqual(len(result['replied_customers']), 1)
  50. self.assertEqual(result['intent_counter']['明确有意向'], 1)