kangshifu-board/scripts/prepare-jd-historical-baseline.py

"""Recover the seven existing JD store pages and compare with the public library.

No collector calls, credentials, database writes, ingestion or publication.
Python 3.9+ standard library. The output directory must not already exist.
"""
import argparse
from collections import Counter
import datetime as dt
import hashlib
import json
from pathlib import Path
import re
import urllib.request
import zipfile


ARCHIVE_SHA = 'ed0e9d60015a28b74b22fe89e5008bd918e8091b0399b5c91a952cae6a320587'
SHOP = '1000084104'
BOARD = 'https://kangshifu.fmode.cn/api/library'
ARRAY_POINTER = '/result/data/data/ItemList'


def require(value, message):
    if not value:
        raise ValueError(message)


def digest(data):
    return hashlib.sha256(data).hexdigest()


def file_digest(path):
    h = hashlib.sha256()
    with Path(path).open('rb') as stream:
        for block in iter(lambda: stream.read(1024 * 1024), b''):
            h.update(block)
    return h.hexdigest()


def encoded(value):
    return (json.dumps(value, ensure_ascii=False, indent=2) + '\n').encode('utf-8')


def extract_pages(archive):
    require(file_digest(archive) == ARCHIVE_SHA, 'ORIGINAL_ARCHIVE_SHA_MISMATCH')
    files, pages, inventory = {}, [], {}
    with zipfile.ZipFile(archive) as source:
        for page in range(1, 8):
            suffix = '/20261104-complete/raw/ksf_pass3_p%d_a1.json' % page
            names = [name for name in source.namelist() if name.endswith(suffix)]
            require(len(names) == 1, 'HISTORICAL_PAGE_MISSING_OR_AMBIGUOUS')
            raw = source.read(names[0])
            doc = json.loads(raw)
            result = doc['result']
            body = result['data']
            data = body['data']
            require(doc['httpStatus'] == 200 and doc['status'] == 'ok' and
                    result['code'] == 200 and body['code'] == 0, 'PAGE_NOT_SUCCESSFUL')
            query = body['requestParams']['query']
            require(str(doc['params']['shopId']) == SHOP and str(query['shopId']) == SHOP and
                    str(data['shopId']) == SHOP, 'STORE_ID_MISMATCH')
            require(int(doc['params']['page']) == page and int(query['page']) == page,
                    'REQUEST_PAGE_MISMATCH')
            require(data['totalResults'] == 165 and data['TotalPage'] == 7,
                    'REPORTED_COVERAGE_CHANGED')
            rows = data['ItemList']
            require(len(rows) == (21 if page == 7 else 24), 'PAGE_ROW_COUNT_MISMATCH')
            name = 'originals/ksf_pass3_p%d_a1.json' % page
            files[name] = raw
            ids = []
            for index, row in enumerate(rows):
                item_id = row['itemId']
                require(isinstance(item_id, str) and item_id.isdigit(), 'INVALID_EXACT_ITEM_ID')
                pid = 'jd:' + item_id
                require(pid not in inventory, 'DUPLICATE_LISTING_ACROSS_PAGES')
                inventory[pid] = {
                    'id': pid, 'itemId': item_id, 'platformFamily': 'jd',
                    'historical': {
                        'shopId': SHOP, 'shopName': data['shopName'],
                        'title': row['title'], 'url': row.get('link'),
                        'sourceTimeRaw': body.get('recordTime'),
                        'sourceTimezone': None,
                        'startedAt': doc.get('startedAt'), 'finishedAt': doc.get('finishedAt'),
                        'archiveMember': names[0], 'archiveSha256': ARCHIVE_SHA,
                        'responsePath': name, 'responseSha256': digest(raw),
                        'jsonPointer': ARRAY_POINTER + '/' + str(index), 'page': page,
                    },
                }
                ids.append(pid)
            pages.append({
                'page': page, 'requestPageRaw': doc['params']['page'], 'shopId': SHOP,
                'shopName': data['shopName'], 'rowCount': len(rows),
                'uniqueCount': len(set(ids)), 'reportedTotal': data['totalResults'],
                'reportedPages': data['TotalPage'], 'sourceTimeRaw': body.get('recordTime'),
                'startedAt': doc.get('startedAt'), 'finishedAt': doc.get('finishedAt'),
                'httpStatus': doc['httpStatus'], 'outerCode': result['code'],
                'businessCode': body['code'], 'archiveMember': names[0],
                'responsePath': name, 'sha256': digest(raw), 'productIds': ids,
            })
    require(len(inventory) == 165, 'HISTORICAL_UNIQUE_TOTAL_MISMATCH')
    return files, pages, inventory


