kangshifu-board/docs/unified-database-migration-20261109/release-r2/package/components/research-runtime/scripts/update_library.py

"""Authorized batch merge into a new immutable generation, then atomic pointer switch.

Incoming databases must first be built with the existing validated VOC importer.
Default is a dry plan. Readers continue using the old generation until --apply succeeds.
"""
import sys
sys.dont_write_bytecode=True
import argparse
import datetime as dt
from decimal import Decimal
import hashlib
import json
import os
import re
from pathlib import Path
import shutil
import sqlite3
import tempfile
import uuid
import voc_db
import retrieval
from product_insights import ANNOTATION_DDL

ORDER=['batches','assets','collection_requests','brands','categories','source_records','source_links','products',
       'classification_assertions','product_observations','specs','prices','price_sources','reviews','social_posts',
       'social_brand_mentions','policy_documents','policy_clauses','ranking_signals','coverage_observations','quality_issues','review_theme_annotations']
STABLE={'products','reviews','social_posts','policy_documents','policy_clauses'}
MERGE_POLICY='ksf.voc-merge.20261106.v1'
MERGE_DDL='''CREATE TABLE IF NOT EXISTS merge_decisions(
 id TEXT PRIMARY KEY, entity TEXT NOT NULL, record_id TEXT NOT NULL, batch_sha256 TEXT NOT NULL,
 prior_source_record_id TEXT NOT NULL REFERENCES source_records(id),
 incoming_source_record_id TEXT NOT NULL REFERENCES source_records(id),
 selected_source_record_id TEXT NOT NULL REFERENCES source_records(id),
 decision TEXT NOT NULL CHECK(decision IN ('kept_existing','used_incoming')),
 reason_codes_json TEXT NOT NULL CHECK(json_valid(reason_codes_json)), policy_version TEXT NOT NULL);
 CREATE INDEX IF NOT EXISTS idx_merge_decisions_record ON merge_decisions(entity,record_id,id);'''

def observation_stamp(value):
    """Keep naive wall time separate from UTC; preserve submicrosecond precision."""
    if not isinstance(value,str):return None
    match=re.fullmatch(r'(\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2})(?:\.(\d{1,9}))?(Z|[+-]\d{2}:\d{2})?',value)
    if not match:return None
    try:stamp=dt.datetime.fromisoformat(match[1]+(match[3] or '').replace('Z','+00:00'))
    except ValueError:return None
    aware=stamp.utcoffset() is not None
    if aware:stamp=stamp.astimezone(dt.timezone.utc).replace(tzinfo=None)
    return aware,stamp,Decimal('0.'+(match[2] or '0'))

def observation_order(old,new):
    before,after=observation_stamp(old),observation_stamp(new)
    if before is None or after is None or before[0]!=after[0]:return None
    # Both naive: compare only their source wall clocks, never label these UTC.
    return (after[1:]>before[1:])-(after[1:]<before[1:])

def unknown(value):
    return value is None or isinstance(value,str) and value.strip() in (
        '', 'null', '{}', '[]', 'unknown', 'unverified', 'unresolved', 'pending',
        'taobao_unverified', 'unspecified_not_verified')

