kangshifu-board/scripts/collect-five-topic-supplement.py

"""Supplement sparse noodle topics with archived and public original articles.

Search results are discovery only. Short community posts keep their full actual
body; no minimum word count is padded with recommendations or page furniture.
"""
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
import hashlib
import importlib.util
import json
from pathlib import Path
import re
import time
import urllib.parse
from lxml import html

ROOT = Path(__file__).resolve().parents[1]
ORIGINAL_WORK = ROOT / 'work/five-topic-collection-20261110'
WORK = ORIGINAL_WORK / 'supplement-01'
PUBLIC = WORK / 'public-sources'
SEARCH = WORK / 'search'
for directory in [WORK, PUBLIC, SEARCH]: directory.mkdir(parents=True, exist_ok=True)
spec = importlib.util.spec_from_file_location('base_public', ROOT / 'scripts/collect-five-topic-public.py')
base = importlib.util.module_from_spec(spec)
spec.loader.exec_module(base)
base.PUBLIC = PUBLIC
POST = re.compile(r'https://post\.smzdm\.com/(?:talk/)?p/([a-z0-9]+)/?$')
QUERIES = [
    ('wellness', '菌菇 方便面 测评 site:post.smzdm.com'),
    ('premium', '拉面说 牛肉 评测 site:post.smzdm.com'),
    ('stockpile', '方便面 囤货 site:post.smzdm.com'),
    ('wellness', '蘑菇 鲜蔬 方便面 site:post.smzdm.com'),
    ('premium', '满汉大餐 试吃 site:post.smzdm.com'),
    ('stockpile', '泡面 家里 常备 site:post.smzdm.com'),
    ('wellness', '五谷道场 非油炸 测评 site:post.smzdm.com'),
    ('premium', '速达面馆 牛肉 试吃 site:post.smzdm.com'),
    ('stockpile', '泡面 整箱 囤 site:post.smzdm.com'),
    ('novelty', '方便面 冬阴功 尝鲜 site:post.smzdm.com'),
    ('wellness', '方便面 枸杞 山药 site:post.smzdm.com'),
    ('premium', '拉面说 满汉宴 真肉 site:post.smzdm.com'),
    ('stockpile', '方便面 囤货 保质期 site:post.smzdm.com'),
    ('novelty', '方便面 藤椒 香菜 测评 site:post.smzdm.com'),
    ('wellness', '方便面 低钠 减盐 site:post.smzdm.com'),
    ('stockpile', '泡面 家庭 囤货 组合 site:post.smzdm.com'),
]
STATE = {'schema': 'ksf.five-topic.supplement-collection.v1', 'startedAt': base.now(),
         'status': 'running', 'searches': [], 'sources': [], 'rejected': [], 'productionWrites': 0}


def persist():
    STATE['updatedAt'] = base.now()
    base.save(WORK / 'collection-result.json', STATE)


def canonical(url):
    parsed = urllib.parse.urlparse(url)
    if parsed.hostname != 'post.smzdm.com': return None
    value = 'https://post.smzdm.com' + parsed.path
    return value if POST.fullmatch(value) else None


def decode(raw, receipt):
    declared = re.search(r'charset\s*=\s*([^;\s]+)', receipt.get('contentType', ''), re.I)
    encoding = declared.group(1).strip('"\'') if declared else 'utf8'
    try: return raw.decode(encoding), encoding
    except (LookupError, UnicodeDecodeError):
        try: return raw.decode('utf8'), 'utf8'
        except UnicodeDecodeError: return raw.decode('gb18030'), 'gb18030'


