"""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))