def choose_record(table,old,new):
    """Select one coherent source row; never synthesize an untraceable hybrid."""
    bookkeeping={'id','source_record_id','collected_at_raw','collected_at_utc','source_time_raw',
                 'source_time_utc','source_calendar_date','review_date_utc','published_at_utc','coverage_type'}
    fields=set(old)-bookkeeping
    missing=sorted(k for k in fields if not unknown(old[k]) and unknown(new[k]))
    different=sorted(k for k in fields if not unknown(old[k]) and not unknown(new[k]) and old[k]!=new[k])
    identity={'products':{'marketplace','shop_id','brand_code','category_code'},
              'reviews':{'product_id','marketplace_claim','content','purchased_sku_text','rating_raw','rating_scale','review_date_raw','is_default'},
              'policy_clauses':{'document_id','clause','excerpt'}}.get(table,set())
    conflicts=sorted(set(different)&identity)
    ranks={'source_attribute':3,'rule_title_prefix':1,'rule_title_explicit':1}
    degraded=sorted(k for k in ('brand_basis','category_basis') if k in old and
                    ranks.get(old[k],0)>ranks.get(new[k],0))
    order=observation_order(old.get('source_time_raw'),new.get('source_time_raw'))
    reasons=[]
    if missing:reasons.append('KNOWN_VALUES_WOULD_BE_LOST')
    if degraded:reasons.append('CLASSIFICATION_EVIDENCE_WOULD_WEAKEN')
    if conflicts:reasons.append('IDENTITY_OR_FACT_CONFLICT')
    if order==-1:reasons.append('OLDER_OBSERVATION')
    if different and order in (None,0):reasons.append('CHANGED_VALUES_WITHOUT_NEWER_OBSERVATION')
    if reasons:decision='kept_existing'
    elif all(old[k]==new[k] or unknown(old[k]) and unknown(new[k]) for k in fields) and order!=1:
        decision='kept_existing';reasons=['EQUIVALENT_RECORD']
    else:decision='used_incoming';reasons=['NEWER_COMPLETE_OBSERVATION' if order==1 else 'NONCONFLICTING_ENRICHMENT']
    return decision,{'codes':reasons,'missingFields':missing,'conflictingFields':conflicts,
                     'weakerEvidenceFields':degraded,'changedKnownFields':different,'sourceTimeOrder':order}
def digest(p):
    with Path(p).open('rb') as f:return voc_db.file_sha256(f)
def canonical(v):return json.dumps(v,ensure_ascii=False,sort_keys=True,separators=(',',':'))
def verified(p):
    with voc_db.readonly(p) as c:
        voc_db.require(c.execute('PRAGMA integrity_check').fetchone()[0]=='ok' and not c.execute('PRAGMA foreign_key_check').fetchall(),'DATABASE_CHECK_FAILED')
        voc_db.require(c.execute('SELECT count(*) FROM retrieval_documents').fetchone()[0]==sum(c.execute('SELECT count(*) FROM '+t).fetchone()[0] for t in retrieval.ENTITIES.values()),'RETRIEVAL_COUNT_MISMATCH')
        for table in ORDER:
            if table=='review_theme_annotations':continue  # Optional, backward-compatible enrichment.
            voc_db.require(c.execute('SELECT sql FROM sqlite_master WHERE type=? AND name=?',('table',table)).fetchone() is not None,'MISSING_TABLE:'+table)
        return dict(c.execute('SELECT * FROM meta'))

