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

"""Collect readable public source pages, keeping search results out of evidence counts."""
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
import re
import threading
import urllib.error
import urllib.parse
import urllib.request
from lxml import html

WORK = Path(__file__).resolve().parents[1] / 'work/five-topic-collection-20261110'
PLAN = json.loads((WORK / 'collection-plan.json').read_bytes())
PUBLIC = WORK / 'public-sources'
PUBLIC.mkdir(exist_ok=True)
SEARCH = WORK / 'public-search'
QUERIES = {
    'wellness': ['方便面 中式养生 当归', '方便面 药食同源', '菌菇汤 方便面 非油炸',
                 '养生方便面 site:thepaper.cn', '方便面 菌菇 测评 site:post.smzdm.com'],
    'student': ['大学生 宿舍 泡面 测评', '学生 夜宵 方便面', '宿舍 方便面 不用锅',
                '宿舍 泡面 site:post.smzdm.com', '方便面 夜宵 site:thepaper.cn'],
    'premium': ['高端 方便面 真实肉块', '方便面 升级 满汉大餐', '方便面 中产 拉面说',
                '高端 方便面 site:thepaper.cn', '拉面说 满汉大餐 测评 site:post.smzdm.com'],
    'stockpile': ['方便面 囤货 测评', '家庭 囤货 方便面 保质期', '方便面 整箱 组合装',
                  '方便面 囤货 site:post.smzdm.com', '方便面 囤货 保质期 site:sohu.com'],
    'novelty': ['方便面 联名 新品', '方便面 新口味 测评', '泡面 尝鲜 冬阴功 香菜',
                '方便面 联名 site:thepaper.cn', '方便面 新口味 测评 site:post.smzdm.com'],
}
NOODLES = re.compile(r'方便面|泡面|即食面|速食面|杯面|桶面|碗面|方便拌面|速食拌面|火鸡面')
PLATFORMS = {'sohu.com': 'sohu', 'thepaper.cn': 'thepaper', 'smzdm.com': 'smzdm',
             'zhihu.com': 'zhihu', 'weixin.qq.com': 'wechat', 'xinhuanet.com': 'xinhua',
             'news.cn': 'xinhua', 'people.com.cn': 'people', 'chinanews.com.cn': 'chinanews',
             'chinanews.com': 'chinanews', 'cctv.com': 'cctv', 'foodaily.com': 'foodaily',
             'foodtalks.cn': 'foodtalks', 'cbndata.com': 'cbndata', 'jiemian.com': 'jiemian',
             'stcn.com': 'stcn', 'yicai.com': 'yicai', 'cnr.cn': 'cnr', 'sina.com.cn': 'sina',
             '163.com': 'netease', 'news.qq.com': 'tencent'}
STATE = {'schema': 'ksf.five-topic.public-collection.v1', 'status': 'running', 'startedAt': None,
         'searches': [], 'sources': [], 'rejected': [], 'productionWrites': 0}
GUARD = threading.Lock()


def now(): return datetime.now(timezone.utc).isoformat()
def digest(raw): return hashlib.sha256(raw).hexdigest()
def save(path, data):
    temporary = path.with_suffix(path.suffix + '.next')
    temporary.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding='utf8')
    temporary.replace(path)


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


def platform(url):
    host = urllib.parse.urlparse(url).hostname or ''
    return next((value for suffix, value in PLATFORMS.items() if host == suffix or host.endswith('.' + suffix)), None)


def fetch(url, name, target):
    previous = target / (name + '.http.json')
    if previous.exists():
        receipt = json.loads(previous.read_bytes())
        body = target / receipt['responseFile']
        if receipt['requestedUrl'] == url and body.exists() and digest(body.read_bytes()) == receipt['sha256']:
            return body.read_bytes(), receipt
    request = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
    try:
        with urllib.request.urlopen(request, timeout=25) as response:
            raw = response.read(6000001)
            status, final, content_type = response.status, response.url, response.headers.get('Content-Type', '')
    except urllib.error.HTTPError as error:
        raw, status, final, content_type = error.read(6000001), error.code, error.url, error.headers.get('Content-Type', '')
    if len(raw) > 6000000: raise ValueError('SOURCE_TOO_LARGE')
    body = name + '.body'
    (target / body).write_bytes(raw)
    receipt = {'requestedUrl': url, 'finalUrl': final, 'httpStatus': status, 'contentType': content_type,
               'collectedAt': now(), 'responseFile': body, 'bytes': len(raw), 'sha256': digest(raw)}
    save(target / (name + '.http.json'), receipt)
    return raw, receipt


def search(topic, index, query):
    key = topic + '-sogou-' + str(index + 1)
    file = SEARCH / (key + '.results.json')
    if file.exists(): return json.loads(file.read_bytes())
    url = 'https://www.sogou.com/web?' + urllib.parse.urlencode({'query': query})
    raw, receipt = fetch(url, key, SEARCH)
    rows = []
    if receipt['httpStatus'] == 200:
        root = html.fromstring(raw)
        for heading in root.xpath('//h3'):
            links = heading.xpath('.//a[@href]')
            if not links: continue
            title = re.sub(r'\s+', ' ', heading.text_content()).strip()
            # Search snippets alone never count as source bodies.
            if not title or len(title) < 6: continue
            rows.append({'title': title, 'url': urllib.parse.urljoin(receipt['finalUrl'], links[0].get('href')),
                         'topic': topic, 'query': query, 'searchReceipt': key + '.http.json'})
    result = {'topic': topic, 'query': query, 'receipt': receipt, 'results': rows}
    save(file, result)
    return result


