"""Build a source-verified v2 research batch for all nine domain entities."""
import collections
import hashlib
import json
import math
import os
from pathlib import Path, PurePosixPath
import sqlite3
import tempfile
import zipfile
import voc_db as v
from intake_contract import VERSION, ENTITIES
from product_insights import ANNOTATION_DDL, THEMES, append_text, text_hash

def typed(value, kind, required=False):
    from agent_batch import identifier
    if value is None:
        v.require(not required, 'REQUIRED_VALUE_MISSING'); return
    if kind == 'id': identifier(value)
    elif kind == 'text': v.require(isinstance(value,str) and (not required or value.strip()), 'TEXT_REQUIRED')
    elif kind == 'date': v.require(isinstance(value,str) and v.calendar_date(value)==value, 'ISO_DATE_REQUIRED')
    elif kind == 'bool': v.require(type(value) is bool, 'BOOLEAN_REQUIRED')
    elif kind in ('count','integer'): v.require(type(value) is int and value >= (1 if kind=='integer' else 0), 'NONNEGATIVE_COUNT_OR_POSITIVE_INTEGER_REQUIRED')
    elif kind == 'positive': v.require(type(value) in (int,float) and math.isfinite(value) and value>0, 'POSITIVE_NUMBER_REQUIRED')
    elif kind == 'amount': v.require(isinstance(value,str) or type(value) in (int,float) and math.isfinite(value), 'RAW_AMOUNT_REQUIRED')
    elif kind == 'json': v.require(isinstance(value,(str,dict,list)), 'TEXT_OR_JSON_REQUIRED')