def extract_competitors(archive):
    """Choose a complete single pass per store, never the union of retries."""
    groups, files = {}, {}
    with zipfile.ZipFile(archive) as source:
        for name in source.namelist():
            match = re.search(r'/20261104-complete/raw/([^/]+)_pass(\d+)_p(\d+)_a1\.json$', name)
            if not match or match[1] == 'ksf':
                continue
            raw = source.read(name)
            doc = json.loads(raw)
            if doc.get('id') != 'jd.get_shop_item_list_v1':
                continue
            result = doc.get('result', {})
            body = result.get('data') or {}
            data = body.get('data') or {}
            require(doc.get('httpStatus') == 200 and result.get('code') == 200 and
                    body.get('code') == 0, 'COMPETITOR_PAGE_FAILURE')
            shop = str(doc['params']['shopId'])
            require(shop == str(data['shopId']) == str(body['requestParams']['query']['shopId']),
                    'COMPETITOR_STORE_MISMATCH')
            page = int(doc['params']['page'])
            require(page == int(match[3]) == int(body['requestParams']['query']['page']),
                    'COMPETITOR_PAGE_MISMATCH')
            groups.setdefault((shop, int(match[2])), []).append({
                'archiveMember': name, 'sha256': digest(raw), 'raw': raw,
                'page': page, 'reportedPages': data['TotalPage'],
                'reportedTotal': data['totalResults'], 'rows': data['ItemList'],
                'sourceTimeRaw': body.get('recordTime'), 'shopName': data['shopName'],
            })
    audits, candidates = [], {}
    for (shop, pass_number), pages in sorted(groups.items()):
        ids = [str(row['itemId']) for page in pages for row in page['rows']]
        totals = {p['reportedTotal'] for p in pages}
        page_counts = {p['reportedPages'] for p in pages}
        got = {p['page'] for p in pages}
        complete = (len(totals) == 1 and len(page_counts) == 1 and
                    len(got) == len(pages) and
                    got == set(range(1, next(iter(page_counts)) + 1)) and
                    len(ids) == len(set(ids)) == next(iter(totals)))
        audit = {
            'shopId': shop, 'shopName': pages[0]['shopName'], 'pass': pass_number,
            'pages': len(pages), 'pageNumbers': sorted(got), 'rows': len(ids),
            'uniqueProducts': len(set(ids)), 'duplicateRows': len(ids) - len(set(ids)),
            'reportedTotals': sorted(totals), 'reportedPageCounts': sorted(page_counts),
            'accessibleHistoricalCatalogComplete': complete,
            'sourceTimeMinRaw': min(p['sourceTimeRaw'] for p in pages),
            'sourceTimeMaxRaw': max(p['sourceTimeRaw'] for p in pages),
            'responses': [{k: v for k, v in p.items() if k not in ('raw', 'rows')} for p in pages],
        }
        audits.append(audit)
        if complete and (shop not in candidates or candidates[shop][0]['pass'] < pass_number):
            candidates[shop] = (audit, pages)
    inventories, selected = [], []
    for shop, (audit, pages) in sorted(candidates.items()):
        selected.append(audit)
        for page in sorted(pages, key=lambda p: p['page']):
            path = 'competitors/originals/' + page['archiveMember'].rsplit('/', 1)[-1]
            files[path] = page['raw']
            for index, row in enumerate(page['rows']):
                item_id = row['itemId']
                require(isinstance(item_id, str) and item_id.isdigit(), 'INVALID_COMPETITOR_ITEM_ID')
                inventories.append({
                    'id': 'jd:' + item_id, 'historicalShopId': shop,
                    'historicalShopName': page['shopName'], 'historicalTitle': row['title'],
                    'historicalUrl': row.get('link'), 'sourceTimeRaw': page['sourceTimeRaw'],
                    'sourceTimezone': None, 'archiveSha256': ARCHIVE_SHA,
                    'archiveMember': page['archiveMember'], 'responsePath': path,
                    'responseSha256': page['sha256'], 'page': page['page'],
                    'jsonPointer': ARRAY_POINTER + '/' + str(index),
                })
    require(len(selected) == 7, 'EXPECTED_SEVEN_COMPETITOR_STORES')
    files['competitors/pass-audit.json'] = encoded(audits)
    return files, selected, inventories


