proposa-ai/phase1_parsing/analyze_similarity.py

"""
Phase 1: 历史招投标相似度分析

对比新招标文件与各历史招标文件的相似度，
决定每组历史数据对最终投标文件的影响权重。

用法:
    from phase1_parsing.analyze_similarity import calculate_similarity
    results = calculate_similarity(new_tender_text, historical_groups)
"""

import sys
import os
import json
import re

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from llm.llm_utils import LLM


def extract_tender_features(text: str) -> dict:
    """从招标文本中提取关键特征用于相似度比较"""
    features = {}

    # 项目类型判断
    type_keywords = {
        "物业管理": ["物业", "保洁", "保安", "绿化", "维修", "会务", "客服"],
        "工程装修": ["装修", "改造", "修缮", "施工", "工程"],
        "IT服务": ["软件", "系统", "信息化", "开发", "运维", "网络安全"],
        "物资采购": ["采购", "设备", "货物", "供应", "物资"],
        "咨询服务": ["咨询", "评估", "审计", "监理", "设计"],
        "餐饮服务": ["餐饮", "食堂", "供餐", "配餐"],
    }
    matched_types = []
    for ttype, kws in type_keywords.items():
        if any(kw in text[:8000] for kw in kws):
            matched_types.append(ttype)
    features["project_types"] = matched_types

    # 采购人（同一采购人很可能项目相似）
    m = re.search(r"采购人[：:]\s*(.{2,30})", text[:5000])
    if m:
        features["buyer"] = m.group(1).strip()

    # 预算规模
    m = re.search(r"预算[^\n]{0,10}[：:]\s*([\d,.]+\s*万)", text[:10000])
    if m:
        budget_str = m.group(1).strip().replace(",", "")
        try:
            budget_val = float(budget_str.replace("万", ""))
            features["budget"] = budget_val
            if budget_val > 1000:
                features["budget_level"] = "大型"
            elif budget_val > 200:
                features["budget_level"] = "中型"
            else:
                features["budget_level"] = "小型"
        except ValueError:
            pass

    # 服务期限
    m = re.search(r"服务期限[：:](.{0,40})", text[:5000])
    if m:
        features["service_period"] = m.group(1).strip()

    # 地域
    m = re.search(r"([一-鿿]{2,8}(?:区|市|县))", text[:3000])
    if m:
        features["region"] = m.group(1).strip()

    # 关键需求关键词密度
    demand_kw = ["服务方案", "质量标准", "人员配置", "管理制度", "应急预案",
                 "考核", "培训", "设备", "安全", "合同"]
    found_kw = [kw for kw in demand_kw if kw in text]
    features["demand_keywords"] = found_kw

    return features


def calculate_similarity(
    new_tender_text: str,
    historical_groups: list,
    use_llm: bool = True,
) -> list:
    """
    计算新招标文件与每组历史招标文件的相似度。

    Args:
        new_tender_text: 新招标文件全文
        historical_groups: [
            {
                "group_id": "project_A",
                "name": "项目名称",
                "tender_text": "历史招标文本",
                "bid_text": "对应投标文本",
                "tender_file": "原始PDF路径",
                "bid_file": "原始PDF路径",
            },
            ...
        ]
        use_llm: 是否用 LLM 做语义级评分（否则只用关键词匹配）

    Returns:
        [
            {
                "group_id": "project_A",
                "name": "项目名称",
                "similarity_score": 0.85,    # 0~1
                "similarity_level": "高",     # 高/中/低
                "similarity_reason": "同为政府机关物业管理...",
                "new_tender_features": {...},
                "history_tender_features": {...},
                "bid_structure": {...},       # 对应投标的目录结构
                "weight": 0.6,                # 归一化后的权重
            },
            ...
        ]
    """
    new_features = extract_tender_features(new_tender_text)
    results = []

    for group in historical_groups:
        hist_features = extract_tender_features(group.get("tender_text", ""))

        if use_llm:
            score, reason = _llm_similarity_score(
                new_tender_text[:8000],
                group.get("tender_text", "")[:8000],
            )
        else:
            score, reason = _keyword_similarity(new_features, hist_features)

        # 提取投标目录
        from phase1_parsing.extract_bid_toc import extract_toc
        bid_toc = extract_toc(group.get("bid_text", ""))

        # 判断相似等级
        if score >= 0.7:
            level = "高"
        elif score >= 0.4:
            level = "中"
        else:
            level = "低"

        results.append({
            "group_id": group["group_id"],
            "name": group.get("name", group["group_id"]),
            "similarity_score": round(score, 4),
            "similarity_level": level,
            "similarity_reason": reason,
            "new_tender_features": new_features,
            "history_tender_features": hist_features,
            "bid_toc": bid_toc,
            "bid_text_len": len(group.get("bid_text", "")),
            "tender_text_len": len(group.get("tender_text", "")),
        })

    # 归一化权重：相似度越高权重越大
    scores = [r["similarity_score"] for r in results]
    total = sum(scores) or 1
    for r in results:
        r["weight"] = round(r["similarity_score"] / total, 4)

    results.sort(key=lambda x: x["similarity_score"], reverse=True)
    return results


