business-assistant/scripts/check-bge-knowledge.py

"""Read-only migration checks. Output only counts/assertions, never source bodies."""

from collections import Counter
import json
from pathlib import Path
import sys

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from openpyxl import load_workbook
from src.step2_knowledge_ingest.ingest import load
from src.step3_question_answer.retrieval import Retriever


def main():
    prepared = ROOT / 'knowledge/prepared'
    records = load(prepared)
    excel = [r for r in records if r['raw_record'].get('workbook_source')]
    counts = Counter(r['raw_record']['workbook_source']['sheet'] for r in excel)
    assert sorted(counts.values()) == [11, 16, 26]
    workbook = load_workbook(ROOT / excel[0]['source']['file'], data_only=False)
    try:
        for sheet_name in counts:
            sheet = workbook[sheet_name]
            represented = {}
            for record in excel:
                source = record['raw_record']['workbook_source']
                if source['sheet'] == sheet_name:
                    for row in source['rows']:
                        represented.update(row['cells'])
            expected = {cell.coordinate: cell.value for row in sheet.iter_rows(min_row=4)
                        for cell in row if cell.value is not None}
            assert all(represented.get(k) == v for k, v in expected.items())
    finally:
        workbook.close()
    retriever = Retriever(ROOT / '.env', prepared / 'ingestion/manifest.json')
    checks = []
    try:
        selected = [next(r for r in excel if r['raw_record']['workbook_source']['sheet'] == sheet) for sheet in counts]
        selected += [next(r for r in records if r['kind'] == kind) for kind in ('policy', 'finance', 'qa')]
        for record in selected:
            query = record['data']['question'] if record['kind'] == 'qa' else record['title']
            result = retriever.search(query)
            checks.append({'complete': result['complete'], 'expected_record_found': any(
                candidate['record']['id'] == record['id'] for candidate in result['candidates'])})
        assert all(c['complete'] and c['expected_record_found'] for c in checks)
    finally:
        retriever.close()
    report = {'records': len(records), 'workbook_counts': dict(counts), 'nonempty_cells_preserved': True,
              'retrieval_checks': checks, 'scope': 'exact-title retrieval smoke; not recall-quality evaluation'}
    path = ROOT / 'output/checks/bge-knowledge.json'
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    print(json.dumps(report, ensure_ascii=False))


if __name__ == '__main__':
    main()