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

"""Supplement sparse topics using precise original pages and publisher search.

Reuse preserved discovery receipts, stop a host when a challenge page appears,
and never count search snippets or recommended content as original evidence.
"""
from datetime import datetime
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]
PARENT = ROOT / 'work/five-topic-collection-20261110'
WORK = PARENT / 'supplement-02'
PUBLIC = WORK / 'sources'
SEARCH = WORK / 'search'
for directory in [WORK, PUBLIC, SEARCH]: directory.mkdir(parents=True, exist_ok=True)
spec = importlib.util.spec_from_file_location('public_base_round3', ROOT / 'scripts/collect-five-topic-public.py')
base = importlib.util.module_from_spec(spec)
spec.loader.exec_module(base)
spec = importlib.util.spec_from_file_location('industry_base_round3', ROOT / 'scripts/collect-five-topic-industry-supplement.py')
industry = importlib.util.module_from_spec(spec)
spec.loader.exec_module(industry)
QUERIES = ['中式养生', '药食同源', '草本', '拉面说', '保质期', '五连包', '组合装', '家庭装']
DISCOVERY = [('wellness', '人参 老母鸡 方便面 测评'),
             ('stockpile', '方便面 囤货 保质期 测评'),
             ('wellness', '当归 花雕鸡 面 统一'),
             ('premium', '速达面馆 牛肉 测评')]
STATE = {'schema': 'ksf.five-topic.round3-collection.v1', 'startedAt': base.now(), 'status': 'running',
         'searches': [], 'sources': [], 'rejected': [], 'hostStops': {}, 'productionWrites': 0}


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


def existing_urls():
    rows = []
    for name in ['five-topics-public-01', 'five-topics-public-02']:
        rows.extend(json.loads((ROOT / 'work/database-enrichment-20261110/batches' / name / 'expected-records.json').read_bytes()))
    return {row['url'] for row in rows}


def capture(url, name, target):
    # Reuse a prior exact response if present in this round's initial probes.
    for receipt_file in (WORK / 'probes').glob('*.http.json'):
        receipt = json.loads(receipt_file.read_bytes())
        if receipt['requestedUrl'] == url and receipt['httpStatus'] == 200:
            raw = (receipt_file.parent / receipt['responseFile']).read_bytes()
            assert base.digest(raw) == receipt['sha256']
            copied = {**receipt, 'responseFile': name + '.body',
                      'reusedReceipt': str(receipt_file.relative_to(ROOT))}
            (target / copied['responseFile']).write_bytes(raw)
            base.save(target / (name + '.http.json'), copied)
            return raw, copied
    return base.fetch(url, name, target)


def challenge(raw, receipt, root):
    title = root.xpath('string(//title)')
    if re.search(r'/geetest|captcha|/verify', receipt.get('finalUrl', ''), re.I): return True
    if receipt['httpStatus'] in (202, 403, 412, 429) or re.search(r'安全验证|访问验证|验证码|访问异常|验证中心', title): return True
    if len(raw) < 10000 and re.search(rb'aliyun|captcha|acw_sc__|var seqid\s*=', raw, re.I): return True
    return False


