"""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:]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)