cms-SiteForge/app/v8_workflows.py

"""V8 portable content import pipeline: deterministic preview, batch audit and guarded rollback.

This is a single-instance pipeline suitable for controlled pilots. It never runs
untrusted scripts, and keeps original import and business data separate.
"""
from __future__ import annotations
import hashlib
import io
import json
import re
import uuid
import csv
from urllib.parse import urlsplit
from fastapi import APIRouter, Body, HTTPException, Request, File, UploadFile, Form
from . import db, enterprise, assembly
from .core import save_entry, route_path

router = APIRouter(prefix='/api/factory/sites', tags=['V8 content workflow'])
MAX_ITEMS = 1000
MAX_UPLOAD = 12_000_000

def setup():
    with db.connect() as c:
        c.executescript('''
        CREATE TABLE IF NOT EXISTS v8_import_batches(
          id TEXT PRIMARY KEY, site_id TEXT NOT NULL REFERENCES sites(id) ON DELETE CASCADE,
          origin TEXT NOT NULL, content_type TEXT NOT NULL, state TEXT NOT NULL,
          mapping_json TEXT NOT NULL, warnings_json TEXT NOT NULL DEFAULT '[]',
          input_count INTEGER NOT NULL, created_at TEXT NOT NULL, committed_at TEXT,
          rolled_back_at TEXT);
        CREATE TABLE IF NOT EXISTS v8_import_items(
          id TEXT PRIMARY KEY, batch_id TEXT NOT NULL REFERENCES v8_import_batches(id) ON DELETE CASCADE,
          source_key TEXT NOT NULL, row_number INTEGER NOT NULL,
          route TEXT NOT NULL, content_json TEXT NOT NULL, state TEXT NOT NULL DEFAULT 'staged',
          entry_id TEXT, before_json TEXT, after_fingerprint TEXT, error TEXT,
          UNIQUE(batch_id, source_key));
        CREATE INDEX IF NOT EXISTS v8_batches_by_site ON v8_import_batches(site_id,created_at);
        CREATE TABLE IF NOT EXISTS v8_site_templates(
          id TEXT PRIMARY KEY, site_id TEXT NOT NULL REFERENCES sites(id) ON DELETE CASCADE,
          name TEXT NOT NULL, version TEXT NOT NULL, blueprint_json TEXT NOT NULL,
          created_at TEXT NOT NULL);
        ''')

def _row_hash(row):
    if row is None:return ''
    data={k:row[k] for k in ('id','site_id','type','title','route','slug','summary','body','cover',
          'seo_title','seo_description','canonical','robots','meta_json','status','source','created_at','updated_at')}
    return hashlib.sha256(json.dumps(data,ensure_ascii=False,sort_keys=True).encode()).hexdigest()

def _require_site(site_id):
    if not db.one('SELECT id FROM sites WHERE id=?',(site_id,)):
        raise HTTPException(404,'网站不存在')

def _mapped_input(rows, mapping, kind):
    if not isinstance(rows,list) or not rows or len(rows)>MAX_ITEMS:
        raise ValueError('每批需要 1-1000 条内容')
    if any(not isinstance(x,dict) for x in rows):raise ValueError('数据行必须是对象')
    if kind not in ('product','vehicle','page','post','case','brand','service','other','mixed'):
        raise ValueError('内容类型不受支持')
    mapping = mapping or assembly.suggest(list(rows[0].keys()))
    if not isinstance(mapping,dict):raise ValueError('映射规则不合法')
    mapped,errors=assembly.transform(rows,mapping,'page' if kind=='mixed' else kind)
    if kind=='mixed':
        kinds={'post':'post','page':'page','product':'product','brand':'brand','vehicle':'vehicle',
               'case':'case','service':'service'}
        for i,item in enumerate(mapped):
            src=str(rows[i].get('类型') or rows[i].get('分类') or 'page').lower()
            item['type']=kinds.get(src,'other')
    if errors:raise ValueError('数据预检失败：'+json.dumps(errors[:8],ensure_ascii=False))
    return mapped,mapping