def collect(url, discovery):
    host = urllib.parse.urlparse(url).hostname
    if host in STATE['hostStops']: raise ValueError('HOST_STOPPED_AFTER_CHALLENGE')
    key = base.digest(url.encode())[:24]
    raw, receipt = capture(url, key, PUBLIC)
    root = html.fromstring(raw.decode('utf8', errors='replace'))
    if challenge(raw, receipt, root):
        STATE['hostStops'][host] = {'reason': 'VERIFICATION_RESPONSE', 'url': url,
                                    'httpStatus': receipt['httpStatus'], 'sha256': receipt['sha256']}
        raise ValueError('VERIFICATION_RESPONSE')
    if receipt['httpStatus'] != 200: raise ValueError('HTTP_' + str(receipt['httpStatus']))
    date = None
    if host == 'post.smzdm.com':
        match = re.fullmatch(r'https://post\.smzdm\.com/(?:talk/)?p/([a-z0-9]+)/', receipt['finalUrl'])
        if not match: raise ValueError('CANONICAL_ORIGINAL_POST_REQUIRED')
        platform, pid = 'smzdm', match.group(1)
        selector = '//article'
        nodes = root.xpath(selector)
        dates = root.xpath('//meta[@property="article:published_time"]/@content')
        if not dates and len(nodes) == 1:
            match_date = re.search(r'\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}(?::\d{2})?', nodes[0].text_content()[:1000])
            dates = [match_date.group(0)] if match_date else []
        date = dates[0] if dates else None
    elif host == 'sspai.com':
        match = re.fullmatch(r'https://sspai\.com/post/(\d+)', receipt['finalUrl'])
        if not match: raise ValueError('CANONICAL_ORIGINAL_POST_REQUIRED')
        platform, pid = 'sspai', match.group(1)
        selector = '//*[contains(concat(" ",normalize-space(@class)," ")," article-body ")]'
        nodes = root.xpath(selector)
        dates = root.xpath('//*[contains(concat(" ",normalize-space(@class)," ")," article__header__date ")]/text()')
        date = dates[0].strip() if dates else None
    elif host == 'www.foodaily.com':
        title, body, selector, date, encoding = industry.extract(raw, receipt)
        platform, pid = 'foodaily', urllib.parse.urlparse(url).path.split('/')[-1]
        nodes = root.xpath(selector)
    else: raise ValueError('PUBLISHER_NOT_IN_CONTROLLED_EXTRACTORS')
    if len(nodes) != 1: raise ValueError('UNIQUE_ORIGINAL_BODY_REQUIRED')
    body = base.paragraphs(nodes[0])
    if len(body) < 60: raise ValueError('INSUFFICIENT_SOURCE_BODY')
    if not re.search(r'方便面|泡面|速食面|拉面说|速达面馆|满汉|即食面', body): raise ValueError('NOT_PACKAGED_NOODLE_CONTENT')
    title = root.xpath('string(//title)').strip()
    h1 = root.xpath('//article//h1')
    if h1: title = re.sub(r'\s+', ' ', h1[0].text_content()).strip() or title
    if not date: raise ValueError('ORIGINAL_DATE_NOT_LOCATED')
    date_match = re.search(r'(\d{4})[-/.年](\d{1,2})[-/.月](\d{1,2})', date)
    if date_match:
        normal = '-'.join([date_match.group(1), date_match.group(2).zfill(2), date_match.group(3).zfill(2)])
        if normal > datetime.now().strftime('%Y-%m-%d'): raise ValueError('FUTURE_PUBLICATION_DATE')
    row = {'schema': 'ksf.five-topic.public-source.v1', 'platform': platform, 'postId': pid,
           'sourceId': platform + ':' + pid, 'sourceType': 'public_industry_article_not_consumer_review' if platform == 'foodaily' else 'public_article_not_ecommerce_review',
           'url': receipt['finalUrl'], 'title': title, 'body': body, 'bodySha256': base.digest(body.encode()),
           'publishedAtRaw': date, 'collectedAt': receipt['collectedAt'], 'httpReceipt': receipt,
           'sourceEncoding': 'utf8', 'bodyExtractionSelector': selector, 'collectionDirectory': 'sources',
           'searchDiscovery': discovery, 'topics': [], 'publishedToDatabase': False,
           'limits': ['保留作者、媒体原文和日期；不等同独立购买评价或已验证消费者身份',
                      '原文报价、营养、工艺及功效声明未独立验证，不写为当前报价或功效结论',
                      '原文可能含推广或转载；独立URL不等于独立证据，关联主题需按实际段落复核']}
    base.save(PUBLIC / (key + '.source.json'), row)
    return row


def prior_discoveries():
    old = PARENT / 'supplement-01'
    finals = {}
    for file in (old / 'public-sources').glob('*redirect-*.http.json'):
        row = json.loads(file.read_bytes())
        if 'post.smzdm.com/' in row['finalUrl']:
            finals[file.stem.split('-redirect')[0]] = row['finalUrl']
    result = {}
    for index in [2, 3, 4]:
        file = old / 'search' / ('query-' + str(index).zfill(2) + '.results.json')
        for row in json.loads(file.read_bytes())['results']:
            key = base.digest(row['url'].encode())[:24]
            url = finals.get(key)
            if not url or not re.fullmatch(r'https://post\.smzdm\.com/(?:talk/)?p/([a-z0-9]+)/?', url): continue
            if not url.endswith('/'): url += '/'
            # Discard general emergency stock and non-noodle alternatives at discovery.
            if re.search(r'方便面终结者|多囤5样|2千字', row['title']): continue
            result[url] = {**row, 'previousDiscoveryReceipt': str(file.relative_to(ROOT))}
    return result