def merge(current,incoming,dest,review_correction=None):
    dest=Path(dest)
    voc_db.require(not dest.exists() and not dest.is_symlink(),'NEW_MERGE_TARGET_REQUIRED')
    im=verified(incoming)
    correction=None
    if review_correction is not None:
        from review_correction import validate
        correction=validate(current,incoming,review_correction)
    if not current:shutil.copyfile(incoming,dest)
    else:
        with voc_db.readonly(current) as old:
            con=sqlite3.connect(dest);old.backup(con);con.close()
    c=sqlite3.connect(dest);c.row_factory=sqlite3.Row;c.execute('PRAGMA foreign_keys=ON')
    maps={};delta={'added':{},'changed':{},'duplicate':{},'retained':{},
        'mergeReview':{'policy':MERGE_POLICY,'decisions':0,'conflictRecords':0,'reasonCounts':{}}}
    if correction is not None:delta['reviewCorrection']=correction
    try:
        c.executescript(ANNOTATION_DDL)
        c.executescript(MERGE_DDL)
        c.execute('CREATE TABLE IF NOT EXISTS source_origins(source_record_id TEXT NOT NULL REFERENCES source_records(id),archive_sha256 TEXT NOT NULL,PRIMARY KEY(source_record_id,archive_sha256))')
        c.execute('CREATE TABLE IF NOT EXISTS record_versions(entity TEXT NOT NULL,record_id TEXT NOT NULL,batch_sha256 TEXT NOT NULL,record_json TEXT NOT NULL,PRIMARY KEY(entity,record_id,batch_sha256))')
        baseline=dict(c.execute('SELECT * FROM meta'))
        c.execute('INSERT OR IGNORE INTO source_origins SELECT id,? FROM source_records s WHERE NOT EXISTS (SELECT 1 FROM source_origins o WHERE o.source_record_id=s.id)',(baseline['archive_sha256'],))
        if current:
            with voc_db.readonly(incoming) as inc:
                for table in ORDER:
                    if not inc.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?",(table,)).fetchone():continue
                    info=c.execute('PRAGMA table_info('+table+')').fetchall()
                    pk=[r['name'] for r in sorted(info,key=lambda r:r['pk']) if r['pk']]
                    fks=[dict(r) for r in c.execute('PRAGMA foreign_key_list('+table+')')]
                    mapping={};maps[table]=mapping
                    for original in inc.execute('SELECT * FROM '+table):
                        row=dict(original)
                        for fk in fks:
                            value=row[fk['from']]
                            if value is not None:row[fk['from']]=maps.get(fk['table'],{}).get(value,value)
                        if table=='quality_issues' and row['entity_type']=='source':
                            row['entity_id']=maps.get('source_records',{}).get(row['entity_id'],row['entity_id'])
                        if table=='quality_issues' and row['entity_type'] in ('product','review'):
                            entity='products' if row['entity_type']=='product' else 'reviews'
                            candidate=c.execute('SELECT incoming_source_record_id,selected_source_record_id FROM merge_decisions WHERE entity=? AND record_id=? AND batch_sha256=? AND decision=?',
                                (entity,row['entity_id'],im['archive_sha256'],'kept_existing')).fetchone()
                            if candidate:
                                row['detail']=canonical({'scope':'incoming_candidate_not_current_row',
                                    'candidateSourceRecordId':candidate[0],'selectedSourceRecordId':candidate[1],'originalDetail':row['detail']})
                        columns=list(row)
                        old=c.execute('SELECT * FROM '+table+' WHERE '+' AND '.join(k+'=?' for k in pk),[row[k] for k in pk]).fetchone()
                        if table=='batches' and old and old['archive_sha256']==row['archive_sha256']:
                            mapping[original['id']]=old['id'];delta['duplicate'][table]=delta['duplicate'].get(table,0)+1;continue
                        # The same timestamped price observation may be re-exported under a new source locator.
                        # Unknown-time observations only deduplicate with identical source bytes.
                        if table=='prices' and not old:
                            fields=[k for k in columns if k not in ('id','source_record_id','collected_at_raw','collected_at_utc')]
                            sql='SELECT p.id FROM prices p JOIN source_records s ON s.id=p.source_record_id WHERE '+' AND '.join('p.'+k+' IS ?' for k in fields)
                            vals=[row[k] for k in fields]
                            if not row.get('source_time_raw'):
                                sql+=' AND s.payload_sha256=(SELECT payload_sha256 FROM source_records WHERE id=?)';vals.append(row['source_record_id'])
                            same_price=c.execute(sql+' LIMIT 1',vals).fetchone()
                            if same_price:
                                mapping[original['id']]=same_price['id'];delta['duplicate'][table]=delta['duplicate'].get(table,0)+1;continue
                        if table in ('brands','categories') and old:
                            voc_db.require(dict(old)==row,'TAXONOMY_CONFLICT:'+table+':'+str(row[pk[0]]))
                        if old and table in STABLE:
                            prior=dict(old)
                            # The current row may have come from an earlier merged batch,
                            # not the original snapshot's meta.archive_sha256.
                            origins=[r[0] for r in c.execute('SELECT archive_sha256 FROM source_origins WHERE source_record_id=?',(prior['source_record_id'],))]
                            for version,batch in [(prior,h) for h in origins]+[(row,im['archive_sha256'])]:
                                c.execute('INSERT OR IGNORE INTO record_versions VALUES(?,?,?,?)',(table,str(row[pk[0]]),batch,canonical(version)))
                            decision,reason=choose_record(table,prior,row)
                            if correction is not None and table=='reviews' and row['id']==correction['recordId']:
                                reason={**reason,'codes':['SOURCE_VERIFIED_MAIN_APPEND_CORRECTION'],
                                    'ordinaryConflictFields':reason['conflictingFields'],
                                    'conflictingFields':[],'correction':correction}
                                decision='used_incoming'
                            selected=prior if decision=='kept_existing' else row
                            c.execute('INSERT OR IGNORE INTO merge_decisions VALUES(?,?,?,?,?,?,?,?,?,?)',(
                                voc_db.key(MERGE_POLICY,table,row[pk[0]],im['archive_sha256']),table,str(row[pk[0]]),im['archive_sha256'],
                                prior['source_record_id'],row['source_record_id'],selected['source_record_id'],decision,canonical(reason),MERGE_POLICY))
                            review=delta['mergeReview'];review['decisions']+=1
                            review['conflictRecords']+=int(bool(reason['conflictingFields']))
                            for code in reason['codes']:review['reasonCounts'][code]=review['reasonCounts'].get(code,0)+1
                            if decision=='kept_existing':
                                key='duplicate' if reason['codes']==['EQUIVALENT_RECORD'] else 'retained'
                                delta[key][table]=delta[key].get(table,0)+1
                                mapping[original[pk[0]]]=prior[pk[0]]
                                continue
                        same=old and all(old[k]==v for k,v in row.items() if k!='source_record_id')
                        if same:
                            delta['duplicate'][table]=delta['duplicate'].get(table,0)+1
                            if len(pk)==1:mapping[original[pk[0]]]=row[pk[0]]
                            continue
                        if old and table not in STABLE:
                            voc_db.require(len(pk)==1 and pk[0] in ('id','path','asset_path'),'CONFLICTING_RELATION:'+table)
                            col=pk[0]
                            if table=='assets':row[col]='batches/'+im['archive_sha256'][:16]+'/'+row[col]
                            elif table=='collection_requests':raise ValueError('CONFLICTING_COLLECTION_REQUEST')
                            else:row[col]=hashlib.sha256((im['generation']+':'+str(original[col])).encode()).hexdigest()
                            existing=c.execute('SELECT * FROM '+table+' WHERE '+col+'=?',(row[col],)).fetchone()
                            voc_db.require(existing is None or dict(existing)==row,'BATCH_ID_COLLISION')
                            if existing:
                                mapping[original[col]]=row[col];delta['duplicate'][table]=delta['duplicate'].get(table,0)+1;continue
                            old=None
                        if old:
                            c.execute('UPDATE '+table+' SET '+','.join(k+'=?' for k in columns if k not in pk)+' WHERE '+' AND '.join(k+'=?' for k in pk),[row[k] for k in columns if k not in pk]+[row[k] for k in pk])
                            delta['changed'][table]=delta['changed'].get(table,0)+1
                        else:
                            c.execute('INSERT INTO '+table+'('+','.join(columns)+') VALUES('+','.join('?' for _ in columns)+')',[row[k] for k in columns])
                            delta['added'][table]=delta['added'].get(table,0)+1
                        if len(pk)==1:mapping[original[pk[0]]]=row[pk[0]]
                for sid in maps['source_records'].values():c.execute('INSERT OR IGNORE INTO source_origins VALUES(?,?)',(sid,im['archive_sha256']))
            # Merge dictionaries from the actual validated rows, retaining prior definitions.
            tax=json.loads(baseline['taxonomy'])
            for key in ('brands','categories'):
                known={r['code'] for r in tax[key]}
                tax[key].extend(r for r in json.loads(im['taxonomy'])[key] if r['code'] not in known)
            c.execute("UPDATE meta SET value=? WHERE key='taxonomy'",(canonical(tax),))
            for name in ('retrieval_fts','retrieval_documents'):c.execute('DROP TABLE '+name)
            retrieval.create_index(c)
            generation=hashlib.sha256(canonical([baseline['generation'],im['generation'],digest(Path(__file__)),delta]).encode()).hexdigest()
            c.execute("UPDATE meta SET value=? WHERE key='generation'",(generation,))
        else:
            delta['added']={t:c.execute('SELECT count(*) FROM '+t).fetchone()[0] for t in ORDER}
        c.commit()
    finally:c.close()
    verified(dest)
    return delta

