124 lines
3.8 KiB
Python
124 lines
3.8 KiB
Python
|
"""公司知识库模块 — 公司匹配和证据排序。"""
|
|||
|
|
import re
|
|||
|
|
from typing import Optional
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _normalize_loose(input_str: str) -> str:
|
|||
|
|
"""模糊标准化：小写 + 去除括号/分隔符。"""
|
|||
|
|
s = input_str.lower()
|
|||
|
|
s = re.sub(r"[\s\-_·.（）()\[\]{}]", "", s)
|
|||
|
|
s = s.replace("x", "")
|
|||
|
|
return s
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _score_profile(input_str: str, aliases: list[str], tags: list[str]) -> int:
|
|||
|
|
"""评分公司匹配度。"""
|
|||
|
|
normalized = _normalize_loose(input_str)
|
|||
|
|
score = 0
|
|||
|
|
for alias in aliases:
|
|||
|
|
if _normalize_loose(alias) in normalized or normalized in _normalize_loose(alias):
|
|||
|
|
score += 4
|
|||
|
|
for tag in tags:
|
|||
|
|
if _normalize_loose(tag) in normalized:
|
|||
|
|
score += 2
|
|||
|
|
return score
|
|||
|
|
|
|||
|
|
|
|||
|
|
def match_company_knowledge(
|
|||
|
|
product_name: Optional[str] = None,
|
|||
|
|
company_hint: Optional[str] = None,
|
|||
|
|
forced_company_id: Optional[str] = None,
|
|||
|
|
companies: list[dict] = None,
|
|||
|
|
products: list[dict] = None,
|
|||
|
|
) -> dict:
|
|||
|
|
"""匹配公司知识库。
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
product_name: 产品名称
|
|||
|
|
company_hint: 公司名称提示
|
|||
|
|
forced_company_id: 强制指定公司 ID
|
|||
|
|
companies: 公司配置列表
|
|||
|
|
products: 产品配置列表
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
CompanyKnowledgeMatch dict
|
|||
|
|
"""
|
|||
|
|
companies = companies or []
|
|||
|
|
products = products or []
|
|||
|
|
|
|||
|
|
# 1. 强制指定
|
|||
|
|
if forced_company_id:
|
|||
|
|
company = next((c for c in companies if c.get("id") == forced_company_id), None)
|
|||
|
|
if company:
|
|||
|
|
return {
|
|||
|
|
"companyId": company["id"],
|
|||
|
|
"companyName": company.get("displayName", ""),
|
|||
|
|
"evidenceFiles": [],
|
|||
|
|
"confidence": 1.0,
|
|||
|
|
"matchedBy": "company_forced",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 2. 产品目录匹配
|
|||
|
|
if product_name:
|
|||
|
|
normalized = product_name.lower()
|
|||
|
|
for prod in products:
|
|||
|
|
aliases = prod.get("aliases", [])
|
|||
|
|
if any(_normalize_loose(a) in normalized for a in aliases):
|
|||
|
|
company_id = prod.get("companyId")
|
|||
|
|
company = next((c for c in companies if c.get("id") == company_id), None)
|
|||
|
|
if company:
|
|||
|
|
return {
|
|||
|
|
"companyId": company["id"],
|
|||
|
|
"companyName": company.get("displayName", ""),
|
|||
|
|
"evidenceFiles": [],
|
|||
|
|
"confidence": 1.0,
|
|||
|
|
"matchedBy": "product_catalog",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 3. 别名/标签模糊匹配
|
|||
|
|
best_match = None
|
|||
|
|
best_score = 0
|
|||
|
|
search_text = f"{product_name or ''} {company_hint or ''}"
|
|||
|
|
|
|||
|
|
for company in companies:
|
|||
|
|
aliases = company.get("aliases", [])
|
|||
|
|
tags = company.get("aliases", []) # 用别名作为标签
|
|||
|
|
score = _score_profile(search_text, aliases, tags)
|
|||
|
|
if score > best_score:
|
|||
|
|
best_score = score
|
|||
|
|
best_match = company
|
|||
|
|
|
|||
|
|
if best_match and best_score > 0:
|
|||
|
|
confidence = min(0.95, 0.45 + best_score / 20)
|
|||
|
|
return {
|
|||
|
|
"companyId": best_match["id"],
|
|||
|
|
"companyName": best_match.get("displayName", ""),
|
|||
|
|
"evidenceFiles": [],
|
|||
|
|
"confidence": confidence,
|
|||
|
|
"matchedBy": "company_alias",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return {
|
|||
|
|
"companyId": "",
|
|||
|
|
"companyName": "",
|
|||
|
|
"evidenceFiles": [],
|
|||
|
|
"confidence": 0,
|
|||
|
|
"matchedBy": "unknown",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def expected_company_id_for_product(
|
|||
|
|
product_name: Optional[str],
|
|||
|
|
products: list[dict] = None,
|
|||
|
|
) -> Optional[str]:
|
|||
|
|
"""通过产品名查找对应公司 ID。"""
|
|||
|
|
if not product_name:
|
|||
|
|
return None
|
|||
|
|
products = products or []
|
|||
|
|
normalized = product_name.lower()
|
|||
|
|
for prod in products:
|
|||
|
|
aliases = prod.get("aliases", [])
|
|||
|
|
if any(_normalize_loose(a) in normalized for a in aliases):
|
|||
|
|
return prod.get("companyId")
|
|||
|
|
return None
|