kangshifu-board/scripts/audit-unavailable-research.py

"""Separate source disappearance from missing prose or restricted access."""
from datetime import datetime, timezone
import hashlib
from html.parser import HTMLParser
import json
from pathlib import Path
import re

WORK = Path(__file__).resolve().parents[1] / 'work/database-enrichment-20261110'


def official_bili_state(doc, http):
    # 62002 means inaccessible. It does not prove deletion and must not become
    # a destructive decision just because a proxy labels it NO VIDEO.
    if http in (401, 403, 412, 429) or doc.get('code') in (-101, -111, -352, 62002):
        return 'ACCESS_RESTRICTED_NOT_CONFIRMED_REMOVED'
    if http == 200 and doc.get('code') == -404:
        return 'OFFICIAL_NOT_FOUND_REQUIRES_PAGE_CONFIRMATION'
    if http == 200 and doc.get('code') == 0 and isinstance(doc.get('data'), dict):
        return 'SOURCE_RECORD_EXISTS'
    return 'UNCONFIRMED_RESPONSE'


class VisibleText(HTMLParser):
    def __init__(self):
        super().__init__()
        self.hidden, self.text = 0, []

    def handle_starttag(self, tag, attrs):
        if tag in ('head', 'script', 'style', 'template'): self.hidden += 1

    def handle_endtag(self, tag):
        if tag in ('head', 'script', 'style', 'template'): self.hidden = max(0, self.hidden - 1)

    def handle_data(self, text):
        if not self.hidden: self.text.append(text)


def page_confirms_removed(raw, meta, iid):
    if meta.get('httpStatus') not in (200, 404): return False
    # An authentication redirect or error text found only in JS is insufficient.
    if not re.fullmatch(r'https://(?:www\.)?bilibili\.com/video/' + re.escape(iid) + r'/?', meta.get('finalUrl', '')):
        return False
    parser = VisibleText()
    parser.feed(raw.decode('utf-8', errors='replace'))
    return bool(re.search(r'视频(?:不见了|已被删除|已删除|不存在)|稿件(?:已删除|不存在)', ''.join(parser.text)))


def audit():
    source = json.loads((WORK / 'social-body-audit.json').read_bytes())
    rows = []
    for post in source['posts']:
        row = {'id': post['id'], 'deleteEligible': False, 'attempts': len(post['attempts'])}
        states = {a['state'] for a in post['attempts']}
        if states & {'UPSTREAM_BODY_EMPTY', 'UPSTREAM_BODY_PLACEHOLDER', 'BODY_ONLY_REPEATS_TITLE'}:
            row.update(state='MISSING_AUTHOR_TEXT_NOT_EVIDENCE_OF_REMOVAL', reason='Exact source object returned; missing description or repeated title does not mean video or images disappeared.')
        elif 'UPSTREAM_DETAIL_UNAVAILABLE' in states:
            row.update(state='SOURCE_REPORTED_UNAVAILABLE_REQUIRES_VERIFICATION')
        else:
            row.update(state='COLLECTION_FAILED_OR_MISMATCH_NOT_EVIDENCE_OF_REMOVAL')
        if post['platform'] == 'bilibili':
            iid = post['id'].split(':', 1)[1]
            meta_path = WORK / 'link-checks' / (iid + '--official-api.json')
            if meta_path.exists():
                meta = json.loads(meta_path.read_bytes())
                body_path = meta_path.parent / meta['responseFile']
                raw = body_path.read_bytes()
                if hashlib.sha256(raw).hexdigest() != meta['sha256']:
                    raise ValueError('OFFICIAL_RESPONSE_HASH_MISMATCH')
                state = official_bili_state(json.loads(raw), meta['httpStatus'])
                row.update(state=state,
                           officialEvidence={**meta, 'shaVerified': True})
                page_path = meta_path.with_name(iid + '--official-page.json')
                if state == 'OFFICIAL_NOT_FOUND_REQUIRES_PAGE_CONFIRMATION' and page_path.exists():
                    page_meta = json.loads(page_path.read_bytes())
                    page_raw = (page_path.parent / page_meta['responseFile']).read_bytes()
                    if hashlib.sha256(page_raw).hexdigest() != page_meta['sha256']:
                        raise ValueError('OFFICIAL_PAGE_HASH_MISMATCH')
                    if page_confirms_removed(page_raw, page_meta, iid):
                        row.update(state='CONFIRMED_ORIGINAL_SOURCE_REMOVED', deleteEligible=True,
                                   pageEvidence={**page_meta, 'shaVerified': True})
        rows.append(row)
    # Audit available product details too. Out-of-stock and a review page with
    # no rows never establish that the original product link has disappeared.
    products = {}
    for meta_path in sorted((WORK / 'collected').glob('*.meta.json')):
        meta = json.loads(meta_path.read_bytes())
        if meta.get('kind') not in ('product-detail', 'product-reviews'): continue
        row = products.setdefault(meta['target'], {'id': meta['target'], 'deleteEligible': False, 'evidence': []})
        attempt = {'endpoint': meta['id'], 'metaFile': meta_path.name, 'httpStatus': meta.get('httpStatus')}
        if not meta.get('ok'):
            attempt['state'] = 'REQUEST_FAILED_NOT_EVIDENCE_OF_REMOVAL'
        else:
            raw = (meta_path.parent / meta['responseFile']).read_bytes()
            if hashlib.sha256(raw).hexdigest() != meta['sha256']:
                raise ValueError('PRODUCT_RESPONSE_HASH_MISMATCH')
            attempt.update(responseFile=meta['responseFile'], sha256=meta['sha256'], shaVerified=True)
            doc = json.loads(raw)
            wrapper = doc.get('data', {})
            payload = wrapper.get('data', {}) if isinstance(wrapper, dict) else {}
            product = payload.get('product' if meta['target'].startswith('jd:') else 'item', {}) if isinstance(payload, dict) else {}
            iid = product.get('skuId', product.get('itemId')) if isinstance(product, dict) else None
            if meta['kind'] == 'product-detail' and str(iid) == meta['target'].split(':', 1)[1]:
                attempt.update(state='EXACT_PRODUCT_OBJECT_EXISTS', title=product.get('skuName', product.get('title')))
                row['state'] = 'SOURCE_PRODUCT_EXISTS_KEEP_RECORD'
            else:
                attempt['state'] = 'REVIEW_RESPONSE_OR_UNCONFIRMED_DETAIL_NOT_EVIDENCE_OF_REMOVAL'
        row['evidence'].append(attempt)
    for row in products.values():
        row.setdefault('state', 'AVAILABILITY_UNCONFIRMED_KEEP_RECORD')
        rows.append(row)
    result = {'schema': 'ksf.unavailable-research-audit.v1', 'checkedAt': datetime.now(timezone.utc).isoformat(),
              'authorization': 'User authorized removing research records only when original link is confirmed to have no remaining source content.',
              'summary': {'recordsChecked': len(rows), 'socialChecked': len(source['posts']), 'productsChecked': len(products),
                          'confirmedDeletionCandidates': sum(r['deleteEligible'] for r in rows),
                          'removedRecords': 0, 'productionWrites': 0}, 'records': rows}
    (WORK / 'unavailable-records-audit.json').write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8')
    print(json.dumps(result['summary']))
    return result


if __name__ == '__main__': audit()