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

"""Collect public industry articles via their publisher's ordinary search form."""
from concurrent.futures import ThreadPoolExecutor, as_completed
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]
WORK = ROOT / 'work/five-topic-collection-20261110/supplement-01'
PUBLIC = WORK / 'industry-sources'
SEARCH = WORK / 'industry-search'
PUBLIC.mkdir(exist_ok=True)
SEARCH.mkdir(exist_ok=True)
spec = importlib.util.spec_from_file_location('public_base', ROOT / 'scripts/collect-five-topic-public.py')
base = importlib.util.module_from_spec(spec)
spec.loader.exec_module(base)
JOBS = [('方便面', page) for page in range(1, 5)] + [
    ('非油炸', 1), ('菌菇', 1), ('囤货', 1), ('拉面', 1), ('五谷道场', 1), ('养生', 1)]
STATE = {'schema': 'ksf.five-topic.industry-supplement.v1', 'status': 'running',
         'startedAt': base.now(), 'searches': [], 'sources': [], 'rejected': [], 'productionWrites': 0}


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


def search(index, query, page):
    key = 'search-' + str(index + 1).zfill(2)
    url = 'https://www.foodaily.com/search/articles?' + urllib.parse.urlencode({'search': query, 'page': page})
    raw, receipt = base.fetch(url, key, SEARCH)
    if receipt['httpStatus'] != 200: raise ValueError('SEARCH_HTTP_' + str(receipt['httpStatus']))
    root = html.fromstring(raw.decode('utf8'))
    if re.search(r'验证|访问异常', root.xpath('string(//title)')): raise ValueError('SEARCH_VERIFICATION')
    rows = {}
    for anchor in root.xpath('//a[@href]'):
        link, title = anchor.get('href'), re.sub(r'\s+', ' ', anchor.text_content()).strip()
        if not re.fullmatch(r'https://www\.foodaily\.com/articles/\d+', link) or not title: continue
        # Publisher search results, not evidence. Read each selected actual article.
        if not re.search(r'方便面|泡面|速食|拉面|非油炸|面条|汤面|面饼|拌面|高汤|菌菇面', title): continue
        if re.search(r'冰淇淋|奶茶|生鲜传奇|咖啡|可口可乐|茶饮', title) and not re.search(r'泡面|面条|速食面|方便面|拉面', title): continue
        rows[link] = {'url': link, 'title': title, 'query': query, 'page': page,
                      'searchUrl': url, 'searchReceipt': key + '.http.json'}
    record = {'query': query, 'page': page, 'httpReceipt': receipt, 'results': list(rows.values())}
    base.save(SEARCH / (key + '.results.json'), record)
    STATE['searches'].append({'query': query, 'page': page, 'count': len(rows), 'sha256': receipt['sha256']})
    return list(rows.values())


def extract(raw, receipt):
    title, body, selector, date, encoding = base.extract(raw, receipt)
    root = html.fromstring(raw.decode(encoding))
    if not date:
        dates = [re.sub(r'\s+', ' ', node.text_content()).strip() for node in root.xpath('//*[contains(concat(" ",normalize-space(@class)," ")," float-right ")]')]
        dates = [value for value in dates if re.fullmatch(r'\d{4}\.\d{2}\.\d{2}', value)]
        date = dates[0] if len(set(dates)) == 1 else None
    if date:
        normal = date[:10].replace('.', '-').replace('/', '-')
        if normal > datetime.now().strftime('%Y-%m-%d'): raise ValueError('FUTURE_PUBLICATION_DATE')
    return title, body, selector, date, encoding


def collect(candidate):
    key = base.digest(candidate['url'].encode())[:24]
    cached = PUBLIC / (key + '.source.json')
    if cached.exists():
        row = json.loads(cached.read_bytes())
        assert base.digest((PUBLIC / row['httpReceipt']['responseFile']).read_bytes()) == row['httpReceipt']['sha256']
        return row
    raw, receipt = base.fetch(candidate['url'], key, PUBLIC)
    if receipt['httpStatus'] != 200: raise ValueError('SOURCE_HTTP_' + str(receipt['httpStatus']))
    if not re.fullmatch(r'https://www\.foodaily\.com/articles/\d+', receipt['finalUrl']):
        raise ValueError('NOT_REQUESTED_PUBLISHER_ARTICLE')
    title, body, selector, published, encoding = extract(raw, receipt)
    topics = []
    for topic in base.PLAN['topics']:
        hits = [term for term in topic['keywords'] if term in title + '\n' + body]
        passages = []
        for term in hits:
            positions = [m.start() for m in re.finditer(re.escape(term), body)]
            for start in positions[:2]: passages.append({'term': term, 'text': body[max(0, start - 90):start + 180]})
        if hits: topics.append({'topic': topic['id'], 'matchedTerms': hits, 'passages': passages[:8]})
    row = {'schema': 'ksf.five-topic.public-source.v1', 'platform': 'foodaily',
           'sourceType': 'public_industry_article_not_consumer_review',
           'sourceId': 'url-sha256:' + base.digest(receipt['finalUrl'].encode()),
           'url': receipt['finalUrl'], 'title': title, 'body': body, 'bodySha256': base.digest(body.encode()),
           'publishedAtRaw': published, 'collectedAt': receipt['collectedAt'], 'httpReceipt': receipt,
           'sourceEncoding': encoding, 'bodyExtractionSelector': selector, 'topics': topics,
           'searchDiscovery': candidate, 'publishedToDatabase': False,
           'limits': ['行业媒体文章，含转载时保留文内来源署名，不冒称独立消费者或购买评价',
                      '报道中的统计、功效、制造工艺和价格仅为报道声明，未在本轮逐一独立验证',
                      '原文保留原日期；堂食/鲜面段落不作为包装即食面产品事实']}
    base.save(cached, row)
    return row


def main():
    candidates = {}
    for index, (query, page) in enumerate(JOBS):
        try:
            for row in search(index, query, page): candidates.setdefault(row['url'], row)
        except Exception as error:
            STATE['searches'].append({'query': query, 'page': page, 'error': str(error)})
            if str(error) == 'SEARCH_VERIFICATION': break
        persist()
        print(json.dumps({'stage': 'publisher-search', 'completed': index + 1, 'candidates': len(candidates)}, ensure_ascii=False), flush=True)
        time.sleep(1)
    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):
            candidate = tasks[task]
            try: STATE['sources'].append(task.result())
            except Exception as error: STATE['rejected'].append({'candidate': candidate, 'reason': str(error)})
            persist()
            print(json.dumps({'stage': 'publisher-body', 'sources': len(STATE['sources']), 'rejected': len(STATE['rejected'])}, ensure_ascii=False), flush=True)
    STATE['status'] = 'collection-complete-awaiting-semantic-review'
    STATE['finishedAt'] = base.now()
    STATE['uniqueSources'] = len(STATE['sources'])
    persist()
    reviews = []
    for row in STATE['sources']:
        reviews.append({'url': row['url'], 'title': row['title'], 'date': row['publishedAtRaw'],
                        'length': len(row['body']), 'start': row['body'][:450], 'topics': row['topics']})
    base.save(WORK / 'industry-review-candidates.json', reviews)
    print(json.dumps({'status': STATE['status'], 'sources': len(reviews)}, ensure_ascii=False), flush=True)


if __name__ == '__main__': main()