def _keyword_similarity(new_feat: dict, hist_feat: dict) -> tuple:
    """基于关键词匹配的相似度计算（快速但粗糙）"""
    score = 0.0
    reasons = []

    # 项目类型匹配
    new_types = set(new_feat.get("project_types", []))
    hist_types = set(hist_feat.get("project_types", []))
    if new_types and hist_types:
        overlap = new_types & hist_types
        if overlap:
            score += 0.3
            reasons.append(f"类型匹配: {overlap}")

    # 采购人匹配
    if (new_feat.get("buyer") and hist_feat.get("buyer")
            and new_feat["buyer"] == hist_feat["buyer"]):
        score += 0.25
        reasons.append("同一采购人")

    # 预算规模匹配
    if new_feat.get("budget_level") and hist_feat.get("budget_level"):
        if new_feat["budget_level"] == hist_feat["budget_level"]:
            score += 0.15
            reasons.append(f"预算规模相当({new_feat['budget_level']})")

    # 需求关键词重叠
    new_kw = set(new_feat.get("demand_keywords", []))
    hist_kw = set(hist_feat.get("demand_keywords", []))
    if new_kw and hist_kw:
        overlap = new_kw & hist_kw
        if overlap:
            kw_score = len(overlap) / max(len(new_kw), 1) * 0.2
            score += kw_score
            reasons.append(f"需求关键词重叠: {overlap}")

    # 地域匹配
    if (new_feat.get("region") and hist_feat.get("region")
            and new_feat["region"] == hist_feat["region"]):
        score += 0.1
        reasons.append("同区域")

    score = min(score, 1.0)
    reason = "；".join(reasons) if reasons else "无明显相似特征"
    return score, reason


def _llm_similarity_score(new_excerpt: str, hist_excerpt: str) -> tuple:
    """用 LLM 评估两个招标文件的相似度（语义级）"""
    try:
        llm = LLM()
        system = (
            "你是一个招标文件比对专家。比较两份招标文件，输出相似度评分(0~1)和理由。\n"
            "输出 JSON: {\"score\": 0.85, \"reason\": \"同类型项目，服务内容重叠度高\"}"
        )
        user = (
            "请比较以下两份招标文件的相似度，考虑项目类型、服务内容、预算规模、采购人等因素：\n\n"
            f"【新招标文件】\n{new_excerpt[:4000]}\n\n"
            f"【历史招标文件】\n{hist_excerpt[:4000]}"
        )
        result = llm._call(system, user, max_tokens=1024, temperature=0.2,
                           response_format={"type": "json_object"})
        parsed = json.loads(result)
        return float(parsed.get("score", 0.5)), parsed.get("reason", "")
    except Exception as e:
        return 0.5, f"LLM评分异常: {e}"


if __name__ == "__main__":
    import sys
    sys.stdout.reconfigure(encoding="utf-8")

    # 简单测试
    text1 = "项目名称：松江区机关事务管理局物业管理服务 预算金额：1739.51万元 服务内容：保洁、保安、绿化、会务"
    text2 = "项目名称：嘉定区机关事务管理局物业管理服务 预算金额：1200万元 服务内容：保洁、保安、绿化"

    groups = [{
        "group_id": "test",
        "name": "测试项目",
        "tender_text": text2,
        "bid_text": "第一章 投标人资格...",
    }]

    results = calculate_similarity(text1, groups, use_llm=False)
    for r in results:
        print(f"{r['name']}: 相似度 {r['similarity_score']} ({r['similarity_level']})")
        print(f"  理由: {r['similarity_reason']}")
        print(f"  权重: {r['weight']}")