kangshifu-board/scripts/audit-library-cleanup.py

"""Conservative, evidence-bearing triage of the current research inventory."""
from collections import Counter
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
import re
import unicodedata
import os

ROOT = Path(__file__).resolve().parents[1]
WORK = ROOT / os.environ.get('KSF_CLEANUP_WORK','work/library-cleanup-20261111')


def canonical(value): return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'))
def sha(value): return hashlib.sha256(value.encode()).hexdigest()
def clean(value): return re.sub(r'\s+', '', unicodedata.normalize('NFKC', str(value or ''))).strip()


def meaningless_body(value, title):
    body = clean(value)
    stripped = re.sub(r'[^\w\u3400-\u9fff]', '', body)
    if not stripped or body.lower() in {'null','none','undefined','n/a','无','暂无','暂无简介','暂无描述','暂无内容','无简介','未填写','简介','视频简介','不详','网络','转载','自制','转自网络','。','-', '--', '...','待补充'}:
        return 'EMPTY_OR_PLACEHOLDER_BODY'
    if body == clean(title): return 'BODY_REPEATS_TITLE'
    if re.fullmatch(r'https?://\S+', str(value or '').strip()): return 'LINK_ONLY_BODY'
    if re.fullmatch(r'(?:#[^#\s]+\s*)+', str(value or '').strip()): return 'HASHTAGS_ONLY_BODY'
    return None


