kangshifu-board/scripts/collect-first-live-sources.py

"""Capture two official noodle pages and build a bounded, source-backed import batch."""
from pathlib import Path
from html.parser import HTMLParser
from datetime import datetime, timezone
from urllib.request import Request, urlopen
import hashlib
import json
import re

ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / 'work/first-live-sources-20261102'
TASK_KEY = 'KSF-真实资料-20261102-01'

class Text(HTMLParser):
    def __init__(self):
        super().__init__(convert_charrefs=True)
        self.skip = 0
        self.parts = []
    def handle_starttag(self, tag, attrs):
        if tag in ('script', 'style'):
            self.skip += 1
    def handle_endtag(self, tag):
        if tag in ('script', 'style'):
            self.skip = max(0, self.skip - 1)
    def handle_data(self, data):
        if not self.skip:
            self.parts.append(data)

def dump(file, value):
    file.write_text(json.dumps(value, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')

def main():
    OUT.mkdir(parents=True, exist_ok=False)
    category = {'dataMode': 'live', 'code': 'KSF-RESEARCH-INSTANT-MIXED-NOODLES-V1', 'name': '速食拌面（研究工作分类）', 'taxonomyVersion': 'ksf-research-food-form-v1', 'axis': '食用方式', 'scopeNote': '研究者按官网明确标为拌面/酱拌面的产品设定工作分类，不是官方行业分类；本批不含汤面、传统现制面或未知食用方式产品。'}
    specs = [
        {'key': 'uni-jiangban', 'url': 'https://www.uni-president.com.cn/products-jiangbanmian.asp', 'title': '统一官网：统一酱拌面', 'family': '统一酱拌面', 'marker': '统一酱拌面 酱浓才够味', 'flavors': ['老干妈豆豉辣酱风味', '卤香牛肉风味', '老坛酸菜牛肉风味', 'XO酱海鲜风味'], 'packaging': '碗、五合一（袋*5）'},
        {'key': 'uni-qiaomianguan', 'url': 'https://www.uni-president.com.cn/products-qiaomianguan.asp', 'title': '统一官网：巧面馆（仅拌面部分）', 'family': '统一巧面馆拌面', 'marker': '统一巧面馆（拌面）', 'flavors': ['老坛酸菜酱拌面', '老成都担担面', '重庆豌豆杂酱风味面', '串串香干碟拌面', '贵州豆豉辣酱风味面'], 'packaging': '碗、袋、五合一(袋*5)'}
    ]
    batches = []
    facts = []
    for spec in specs:
        with urlopen(Request(spec['url'], headers={'User-Agent': 'KSF-Workshop-Research/1.0'}), timeout=25) as response:
            assert response.status == 200
            assert response.url == spec['url'], 'unexpected redirect'
            raw = response.read(2_000_001)
            assert len(raw) <= 2_000_000
            content_type = response.headers.get('Content-Type', '')
        collected = datetime.now(timezone.utc).isoformat(timespec='milliseconds').replace('+00:00', 'Z')
        html = raw.decode('gb18030', errors='strict')
        parser = Text()
        parser.feed(html)
        text = re.sub(r'\s+', ' ', ' '.join(parser.parts)).strip()
        assert spec['marker'] in text
        section = text[text.index(spec['marker']):]
        quote_match = re.search(r'产品口味：.*?包装规格：[^ ]+', section)
        assert quote_match, 'flavor/packaging block missing'
        quote = quote_match.group(0)
        assert all(flavor in quote for flavor in spec['flavors']), 'official flavor list changed'
        assert 'sample' not in category.values()
        (OUT / (spec['key'] + '.response.bin')).write_bytes(raw)
        (OUT / (spec['key'] + '.html')).write_text(html, encoding='utf-8')
        (OUT / (spec['key'] + '.txt')).write_text(text, encoding='utf-8')
        source = {'title': spec['title'], 'sourceKind': 'web', 'sourceUrl': spec['url'], 'platform': 'official_website', 'dataMode': 'live', 'rawText': text, 'collectedAt': collected}
        records = [{'itemKey': 'category', 'className': 'KfCategory', 'externalKey': 'KSF-RESEARCH-INSTANT-MIXED-NOODLES-V1', 'data': category}]
        for index, flavor in enumerate(spec['flavors'], 1):
            item = 'product-' + str(index)
            name = spec['family'] + '·' + flavor
            product = {'dataMode': 'live', 'name': name, 'brandName': '统一', 'categoryRef': {'$item': 'category'}, 'flavor': flavor, 'packaging': spec['packaging'], 'isOwnBrand': False, 'productLevel': 'product', 'attributes': {'sourceStatement': '官网品牌自述；不等同于在售或第三方验证', 'productGranularity': '口味条目，不展开为SKU，不假定各口味拥有全部包装组合', 'pagePublicationDate': None}}
            records.append({'itemKey': item, 'className': 'KfProduct', 'externalKey': TASK_KEY + ':' + spec['key'] + ':' + item, 'dependsOn': ['category'], 'data': product})
            records.append({'itemKey': 'evidence-' + str(index), 'className': 'KfEvidence', 'externalKey': TASK_KEY + ':' + spec['key'] + ':evidence-' + str(index), 'dependsOn': [item], 'data': {'dataMode': 'live', 'productRef': {'$item': item}, 'quote': quote, 'locator': {'url': spec['url'], 'section': spec['marker'], 'startCharacter': text.index(quote), 'endCharacter': text.index(quote) + len(quote), 'unit': 'Unicode code points; captured text UTF-8'}, 'evidenceKind': 'manufacturer_product_statement', 'collectedAt': collected, 'verificationState': 'page_fetched_claim_not_independently_verified', 'platform': 'official_website', 'quoteForm': 'verbatim_excerpt'}})
            facts.append({'name': name, 'flavor': flavor, 'packagingStatement': spec['packaging'], 'sourceUrl': spec['url']})
        task = {'source': source, 'records': records}
        dump(OUT / (spec['key'] + '.import.json'), task)
        capture = {'url': spec['url'], 'status': 200, 'contentType': content_type, 'encoding': 'gb18030', 'collectedAt': collected, 'responseSha256': hashlib.sha256(raw).hexdigest(), 'rawTextSha256': hashlib.sha256(text.encode()).hexdigest(), 'collector': 'Codex local urllib HTTPS GET; archived snapshot imported by workshop Agent', 'publishedAt': None}
        dump(OUT / (spec['key'] + '.capture.json'), capture)
        batches.append({'key': spec['key'], 'task': task, 'capture': capture})
    gaps = ['康师傅自有拌面同口径产品资料尚待补齐，暂不能作完整自有/竞品比较', '官网页面无明确发布日期，不能据此判断新品上市时间或当前在售状态', '缺单个SKU净含量、配料表、营养信息及每个口味对应包装组合', '缺渠道/店铺、实际成交或观察价格、优惠口径与价格观察时间', '缺真实消费者逐字评价、使用场景与需求强度，不从品牌文案推断用户痛点', '缺供应商报价、配方比例、工艺参数、成本与法规验证，不能推出量产结论']
    bundle = {'version': 'ksf.first-live-source-batch.v1', 'taskKey': TASK_KEY, 'batches': batches, 'facts': facts, 'pendingQuestions': gaps, 'collectionPerformedBy': 'Codex', 'platformScope': 'two publicly accessible official website pages; no ecommerce/social collection'}
    dump(OUT / 'batch.json', bundle)
    print(json.dumps({'ok': True, 'output': str(OUT), 'sourcePages': len(batches), 'flavorEntries': len(facts), 'evidenceEntries': len(facts), 'databaseWrites': 0}, ensure_ascii=False))

if __name__ == '__main__':
    main()