def publisher_search(index, query):
    url = 'https://www.foodaily.com/search/articles?' + urllib.parse.urlencode({'search': query})
    key = 'publisher-' + str(index + 1).zfill(2)
    raw, receipt = base.fetch(url, key, SEARCH)
    root = html.fromstring(raw.decode('utf8'))
    if challenge(raw, receipt, root): raise ValueError('PUBLISHER_SEARCH_VERIFICATION')
    candidates = {}
    for anchor in root.xpath('//a[@href]'):
        link = anchor.get('href')
        title = re.sub(r'\s+', ' ', anchor.text_content()).strip()
        if not re.fullmatch(r'https://www\.foodaily\.com/articles/\d+', link) or not title: continue
        if not re.search(r'泡面|方便面|拉面|面饼|速食|面条|汤面|拌面', title): continue
        candidates[link] = {'url': link, 'title': title, 'query': query, 'receipt': key + '.http.json'}
    base.save(SEARCH / (key + '.results.json'), {'query': query, 'receipt': receipt, 'results': list(candidates.values())})
    STATE['searches'].append({'kind': 'publisher-search', 'query': query, 'count': len(candidates), 'sha256': receipt['sha256']})
    return candidates


def general_search(index, topic, query):
    url = 'https://www.so.com/s?' + urllib.parse.urlencode({'q': query})
    key = 'web-' + str(index + 1).zfill(2)
    raw, receipt = base.fetch(url, key, SEARCH)
    root = html.fromstring(raw.decode('utf8'))
    if challenge(raw, receipt, root): raise ValueError('WEB_SEARCH_VERIFICATION')
    result = []
    for anchor in root.xpath('//a[@data-mdurl]'):
        link, title = anchor.get('data-mdurl'), re.sub(r'\s+', ' ', anchor.text_content()).strip()
        if not link or len(title) < 8: continue
        if re.fullmatch(r'https://(?:post\.smzdm\.com/(?:talk/)?p/[a-z0-9]+/?|sspai\.com/post/\d+)', link):
            result.append({'url': link if 'sspai.com' in link or link.endswith('/') else link + '/',
                           'title': title, 'topic': topic, 'query': query, 'receipt': key + '.http.json'})
    base.save(SEARCH / (key + '.results.json'), {'query': query, 'receipt': receipt, 'results': result})
    STATE['searches'].append({'kind': 'web-search', 'query': query, 'count': len(result), 'sha256': receipt['sha256']})
    return result


def main():
    existing = existing_urls()
    candidates = prior_discoveries()
    candidates['https://sspai.com/post/73701'] = {'type': 'specific_original_review', 'topic': 'wellness'}
    for index, (topic, query) in enumerate(DISCOVERY):
        try:
            for row in general_search(index, topic, query): candidates.setdefault(row['url'], row)
        except Exception as error: STATE['rejected'].append({'query': query, 'reason': str(error)})
        persist()
        time.sleep(1.3)
    for index, query in enumerate(QUERIES):
        try: candidates.update(publisher_search(index, query))
        except Exception as error: STATE['rejected'].append({'query': query, 'reason': str(error)})
        persist()
        time.sleep(1.1)
    # Aim at sparse topics first. A host's verification response stops it.
    candidates = {url: row for url, row in candidates.items() if url not in existing}
    STATE['candidateCount'] = len(candidates)
    print(json.dumps({'stage': 'discovery', 'newCandidates': len(candidates)}, ensure_ascii=False), flush=True)
    for index, (url, discovery) in enumerate(candidates.items()):
        try:
            result = collect(url, discovery)
            STATE['sources'].append(result)
        except Exception as error: STATE['rejected'].append({'url': url, 'reason': str(error)})
        persist()
        print(json.dumps({'stage': 'body', 'processed': index + 1, 'target': len(candidates),
                          'sources': len(STATE['sources']), 'rejected': len(STATE['rejected'])}, ensure_ascii=False), flush=True)
        time.sleep(1.5)
    STATE['status'] = 'collection-complete-awaiting-semantic-review'
    STATE['finishedAt'] = base.now()
    STATE['uniqueSources'] = len(STATE['sources'])
    persist()
    base.save(WORK / 'review-candidates.json', [{'url': r['url'], 'platform': r['platform'], 'title': r['title'],
               'date': r['publishedAtRaw'], 'body': r['body']} for r in STATE['sources']])
    print(json.dumps({'status': STATE['status'], 'sources': len(STATE['sources']), 'hostStops': STATE['hostStops']}, ensure_ascii=False), flush=True)


if __name__ == '__main__': main()