def merge_plan(current,incoming,review_correction=None):
    """Run the exact merge in a disposable database; never write the registry."""
    fingerprints={str(p):digest(p) for p in (current,incoming) if p}
    with tempfile.TemporaryDirectory(prefix='ksf-merge-plan-') as td:
        target=Path(td)/'planned.sqlite';delta=merge(current,incoming,target,review_correction)
        with voc_db.readonly(target) as c:
            # Report only decisions made by this incoming archive, not old history.
            archive=verified(incoming)['archive_sha256']
            rows=[dict(r) for r in c.execute('SELECT entity,record_id,decision,reason_codes_json,prior_source_record_id,incoming_source_record_id,selected_source_record_id FROM merge_decisions WHERE batch_sha256=? ORDER BY entity,record_id LIMIT 101',(archive,))]
            for row in rows:row['reasons']=json.loads(row.pop('reason_codes_json'))
            result={'delta':delta,'plannedGeneration':dict(c.execute('SELECT key,value FROM meta'))['generation'],
                'plannedCounts':voc_db.table_counts(c),'recordDecisions':rows[:100],'recordDecisionsTruncated':len(rows)>100}
    voc_db.require(all(digest(p)==h for p,h in fingerprints.items()),'INPUT_CHANGED_DURING_MERGE_PLAN')
    return {**result,'temporaryFilesUsed':True,'persistentFilesWritten':False,'activeLibraryModified':False,
            'databaseWrites':'temporary-only','inputsUnchanged':True}

