brand_detector.py 2.3 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374
  1. """
  2. 品牌检测与排他协议辅助判断
  3. """
  4. import re
  5. from typing import List, Dict, Any
  6. # 中国品牌
  7. CHINESE_BRANDS = [
  8. "byd", "chery", "dfsk", "dongfeng", "foton", "gac", "geely",
  9. "great wall", "haval", "jac", "jetour", "li auto", "mg", "nio",
  10. "omoda", "ora", "wuling", "xpeng", "zeekr", "changan", "baic",
  11. " brilliance", "maxus", "saic"
  12. ]
  13. # 常见非中国品牌(用于判断“无排他”)
  14. NON_CHINESE_BRANDS = [
  15. "audi", "bmw", "citroen", "dacia", "fiat", "ford", "honda",
  16. "hyundai", "kia", "mercedes", "nissan", "opel", "peugeot",
  17. "renault", "seat", "skoda", "toyota", "volkswagen", "volvo",
  18. "jeep", "chevrolet", "mazda", "mitsubishi", "suzuki"
  19. ]
  20. def detect_brands(text: str) -> Dict[str, List[str]]:
  21. """从文本中检测品牌词"""
  22. if not text:
  23. return {"chinese": [], "non_chinese": [], "all": []}
  24. text_lower = text.lower()
  25. found_chinese = []
  26. found_non_chinese = []
  27. for brand in CHINESE_BRANDS:
  28. if re.search(r'\b' + re.escape(brand) + r'\b', text_lower):
  29. found_chinese.append(brand)
  30. for brand in NON_CHINESE_BRANDS:
  31. if re.search(r'\b' + re.escape(brand) + r'\b', text_lower):
  32. found_non_chinese.append(brand)
  33. return {
  34. "chinese": found_chinese,
  35. "non_chinese": found_non_chinese,
  36. "all": found_chinese + found_non_chinese
  37. }
  38. def assess_exclusivity(text: str) -> str:
  39. """
  40. 辅助判断排他协议状态
  41. 返回: 疑似无排他,可合作 / 已代理中国品牌,需评估 / 信息不足,待判断
  42. """
  43. brands = detect_brands(text)
  44. if brands["chinese"]:
  45. return "已代理中国品牌,需评估"
  46. if brands["non_chinese"] and len(brands["non_chinese"]) >= 1:
  47. return "疑似无排他,可合作"
  48. return "信息不足,待判断"
  49. def enrich_record_with_brands(record: Dict[str, Any]) -> Dict[str, Any]:
  50. """给记录添加品牌检测字段"""
  51. combined_text = " ".join([
  52. str(record.get("主营业务", "")),
  53. str(record.get("备注", "")),
  54. str(record.get("客户类型", ""))
  55. ])
  56. brands = detect_brands(combined_text)
  57. record["detected_brands"] = brands["all"]
  58. record["exclusivity_assessment"] = assess_exclusivity(combined_text)
  59. return record