business-assistant/src/step3_question_answer/dms_company_check.py

"""Explicit read-only DMS integration check; output contains no company values."""

import json
from pathlib import Path

from .dms_company import BASIC_ID, HONOR_ID, create_dms_company_client
from .company import CompanyClient
from .graph import build_graph
from src.step5_session_analysis.dms import encode


def main():
    path = Path(__file__).resolve().parents[2] / '.env'
    client = create_dms_company_client(path)
    client.connect()
    def seed(identity):
        return client.call('/content/selectContentList', {
            'columnId': identity, 'page': 0, 'pageSize': 1, 'states': 0,
            'search': '[]', 'orderBy': encode([{'field': 'id', 'orderByType': 1}])})['data'][0]
    row = seed(BASIC_ID)
    result = client.search(row['c_name'])
    assert result['exact_unique'] and HONOR_ID not in client.columns
    company = result['candidates'][0]
    profile = client.basic(company)
    assert HONOR_ID not in client.columns
    assert profile['data']['CreditCode'] == row['c_credit_code']
    assert profile['source']['provider'] == 'DMS'
    assert client.search(row['c_credit_code'])['exact_unique']
    honors = client.honors(company)
    assert honors['complete']
    fuzzy = client.search(row['c_name'][:4])
    assert 1 <= len(fuzzy['candidates']) <= 3 and fuzzy['can_search_more']
    empty = client.search('不存在的合成校验企业XYZ123')
    assert not empty['candidates'] and empty['complete']

    # Test fixture discovery only: locate a populated honor owner in memory.
    # The production graph below still resolves its basic identity first.
    honor_seed = seed(HONOR_ID)
    populated = client.search(honor_seed['c_credit_code'])
    assert populated['exact_unique']
    selected = populated['candidates'][0]
    client.basic(selected)
    multi = client.honors(selected)
    assert multi['complete'] and len(multi['records']) > 1
    assert all(r['data']['Name'] != selected['Name'] for r in multi['records'])
    def no_qcc():
        raise AssertionError('unexpected_qcc_call')
    combined = CompanyClient(path, dms_factory=lambda: create_dms_company_client(path), qcc_factory=no_qcc)
    graph = build_graph(combined, lambda q: {'conversation_kind': 'specific_need',
                        'needs_company': True, 'company_hint': q})
    answer = graph.invoke({'question': selected['Name']}, {'configurable': {'thread_id': 'read-only-dms-check'}})
    assert '__interrupt__' not in answer and answer['profile']['source']['provider'] == 'DMS'
    assert answer['honors']['complete'] and len(answer['honors']['records']) > 1
    print(json.dumps({'exact_name': True, 'exact_credit_code': True, 'basic_before_honors': True,
                      'fuzzy_candidates': len(fuzzy['candidates']), 'fuzzy_complete': fuzzy['complete'],
                      'empty_search': True, 'seed_honor_count': len(honors['records']),
                      'multiple_honor_count': len(multi['records']), 'graph_without_qcc': True,
                      'business_records_written': False}, ensure_ascii=False))


if __name__ == '__main__':
    try:
        main()
    except Exception:
        print(json.dumps({'status': 'dms_company_check_failed'}))
        raise SystemExit(1) from None