def extract(raw, receipt):
    text, encoding = decode(raw, receipt)
    root = html.fromstring(text)
    title = re.sub(r'\s+', ' ', root.xpath('string(//title)')).strip()
    if re.search(r'验证|验证码|页面不存在|访问异常|系统繁忙|404', title):
        raise ValueError('VERIFICATION_OR_ERROR_PAGE')
    articles = root.xpath('//article')
    if len(articles) != 1: raise ValueError('ORIGINAL_ARTICLE_NOT_UNIQUE')
    article = articles[0]
    heading = article.xpath('.//h1')
    if heading: title = re.sub(r'\s+', ' ', heading[0].text_content()).strip() or title
    selector = '//article'
    body = base.paragraphs(article)
    # SMZDM talk posts can be short but complete. Exclude a title-only shell.
    body_only = body.removeprefix(title).strip()
    # This only rejects an empty shell; semantic review decides usefulness.
    if len(body_only) < 50: raise ValueError('INSUFFICIENT_ACTUAL_ARTICLE_BODY')
    if not re.search(r'方便面|泡面|即食面|速食面|拌面|火鸡面|拉面说|速达面馆|满汉大餐|满汉宴', title + '\n' + body):
        raise ValueError('NOT_PACKAGED_NOODLE_EXPERIENCE')
    dates = root.xpath('//meta[@property="article:published_time" or @name="pubdate" or @name="weibo:article:create_at"]/@content')
    if not dates:
        match = re.search(r'\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}(?::\d{2})?', article.text_content()[:1200])
        dates = [match.group(0)] if match else []
    published = dates[0] if dates else None
    if published:
        match = re.match(r'(\d{4}-\d{2}-\d{2})', published)
        if match and match.group(1) > datetime.now().strftime('%Y-%m-%d'):
            raise ValueError('FUTURE_PUBLICATION_DATE')
    return title, body, selector, published, encoding


def make_source(raw, receipt, candidate, reused):
    url = canonical(receipt['finalUrl'])
    if not url: raise ValueError('NOT_CANONICAL_SMZDM_ORIGINAL')
    title, body, selector, published, encoding = extract(raw, receipt)
    matches = []
    for topic in base.PLAN['topics']:
        hits = [term for term in topic['keywords'] if term in title + '\n' + body]
        if hits: matches.append({'topic': topic['id'], 'matchedTerms': hits})
    return {'schema': 'ksf.five-topic.public-source.v1', 'platform': 'smzdm',
            'sourceType': 'public_article_not_ecommerce_review', 'sourceId': 'url-sha256:' + base.digest(url.encode()),
            'url': url, 'title': title, 'body': body, 'bodySha256': base.digest(body.encode()),
            'publishedAtRaw': published, 'collectedAt': receipt['collectedAt'], 'httpReceipt': receipt,
            'sourceEncoding': encoding, 'bodyExtractionSelector': selector, 'topics': matches,
            'searchDiscovery': candidate, 'previouslyArchivedOriginalReused': reused,
            'publishedToDatabase': False,
            'limits': ['公开社区作者原文，不等同电商购买评价或独立消费者',
                       '作者功效、营养、原料、工艺及价格声明未在本轮独立验证',
                       '保留原日期，不把历史原文当作当前趋势']}


def archived_sources(existing):
    rows = {}
    for file in sorted((ORIGINAL_WORK / 'public-sources').glob('*.http.json')):
        receipt = json.loads(file.read_bytes())
        url = canonical(receipt['finalUrl'])
        if not url or url in existing or receipt['httpStatus'] != 200: continue
        raw = (file.parent / receipt['responseFile']).read_bytes()
        if base.digest(raw) != receipt['sha256']: raise ValueError('ARCHIVED_ORIGINAL_SHA_MISMATCH')
        key = 'archive-' + base.digest(url.encode())[:24]
        (PUBLIC / (key + '.body')).write_bytes(raw)
        copied = {**receipt, 'responseFile': key + '.body', 'previousReceipt': str(file.relative_to(ROOT))}
        base.save(PUBLIC / (key + '.http.json'), copied)
        try:
            record = make_source(raw, copied, {'type': 'prior_preserved_original'}, True)
            base.save(PUBLIC / (key + '.source.json'), record)
            rows[url] = record
        except ValueError as error:
            STATE['rejected'].append({'url': url, 'reason': str(error), 'archiveReused': True})
    return rows


