kangshifu-board/scripts/prepare-industry-handoff.py

"""Build a bounded, read-only scope/quotation handoff from frozen local evidence. Never dispatch."""
import argparse
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
import sqlite3
import sys

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'server/voc-library'))
from industry_monitor import Monitor, load
from monitoring_store import need, read_state
from monitoring_dispatch import projection as budget_projection
from ranking_discovery import projection as ranking_projection
from monitoring_details import projection as detail_projection


def digest(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()


def build(snapshot, store, catalog, out):
    need(not out.exists(), 'HANDOFF_OUTPUT_ALREADY_EXISTS')
    snapshot, store, catalog = Path(snapshot), Path(store), Path(catalog)
    before = digest(snapshot)
    data = load({'snapshot': str(snapshot)})
    generation = data['generation']
    runs = read_state(store)
    monitor = Monitor(data, runs)
    primary = monitor.selected({'platform': 'jd', 'shopId': '1000084104'})
    primary_ids = {p['record_id'] for p in primary}
    product_ids = {p['record_id'] for p in monitor.selected({})}
    catalog_data = json.loads(catalog.read_bytes())
    need(catalog_data['schema'] == 'ksf.industry-connectors.v1' and catalog_data['networkCalls'] == 0,
         'READ_ONLY_CONNECTOR_CATALOG_REQUIRED')
    budget = budget_projection(store, generation)
    need(budget['paidDispatchStopped'] is True and budget['remainingDispatchMinor'] == 0,
         'STOPPED_BUDGET_REQUIRED')
    listing_runs = [r for r in runs if r['kind'] == 'listings']
    shops = []
    for r in listing_runs:
        current = monitor.selected({'platform': r['platform'], 'shopId': r['scope']['shopId']})
        shops.append({'platform': r['platform'], 'shopId': r['scope']['shopId'], 'name': r.get('scopeName'),
                      'historicalProducts': len({e['productId'] for e in r['entries']}),
                      'historicalPages': len(r['pages']), 'historicalObservedAtRaw': r['observedAt'],
                      'historicalRunId': r['runId'], 'currentLibraryProducts': len(current),
                      'sourcePages': [{'page': p['page'], 'requestId': p['requestKey'], 'sha256': p.get('rawSha256')} for p in r['pages']],
                      'freshCollectionAuthorized': False})
    reviews_by_product = {}
    for r in runs:
        if r['kind'] == 'reviews':
            reviews_by_product.setdefault(r['scope']['productId'], []).append(r)
    from contextlib import closing
    with closing(sqlite3.connect((store / 'monitoring.sqlite').resolve().as_uri() + '?mode=ro', uri=True)) as c:
        c.execute('PRAGMA query_only=ON')
        charged = c.execute('SELECT product_id,page,request_id,charged_minor,business_result FROM dispatch_history ORDER BY product_id,page').fetchall()
        c.row_factory = sqlite3.Row
        from monitoring_ledger import read as read_ledger
        _, ledger_rows = read_ledger(c)
    charged_by_product = {}
    if ledger_rows:
        bound_charges = [(('jd:' if row['endpointId'].startswith('jd.') else 'taobao:') + row['params']['itemId'],
                          row['params']['page'], row['requestId'], row['chargedMinor'], row['businessResult'])
                         for row in ledger_rows if row['params']]
    else:
        bound_charges = charged
    for pid, page, request, cost, result in bound_charges:
        charged_by_product.setdefault(pid, []).append({'page': page, 'requestId': request,
                                                       'chargedMinor': cost, 'businessResult': result,
                                                       'automaticRetryAllowed': False})
    products = []
    for p in sorted(primary, key=lambda p: p['record_id']):
        pid = p['record_id']
        observations = []
        for r in reviews_by_product.get(pid, []):
            # This is raw interface metadata, not a promise that all pages are retrievable or billable.
            limits = []
            from monitoring_store import pointer
            for page in r['pages']:
                loc = page.get('mapping', {}).get('pageCountPointer')
                if page['status'] == 'success' and loc:
                    doc = json.loads((store / 'assets' / (page['rawSha256'] + '.json')).read_bytes())
                    limits.append(pointer(doc, loc))
            observations.append({'runId': r['runId'], 'observedAtRaw': r['observedAt'],
                                 'receivedPages': [v['page'] for v in r['pages']],
                                 'declaredPageLimits': limits, 'reportedTotal': r.get('reportedTotal'),
                                 'reasons': r['verification']['reasons'],
                                 'completeAccessibleScope': r['verification']['completeAccessibleScope'],
                                 'identityConflicts': r['verification']['identityConflicts']})
        products.append({'productId': pid, 'title': p['title'], 'category': p.get('category'),
                         'libraryReviewRows': len(monitor.reviews[pid]), 'registeredReviewRuns': observations,
                         'settledRequests': charged_by_product.get(pid, []),
                         'dispatchAllowed': False,
                         'neededAction': 'reuse_originals_then_quote_missing_accessible_pages'})
    ranks = ranking_projection(store, generation, {}, product_ids)
    detail = detail_projection(store, generation, product_ids, {'state': 'IMAGE_ONLY'})
    from monitoring_policy import projection as policy_projection
    checked_policy = policy_projection(store, generation).get('documents', {})
    policy = [{'documentId': p['record_id'], 'title': p.get('title'),
               'declaredStatus': p.get('declared_status'), 'declaredStatusAsOf': p.get('status_as_of'),
               'sourceRecordId': p.get('source_record_id'), 'officialCheck': checked_policy.get(p['record_id']),
               'reverificationNeeded': not checked_policy.get(p['record_id'], {}).get('officialMetadataVerified', False)}
              for p in data['entities']['documents']]
    result = {'schema': 'ksf.mainline-handoff.v1', 'createdAt': datetime.now(timezone.utc).isoformat(),
              'generation': generation, 'snapshotSha256': before, 'catalogSha256': digest(catalog),
              'dispatchAllowed': False, 'paidCalls': 0, 'researchWrites': 0, 'budget': budget,
              'primaryShop': {'platform': 'jd', 'shopId': '1000084104', 'products': len(primary),
                              **{k: monitor.review_coverage({'platform': 'jd', 'shopId': '1000084104'})[k]
                                 for k in ('reviewRows', 'uniqueReadableTexts')}},
              'shops': shops, 'primaryProducts': products,
              'chargedHistory': {'identifiedRows': budget['identifiedSettledRequests'],
                                'parameterBoundRows': budget.get('parameterBoundSettledRequests', len(charged)),
                                'queueUnionRows': len(charged), 'priorAggregateRows': budget['aggregateOnlyRows'],
                                'all154IdentitiesVerified': budget['fullHistoricalIdentitiesVerified'],
                                'historyOutsidePrimaryShop': sorted(set(charged_by_product) - primary_ids)},
              'rankingIdentities': ranks, 'detailResponseCoverage': detail['coverage'],
              'policyDocumentsForStatusReview': policy,
              'connectors': catalog_data['existingConnectors'],
              'rankingConnectors': catalog_data['jdTmallRankingCandidates'],
              'quotation': {'status': 'NOT_OBTAINED', 'currency': 'CNY', 'authorizedNewMinor': 0,
                            'reviewsHistoricalUnitMinor': 30, 'historicalRateIsCurrentQuote': False,
                            'estimatedTotalMinor': None,
                            'reasons': ['current_unit_prices_not_confirmed', 'accessible_page_limits_not_confirmed',
                                        'ranking_endpoint_not_confirmed'] +
                                       (['remaining_15_request_parameter_bindings_incomplete'] if ledger_rows else
                                        ['prior_78_request_identities_incomplete']),
                            'requiredBreakdown': ['already_paid_original_recovery', 'current_8shop_listings',
                                                   'primary_shop_missing_accessible_review_pages',
                                                   'detail_claim_text_and_images', 'ranking_top100_or_top400',
                                                   'monitoring_cost_per_day_and_stop_limit']}}
    need(digest(snapshot) == before, 'SNAPSHOT_CHANGED')
    out.mkdir(parents=True)
    (out / 'collector-scope.json').write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n', 'utf-8')
    summary = {'ok': True, 'generation': generation, 'snapshotSha256': before, 'shops': len(shops),
               'primaryProducts': len(products), 'identifiedChargedRequests': budget['identifiedSettledRequests'],
               'parameterBoundChargedRequests': budget.get('parameterBoundSettledRequests', len(charged)),
               'queueUnionRequests': len(charged),
               'reviewProductsWithRegisteredResponse': sum(bool(p['registeredReviewRuns']) for p in products),
               'reviewProductsWithoutRegisteredResponse': sum(not p['registeredReviewRuns'] for p in products),
               'rankingIdentities': ranks['knownRankIds'], 'estimatedTotalMinor': None,
               'newBudgetAuthorizedMinor': 0, 'paidCalls': 0, 'researchWrites': 0}
    (out / 'scope-receipt.json').write_text(json.dumps(summary, ensure_ascii=False, indent=2) + '\n', 'utf-8')
    return summary


if __name__ == '__main__':
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('--snapshot', type=Path, required=True)
    p.add_argument('--store', type=Path, required=True)
    p.add_argument('--catalog', type=Path, required=True)
    p.add_argument('--out', type=Path, required=True)
    args = p.parse_args()
    print(json.dumps(build(args.snapshot, args.store, args.catalog, args.out), ensure_ascii=False))