def publish(incoming,registry,apply=False,label='资料更新',rollback=None,expected_generation=None,review_correction=None):
    import unified_store
    if unified_store.configured():
        return unified_store.publish(incoming, registry, apply, label, rollback, expected_generation, review_correction)
    voc_db.require(review_correction is None or not rollback,'CORRECTION_ROLLBACK_NOT_ALLOWED')
    root=Path(registry).resolve();source=Path(incoming).resolve() if incoming else None
    pointer=root/'current.json'
    prior=json.loads(pointer.read_text(encoding='utf-8')) if pointer.exists() else None
    if not apply:
        voc_db.require(expected_generation is None or (prior or {}).get('generation')==expected_generation,'LIBRARY_GENERATION_CHANGED')
        if rollback:
            voc_db.require(prior and any(x['generation']==rollback for x in prior['history']),'VERSION_NOT_FOUND')
            return {'ok':True,'readOnly':True,'rollbackGeneration':rollback}
        meta=verified(source)
        current=(root/prior['file']).resolve() if prior else None
        if current:voc_db.require(current.is_relative_to(root) and digest(current)==prior['databaseSha256'],'CURRENT_HASH_MISMATCH')
        if prior and any(x.get('incomingGeneration')==meta['generation'] for x in prior['history']):
            with voc_db.readonly(current) as c:counts=voc_db.table_counts(c)
            return {'ok':True,'readOnly':True,'operation':'already-imported','incomingGeneration':meta['generation'],
                'currentGeneration':prior['generation'],'plannedGeneration':prior['generation'],'plannedCounts':counts,
                'temporaryFilesUsed':False,'persistentFilesWritten':False,'activeLibraryModified':False,'databaseWrites':0}
        return {'ok':True,'readOnly':True,'incomingGeneration':meta['generation'],'currentGeneration':prior['generation'] if prior else None,
            'mode':'merge-batch' if prior else 'initial','label':label,**merge_plan(current,source,review_correction)}
    root.mkdir(parents=True,exist_ok=True);os.chmod(root,0o700)
    lock=root/'.update.lock';fd=os.open(lock,os.O_CREAT|os.O_EXCL|os.O_WRONLY,0o600)
    temp=None;tmp=None
    try:
        # Re-read after the exclusive lock. Query readers never need this lock.
        prior=json.loads(pointer.read_text(encoding='utf-8')) if pointer.exists() else None
        voc_db.require(expected_generation is None or (prior or {}).get('generation')==expected_generation,'LIBRARY_GENERATION_CHANGED')
        if rollback:
            entry=next((x for x in (prior or {}).get('history',[]) if x['generation']==rollback),None)
            voc_db.require(entry is not None,'VERSION_NOT_FOUND')
            target=(root/entry['file']).resolve();voc_db.require(target.is_relative_to(root),'INVALID_VERSION_PATH')
            voc_db.require(digest(target)==entry['databaseSha256'],'ROLLBACK_HASH_MISMATCH');verified(target)
            result={**entry,'label':'回退至 '+rollback[:12],'updatedAt':voc_db.now(),'history':prior['history']}
        else:
            meta=verified(source)
            if prior and any(x.get('incomingGeneration')==meta['generation'] for x in prior['history']):
                return {'ok':True,'applied':False,'operation':'already-imported','generation':prior['generation']}
            current=(root/prior['file']).resolve() if prior else None
            if current:
                voc_db.require(current.is_relative_to(root) and digest(current)==prior['databaseSha256'],'CURRENT_HASH_MISMATCH')
            temp=root/('.build-'+str(uuid.uuid4())+'.sqlite')
            delta=merge(current,source,temp,review_correction)
            finalmeta=verified(temp);generation=finalmeta['generation']
            versions=root/'versions';versions.mkdir(exist_ok=True);os.chmod(versions,0o700)
            target=versions/(generation+'.sqlite')
            if target.exists():
                voc_db.require(digest(target)==digest(temp),'VERSION_ALREADY_EXISTS_DIFFERENT_CONTENT');temp.unlink();temp=None
            else:
                os.chmod(temp,0o600);temp.rename(target);temp=None
            names={'products':'商品','reviews':'评价','prices':'价格观察','specs':'规格','social_posts':'社媒内容','policy_documents':'政策文件','policy_clauses':'条款','ranking_signals':'榜单线索','coverage_observations':'覆盖记录'}
            summary='、'.join(names.get(t,t)+'新增'+str(delta['added'].get(t,0))+'／变更'+str(delta['changed'].get(t,0)) for t in retrieval.ENTITIES.values() if delta['added'].get(t,0) or delta['changed'].get(t,0)) or '重复资料，无实体新增'
            entry={'generation':generation,'incomingGeneration':meta['generation'],'file':target.relative_to(root).as_posix(),'databaseSha256':digest(target),'updatedAt':voc_db.now(),'label':label,'summary':summary,'delta':delta}
            result={**entry,'history':[entry,*((prior or {}).get('history',[]))]}
        tmp=root/('.pointer-'+str(uuid.uuid4())+'.json')
        with tmp.open('x',encoding='utf-8') as f:f.write(canonical(result)+'\n');f.flush();os.fsync(f.fileno())
        os.chmod(tmp,0o600);os.replace(tmp,pointer);tmp=None
        return {'ok':True,'applied':True,**result}
    finally:
        os.close(fd);lock.unlink()
        if temp and temp.exists():temp.unlink()
        if tmp and tmp.exists():tmp.unlink()

if __name__=='__main__':
    p=argparse.ArgumentParser(description=__doc__);p.add_argument('--incoming');p.add_argument('--registry',required=True);p.add_argument('--label',default='资料更新');p.add_argument('--apply',action='store_true');p.add_argument('--rollback');p.add_argument('--expected-generation')
    args=p.parse_args()
    try:print(json.dumps(publish(args.incoming,args.registry,args.apply,args.label,args.rollback,args.expected_generation),ensure_ascii=False,indent=2))
    except Exception as e:
        code = str(e)
        if os.environ.get('KSF_UNIFIED_DB_CONFIG') and not __import__('re').fullmatch('[A-Z][A-Z0-9_:.-]{0,150}', code):
            code = 'UNIFIED_PUBLICATION_FAILED'
        print(json.dumps({'ok':False,'error':code},ensure_ascii=False),file=sys.stderr);sys.exit(1)