def load(batch_path):
    from agent_batch import identifier, pointer
    path = Path(batch_path).absolute()
    v.require(path.is_file() and not any(p.is_symlink() for p in (path,*path.parents)), 'BATCH_NOT_FOUND_OR_REDIRECTED')
    raw_manifest = path.read_bytes(); v.require(len(raw_manifest)<=20000000, 'BATCH_TOO_LARGE')
    batch=json.loads(raw_manifest.decode('utf-8-sig'))
    v.require(isinstance(batch,dict) and batch.get('schema')==VERSION and not set(batch)-({'schema','batchId','producer','sources','attachments','themes'}|set(ENTITIES)), 'INVALID_BATCH_SCHEMA_OR_FIELDS')
    identifier(batch.get('batchId')); identifier(batch.get('producer'))
    v.require(isinstance(batch.get('sources'),list) and 0<len(batch['sources'])<=1000, 'SOURCES_REQUIRED')
    sources={}; assets={}
    for source in batch['sources']:
        v.require(isinstance(source,dict) and not set(source)-{'id','path','sha256','platform','marketplace','requestItemId','observedAt','collectedAt','url'}, 'INVALID_SOURCE_FIELDS')
        sid=identifier(source.get('id')); rel=source.get('path')
        v.require(sid not in sources, 'DUPLICATE_SOURCE_ID')
        v.require(isinstance(rel,str) and rel.startswith('raw/') and '\\' not in rel and ':' not in rel and '..' not in PurePosixPath(rel).parts, 'INVALID_SOURCE_PATH')
        p=path.parent/rel
        v.require(p.is_file() and not any(x.is_symlink() for x in (p,*p.parents)) and p.resolve().is_relative_to(path.parent.resolve()), 'SOURCE_MISSING_OR_REDIRECTED')
        v.require(p.stat().st_size<=50000000, 'SOURCE_TOO_LARGE')
        raw=p.read_bytes(); h=hashlib.sha256(raw).hexdigest()
        v.require(h==source.get('sha256'), 'SOURCE_HASH_MISMATCH')
        identifier(source.get('platform'))
        for name in ('observedAt','collectedAt','url'): typed(source.get(name),'text')
        if source.get('requestItemId') is not None: identifier(source['requestItemId'])
        v.require(rel not in assets or assets[rel][1]==h, 'SOURCE_PATH_HASH_CONFLICT')
        assets[rel]=(p,h); sources[sid]=(source,json.loads(raw.decode('utf-8-sig')))
    attachments=batch.get('attachments',[])
    v.require(isinstance(attachments,list) and len(attachments)<=1000, 'INVALID_ATTACHMENTS')
    for entry in attachments:
        v.require(isinstance(entry,dict) and set(entry)=={'path','sha256'}, 'INVALID_ATTACHMENT_FIELDS')
        rel=entry['path']
        v.require(isinstance(rel,str) and rel.startswith('raw/') and '\\' not in rel and ':' not in rel and '..' not in PurePosixPath(rel).parts and rel not in assets, 'INVALID_ATTACHMENT_PATH')
        p=path.parent/rel
        v.require(p.is_file() and not any(x.is_symlink() for x in (p,*p.parents)) and p.resolve().is_relative_to(path.parent.resolve()) and p.stat().st_size<=50000000, 'ATTACHMENT_MISSING_OR_REDIRECTED')
        h=hashlib.sha256(p.read_bytes()).hexdigest();v.require(h==entry['sha256'], 'ATTACHMENT_HASH_MISMATCH')
        assets[rel]=(p,h)
    rows={name:[] for name in ENTITIES}; keys=collections.defaultdict(set)
    for entity,(_,required,fields) in ENTITIES.items():
        records=batch.get(entity,[])
        v.require(isinstance(records,list) and len(records)<=100000, 'INVALID_BATCH_ROWS')
        for record in records:
            v.require(isinstance(record,dict) and set(record)=={'sourceId','pointer','fields'} and record['sourceId'] in sources, 'INVALID_RECORD_MAPPING')
            mapping=record['fields']; v.require(isinstance(mapping,dict) and required<=set(mapping) and not set(mapping)-set(fields), 'INVALID_FIELD_MAPPING')
            source,payload=sources[record['sourceId']]; original=pointer(payload,record['pointer'])
            values={name:pointer(original,loc) for name,loc in mapping.items()}
            for name,value in values.items(): typed(value, fields[name][1], name in required)
            row={fields[name][0]:value for name,value in values.items()}
            row.update(platform=source['platform'],source_file=source['path'],source_time=row.get('source_time',source.get('observedAt')),
                collected_at=source.get('collectedAt'),source_locator=record['pointer'],source_field_map=mapping,producer=batch['producer'])
            if entity in ('products','reviews','prices','specs','rankings','coverage'):
                v.require(source['platform'] in ('jd','taobao') and source.get('marketplace') in (('jd',) if source['platform']=='jd' else ('tmall','taobao_unverified')), 'SOURCE_PLATFORM_MISMATCH')
                row['marketplace']=source['marketplace']; pid=source['platform']+':'+row['item_id']
                if entity!='reviews': v.require(source.get('requestItemId') in (None,row['item_id']), 'PRODUCT_REQUEST_ID_MISMATCH')
                if entity=='products': key=pid
                elif entity=='reviews':
                    key=pid+':'+row['review_id']; row.update(observed_item_id=source.get('requestItemId'),
                        item_identity_match=(row['item_id']==source['requestItemId']) if source.get('requestItemId') else None,review_id_method='source_field')
                else: key=v.key(entity,pid,source['path'],record['pointer'])
            elif entity=='social':
                v.require(source['platform'] not in ('jd','taobao','web','document'), 'SOCIAL_PLATFORM_REQUIRED')
                v.require(bool(row.get('title') or row.get('content')), 'SOCIAL_TEXT_REQUIRED'); key=source['platform']+':'+row['id']
            elif entity=='documents':
                v.require(not row.get('declared_status') or row['declared_status']=='unknown' or row.get('status_as_of'), 'POLICY_STATUS_AS_OF_REQUIRED')
                v.require(all(not row.get(k) or row[k] in assets for k in ('pdf_file','text_file')), 'DECLARED_ATTACHMENT_NOT_REGISTERED')
                key=row['document_id']
            else: key=v.key(row['document_id'],row['clause'])
            v.require(key not in keys[entity], 'DUPLICATE_RECORD_ID'); keys[entity].add(key)
            row['_record_key']=key
            if entity=='prices': v.require(row.get('currency') in (None,'CNY'), 'UNSUPPORTED_CURRENCY')
            if entity=='specs':
                v.require(row.get('verification') in (None,'source_attribute','title_candidate','unknown'), 'INVALID_SPEC_VERIFICATION')
                if any(row.get(k) is not None for k in ('net_weight_g','pack_count','total_weight_g')):
                    v.require(bool(row.get('basis')), 'SPEC_BASIS_REQUIRED')
            rows[entity].append(row)
    v.require(any(rows.values()), 'EMPTY_BATCH')
    for entity in ('reviews','prices','specs','rankings','coverage'):
        v.require(all(r['platform']+':'+r['item_id'] in keys['products'] for r in rows[entity]), 'PRODUCT_CONTEXT_REQUIRED')
    v.require(all(r['document_id'] in keys['documents'] for r in rows['clauses']), 'DOCUMENT_CONTEXT_REQUIRED')
    tags=batch.get('themes',[]); v.require(isinstance(tags,list), 'INVALID_THEMES')
    by_review={r['_record_key']:r for r in rows['reviews']}; seen=set()
    for tag in tags:
        v.require(isinstance(tag,dict) and set(tag)=={'reviewId','theme','quote','quoteField'}, 'INVALID_THEME_FIELDS')
        v.require(tag['reviewId'] in by_review and tag['theme'] in THEMES and tag['quoteField'] in ('content','append') and isinstance(tag['quote'],str) and 0<len(tag['quote'])<=500, 'INVALID_THEME_EVIDENCE')
        key=(tag['reviewId'],tag['theme']); v.require(key not in seen, 'DUPLICATE_REVIEW_THEME'); seen.add(key)
        r=by_review[tag['reviewId']]; text=r['content'] if tag['quoteField']=='content' else append_text(v.canonical(r.get('append_content')))
        v.require(tag['quote'] in text, 'THEME_QUOTE_NOT_VERBATIM')
    return path,raw_manifest,batch,rows,assets