def resolve(raw, receipt, key):
    for step in range(2):
        host = urllib.parse.urlparse(receipt['finalUrl']).hostname or ''
        if not host.endswith('sogou.com'): return raw, receipt
        root = html.fromstring(raw)
        urls = root.xpath('//meta[translate(@http-equiv,"REFSH","refsh")="refresh"]/@content')
        target = None
        for value in urls:
            found = re.search(r'url\s*=\s*[\'\"]?(https?://[^\'\"\s>]+)', value, re.I)
            if found: target = found.group(1)
        if not target:
            text = raw.decode('utf8', errors='replace')
            found = re.search(r'(?:location\.(?:replace|assign)\s*\(|(?:window\.)?location(?:\.href)?\s*=\s*)[\'\"](https?://[^\'\"\s]+)', text)
            if found: target = found.group(1).replace('\\/', '/')
        if not target: raise ValueError('SEARCH_REDIRECT_NOT_ORIGINAL_PAGE')
        raw, receipt = fetch(target, key + '-redirect-' + str(step + 1), PUBLIC)
    return raw, receipt


def paragraphs(node):
    clone = html.fromstring(html.tostring(node, encoding='unicode'))
    for element in clone.xpath('.//script|.//style|.//noscript|.//nav|.//footer|.//form|.//button|.//*[contains(concat(" ",normalize-space(@class)," ")," recommend-tab ")]'):
        element.drop_tree()
    nodes = clone.xpath('.//p|.//h1|.//h2|.//h3|.//blockquote|.//li')
    chunks = [re.sub(r'\s+', ' ', element.text_content()).strip() for element in nodes]
    chunks = [part for part in chunks if part]
    if not chunks: chunks = [re.sub(r'\s+', ' ', clone.text_content()).strip()]
    return '\n'.join(dict.fromkeys(chunks))