def search(index, topic, query):
    key = 'query-' + str(index + 1).zfill(2)
    url = 'https://www.sogou.com/web?' + urllib.parse.urlencode({'query': query})
    raw, receipt = base.fetch(url, key, SEARCH)
    decoded, _ = decode(raw, receipt)
    root = html.fromstring(decoded)
    title = root.xpath('string(//title)')
    if receipt['httpStatus'] != 200 or re.search(r'验证|验证码|访问异常', title):
        raise ValueError('SEARCH_UNAVAILABLE_OR_VERIFICATION')
    rows = []
    for heading in root.xpath('//h3'):
        links = heading.xpath('.//a/@href')
        if not links: continue
        text = re.sub(r'\s+', ' ', heading.text_content()).strip()
        if re.search(r'高清大图|做法步骤|菜谱大全|价格表|批发|青团', text): continue
        rows.append({'topic': topic, 'query': query, 'title': text,
                     'url': urllib.parse.urljoin(receipt['finalUrl'], links[0]), 'searchReceipt': key + '.http.json'})
    base.save(SEARCH / (key + '.results.json'), {'query': query, 'receipt': receipt, 'results': rows})
    STATE['searches'].append({'query': query, 'topic': topic, 'candidateCount': len(rows), 'sha256': receipt['sha256']})
    return rows


def collect(candidate):
    key = base.digest(candidate['url'].encode())[:24]
    cached = PUBLIC / (key + '.source.json')
    if cached.exists():
        result = json.loads(cached.read_bytes())
        raw = (PUBLIC / result['httpReceipt']['responseFile']).read_bytes()
        assert base.digest(raw) == result['httpReceipt']['sha256']
        return result
    raw, receipt = base.fetch(candidate['url'], key, PUBLIC)
    if receipt['httpStatus'] != 200: raise ValueError('HTTP_' + str(receipt['httpStatus']))
    raw, receipt = base.resolve(raw, receipt, key)
    if receipt['httpStatus'] != 200: raise ValueError('ORIGINAL_HTTP_' + str(receipt['httpStatus']))
    result = make_source(raw, receipt, candidate, False)
    base.save(cached, result)
    return result


def main():
    prior = json.loads((ORIGINAL_WORK / 'reviewed-public-sources.json').read_bytes())
    existing = {r['url'] for r in prior['selected']}
    unique = archived_sources(existing)
    STATE['sources'] = list(unique.values())
    persist()
    print(json.dumps({'stage': 'archive', 'recoveredOriginalBodies': len(unique)}, ensure_ascii=False), flush=True)
    candidates = {}
    for index, (topic, query) in enumerate(QUERIES):
        try:
            for row in search(index, topic, query): candidates.setdefault(row['url'], row)
        except Exception as error:
            STATE['searches'].append({'query': query, 'topic': topic, 'error': str(error)})
            if str(error) == 'SEARCH_UNAVAILABLE_OR_VERIFICATION':
                STATE['searchStopReason'] = str(error)
                break
        persist()
        print(json.dumps({'stage': 'search', 'done': index + 1, 'candidates': len(candidates)}, ensure_ascii=False), flush=True)
        time.sleep(1.3)
    STATE['candidateCount'] = len(candidates)
    with ThreadPoolExecutor(max_workers=2) as pool:
        tasks = {pool.submit(collect, row): row for row in candidates.values()}
        for task in as_completed(tasks):
            row = tasks[task]
            try:
                result = task.result()
                if result['url'] not in existing: unique.setdefault(result['url'], result)
            except Exception as error:
                STATE['rejected'].append({'candidate': row, 'reason': str(error)})
            STATE['sources'] = list(unique.values())
            persist()
            print(json.dumps({'stage': 'body', 'originalBodies': len(unique), 'rejected': len(STATE['rejected'])}, ensure_ascii=False), flush=True)
    STATE['status'] = 'collection-complete-awaiting-semantic-review'
    STATE['finishedAt'] = base.now()
    STATE['uniqueSources'] = len(unique)
    persist()
    review = [{'index': index, 'url': row['url'], 'title': row['title'], 'date': row['publishedAtRaw'],
               'body': row['body'], 'keywordTopics': row['topics']} for index, row in enumerate(STATE['sources'])]
    base.save(WORK / 'review-candidates.json', review)
    print(json.dumps({'status': STATE['status'], 'uniqueSources': len(unique)}, ensure_ascii=False), flush=True)


if __name__ == '__main__': main()