kangshifu-board/server/voc-library/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 hashlib
import json
import os
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'}
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):
    im=verified(incoming)
    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':{}}
    try:
        c.executescript(ANNOTATION_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',(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'])
                        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]]))
                        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:
                            for version,batch in [(dict(old),baseline['archive_sha256']),(row,im['archive_sha256'])]:
                                c.execute('INSERT OR IGNORE INTO record_versions VALUES(?,?,?,?)',(table,str(row[pk[0]]),batch,canonical(version)))
                            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 publish(incoming,registry,apply=False,label='资料更新',rollback=None,expected_generation=None):
    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)
        return {'ok':True,'readOnly':True,'incomingGeneration':meta['generation'],'currentGeneration':prior['generation'] if prior else None,'mode':'merge-batch' if prior else 'initial','label':label}
    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)
            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:print(json.dumps({'ok':False,'error':str(e)},ensure_ascii=False),file=sys.stderr);sys.exit(1)