class PublicBoard:
    def __init__(self):
        self.files = {}
        self.calls = 0

    def read(self, name, request):
        self.calls += 1
        req = urllib.request.Request(BOARD, data=encoded(request),
                                     headers={'Content-Type': 'application/json',
                                              'Accept': 'application/json'}, method='POST')
        with urllib.request.urlopen(req, timeout=30) as response:
            raw = response.read(8 * 1024 * 1024 + 1)
        require(len(raw) <= 8 * 1024 * 1024, 'BOARD_RESPONSE_TOO_LARGE')
        result = json.loads(raw)
        require(result.get('ok') is True, 'BOARD_READ_FAILED')
        self.files['board/' + name + '-request.json'] = encoded(request)
        self.files['board/' + name + '.json'] = raw
        return result

    def browse_all(self, entity, generation, shop=SHOP):
        rows, cursors, total = {}, set(), None
        cursor = None
        for page in range(1, 101):
            query = {'op': 'browse', 'entity': entity,
                     'filters': {'platform': 'jd', 'shopId': shop},
                     'quality': 'audit', 'limit': 100, 'generation': generation}
            if cursor:
                query['cursor'] = cursor
            result = self.read('%s-%s-p%d' % (shop, entity, page), query)
            require(result['generation'] == generation, 'BOARD_GENERATION_CHANGED')
            total = result['totalMatched'] if total is None else total
            require(total == result['totalMatched'], 'BOARD_MATCH_COUNT_CHANGED')
            for row in result['records']:
                require(row['record_id'] not in rows, 'DUPLICATE_BOARD_RECORD')
                require(row['platform'] == 'jd' and row['shop_id'] == shop,
                        'BOARD_RECORD_OUTSIDE_STORE_SCOPE')
                rows[row['record_id']] = row
            cursor = result.get('nextCursor')
            if not cursor:
                require(len(rows) == total, 'INCOMPLETE_BOARD_PAGINATION')
                return rows
            require(cursor not in cursors, 'BOARD_CURSOR_LOOP')
            cursors.add(cursor)
        raise ValueError('BOARD_PAGINATION_LIMIT')