def stage(site_id, rows, mapping=None, kind='product', origin='manual-import', source_keys=None):
    _require_site(site_id)
    mapped,mapping=_mapped_input(rows,mapping,kind)
    if len(json.dumps(mapped,ensure_ascii=False).encode('utf-8'))>MAX_UPLOAD:
        raise ValueError('映射后的批次超过12MB上限，请拆分导入')
    if source_keys is not None and len(source_keys)!=len(rows):raise ValueError('源记录主键数量不一致')
    existing_routes={x['route'] for x in db.rows('SELECT route FROM entries WHERE site_id=?',(site_id,))}
    conflicts=[x['route'] for x in mapped if x['route'] in existing_routes]
    batch_id=uuid.uuid4().hex
    now=db.now()
    keys=[]
    for i,item in enumerate(mapped):
        source_key=str(source_keys[i] if source_keys else item['route'])
        if not source_key or len(source_key)>250:raise ValueError('来源主键缺失或过长')
        keys.append(source_key)
    if len(set(keys))!=len(keys):raise ValueError('导入文件的来源主键重复')
    with db.connect() as c:
        c.execute('''INSERT INTO v8_import_batches(id,site_id,origin,content_type,state,mapping_json,
            input_count,created_at) VALUES(?,?,?,?,?,?,?,?)''',
            (batch_id,site_id,origin[:160],kind,'staged',json.dumps(mapping,ensure_ascii=False),len(mapped),now))
        for i,item in enumerate(mapped):
            c.execute('''INSERT INTO v8_import_items(id,batch_id,source_key,row_number,route,content_json)
                VALUES(?,?,?,?,?,?)''',(uuid.uuid4().hex,batch_id,keys[i],i+1,item['route'],
                json.dumps(item,ensure_ascii=False)))
    return {'batch_id':batch_id,'state':'staged','mapped':len(mapped),'conflicts':conflicts,
            'sample':mapped[:5],'mapping':mapping,'requires_conflict_choice':bool(conflicts)}

def list_batches(site_id):
    _require_site(site_id)
    return db.rows('SELECT id,origin,content_type,state,input_count,created_at,committed_at,rolled_back_at FROM v8_import_batches WHERE site_id=? ORDER BY created_at DESC LIMIT 100',(site_id,))

def batch_detail(site_id,batch_id):
    result=db.one('SELECT * FROM v8_import_batches WHERE site_id=? AND id=?',(site_id,batch_id))
    if not result:raise HTTPException(404,'导入批次不存在')
    result['items']=db.rows('SELECT id,source_key,row_number,route,state,entry_id,error FROM v8_import_items WHERE batch_id=? ORDER BY row_number LIMIT 1100',(batch_id,))
    return result

def commit(site_id,batch_id,on_conflict='skip'):
    if on_conflict not in ('skip','replace'):raise ValueError('冲突策略只能为 skip 或 replace')
    batch=db.one('SELECT * FROM v8_import_batches WHERE site_id=? AND id=?',(site_id,batch_id))
    if not batch:raise HTTPException(404,'导入批次不存在')
    if batch['state']=='committed':return {'batch_id':batch_id,'state':'committed','existing':True}
    if batch['state']!='staged':raise HTTPException(409,'批次已经开始处理，不能重复执行')
    locked=db.execute("UPDATE v8_import_batches SET state='processing' WHERE id=? AND state='staged'",(batch_id,))
    if not locked:raise HTTPException(409,'批次正在由其他请求处理')
    items=db.rows('SELECT * FROM v8_import_items WHERE batch_id=? ORDER BY row_number',(batch_id,))
    created=updated=skipped=0
    failed=[]
    for item in items:
        original=db.one('SELECT * FROM entries WHERE site_id=? AND route=?',(site_id,item['route']))
        if original and on_conflict=='skip':
            db.execute("UPDATE v8_import_items SET state='skipped', entry_id=? WHERE id=?",(original['id'],item['id']))
            skipped+=1
            continue
        try:
            payload=json.loads(item['content_json'])
            meta=payload.get('meta') or {}
            meta.setdefault('provenance',{})
            meta['provenance'].update({'import_batch':batch_id,'source_key':item['source_key'],'origin':batch['origin']})
            payload['meta']=meta
            before=json.dumps(original,ensure_ascii=False) if original else None
            eid=save_entry(site_id,payload,existing_id=original['id'] if original else None,source='v8-import')
            after=db.one('SELECT * FROM entries WHERE site_id=? AND id=?',(site_id,eid))
            db.execute('''UPDATE v8_import_items SET state='applied',entry_id=?,before_json=?,after_fingerprint=? WHERE id=?''',
                     (eid,before,_row_hash(after),item['id']))
            if original:updated+=1
            else:created+=1
        except Exception as exc:
            msg=str(exc)[:250]
            db.execute("UPDATE v8_import_items SET state='error',error=? WHERE id=?",(msg,item['id']))
            failed.append({'row':item['row_number'],'message':msg})
    db.execute('UPDATE v8_import_batches SET state=?,committed_at=? WHERE id=?',
               ('partial' if failed else 'committed',db.now(),batch_id))
    # Keep unsupported Elementor/WooCommerce details in the existing release gate.
    if batch['origin']=='wordpress-wxr':
        for finding in json.loads(batch.get('warnings_json') or '[]')[:200]:
            db.insert('''INSERT INTO cms_import_reviews(id,site_id,platform,row_number,issue,created_at)
                VALUES(?,?,?,?,?,?)''',(uuid.uuid4().hex,site_id,'wordpress-wxr',int(finding.get('row',0)),
                ('v8:'+batch_id+': '+str(finding.get('issue','Requires review')))[:600],db.now()))
    return {'batch_id':batch_id,'state':'partial' if failed else 'committed',
            'created':created,'updated':updated,'skipped':skipped,'errors':failed}

