kangshifu-board/scripts/prepare-workshop-self-run-ledger.py

"""Compute rehearsal tables from explicitly reviewed record membership, never keyword counts."""
import csv
import hashlib
import json
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / 'docs/workshop-self-run-20261110'
EVIDENCE = ROOT / 'work/workshop-self-run-20261110/evidence'
load = lambda name: json.loads((EVIDENCE / (name + '.json')).read_bytes())

# Membership assigned by this assistant after reading all 49 full normalized review bodies.
# This is one model's coding, not independent human validation or a second model review.
TAGS = {
    'preparation_positive': {2, 5, 9, 11, 14, 19, 22, 23, 36, 44},
    'no_wash_positive': {2, 5},
    'draining_procedure': {19},
    'draining_complaint': set(),
    'texture_positive': {2, 5, 14, 19, 30, 31, 32, 36},
    'flavour_negative': {47},
}
TAG_NAMES = {'preparation_positive': '明确表达准备/使用便利的正向意见', 'no_wash_positive': '不用洗碗的正向表达',
             'draining_procedure': '明确描述沥水步骤（不是抱怨）', 'draining_complaint': '明确表达沥水困扰',
             'texture_positive': '口感/面条结构正向表达', 'flavour_negative': '明确口味负向表达'}
EXCLUDED = {35: '只有“收到”，无产品意见；不进入主题分母', 48: '只有“收到”，无产品意见；不进入主题分母'}
PENDING = {18: '拌面购买规格与“汤头”描述存在用途歧义；先不用于主题统计',
           49: '仅“🉑”，对本题主题判定含义不足；待核，不当作明确未提及'}

reviews = load('banmian-reviews')['records']
assert len(reviews) == 49 and load('banmian-reviews')['nextCursor'] is None
rows = []
for i, original in enumerate(reviews, 1):
    d = load(f'review-{i:02}')['record']
    assert d['id'] == original['record_id'] and d['content'] == original['content']
    status = 'excluded' if i in EXCLUDED else 'pending' if i in PENDING else 'usable'
    row = {'key': f'E{i:02}', 'entity': 'reviews', 'recordId': d['id'], 'productId': d['product_id'],
           'platform': original['platform'], 'brand': original['brand'], 'purchasedSku': d['purchased_sku_text'],
           'dateRaw': d['review_date_raw'], 'quote': d['content'], 'appendText': d.get('append_content_json', ''),
           'status': status, 'reason': EXCLUDED.get(i) or PENDING.get(i) or '规范化原文/归属可读；仅用于本批自述的主题判断，不验证真实性或性能',
           'tags': [name for name, members in TAGS.items() if i in members],
           'source': d['source'], 'currentDefault': d.get('is_default'), 'historicalIssues': d.get('issues', []),
           'responseFile': f'../../work/workshop-self-run-20261110/evidence/review-{i:02}.json'}
    assert row['status'] == 'usable' or not row['tags']
    rows.append(row)
usable = [r for r in rows if r['status'] == 'usable']
assert len(usable) == 45
duplicate_sets = []
by_text = {}
for r in usable:
    by_text.setdefault(r['quote'].strip(), []).append(r['key'])
for members in by_text.values():
    if len(members) > 1:
        duplicate_sets.append(members)
assert len(by_text) == 43
statistics = []
for platform, pid in dict.fromkeys((r['platform'], r['productId']) for r in rows):
    group = [r for r in usable if r['platform'] == platform and r['productId'] == pid]
    for key, label in TAG_NAMES.items():
        members = [r['key'] for r in group if key in r['tags']]
        texts = {r['quote'].strip() for r in group}
        positive_texts = {r['quote'].strip() for r in group if key in r['tags']}
        statistics.append({'platform': platform, 'productId': pid, 'theme': key, 'label': label,
                           'n': len(members), 'N': len(group), 'percent': round(len(members) / len(group) * 100, 1),
                           'members': members, 'denominatorMembers': [r['key'] for r in group],
                           'distinctTextN': len(texts), 'distinctTextn': len(positive_texts),
                           'unit': '本次已判定的评价记录；不是独立消费者',
                           'reviewState': '同一助手编码和回读；无独立第二模型/人工复核'})
social = []
SOCIAL_USE = {
    1: ('context', '2022年视频简介；可见无锅表述，不等于看完视频或本品牌操作实测'),
    2: ('title-only', '正文“-”；标题推广与营多相关，但不能将营销语当消费者体验'),
    3: ('counterexample', '明确有小锅可煮、无冰箱；是本题无锅范围反例与耐放线索'),
    4: ('context', '明确夜间食堂无饭、找不到热水、借水/被帮助煮的自述；无品牌归属'),
    5: ('adjacent-claim', '热干面相邻品类的推广表达；不迁移为本款拌面性能'),
    6: ('excluded', '明确电煮锅带货佣金，不作为面食消费者需求证词'),
    7: ('title-only', '正文“-”；气味只为题名线索，不形成已核痛点'),
    8: ('context-tags', '主要为汤面/宿舍标签，缺行为细节；不当完整使用事件'),
    9: ('adjacent-context', '2023年面皮体验与室友推荐；相邻替代线索，不证明2026增长'),
    10: ('excluded', '无正文、未看视频，只保留检索索引'),
}
for i in range(1, 11):
    d = load(f'social-{i:02}')['record']
    social.append({'key': f'S{i:02}', 'entity': 'social', 'recordId': d['id'], 'title': d['title'], 'quote': d['content'],
                   'dateRaw': d['published_at_raw'], 'url': d['url'], 'source': d['source'],
                   'use': SOCIAL_USE[i][0], 'reason': SOCIAL_USE[i][1], 'fulltextFlag': d['fulltext_available']})