def build(batch_path, output, apply=False):
    from agent_batch import digest
    path,manifest,batch,rows,assets=load(batch_path)
    target=Path(output).absolute(); archive=target.with_suffix('.source.zip')
    v.require(target.suffix=='.sqlite' and not any(p.is_symlink() for p in (target,*target.parents)) and not target.exists() and not archive.exists() and not archive.is_symlink(), 'NEW_DATABASE_TARGET_REQUIRED')
    receipt={'ok':True,'schema':VERSION,'batchId':batch['batchId'],'readOnly':not apply,'applied':False,
        'counts':{k:len(x) for k,x in rows.items()},'sourceFilesVerified':len(assets),'themeAnnotations':len(batch.get('themes',[])),
        'databaseWrites':0,'networkRequests':0,'activeLibraryModified':False,'productionImported':False}
    if not apply: return receipt
    target.parent.mkdir(parents=True,exist_ok=True,mode=0o700)
    created=[]
    with tempfile.TemporaryDirectory(prefix='.structured-batch-',dir=target.parent) as td:
        td=Path(td); packed=td/'source.zip'; staged=td/'staging.sqlite'
        with zipfile.ZipFile(packed,'x',zipfile.ZIP_DEFLATED) as z:
            def write(name,body):
                info=zipfile.ZipInfo('agent-batch/'+name,(2026,1,1,0,0,0));info.external_attr=0o100600<<16
                z.writestr(info,body,compress_type=zipfile.ZIP_DEFLATED)
            for entity,filename in v.PRIMARY.items():write('01_最新底库/'+filename,'\n'.join(v.canonical(r) for r in rows[entity]))
            for entity in ('prices','specs','rankings','coverage'):write('01_最新底库/agent-'+entity+'.jsonl','\n'.join(v.canonical(r) for r in rows[entity]))
            v.require(path.read_bytes()==manifest,'BATCH_CHANGED_DURING_BUILD');write('batch.json',manifest)
            for rel,(p,h) in sorted(assets.items()):
                raw=p.read_bytes();v.require(hashlib.sha256(raw).hexdigest()==h,'SOURCE_CHANGED_DURING_BUILD');write('02_采集档案/'+rel,raw)
        built=v.build(packed,staged)
        c=sqlite3.connect(staged);c.row_factory=sqlite3.Row
        try:
            c.execute('PRAGMA foreign_keys=ON');c.executescript(ANNOTATION_DDL)
            def issue(entity,eid,code,detail):
                c.execute('INSERT OR IGNORE INTO quality_issues VALUES(?,?,?,?,?)',(v.key(entity,eid,code,detail),entity,eid,code,detail))
            for entity in ('prices','specs','rankings','coverage'):
                for line,row in enumerate(rows[entity],1):
                    asset='agent-batch/01_最新底库/agent-'+entity+'.jsonl';raw=v.canonical(row);sid=v.key(asset,line)
                    v.insert(c,'source_records',dict(id=sid,kind='agent_'+entity,asset_path=asset,line_number=line,payload_sha256=v.sha(raw),payload_json=raw))
                    source='agent-batch/02_采集档案/'+row['source_file']
                    v.insert(c,'source_links',dict(record_id=sid,role='source_file',original_path=row['source_file'],asset_path=source,resolution='matched'))
                    pid=row['platform']+':'+row['item_id']; rid=row['_record_key']
                    if entity=='prices':v.add_price(c,pid,row['amount'],row['price_type'],row,sid,issue)
                    elif entity=='specs':
                        v.insert(c,'specs',dict(id=rid,product_id=pid,upstream_sku_id=row.get('upstream_sku_id'),spec_text=row['spec_text'],
                            net_weight_g=row.get('net_weight_g'),pack_count=row.get('pack_count'),total_weight_g=row.get('total_weight_g'),
                            verification=row.get('verification') or 'unknown',source_record_id=sid,basis=row.get('basis') or '仅保留来源规格原文，计量依据待核对'))
                    elif entity=='rankings':v.insert(c,'ranking_signals',dict(id=rid,product_id=pid,rank_id=row.get('rank_id'),rank_number=row.get('rank_number'),title=row.get('title'),scope=row['scope'],source_time_raw=row.get('source_time'),source_record_id=sid))
                    else:v.insert(c,'coverage_observations',dict(id=rid,product_id=pid,status=row['status'],available_reviews=row.get('available_reviews'),reported_total=str(row['reported_total']) if row.get('reported_total') is not None else None,historical_total_complete=int(row.get('historical_total_complete') is True),source_record_id=sid))
            for tag in batch.get('themes',[]):
                r=c.execute('SELECT * FROM reviews WHERE id=?',(tag['reviewId'],)).fetchone()
                v.insert(c,'review_theme_annotations',dict(id=v.key(tag,batch['producer']),review_id=r['id'],product_id=r['product_id'],
                    theme_code=tag['theme'],quote=tag['quote'],quote_field=tag['quoteField'],text_sha256=text_hash(r),method='agent',producer=batch['producer']))
            c.execute('DROP TABLE retrieval_fts');c.execute('DROP TABLE retrieval_documents')
            import retrieval
            retrieval.create_index(c)
            generation=v.key(built['generation'],digest(Path(__file__)),digest(Path(__file__).with_name('intake_contract.py')),
                digest(Path(__file__).with_name('product_insights.py')),batch.get('themes',[]))
            c.execute("UPDATE meta SET value=? WHERE key='generation'",(generation,));c.commit()
        finally:c.close()
        from update_library import verified
        verified(staged)
        try:
            for src,dest in ((packed,archive),(staged,target)):
                os.chmod(src,0o600);os.link(src,dest);created.append((dest,digest(src)))
        except Exception:
            for p,h in reversed(created):
                if p.is_file() and digest(p)==h:p.unlink()
            raise
    return {**receipt,'applied':True,'readOnly':False,'databaseWrites':'staging-only','generation':generation,
        'database':str(target),'databaseSha256':digest(target),'sourceArchive':str(archive),'archiveSha256':digest(archive)}