business-assistant/src/step3_question_answer/matching.py

"""Literal subject/policy comparisons used to prioritize, never approve, recommendations."""

from datetime import date
from .dialogue import user_history


MATCH_PROMPT = '''
另输出match_basis数组，提供主体侧依据与政策条款的比较：
[{dimension:需求/地区/行业/资质/主体/时间/其他,fact_evidence_id:subject_facts编号,
policy_evidence_id:本批evidence编号,status:supports/conflicts/unknown,explanation:简短比较结论}]。
supports表示有针对性的相关线索，不表示已满足全部资格；conflicts用于明确冲突；缺失信息或无法推出结论用unknown。
优先比较已获取的工商信息、荣誉及用户自述个人信息，不得仅以“是企业”“存续”“都在青浦”作为推荐理由。
公司的荣誉认定与原文资质要求可形成较强线索，但需核对有效期及其他条件；过期或日期未知不能写成现行有效资质。
经营范围仅为登记范围，不能直接证明实际项目、行业收入占比或其他资格。注册地址不等于项目地点、户籍或纳税地。
用户只想找惠企政策时，结合已有企业信息寻找确有关系的方向，不凭空猜测其正计划融资、招人或投资。
泛问“有什么政策”本身不算具体需求匹配，不得仅凭这句话填写需求supports；需比较实际工商或荣誉线索。
个人自述的地区、就业或创业情况可以比较相应条件；不存在的个人字段不得猜测。
没有足够主体事实支持针对性比较时返回空数组。纯政策了解无需做用户资格比较，可返回空数组。
reason必须概括具体的主体事实与政策对应关系以及关键缺口，不得只复述原问题或泛称相关。
严格区分两个数组：conditions是资格核验，dimension仅地区/主体/行业/时间/其他条件，status仅满足/不满足/待确认；
match_basis是相关依据比较，dimension仅需求/地区/行业/资质/主体/时间/其他，status仅supports/conflicts/unknown。
两者枚举不可混用；需求和资质不是conditions的dimension，英文状态不能用于conditions。
'''


def subject_facts(handoff):
    facts = [{'path': '/question', 'text': handoff['original_question'], 'origin': 'user',
              'provenance': {'type': 'user_statement'}}]
    facts.extend({'path': f'/history/{i}/question', 'text': text, 'origin': 'user',
                  'provenance': {'type': 'user_statement', 'context': 'same_thread_history'}}
                 for i, text in enumerate(user_history(handoff)))
    facts.extend({'path': '/attachments/' + row['id'], 'text': row['excerpt'], 'origin': 'attachment',
                  'provenance': {'type': 'user_attachment', 'verified': False, 'excerpt': True}}
                 for row in (handoff.get('conversation_context') or {}).get('attachments', []))
    profile = (handoff.get('profile') or {}) if handoff.get('needs_company') else {}
    source = profile.get('source') or {}
    provenance = {k: source[k] for k in ('api_code', 'fetched_at') if k in source}

    def visit(value, path, origin, metadata):
        if isinstance(value, dict):
            for key, child in value.items():
                visit(child, path + '/' + str(key).replace('~', '~0').replace('/', '~1'), origin, metadata)
        elif value is not None and str(value).strip():
            facts.append({'path': path, 'text': str(value), 'origin': origin, **metadata})

    for field in ('Area', 'Address', 'Scope', 'EconKind', 'Status', 'StartDate', 'RegistCapi'):
        visit(profile.get('data', {}).get(field), '/company/' + field, 'company', {'provenance': provenance})
    honors = (handoff.get('honors') or {}) if handoff.get('needs_company') else {}
    for index, row in enumerate(honors.get('records', [])):
        data = row['data']
        current = False
        try:
            start = date.fromisoformat(str(data.get('BegingDate'))[:10])
            end = date.fromisoformat(str(data.get('DeadLine'))[:10])
            current = start <= date.today() <= end
        except ValueError:
            pass
        source = row.get('source') or {}
        metadata = {'provenance': {k: source[k] for k in ('api_code', 'fetched_at', 'page_index') if k in source},
                    'current_validity_evidenced': current}
        for field in ('Name', 'Level', 'BegingDate', 'DeadLine'):
            visit(data.get(field), f'/honors/{index}/{field}', 'honor', metadata)
    return facts


def validate_basis(value, policy_units, facts):
    if not isinstance(value, list):
        raise ValueError('invalid_match_basis')
    result = []
    for row in value:
        if not isinstance(row, dict):
            raise ValueError('invalid_match_dimension')
        row = dict(row)
        if not isinstance(row.get('dimension'), str):
            raise ValueError('invalid_match_dimension')
        row['dimension'] = {'其他条件': '其他', '资格': '资质'}.get(row.get('dimension'), row.get('dimension'))
        if row.get('dimension') not in ('需求', '地区', '行业', '资质', '主体', '时间', '其他'):
            raise ValueError('invalid_match_dimension')
        if row.get('status') not in ('supports', 'conflicts', 'unknown'):
            raise ValueError('invalid_match_status')
        if any(not isinstance(row.get(k), str) or not row[k].strip() for k in ('fact_quote', 'policy_quote', 'explanation')):
            raise ValueError('invalid_match_text')
        fact = next((f for f in facts if row.get('fact_path') == f['path'] and row['fact_quote'] in f['text']), None)
        if fact is None or not any(row.get('policy_path') == unit['path'] and row['policy_quote'] in unit['text'] for unit in policy_units):
            raise ValueError('invented_match_evidence')
        clean = {k: row[k] for k in ('dimension', 'status', 'fact_path', 'fact_quote', 'policy_path', 'policy_quote', 'explanation')}
        clean['fact_origin'] = fact['origin']
        clean['provenance'] = fact['provenance']
        if fact['origin'] == 'attachment' and clean['status'] == 'supports':
            clean['status'] = 'unknown'
            clean['explanation'] += '用户附件尚未独立核实，不能据此确认资格。'
        if fact['origin'] == 'honor' and not fact.get('current_validity_evidenced') and clean['status'] == 'supports':
            clean['status'] = 'unknown'
            clean['explanation'] += '该荣誉当前有效性没有完整日期证据，不能视为当前资质符合。'
        result.append(clean)
    return result


def priority_key(item):
    """Evidence coverage, not an approval score; stable ties keep recall order."""
    supported = [b for b in item.get('match_basis', []) if b['status'] == 'supports']
    strong = {b['dimension'] for b in supported if b['dimension'] in ('需求', '行业', '资质')}
    company_or_honor = {b['dimension'] for b in supported if b['fact_origin'] in ('company', 'honor')
                        and b['dimension'] in ('行业', '资质')}
    return ('需求' in strong), len(company_or_honor), len(strong), len({b['dimension'] for b in supported})