def main():
    inventory = json.loads((WORK / 'inventory-result.json').read_bytes())['body']['result']
    # pg-promise serializes bigint as text. Bind plans to the actual typed rows,
    # and never mistake the string "0" for an enabled fulltext flag.
    definitions = {t['name']:t['columns'] for t in inventory['manifest']['tables']}
    for entity,table in [('products','products'),('social','social_posts')]:
        for row in inventory[entity]:
            for field in definitions[table]:
                value=row.get(field['name'])
                if value is not None and 'INT' in field['type'].upper():
                    row[field['name']]=int(value)
    protected = set()
    for b in ['five-topics-public-01','five-topics-public-02','five-topics-public-03']:
        protected.update(r['id'] for r in json.loads((ROOT/'work/database-enrichment-20261110/batches'/b/'expected-records.json').read_bytes()))
    removals, repairs, review = [], [], []
    core_food = re.compile(r'方便面|泡面|拌面|汤面|拉面|速食面|即食面|面饼|干脆面|挂面|面条|面皮|热干面|乌冬|乌龙面|河粉|米粉|米线|螺蛳粉|酸辣粉|酱|调料|调味|配料|面粉')
    beverages = re.compile(r'豆奶|豆乳|奶茶|果汁|饮料|饮品|绿茶|冰红茶|柠檬水|咖啡|凉茶|海之言|鲜橙多|橙汁|苹果汁|奶绿|冰糖雪梨|金桔柠檬|葡萄汁|无糖茶|乌龙|茶里王|双萃|蜜桃多|果漾|矿泉水|椰子水|果蔬汁|番茄汁|酸梅汤|春拂|爱夸')
    hard_offtopic = re.compile(r'街霸|街头霸王|最强杰米|无期迷途|[Pp]higros|猫咪大战争|喵星人大战|红莲的弓矢|鬼灭之刃|空耳翻唱|中文进食曲|原创编舞|花衬衫和油泼面|新歌|翻唱|伴奏|速食面刷新|方便面滑板|泡面滑板|502胶水|修复万物|泡面能复制|郭德纲|于谦|家有儿女|风犬少年的天空|波兰球|AI全民制作人|AI剧情|泡面半生的代价|赛博人物志')
    protection = re.compile(r'配料|口感|口味|汤底|面饼|面条|料包|份量|分量|包装|制作|烹饪|工艺|营养|钠|蛋白|保质期|储存|囤货|非油炸|干拌|酸甜|麻辣|咸|辣度')
    generic = re.compile(r'^(?:方便面|泡面|速食面|拌面|汤面|拉面|炸酱面|担担面|热干面|葱油拌面|油泼面|白象方便面|康师傅方便面|统一方便面|今天吃面|今天吃泡面|吃面|吃泡面|嗦面|一碗面|面面面)+$')
    # Individually reviewed against the original title/tags, never tags alone.
    reviewed_irrelevant = {
        'bilibili:BV1BSJyzxELa','bilibili:BV1HXHm6ZEG2','bilibili:BV1JL41187mi',
        'bilibili:BV1Jnno6mEwC','bilibili:BV1PC4y1F7Qi','bilibili:BV1QaoMBQE22',
        'bilibili:BV1W2xS6LEur','bilibili:BV1bV4367E6g','bilibili:BV1f741177xs',
        'bilibili:BV1k24y1c7TA','bilibili:BV1kr9mBMEMz','bilibili:BV1rK4y1V7EU',
        'bilibili:BV1v8x96kEwy','bilibili:BV1yKNG6sETM',
        'bilibili:BV111421B75m','bilibili:BV1AnQLB1ECA','bilibili:BV1CU99BoECj',
        'bilibili:BV1CX4y1y7ED','bilibili:BV1DG4y1B7qu','bilibili:BV1GTeP69ES4',
        'bilibili:BV1Kse466ELq','bilibili:BV1LKMn6uETD','bilibili:BV1Si4y1E72L',
        'bilibili:BV1UTXnYVEsj','bilibili:BV1WEBMBoE4G','bilibili:BV1Ye6KYLEi9',
        'bilibili:BV1gv2MBsES6','bilibili:BV1hT4y1A7C2','bilibili:BV1yoVo6qEY1',
    }

    def entry(row, entity, reason, evidence):
        # Hash only stable entity columns, excluding the audit-only source join.
        before = {k:v for k,v in row.items() if k not in ('payload_sha256','source_payload','_ksf_source_rowid')}
        return {'entity':entity,'id':row['id'],'title':row.get('title'),'reason':reason,'evidence':evidence,
                'sourceRecordId':row['source_record_id'],'sourcePayloadSha256':row['payload_sha256'],
                'recordSha256':sha(canonical(before)), 'before':before}

    for row in inventory['products']:
        title = row['title'] or ''
        if not clean(title) or clean(title) in {'商品','测试商品','暂无','未知','null','None'}:
            removals.append(entry(row,'products','INVALID_PRODUCT_TITLE',{'title':title}))
        elif beverages.search(title) and not core_food.search(title):
            removals.append(entry(row,'products','UNRELATED_STANDALONE_BEVERAGE',{'title':title,'matched':beverages.search(title).group(0)}))
        elif re.search(r'补邮专用|邮费链接',title) and not core_food.search(title):
            removals.append(entry(row,'products','NON_PRODUCT_POSTAGE_LINK',{'title':title}))
        elif row['category_code'] is None and not core_food.search(title):
            review.append({'entity':'products','id':row['id'],'title':title,'reason':'UNCATEGORIZED_REQUIRES_SEMANTIC_REVIEW'})
    for row in inventory['social']:
        if row['id'] in protected: continue
        title, body = row['title'] or '', row['content'] or ''
        source = json.loads(row['source_payload'])
        tags = str(source.get('tags') or '')
        missing = meaningless_body(body,title)
        title_norm = re.sub(r'[\W_]+','',title)
        off = hard_offtopic.search(title)
        reason = None
        if row['id']=='bilibili:BV11ENX6BE6p': reason='USER_IDENTIFIED_EMPTY_ENTERTAINMENT'
        elif row['id'] in reviewed_irrelevant: reason='REVIEWED_FICTION_GAME_OR_CRAFT'
        elif off: reason='UNRELATED_GAME_MUSIC_DRAMA_OR_DIY'
        elif generic.fullmatch(title_norm) and (missing or len(clean(body))<12): reason='GENERIC_TITLE_WITHOUT_RESEARCH_EVIDENCE'
        elif re.fullmatch(r'吃遍全球所有泡面，成为世界泡面先生【第\d+天】',title) and missing:
            reason='GENERIC_TITLE_WITHOUT_RESEARCH_EVIDENCE'
        elif not clean(title) and missing: reason='EMPTY_RESEARCH_RECORD'
        if reason:
            removals.append(entry(row,'social',reason,{'title':title,'content':body,'tags':tags,'matched':off.group(0) if off else None}))
        elif missing:
            if row['fulltext_available'] != 0:
                repairs.append({**entry(row,'social',missing,{'title':title,'content':body,'tags':tags}),
                                'changes':{'content':None,'fulltext_available':0}})
            review.append({'entity':'social','id':row['id'],'title':title,'content':body,'tags':tags,
                           'reason':missing,'informativeTitle':bool(protection.search(title))})
        elif len(clean(body))<15 and not protection.search(title+' '+body):
            review.append({'entity':'social','id':row['id'],'title':title,'content':body,'tags':tags,'reason':'SHORT_NON_SUBSTANTIVE_BODY'})
    result = {'schema':'ksf.library-cleanup.audit.v1','generation':inventory['generation'],
              'checkedAt':datetime.now(timezone.utc).isoformat(),'scope':{'products':len(inventory['products']),'social':len(inventory['social'])},
              'protectedDedicatedArticles':len(protected),'proposedRemovals':removals,'proposedRepairs':repairs,'requiresReview':review,
              'counts':{'removeByEntity':dict(Counter(r['entity'] for r in removals)),
                        'removeByReason':dict(Counter(r['reason'] for r in removals)),
                        'repairFulltextFlags':len(repairs),'review':len(review)},
              'limits':['缺少价格、评论、图片或品牌分类不自动删除有效商品。','访问受限不等于原内容删除。',
                        '保留面体原料、调味工艺和包装研发线索；不把原料当即食竞品。','纯标题的视频线索不当作已获取正文或消费者证言。']}
    (WORK/'audit-proposals.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf8')
    for name,rows in [('proposed-removals',removals),('requires-review',review)]:
        (WORK/(name+'.jsonl')).write_text('\n'.join(json.dumps({k:v for k,v in r.items() if k not in ('before','recordSha256','sourcePayloadSha256')},ensure_ascii=False) for r in rows),encoding='utf8')
    print(json.dumps(result['counts'],ensure_ascii=False))


if __name__=='__main__': main()