def prepare(archive, out):
    require(not out.exists(), 'OUTPUT_ALREADY_EXISTS')
    files, pages, historical = extract_pages(archive)
    competitor_files, competitor_pages, competitor_inventory = extract_competitors(archive)
    files.update(competitor_files)
    board = PublicBoard()
    before = board.read('summary-before', {'op': 'summary'})
    generation = before['generation']
    current = board.browse_all('products', generation)
    prices = board.browse_all('prices', generation)
    competitor_summary = []
    for store in competitor_pages:
        shop = store['shopId']
        live = board.browse_all('products', generation, shop=shop)
        hist = [r for r in competitor_inventory if r['historicalShopId'] == shop]
        hist_ids, live_ids = {r['id'] for r in hist}, set(live)
        for row in hist:
            context = live.get(row['id'])
            row['current'] = ({'generation': generation, **{key: context.get(key) for key in (
                'title', 'shop_id', 'shop_name', 'brand', 'category', 'source_day', 'relatedCounts')
            }} if context else None)
        competitor_summary.append({
            'shopId': shop, 'shopName': store['shopName'], 'selectedHistoricalPass': store['pass'],
            'historicalPages': store['pages'], 'historicalUniqueProducts': len(hist_ids),
            'historicalSourceTimeMinRaw': store['sourceTimeMinRaw'],
            'historicalSourceTimeMaxRaw': store['sourceTimeMaxRaw'],
            'accessibleHistoricalCatalogComplete': True, 'currentStoreProducts': len(live),
            'sameIds': len(hist_ids & live_ids),
            'historicalIdsAbsentFromCurrentStore': sorted(hist_ids - live_ids),
            'currentIdsAbsentFromHistoricalPages': sorted(live_ids - hist_ids),
            'currentProductsWithReviews': sum(r.get('relatedCounts', {}).get('reviews', 0) > 0
                                             for r in live.values()),
            'currentStoredReviewCount': sum(r.get('relatedCounts', {}).get('reviews', 0)
                                           for r in live.values()),
            'currentCategoryCounts': dict(Counter(r.get('category') or '(unknown)' for r in live.values())),
            'brandOwnershipInferredFromStoreName': False,
        })
    files['competitors/coverage.json'] = encoded(competitor_summary)
    files['competitors/inventory.json'] = encoded(competitor_inventory)
    after = board.read('summary-after', {'op': 'summary'})
    require(after['generation'] == generation, 'BOARD_CHANGED_DURING_READBACK')
    files.update(board.files)
    hist_ids, current_ids = set(historical), set(current)
    # Save product context separately from historical observations; never overwrite it.
    rows = []
    price_by_product = Counter(row['product_id'] for row in prices.values())
    for pid in sorted(hist_ids | current_ids):
        item = dict(historical.get(pid, {'id': pid}))
        live = current.get(pid)
        item['current'] = ({
            'generation': generation,
            **{key: live.get(key) for key in (
                'title', 'shop_id', 'shop_name', 'brand', 'category', 'source_day',
                'classification_state', 'source_record_id', 'relatedCounts')},
        } if live else None)
        item['historicalOnly'] = pid not in current_ids
        item['currentOnly'] = pid not in hist_ids
        item['currentReviewCount'] = (live or {}).get('relatedCounts', {}).get('reviews', 0)
        item['currentPriceCount'] = price_by_product.get(pid, 0)
        rows.append(item)
    category_groups = []
    for category in sorted({row.get('category') or '(unknown)' for row in current.values()}):
        selected = [r for r in rows if r['current'] and
                    (r['current'].get('category') or '(unknown)') == category]
        category_groups.append({
            'category': None if category == '(unknown)' else category,
            'products': len(selected),
            'productsWithReviews': sum(r['currentReviewCount'] > 0 for r in selected),
            'storedReviewCount': sum(r['currentReviewCount'] for r in selected),
            'productsWithPrices': sum(r['currentPriceCount'] > 0 for r in selected),
            'storedPriceCount': sum(r['currentPriceCount'] for r in selected),
        })
    price_groups = []
    # Different price bases and dates remain distinct; no per-unit normalization.
    keys = {(p.get('price_basis'), p.get('source_day')) for p in prices.values()}
    for basis, day in sorted(keys, key=lambda k: (str(k[0]), str(k[1]))):
        selected = [p for p in prices.values()
                    if p.get('price_basis') == basis and p.get('source_day') == day]
        known = [p['amount_minor'] for p in selected if p.get('amount_minor') is not None]
        price_groups.append({
            'priceBasis': basis, 'sourceCalendarDate': day, 'observations': len(selected),
            'products': len({p['product_id'] for p in selected}),
            'knownAmountObservations': len(known),
            'minAmountMinor': min(known) if known else None,
            'maxAmountMinor': max(known) if known else None,
            'interpretation': '历史整件展示价格观察；非当前成交价或同规格单位价',
        })
    with_reviews = [r for r in rows if r['current'] and r['currentReviewCount'] > 0]
    report = {
        'schema': 'ksf.jd-historical-store-baseline.v1', 'ok': True,
        'generatedAt': dt.datetime.now(dt.timezone.utc).isoformat(),
        'collectorCalls': 0, 'newPaidCalls': 0, 'newCollectionCostCny': '0.00',
        'productionWrites': 0, 'databaseWrites': 0, 'publicBoardReads': board.calls,
        'shop': {'platformFamily': 'jd', 'shopId': SHOP,
                 'shopName': '康师傅方便面京东自营旗舰店'},
        'historical': {
            'archiveSha256': ARCHIVE_SHA, 'archiveBytes': archive.stat().st_size,
            'pages': 7, 'rows': 165, 'uniqueProducts': 165, 'duplicateRows': 0,
            'reportedTotal': 165, 'reportedPages': 7,
            'accessibleHistoricalCatalogComplete': True,
            'completenessBasis': '请求与响应店铺/页码一致，1至7页齐全，165唯一ID等于各页声明总数',
            'sourceTimeMinRaw': min(p['sourceTimeRaw'] for p in pages),
            'sourceTimeMaxRaw': max(p['sourceTimeRaw'] for p in pages),
            'sourceTimezone': None,
            'currentLiveCatalogComplete': False,
            'transactionalSnapshotProven': False,
        },
        'production': {
            'url': BOARD, 'generation': generation, 'generationUnchanged': True,
            'libraryCounts': before['counts'], 'storeProducts': len(current),
            'sameIds': len(hist_ids & current_ids),
            'historicalIdsAbsentFromCurrentStore': sorted(hist_ids - current_ids),
            'currentStoreIdsAbsentFromHistoricalPages': sorted(current_ids - hist_ids),
            'productsWithStoredReviews': len(with_reviews),
            'productsWithoutStoredReviews': len(current) - len(with_reviews),
            'storedReviewCount': sum(r['currentReviewCount'] for r in with_reviews),
            'productsWithStoredPrices': len(price_by_product),
            'storedPriceCount': len(prices),
            'productsWithNumericPrice': len({r['product_id'] for r in prices.values()
                                             if r.get('amount_minor') is not None}),
            'numericPriceObservations': sum(r.get('amount_minor') is not None for r in prices.values()),
            'noReviewsDoesNotMeanPlatformZero': True,
            'reviewCountBasis': '公开商品检索relatedCounts；含待确认评价，不是analysis数量',
        },
        'categories': category_groups, 'historicalPriceGroups': price_groups,
        'competitors': {
            'stores': len(competitor_summary),
            'historicalListingsAcrossStores': len(competitor_inventory),
            'historicalPages': sum(r['historicalPages'] for r in competitor_summary),
            'storeCoverage': competitor_summary,
            'completeIndustryUniverseClaimed': False,
            'currentLiveCatalogComplete': False,
        },
        'gatesStillOpen': [
            '不同观察轮次的采集稳定性与主店最终选择',
            '未知/待细分品类、SKU验证、商品宣称与卖点证据',
            '逐商品评价接口可访问范围与结束条件',
            '项目全部窗口费用原账本去重；不能拿ksf前缀子集推总余额',
        ],
    }
    files['coverage.json'] = encoded(report)
    files['pages.json'] = encoded(pages)
    files['inventory.json'] = encoded(rows)
    files['price-observations.json'] = encoded(list(prices.values()))
    files['review-gaps.json'] = encoded([
        {'id': r['id'], 'category': r['current']['category'],
         'storedReviews': r['currentReviewCount'], 'storedPrices': r['currentPriceCount'],
         'title': r['current']['title']}
        for r in rows if r['current'] and not r['currentReviewCount']
    ])
    lines = [
        '# 京东康师傅候选店历史目录基线', '',
        '用途：复用已采原件建立历史目录与评价缺口，不再为这7页历史资料付费重采。', '',
        '- 店铺：1000084104（康师傅方便面京东自营旗舰店）。',
        '- 历史：2026-11-04，7页，24×6＋21＝165条，165唯一商品，无重叠；与接口总数一致。',
        '- 现库：%s代，店铺%d商品；历史/现库交集%d。' %
        (generation, len(current), len(hist_ids & current_ids)),
        '- 店内%d商品有收录评价、%d商品未收录，评价%d条（含待确认）。' %
        (len(with_reviews), len(current) - len(with_reviews), report['production']['storedReviewCount']),
        '- %d商品有%d条价格观察；按价格口径和日期分组，不能混作当前售价。' %
        (len(price_by_product), len(prices)),
        '- 本次新增付费0元、生产写入0；%d次公开看板只读请求。' % board.calls,
        '- 另核7家京东竞品店历史目录，57页/1279个店内商品观察；按独立pass验证，未拼接重试轮次。',
        '', '## 文件', '',
        '`originals/`保留7页原字节；`pages.json`含页码/来源/SHA/Pointer；',
        '`inventory.json`分别保留历史与现库字段；`review-gaps.json`列评价空白商品；',
        '`coverage.json`给出分类覆盖与口径；`price-observations.json`保留已有价格；',
        '`board/`为代际固定的只读核对原响应。', '',
        '`competitors/`含7店历史目录原件、逐轮分页审计、现库核对与商品清单。', '',
        '## 范围', '',
        '已验收的是该接口当时可访问的历史店铺目录，不是今天在售全集、原子快照、完整SKU、',
        '全历史评价或已选定长期主店。未知分类不按标题补齐；不能用店铺归属代替品牌证据。',
        '本包是覆盖证据与执行输入，不是待发布batch，不要将旧目录再次覆盖现库。',
        '本包不包含项目总费用核实；账本去重后再决定新增付费采集范围。', '',
    ]
    files['README.md'] = '\n'.join(lines).encode('utf-8')
    out.mkdir(parents=True, exist_ok=False)
    for name, raw in files.items():
        target = out / name
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_bytes(raw)
    manifest = ''.join('%s  %s\n' % (digest(raw), name) for name, raw in sorted(files.items()))
    (out / 'FILES.sha256').write_bytes(manifest.encode('utf-8'))
    return report


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--archive', type=Path, required=True)
    parser.add_argument('--out', type=Path, required=True)
    args = parser.parse_args()
    result = prepare(args.archive.resolve(), args.out.resolve())
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    main()