def extract(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: decoded = raw.decode(encoding)
    except (LookupError, UnicodeDecodeError):
        try: decoded = raw.decode('utf8'); encoding = 'utf8'
        except UnicodeDecodeError: decoded = raw.decode('gb18030'); encoding = 'gb18030'
    root = html.fromstring(decoded)
    # Never treat a verification/login response as an article.
    title = ''.join(root.xpath('//meta[@property="og:title"]/@content')[:1]) or root.xpath('string(//title)')
    title = re.sub(r'\s+', ' ', title).strip()
    if re.search(r'安全验证|访问验证|验证码|系统繁忙|访问异常|扫码登录|页面不存在|404|没有找到', title, re.I):
        raise ValueError('VERIFICATION_OR_ERROR_PAGE')
    selectors = [
        '//*[@id="js_content"]', '//*[@id="artibody"]',
        '//*[contains(concat(" ",normalize-space(@class)," ")," article-content ")]',
        '//*[contains(concat(" ",normalize-space(@class)," ")," article_content ")]',
        '//*[contains(concat(" ",normalize-space(@class)," ")," article-main ")]',
        '//*[@id="detail_content"]', '//*[@id="article-content"]',
        '//*[contains(concat(" ",normalize-space(@class)," ")," rich_media_content ")]',
        '//article', '//*[@itemprop="articleBody"]',
        '//*[contains(concat(" ",normalize-space(@class)," ")," article_content_wrapper ")]',
        '//*[contains(concat(" ",normalize-space(@class)," ")," post_text ")]',
        '//*[contains(concat(" ",normalize-space(@class)," ")," entry-content ")]',
    ]
    body, selector = None, None
    for candidate in selectors:
        texts = [paragraphs(node) for node in root.xpath(candidate)]
        texts = [value for value in texts if len(value) >= 300]
        if texts: body, selector = max(texts, key=len), candidate; break
    if not body: raise ValueError('ARTICLE_BODY_NOT_LOCATED')
    if not NOODLES.search(title + '\n' + body): raise ValueError('NOT_PACKAGED_NOODLE_CONTENT')
    date_values = root.xpath('//meta[@property="article:published_time" or @name="pubdate" or @name="publishdate" or @name="publish-date" or @name="PublishDate" or @name="weibo:article:create_at"]/@content')
    dates = [value for value in date_values if re.search(r'\d{4}[-/]\d{1,2}[-/]\d{1,2}', value)]
    if not dates:
        headers = root.xpath('//article')
        header = headers[0].text_content()[:1100] if headers else ''
        visible = re.search(r'\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}(?::\d{2})?', header)
        if visible: dates = [visible.group(0)]
    return title, body, selector, dates[0] if dates else None, encoding


def collect(candidate):
    key = digest(candidate['url'].encode())[:24]
    record_file = PUBLIC / (key + '.source.json')
    if record_file.exists(): return json.loads(record_file.read_bytes())
    raw, receipt = fetch(candidate['url'], key, PUBLIC)
    if receipt['httpStatus'] != 200: raise ValueError('HTTP_' + str(receipt['httpStatus']))
    raw, receipt = resolve(raw, receipt, key)
    site = platform(receipt['finalUrl'])
    if not site: raise ValueError('UNSUPPORTED_OR_SEARCH_RESULT_HOST')
    title, body, selector, published, encoding = extract(raw, receipt)
    source_topics = []
    for topic in PLAN['topics']:
        hit = [term for term in topic['keywords'] if term in title + '\n' + body]
        if not hit: continue
        passages = []
        for term in hit:
            start = body.find(term)
            if start >= 0: passages.append(body[max(0, start - 65):start + 150])
        source_topics.append({'topic': topic['id'], 'matchedTerms': hit, 'passages': passages[:5]})
    if not source_topics: raise ValueError('NO_REQUESTED_TOPIC_IN_SOURCE_BODY')
    # Drop referrer/tracking fields from the identity; preserve the exact URL in the receipt.
    parsed = urllib.parse.urlparse(receipt['finalUrl'])
    query = urllib.parse.parse_qs(parsed.query)
    stable_query = urllib.parse.urlencode({k: query[k][0] for k in ['__biz', 'mid', 'idx'] if k in query})
    canonical = urllib.parse.urlunparse((parsed.scheme, parsed.netloc, parsed.path, '', stable_query, ''))
    record = {'schema': 'ksf.five-topic.public-source.v1', 'sourceType': 'public_article_not_ecommerce_review',
              'sourceId': 'url-sha256:' + digest(canonical.encode()), 'identityBasis': 'canonical_original_url_sha256',
              'platform': site, 'title': title, 'url': canonical, 'originalResolvedUrl': receipt['finalUrl'],
              'publishedAtRaw': published, 'collectedAt': receipt['collectedAt'], 'body': body,
              'bodySha256': digest(body.encode()), 'bodyExtractionSelector': selector, 'sourceEncoding': encoding, 'httpReceipt': receipt,
              'searchDiscovery': candidate, 'topics': source_topics,
              'verification': 'readable_original_article_with_preserved_http_bytes',
              'publishedToDatabase': False,
              'limits': ['原作者/媒体观点，不当作电商购买评价', '文中健康或市场数字未在此轮独立验证',
                         '保留原文字段与原件，不将采集时刻写成发布时间']}
    save(record_file, record)
    return record


def main():
    STATE['startedAt'] = now()
    candidates = {}
    for topic, queries in QUERIES.items():
        for index, query in enumerate(queries):
            try:
                result = search(topic, index, query)
                STATE['searches'].append({'topic': topic, 'query': query, 'candidateCount': len(result['results']),
                                          'sha256': result['receipt']['sha256']})
                # Prefer text articles and actual experiences, excluding obvious recipe/video/category pages.
                filtered = [row for row in result['results'] if not re.search(r'做法步骤|菜谱大全|怎么做好吃|视频大全|中毒|健康知识|价格表|批发价格', row['title'])]
                for row in filtered[:10]: candidates.setdefault(row['url'], row)
            except Exception as error:
                STATE['searches'].append({'topic': topic, 'query': query, 'error': type(error).__name__})
            persist()
            print(json.dumps({'stage': 'search', 'topic': topic, 'query': query, 'candidates': len(candidates)}, ensure_ascii=False), flush=True)
    with ThreadPoolExecutor(max_workers=3) as pool:
        tasks = {pool.submit(collect, row): row for row in candidates.values()}
        for task in as_completed(tasks):
            candidate = tasks[task]
            try:
                record = task.result()
                with GUARD: STATE['sources'].append(record)
            except Exception as error:
                with GUARD: STATE['rejected'].append({'candidate': candidate, 'reason': str(error) if isinstance(error, ValueError) else type(error).__name__})
            with GUARD: persist()
            print(json.dumps({'stage': 'body', 'accepted': len(STATE['sources']), 'rejected': len(STATE['rejected']),
                              'candidates': len(candidates)}, ensure_ascii=False), flush=True)
    unique = {}
    for row in STATE['sources']:
        key = row['sourceId']
        if key in unique:
            known = {item['topic'] for item in unique[key]['topics']}
            unique[key]['topics'].extend(item for item in row['topics'] if item['topic'] not in known)
        else: unique[key] = row
    STATE['sources'] = list(unique.values())
    STATE['status'] = 'complete'
    STATE['finishedAt'] = now()
    STATE['uniqueSources'] = len(STATE['sources'])
    STATE['countsByTopic'] = {topic['id']: sum(any(item['topic'] == topic['id'] for item in row['topics']) for row in STATE['sources']) for topic in PLAN['topics']}
    STATE['databaseImported'] = 0
    persist()
    print(json.dumps({'status': STATE['status'], 'uniqueSources': STATE['uniqueSources'],
                      'countsByTopic': STATE['countsByTopic'], 'databaseImported': 0}, ensure_ascii=False), flush=True)


if __name__ == '__main__': main()