def rollback(site_id,batch_id):
    # Hold one write transaction across the conflict check and all reversions.
    # A manual edit racing with rollback must not be accidentally overwritten.
    with db.connect() as c:
        c.execute('BEGIN IMMEDIATE')
        batch=c.execute('SELECT * FROM v8_import_batches WHERE site_id=? AND id=?',(site_id,batch_id)).fetchone()
        if not batch:raise HTTPException(404,'导入批次不存在')
        if batch['state']=='rolled_back':return {'batch_id':batch_id,'state':'rolled_back','existing':True}
        if batch['state'] not in ('committed','partial'):raise HTTPException(409,'批次还没有完成提交')
        applied=c.execute("SELECT * FROM v8_import_items WHERE batch_id=? AND state='applied' ORDER BY row_number DESC",(batch_id,)).fetchall()
        conflicts=[]
        for item in applied:
            current=c.execute('SELECT * FROM entries WHERE site_id=? AND id=?',(site_id,item['entry_id'])).fetchone()
            if _row_hash(dict(current) if current else None)!=item['after_fingerprint']:
                conflicts.append(item['route'])
        if conflicts:raise HTTPException(409,{'message':'内容在导入后被修改，拒绝覆盖人工编辑','routes':conflicts})
        cols=('type','title','route','slug','summary','body','cover','seo_title','seo_description',
              'canonical','robots','meta_json','status','source','created_at','updated_at')
        for item in applied:
            if not item['before_json']:
                c.execute('DELETE FROM entries WHERE site_id=? AND id=?',(site_id,item['entry_id']))
            else:
                before=json.loads(item['before_json'])
                assignments=','.join(k+'=?' for k in cols)
                c.execute(f'UPDATE entries SET {assignments} WHERE site_id=? AND id=?',
                          (*[before[k] for k in cols],site_id,item['entry_id']))
            c.execute("UPDATE v8_import_items SET state='rolled_back' WHERE id=?",(item['id'],))
        c.execute("UPDATE v8_import_batches SET state='rolled_back',rolled_back_at=? WHERE id=?",(db.now(),batch_id))
        if batch['origin']=='wordpress-wxr':
            c.execute("UPDATE cms_import_reviews SET state='resolved',reviewer='system',resolution='Source import batch rolled back',resolved_at=? WHERE site_id=? AND platform='wordpress-wxr' AND state='pending' AND issue LIKE ?",
                      (db.now(),site_id,'v8:'+batch_id+': %'))
    return {'batch_id':batch_id,'state':'rolled_back','restored':len(applied)}