products = []
for i in range(1, 8):
    d = load(f'product-{i:02}')['record']
    products.append({'key': f'P{i:02}', 'entity': 'products', 'recordId': d['id'], 'title': d['title'], 'url': d['url'],
                     'dateRaw': d.get('source_time_raw'), 'source': d['source'],
                     'prices': [{k: r.get(k) for k in ['record_id', 'amount_minor', 'price_basis', 'source_time_raw', 'source_day']}
                                for r in load(f'product-{i:02}-prices')['records']],
                     'specs': [{k: r.get(k) for k in ['record_id', 'spec_text', 'net_weight_g', 'pack_count', 'spec_verification']}
                               for r in load(f'product-{i:02}-specs')['records']]})
ledger = {'schema': 'ksf.workshop-self-run.evidence.v1', 'dataNature': 'real read-only database records, not fabricated review samples',
          'scope': {'reviewsRetrievedAndRead': 49, 'reviewUsable': 45, 'reviewPending': 2, 'reviewExcluded': 2,
                    'distinctUsableTexts': 43, 'productsDetailed': 7, 'socialCandidatesShown': 50,
                    'socialQueryMatches': 100, 'socialDetailedPurposivelySelected': 10,
                    'socialRepresentativeSample': False, 'newCollection': False, 'sourceArchiveRehashed': False},
          'duplicateTextGroups': duplicate_sets, 'reviews': rows, 'social': social, 'products': products, 'statistics': statistics}
(OUT / '证据与编码.json').write_text(json.dumps(ledger, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
with (OUT / '评价逐条编码.csv').open('w', encoding='utf-8-sig', newline='') as f:
    fields = ['key', 'recordId', 'productId', 'platform', 'purchasedSku', 'dateRaw', 'quote', 'status', 'reason', 'tags', 'sourceRecordId', 'archiveSha256', 'assetSha256', 'lineNumber']
    w = csv.DictWriter(f, fieldnames=fields)
    w.writeheader()
    for r in rows:
        w.writerow({**{k: r[k] for k in fields if k in r and k != 'tags'}, 'tags': ' | '.join(r['tags']),
                    'sourceRecordId': r['source']['id'], 'archiveSha256': r['source'].get('archiveSha256'),
                    'assetSha256': r['source'].get('assetSha256'), 'lineNumber': r['source'].get('line_number')})
with (OUT / '提及率复算.csv').open('w', encoding='utf-8-sig', newline='') as f:
    fields = ['platform', 'productId', 'label', 'n', 'N', 'percent', 'distinctTextn', 'distinctTextN', 'members', 'denominatorMembers']
    w = csv.DictWriter(f, fieldnames=fields)
    w.writeheader()
    for s in statistics:
        w.writerow({k: ' | '.join(s[k]) if isinstance(s[k], list) else s[k] for k in fields})
lines = ['# 真实资料与本地分析索引', '', '全部来自同一正式代际的只读API；E/P/S为本地引用键，不是业务库编号。来源指纹来自服务器回读，本轮未重新下载源ZIP验证。编码为同一助手所作，未经独立第二模型或人工复核。', '',
         '## 覆盖', '', json.dumps(ledger['scope'], ensure_ascii=False), '', '## 商品', '']
for r in products:
    lines += [f"- **{r['key']}** [{r['title']}]({r['url']})；`{r['recordId']}`；来源 `{r['source']['id']}`。"]
lines += ['', '## 社媒（仅定向回读的10条，不计算人群占比）', '']
for r in social:
    lines += [f"### {r['key']}｜{r['title']}", '', f"[{r['recordId']}]({r['url']}) · {r['dateRaw']}", '',
              '> ' + (r['quote'] or '（正文为空）').replace('\n', '\n> '), '', f"用途：{r['reason']}。来源 `{r['source']['id']}`。", '']
lines += ['## 电商评价', '']
for r in rows:
    lines += [f"### {r['key']}｜{r['recordId']}", '', f"{r['platform']} · {r['purchasedSku']} · {r['dateRaw']}", '',
              '> ' + r['quote'].replace('\n', '\n> '), '', f"用途：{r['status']}；{r['reason']}。标签：{'、'.join(r['tags']) or '本题主题未明确提及'}。", '',
              f"来源 `{r['source']['id']}`；原件行号 {r['source'].get('line_number')}；归档指纹 `{r['source'].get('archiveSha256')}`。", '']
(OUT / '证据索引.md').write_text('\n'.join(lines), encoding='utf-8')
print(json.dumps({'ok': True, **ledger['scope'], 'duplicateTextGroups': duplicate_sets,
                  'convenience': [{k: s[k] for k in ['platform', 'productId', 'n', 'N', 'distinctTextn', 'distinctTextN']} for s in statistics if s['theme'] == 'preparation_positive']}, ensure_ascii=False))