kangshifu-board/scripts/verify-agent-read-repair.py

"""Local regression and independent read-only public checks; no Agent deployment."""
import contextlib
import importlib.util
import io
import json
from pathlib import Path
import sys
from unittest.mock import patch
import urllib.request

sys.dont_write_bytecode = True
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'server/voc-library'))
import library_channel as current


def main():
    spec = importlib.util.spec_from_file_location('old_skill_channel', ROOT / 'skills/ksf-voc-library/scripts/library_channel.py')
    old = importlib.util.module_from_spec(spec); spec.loader.exec_module(old)
    query = {'entities': ['reviews'], 'text': {'any': ['份量']}}
    comparisons = []
    for module, label in ((old, 'installed-skill-source-before'), (current, 'local-candidate-after')):
        output, error = io.StringIO(), io.StringIO()
        with patch('sys.stdin', io.StringIO('\ufeff' + json.dumps(query))), \
                patch.object(module, 'send', return_value={'ok': True, 'readOnly': True, 'databaseWrites': 0}) as send, \
                contextlib.redirect_stdout(output), contextlib.redirect_stderr(error):
            status = module.read_main(['search', '--channel', 'unused', '--input', '-'])
        comparisons.append(dict(version=label, exitCode=status, sendCalled=send.called,
            result=json.loads(output.getvalue() or error.getvalue())))
    assert comparisons[0]['result']['error']['code'] == 'CHANNEL_READ_FAILED'
    assert not comparisons[0]['sendCalled']
    assert comparisons[1]['result']['ok'] and comparisons[1]['sendCalled']

    def public(request):
        req = urllib.request.Request('https://kangshifu.fmode.cn/api/library',
            data=json.dumps(request).encode(), headers={'Content-Type': 'application/json'}, method='POST')
        with urllib.request.urlopen(req, timeout=40) as response: data = json.load(response)
        assert data['ok'] and data['readOnly'] and data['databaseWrites'] == 0
        return data

    summary = public({'op': 'summary'}); generation = summary['generation']
    detail = public({'op': 'detail', 'entity': 'reviews',
        'id': 'jd:100297147639:106531290235443828', 'generation': generation})
    record = detail['record']
    assert record['review_date_raw'] == '2026-11-06 10:49:00' and record['rating_raw'] == '5'
    assert record['rating_scale'] is None and record['is_default'] is None
    browse = public({'op': 'browse', 'entity': 'reviews', 'filters': {
        'platform': 'jd', 'brand': 'ksf', 'category': 'banmian'},
        'keyword': '味道', 'limit': 5, 'generation': generation})
    assert browse['generation'] == detail['generation'] == generation
    result = dict(ok=True, bomRegression=comparisons, historicalFailureRootCauseConfirmed=False,
        publicRead=dict(generation=generation, detail={k: record[k] for k in (
            'id', 'platform_family', 'review_date_raw', 'review_calendar_date', 'rating_raw',
            'rating_scale', 'is_default', 'source_record_id')},
            browseTotal=browse['totalMatched'], effectiveQuery=browse['effectiveQuery']),
        databaseWrites=0, paidCalls=0, deployedToAgent=False, originalQueryVerified=False)
    output = ROOT / 'docs/agent-library-read-repair-20261108/local-verification.json'
    output.write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    print(json.dumps(result, ensure_ascii=False))


if __name__ == '__main__': main()