def parse_wxr(content):
    """Extract conservative WordPress records without writing into the site.
    Unsupported builder configuration stays in source metadata for review.
    """
    from defusedxml import ElementTree as ET
    if len(content)>MAX_UPLOAD:raise ValueError('XML 文件超过限制')
    root=ET.fromstring(content)
    if root.tag!='rss':raise ValueError('这不是 WordPress WXR RSS 导出')
    # Stage only unambiguous permanent paths; dynamic functions remain review items.
    from .wp_complex_v820 import inspect_wxr
    checks=inspect_wxr(content)
    blockers=[x for x in checks['issues'] if x['code'] in ('invalid_permalink','duplicate_route')]
    if blockers or (checks['findings_truncated'] and checks['blocking_issues']):
        raise ValueError('存在不能安全映射的 WordPress URL，请先运行 V8.20 WXR 预检并修正：'+
                         ', '.join(x['code']+':'+x['route'] for x in blockers[:5]))
    rows=[];ids=[];warnings=[]
    for it in root.findall('./channel/item'):
        get=lambda name:next((str(node.text or '') for node in it if node.tag.endswith('}'+name) or node.tag==name),'')
        raw_typ=get('post_type') or 'post'
        status=get('status') or 'publish'
        if raw_typ in ('attachment','revision','nav_menu_item') or status not in ('publish','draft','private'):continue
        if len(rows)>=MAX_ITEMS:raise ValueError('单批 WXR 最多1000条，请拆分文件')
        wid=get('post_id')
        title=get('title').strip() or get('post_name').strip()
        if not title:continue
        url=get('link')
        slug=get('post_name') or re.sub(r'[^a-z0-9-]','-',title.lower())[:65] or uuid.uuid4().hex[:8]
        route=urlsplit(url).path if url.startswith(('http://','https://')) else ''
        if not route:route=f'/{slug}/' if raw_typ=='page' else f'/{raw_typ}/{slug}/'
        metas={}
        for pm in it:
            if not pm.tag.endswith('}postmeta'):continue
            k=v=''
            for n in pm:
                if n.tag.endswith('}meta_key'):k=n.text or ''
                if n.tag.endswith('}meta_value'):v=n.text or ''
            if k:metas[k]=v[:10000]
        if any(k in metas for k in ('_elementor_data','_product_attributes')):
            warnings.append({'row':len(rows)+1,'issue':'含复杂 Elementor/WooCommerce 组件，导入后需人工核对'})
        rows.append({'标题':title,'路径':route,'正文':get('encoded'),'类型':raw_typ,

          'SEO标题':metas.get('_yoast_wpseo_title') or metas.get('rank_math_title',''),
          'SEO描述':metas.get('_yoast_wpseo_metadesc') or metas.get('rank_math_description',''),
          '分类':raw_typ})
        ids.append(f'wordpress:{wid}' if wid else f'wordpress:path:{route}')
    for finding in checks['issues']:
        if finding['severity'] not in ('review','blocker'):continue
        if finding['code'] in ('invalid_permalink','duplicate_route'):continue
        route=finding['route']
        index=next((i+1 for i,row in enumerate(rows) if row['路径']==route),0)
        if index:
            if finding['code'].startswith('elementor') and any(w.get('row')==index and 'Elementor' in w.get('issue','') for w in warnings):
                continue
            warnings.append({'row':index,'issue':finding['code']+': '+finding['message']})
    # Bound warnings written into batch metadata and shown for human review.
    return rows,ids,warnings[:200]

@router.post('/{site_id}/staging')
def api_stage(request:Request,site_id:str,payload:dict=Body(...)):
    enterprise.require(request,site_id,'editor')
    try:return stage(site_id,payload.get('rows'),payload.get('mapping'),payload.get('content_type','product'),payload.get('origin','manual-import'))
    except ValueError as e:raise HTTPException(400,str(e))

@router.post('/{site_id}/staging/file')
async def api_table_file(request:Request,site_id:str,file:UploadFile=File(...),
                         content_type:str=Form('product')):
    enterprise.require(request,site_id,'editor')
    from .table_loader import parse_table
    blob=await file.read(MAX_UPLOAD+1)
    try:
        parsed=parse_table(file.filename or '',blob)
        result=stage(site_id,parsed['rows'],kind=content_type,origin='table:'+parsed['filename'][:120])
        result['columns']=parsed['columns']
        return result
    except (ValueError,UnicodeError,json.JSONDecodeError) as e:
        raise HTTPException(400,str(e))

@router.post('/{site_id}/staging/wxr')
async def api_wxr(request:Request,site_id:str,file:UploadFile=File(...)):
    enterprise.require(request,site_id,'editor')
    if not (file.filename or '').lower().endswith('.xml'):raise HTTPException(400,'仅支持 .xml 导出')
    blob=await file.read(MAX_UPLOAD+1)
    try:
        rows,keys,warnings=parse_wxr(blob)
        if not rows:raise ValueError('没有可导入的 WordPress 页面或文章')
        result=stage(site_id,rows,{'title':'标题','route':'路径','body':'正文',
              'seo_title':'SEO标题','seo_description':'SEO描述'},'mixed','wordpress-wxr',keys)
        db.execute('UPDATE v8_import_batches SET warnings_json=? WHERE id=?',
                    (json.dumps(warnings,ensure_ascii=False),result['batch_id']))
        return {**result,'warnings':warnings[:30]}
    except ValueError as e:raise HTTPException(400,str(e))

@router.get('/{site_id}/staging')
def api_list(request:Request,site_id:str):
    enterprise.require(request,site_id,'viewer')
    return {'items':list_batches(site_id)}

@router.get('/{site_id}/staging/{batch_id}')
def api_detail(request:Request,site_id:str,batch_id:str):
    enterprise.require(request,site_id,'viewer')
    return batch_detail(site_id,batch_id)

@router.post('/{site_id}/staging/{batch_id}/commit')
def api_commit(request:Request,site_id:str,batch_id:str,payload:dict=Body(default_factory=dict)):
    enterprise.require(request,site_id,'editor')
    try:return commit(site_id,batch_id,payload.get('on_conflict','skip'))
    except ValueError as e:raise HTTPException(400,str(e))

@router.post('/{site_id}/staging/{batch_id}/rollback')
def api_rollback(request:Request,site_id:str,batch_id:str):
    enterprise.require(request,site_id,'owner')
    return rollback(site_id,batch_id)