"""Resumable, bounded WordPress REST migration sessions for SiteForge V8.1. Each advance fetches at most one WP REST page. No unknown remote HTML or JS is executed, and no imported content is published until an explicit finalization. The outbound client pins validated public DNS results to the connected socket. This is deliberately a single-instance pilot workflow, not a background bulk crawler. Progress survives HTTP failures and server restarts. """ from __future__ import annotations import json import uuid from urllib.parse import urlsplit, urlencode from bs4 import BeautifulSoup from fastapi import APIRouter, Body, HTTPException, Request from . import db, enterprise, safe_http, migration_audit, v8_workflows from .core import route_path router=APIRouter(prefix='/api/migrations/sites',tags=['V8.1 WordPress migration']) PAGE_SIZE=50 MAX_WARNINGS=500 def setup(): with db.connect() as c: c.executescript(''' CREATE TABLE IF NOT EXISTS wp_migration_sessions ( id TEXT PRIMARY KEY, site_id TEXT NOT NULL REFERENCES sites(id) ON DELETE CASCADE, origin TEXT NOT NULL, state TEXT NOT NULL, phase TEXT NOT NULL DEFAULT 'pages', next_page INTEGER NOT NULL DEFAULT 1, max_items INTEGER NOT NULL, fetched INTEGER NOT NULL DEFAULT 0, reviewed INTEGER NOT NULL DEFAULT 0, warnings_json TEXT NOT NULL DEFAULT '[]', last_error TEXT NOT NULL DEFAULT '', cancelled INTEGER NOT NULL DEFAULT 0, lease_until TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL); CREATE INDEX IF NOT EXISTS wp_sessions_by_site ON wp_migration_sessions(site_id,created_at); CREATE TABLE IF NOT EXISTS wp_migration_pages( session_id TEXT NOT NULL REFERENCES wp_migration_sessions(id) ON DELETE CASCADE, phase TEXT NOT NULL, page INTEGER NOT NULL, batch_id TEXT, source_count INTEGER NOT NULL DEFAULT 0, PRIMARY KEY(session_id,phase,page)); CREATE TABLE IF NOT EXISTS wp_migration_urls( session_id TEXT NOT NULL REFERENCES wp_migration_sessions(id) ON DELETE CASCADE, route TEXT NOT NULL, original_url TEXT NOT NULL, seo_title TEXT NOT NULL DEFAULT '', description TEXT NOT NULL DEFAULT '', canonical TEXT NOT NULL DEFAULT '', robots TEXT NOT NULL DEFAULT '', h1 TEXT NOT NULL DEFAULT '', status INTEGER NOT NULL DEFAULT 200, PRIMARY KEY(session_id,route)); ''') # Keep old migration setup callers forward-compatible with V8.2 schema. from . import wp_migration_v82 wp_migration_v82.setup() from . import wp_migration_v83 wp_migration_v83.setup() def _origin(url): p=urlsplit(str(url).strip()) if p.scheme not in ('http','https') or not p.hostname or p.username or p.password or p.query or p.fragment: raise ValueError('请输入公开可访问的 WordPress 站点 HTTP(S) 地址（不要带账号、参数或片段）') if len(url)>500 or (p.path and ('..' in p.path.split('/') or '\\' in p.path)): raise ValueError('原网站地址不合法') if p.port not in (None,80,443):raise ValueError('仅允许标准 HTTP/HTTPS 端口') return f'{p.scheme}://{p.netloc}{p.path.rstrip("/")}/' def _one(site_id,session_id): s=db.one('SELECT * FROM wp_migration_sessions WHERE site_id=? AND id=?',(site_id,session_id)) if s is None:raise HTTPException(404,'WordPress 迁移会话不存在') return s def create(site_id,origin,max_items=2000): v8_workflows._require_site(site_id) origin=_origin(origin) if not 1<=int(max_items)<=20_000:raise ValueError('迁移上限须为 1–20,000 条') sid=uuid.uuid4().hex db.insert('''INSERT INTO wp_migration_sessions(id,site_id,origin,state,phase,max_items,created_at,updated_at) VALUES(?,?,?,'created','pages',?,?,?)''',(sid,site_id,origin,int(max_items),db.now(),db.now())) return details(site_id,sid) def details(site_id,session_id): s=_one(site_id,session_id) s['warnings']=json.loads(s.pop('warnings_json') or '[]') s['batches']=db.rows('''SELECT p.phase,p.page,p.source_count,b.state AS batch_state,p.batch_id FROM wp_migration_pages p LEFT JOIN v8_import_batches b ON b.id=p.batch_id WHERE p.session_id=? ORDER BY CASE p.phase WHEN 'pages' THEN 0 ELSE 1 END,p.page''',(session_id,)) s['baseline_count']=db.one('SELECT COUNT(*) n FROM wp_migration_urls WHERE session_id=?',(session_id,))['n'] return s def list_sessions(site_id): v8_workflows._require_site(site_id) return db.rows('''SELECT id,origin,state,phase,next_page,max_items,fetched,reviewed,created_at,updated_at,last_error FROM wp_migration_sessions WHERE site_id=? ORDER BY created_at DESC LIMIT 100''',(site_id,)) def _plain(html): return BeautifulSoup(html or '','html.parser').get_text(' ',strip=True)[:600] def _transform(source,kind,origin): """Returns a staged row + audit row, or a manual review warning.""" if not isinstance(source,dict):raise ValueError('WordPress 返回了非对象记录') wp_id=source.get('id') if not isinstance(wp_id,int) or wp_id<1:raise ValueError('WordPress 记录没有合法数字 ID') link=source.get('link') or '' p=urlsplit(link) base=urlsplit(origin) if (p.scheme not in ('http','https') or p.hostname!=base.hostname or p.port!=base.port or p.username or p.password or p.query or p.fragment or not p.path): raise ValueError(f'记录 #{wp_id} 没有本站的固定 URL（可能使用了 ?p= 参数）') route=route_path(p.path) title=_plain((source.get('title') or {}).get('rendered','')) if not title:raise ValueError(f'记录 #{wp_id} 缺少标题') content=(source.get('content') or {}).get('rendered','') excerpt=_plain((source.get('excerpt') or {}).get('rendered','')) seo=source.get('yoast_head_json') or {} if not isinstance(seo,dict):seo={} embedded=source.get('_embedded') or {} media=embedded.get('wp:featuredmedia') or [] cover=media[0].get('source_url','') if isinstance(media,list) and media and isinstance(media[0],dict) else '' record={'标题':title,'路径':route,'正文':content,'摘要':excerpt, 'SEO标题':seo.get('title') or '', 'SEO描述':seo.get('description') or '', 'Canonical':seo.get('canonical') or '', '封面':cover,'类型':'page' if kind=='pages' else 'post'} baseline={'route':route,'url':link,'seo_title':record['SEO标题'], 'description':record['SEO描述'],'canonical':record['Canonical'], 'robots':'noindex,follow' if seo.get('robots',{}).get('index')=='noindex' else '', 'h1':title,'status':200} return record,baseline,f'wp-rest:{kind}:{wp_id}' def _fetch(origin,phase,page): # A fixed page size makes the cursor stable for restart and retries. query=urlencode({'per_page':PAGE_SIZE,'page':page,'_embed':'1','status':'publish'}) url=origin+'wp-json/wp/v2/'+phase+'?'+query r=safe_http.fetch(url,max_bytes=8_000_000,timeout=18,redirects=0,headers={'Accept':'application/json'}) if r.status_code==400 and page>1:return [] # WP returns 400 on page > available total r.raise_for_status() value=r.json() if not isinstance(value,list):raise ValueError('WordPress REST 返回的不是记录数组') if len(value)>PAGE_SIZE:raise ValueError('WordPress REST 单页数据超过限制') return value def advance(site_id,session_id): s=_one(site_id,session_id) if s['state']=='ready':return details(site_id,session_id) if s['state'] in ('completed','completed_with_warnings','cancelled','limit_reached'): raise HTTPException(409,'迁移会话已结束或达到采集上限') # Lease makes concurrent browser clicks / retries safe. Expired claims can resume. from .jobs import iso_after with db.connect() as c: c.execute('BEGIN IMMEDIATE') item=c.execute('SELECT * FROM wp_migration_sessions WHERE id=? AND site_id=?',(session_id,site_id)).fetchone() if item['cancelled']:raise HTTPException(409,'迁移会话已取消') if item['state']=='running' and item['lease_until'] and item['lease_until']>db.now(): raise HTTPException(409,'迁移采集正在执行，请勿重复提交') c.execute("UPDATE wp_migration_sessions SET state='running',lease_until=?,last_error='',updated_at=? WHERE id=?", (iso_after(60),db.now(),session_id)) phase=s['phase'];page=s['next_page'] try: result=_fetch(s['origin'],phase,page) if db.one('SELECT cancelled FROM wp_migration_sessions WHERE id=?',(session_id,))['cancelled']: raise HTTPException(409,'迁移已取消') existing=db.one('SELECT * FROM wp_migration_pages WHERE session_id=? AND phase=? AND page=?',(session_id,phase,page)) candidates=[];audit=[];keys=[];warnings=[] if not existing: seen=set() for source in result: try: row,baseline,key=_transform(source,phase,s['origin']) if row['路径'] in seen:raise ValueError('同一页中源 URL 重复') if db.one('SELECT route FROM wp_migration_urls WHERE session_id=? AND route=?',(session_id,row['路径'])): raise ValueError('原站多个内容共用路径 '+row['路径']+'；已保留第一次采集的记录') seen.add(row['路径']);candidates.append(row);audit.append(baseline);keys.append(key) if not row['SEO标题'] or not row['SEO描述']: warnings.append({'phase':phase,'page':page,'issue':'REST 没有提供完整 SEO 元数据：'+row['路径']+'；需要与原网页 HTML 对比'}) if row['封面']: warnings.append({'phase':phase,'page':page,'issue':'外部特色图片仍需本地化：'+row['路径']}) if any(tag.get('src','').startswith(('http://','https://','//')) for tag in BeautifulSoup(row['正文'],'html.parser').find_all('img',src=True)): warnings.append({'phase':phase,'page':page,'issue':'正文中存在远端图片，需要本地化：'+row['路径']}) if source.get('acf') or 'elementor' in str(source.get('content',{})).lower(): warnings.append({'phase':phase,'page':page,'issue':'包含 ACF/页面构建器信息，必须人工复核布局：'+row['路径']}) except (ValueError,TypeError,AttributeError) as e: warnings.append({'phase':phase,'page':page,'issue':str(e)[:230]}) if s['fetched']+len(result)>s['max_items']: # Do not commit a partial last page and falsely claim an exhaustive scan. db.execute("UPDATE wp_migration_sessions SET state='limit_reached',lease_until=NULL,last_error=?,updated_at=? WHERE id=?", ('达到本会话最大采集上限；请建立更大上限的新迁移会话',db.now(),session_id)) return details(site_id,session_id) origin_key=f'wp-session:{session_id}:{phase}:{page}' batch=db.one('SELECT id FROM v8_import_batches WHERE site_id=? AND origin=?',(site_id,origin_key)) if candidates and not batch: staged=v8_workflows.stage(site_id,candidates, {'title':'标题','route':'路径','body':'正文','summary':'摘要', 'seo_title':'SEO标题','seo_description':'SEO描述','canonical':'Canonical','cover':'封面'}, 'mixed',origin_key,keys) batch_id=staged['batch_id'] else:batch_id=batch['id'] if batch else None # Keep bounded WordPress source IDs/ACF refs with the staged content. from . import wp_migration_v83 if batch_id and not existing: wp_migration_v83.staged_relations(site_id,session_id,result,phase,batch_id) with db.connect() as c: c.execute('BEGIN IMMEDIATE') for item in audit: c.execute('''INSERT OR IGNORE INTO wp_migration_urls(session_id,route,original_url,seo_title,description,canonical,robots,h1,status) VALUES(?,?,?,?,?,?,?,?,?)''',(session_id,item['route'],item['url'],item['seo_title'], item['description'],item['canonical'],item['robots'],item['h1'],200)) c.execute('''INSERT OR IGNORE INTO wp_migration_pages(session_id,phase,page,batch_id,source_count) VALUES(?,?,?,?,?)''',(session_id,phase,page,batch_id,len(result))) if batch_id and not existing: wp_migration_v83.refresh_staged_relations(site_id,session_id) else:warnings=[] previous_warnings=json.loads(s['warnings_json'] or '[]') next_phase=phase;next_page=page+1;state='paused' if len(result)=s['max_items'] and state!='ready':state='limit_reached' db.execute('''UPDATE wp_migration_sessions SET state=?,phase=?,next_page=?,fetched=fetched+?, warnings_json=?,reviewed=reviewed+?,lease_until=NULL,updated_at=? WHERE id=?''', (state,next_phase,next_page,len(result) if not existing else 0, json.dumps((previous_warnings+warnings)[:MAX_WARNINGS],ensure_ascii=False),len(warnings),db.now(),session_id)) return details(site_id,session_id) except Exception as exc: # Do not overwrite a concurrent cancel. Preserve cursor; retries are idempotent. if db.one('SELECT cancelled FROM wp_migration_sessions WHERE id=?',(session_id,))['cancelled']: raise HTTPException(409,'迁移会话已取消') from exc db.execute("UPDATE wp_migration_sessions SET state='paused',lease_until=NULL,last_error=?,updated_at=? WHERE id=?", (str(exc)[:500],db.now(),session_id)) if isinstance(exc,HTTPException):raise raise HTTPException(502,'WordPress 采集失败，可重试当前页：'+str(exc)[:200]) from exc def finalize(site_id,session_id): s=_one(site_id,session_id) if s['state'] in ('completed','completed_with_warnings'): return details(site_id,session_id) if s['state']!='ready':raise HTTPException(409,'必须先完成 WordPress 页面与文章的全部采集') pages=db.rows('SELECT * FROM wp_migration_pages WHERE session_id=? ORDER BY CASE phase WHEN ? THEN 0 ELSE 1 END,page',(session_id,'pages')) if not pages:raise HTTPException(409,'尚未读取任何页面') results=[] for p in pages: if not p['batch_id']:continue outcome=v8_workflows.commit(site_id,p['batch_id'],'skip') results.append(outcome) if outcome['state']!='committed': raise HTTPException(409,'部分内容导入失败，请处理批次后重试') # Source taxonomy archive routes enter the SEO baseline, never silently generated. from . import wp_migration_v83 wp_migration_v83.register_archive_baseline(site_id,session_id) source=db.rows('SELECT * FROM wp_migration_urls WHERE session_id=?',(session_id,)) if not source:raise HTTPException(409,'没有可以迁移的有效原站 URL') manifest={'records':[{'route':r['route'],'original_url':r['original_url'], 'seo_title':r['seo_title'],'description':r['description'], 'canonical':r['canonical'],'robots':r['robots'],'h1':r['h1'],'status':200} for r in source], 'assets':[],'errors':[],'warnings':[]} migration_audit.save(site_id,manifest,replace=False) # Compare discovered Sitemap URLs and unresolved media before release. from . import wp_migration_v82 extra_findings=wp_migration_v82._findings(site_id,session_id) from . import wp_migration_v83 extra_findings += wp_migration_v83.findings(site_id,session_id) # Findings are reviewable in the existing enterprise gate. Do not silently # treat a 'completed_with_warnings' migration as releasable. if s['reviewed'] or extra_findings: with db.connect() as c: c.execute('BEGIN IMMEDIATE') for index, finding in enumerate(json.loads(s['warnings_json'] or '[]')+[{'issue':x} for x in extra_findings]): issue=('wp-rest:'+session_id+': '+str(finding.get('issue','需要人工核对')))[:600] if c.execute('SELECT id FROM cms_import_reviews WHERE site_id=? AND platform=? AND issue=?', (site_id,'wordpress-rest',issue)).fetchone():continue c.execute('''INSERT INTO cms_import_reviews(id,site_id,platform,row_number,issue,state,created_at) VALUES(?,?,?,?,?,?,?)''',(uuid.uuid4().hex,site_id,'wordpress-rest',index+1,issue,'pending',db.now())) final_state='completed_with_warnings' if s['reviewed'] or extra_findings else 'completed' db.execute('UPDATE wp_migration_sessions SET state=?,updated_at=? WHERE id=?',(final_state,db.now(),session_id)) return {'session':details(site_id,session_id),'batch_results':results,'baseline_saved':len(source), 'requires_review':bool(s['reviewed'] or extra_findings),'sitemap_audit':wp_migration_v82.sitemap_status(site_id,session_id)} def cancel(site_id,session_id): s=_one(site_id,session_id) if s['state'] in ('completed','completed_with_warnings'):raise HTTPException(409,'已完成的迁移不能取消；可使用批次回滚') db.execute("UPDATE wp_migration_sessions SET state='cancelled',cancelled=1,lease_until=NULL,updated_at=? WHERE id=?", (db.now(),session_id)) return details(site_id,session_id) @router.post('/{site_id}/wordpress/sessions') def api_create(request:Request,site_id:str,payload:dict=Body(...)): enterprise.require(request,site_id,'editor') try:return create(site_id,payload.get('origin',''),payload.get('max_items',2000)) except (ValueError,TypeError) as ex:raise HTTPException(400,str(ex)) @router.get('/{site_id}/wordpress/sessions') def api_list(request:Request,site_id:str): enterprise.require(request,site_id,'viewer') return {'items':list_sessions(site_id)} @router.get('/{site_id}/wordpress/sessions/{session_id}') def api_detail(request:Request,site_id:str,session_id:str): enterprise.require(request,site_id,'viewer') return details(site_id,session_id) @router.post('/{site_id}/wordpress/sessions/{session_id}/advance') def api_advance(request:Request,site_id:str,session_id:str): enterprise.require(request,site_id,'editor') return advance(site_id,session_id) @router.post('/{site_id}/wordpress/sessions/{session_id}/finalize') def api_finalize(request:Request,site_id:str,session_id:str): enterprise.require(request,site_id,'editor') return finalize(site_id,session_id) @router.post('/{site_id}/wordpress/sessions/{session_id}/cancel') def api_cancel(request:Request,site_id:str,session_id:str): enterprise.require(request,site_id,'editor') return cancel(site_id,session_id)