184 lines
9.9 KiB
Python
184 lines
9.9 KiB
Python
|
"""V8.20 offline WordPress WXR compatibility preflight.
|
||
|
|
|
||
|
|
No network access, PHP evaluation or database writes. Findings are blockers or
|
||
|
|
explicit review items, never evidence of pixel-identical migration.
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import re
|
||
|
|
from collections import Counter
|
||
|
|
from urllib.parse import urlsplit
|
||
|
|
from defusedxml import ElementTree as SafeET
|
||
|
|
from bs4 import BeautifulSoup
|
||
|
|
from fastapi import APIRouter, File, HTTPException, Request, UploadFile
|
||
|
|
from . import enterprise
|
||
|
|
from .core import route_path
|
||
|
|
|
||
|
|
router = APIRouter(prefix='/api/migrations/sites',tags=['WordPress V8.20 preflight'])
|
||
|
|
MAX_XML = 16_000_000
|
||
|
|
MAX_ITEMS = 20_000
|
||
|
|
MAX_FINDINGS = 500
|
||
|
|
MAX_META = 500
|
||
|
|
SHORTCODE = re.compile(r'\[(?!/)(contact-form-7|gravityform|woocommerce|products|wpforms|elementor-template|rev_slider)\b',re.I)
|
||
|
|
SERIALIZED = re.compile(r'^(?:a:\d+:\{|O:\d+:|C:\d+:|s:\d+:|b:[01];)',re.S)
|
||
|
|
|
||
|
|
|
||
|
|
def postmeta(item):
|
||
|
|
"""Preserve duplicate keys as a finding instead of silently overwriting them."""
|
||
|
|
result={};duplicate=[]
|
||
|
|
for node in item:
|
||
|
|
if node.tag.rsplit('}',1)[-1]!='postmeta':continue
|
||
|
|
key=value=''
|
||
|
|
for field in node:
|
||
|
|
label=field.tag.rsplit('}',1)[-1]
|
||
|
|
if label=='meta_key':key=field.text or ''
|
||
|
|
elif label=='meta_value':value=field.text or ''
|
||
|
|
if key:
|
||
|
|
if key in result:duplicate.append(key[:100])
|
||
|
|
result[key]=value
|
||
|
|
if len(result)>MAX_META:raise ValueError('单篇文章元数据字段超过安全上限')
|
||
|
|
return result,duplicate[:30]
|
||
|
|
|
||
|
|
|
||
|
|
def extract_acf(meta):
|
||
|
|
"""Bounded, literal ACF postmeta extraction. Never PHP-unserialize or infer field schema.
|
||
|
|
|
||
|
|
Can safely represent flat fields and simple `repeater_0_subfield` records.
|
||
|
|
Repeater field types are not guaranteed by the exported meta alone.
|
||
|
|
"""
|
||
|
|
if len(meta)>MAX_META:raise ValueError('ACF meta exceeds safety limit')
|
||
|
|
scalar={};repeaters={};warnings=[]
|
||
|
|
referenced={k[1:] for k,v in meta.items() if k.startswith('_') and isinstance(v,str) and v.startswith('field_')}
|
||
|
|
for key in sorted(referenced):
|
||
|
|
if not key or key not in meta:continue
|
||
|
|
value=meta[key]
|
||
|
|
if not isinstance(value,str):continue
|
||
|
|
if SERIALIZED.match(value.strip()):
|
||
|
|
warnings.append(f'ACF 字段 {key[:75]} 含 PHP 序列化结构，未进行反序列化')
|
||
|
|
continue
|
||
|
|
if len(value.encode('utf-8'))>10_000:
|
||
|
|
warnings.append(f'ACF 字段 {key[:75]} 超出单值安全上限')
|
||
|
|
continue
|
||
|
|
match=re.fullmatch(r'(.{1,80})_(\d{1,2})_([A-Za-z][A-Za-z0-9_]*)',key)
|
||
|
|
if match:
|
||
|
|
group,index,field=match.group(1),int(match.group(2)),match.group(3)
|
||
|
|
if index>=50:continue
|
||
|
|
repeaters.setdefault(group,{}).setdefault(index,{})[field]=value
|
||
|
|
elif re.fullmatch(r'[A-Za-z][A-Za-z0-9_]{0,100}',key):
|
||
|
|
scalar[key]=value
|
||
|
|
result={}
|
||
|
|
for group,values in sorted(repeaters.items()):
|
||
|
|
count=meta.get(group)
|
||
|
|
if not (isinstance(count,str) and count.isdigit() and 1<=int(count)<=50):
|
||
|
|
warnings.append(f'ACF 类重复字段 {group} 缺少有效行数，保留原始键供人工核对')
|
||
|
|
continue
|
||
|
|
expected=int(count)
|
||
|
|
if any(i>=expected for i in values):
|
||
|
|
warnings.append(f'ACF 类重复字段 {group} 存在超出声明行数的字段')
|
||
|
|
# Only expose a complete contiguous set. Partial repeaters remain metadata-only.
|
||
|
|
if any(i not in values for i in range(expected)):
|
||
|
|
warnings.append(f'ACF 类重复字段 {group} 行不连续，必须人工复核')
|
||
|
|
continue
|
||
|
|
if sum(len(x) for x in values.values())>1500:
|
||
|
|
warnings.append(f'ACF 类重复字段 {group} 子字段超出安全上限')
|
||
|
|
continue
|
||
|
|
result[group]=[values[i] for i in range(expected)]
|
||
|
|
if repeaters:
|
||
|
|
warnings.append('ACF Repeater 由字段命名模式推断；必须与源站字段组定义核对')
|
||
|
|
return {'fields':scalar,'repeaters':result,'warnings':list(dict.fromkeys(warnings))[:80]}
|
||
|
|
|
||
|
|
|
||
|
|
def inspect_wxr(data:bytes):
|
||
|
|
if not data or len(data)>MAX_XML:raise ValueError('WXR 必须非空且不超过 16 MB')
|
||
|
|
root=SafeET.fromstring(data)
|
||
|
|
if root.tag.rsplit('}',1)[-1]!='rss':raise ValueError('需要 WordPress WXR RSS XML')
|
||
|
|
items=root.findall('./channel/item')
|
||
|
|
if not items:raise ValueError('WXR 缺少 channel/item 记录')
|
||
|
|
if len(items)>MAX_ITEMS:raise ValueError('WXR 超过 20000 条记录上限')
|
||
|
|
counts=Counter();issues=[];routes={};features=Counter();example=[];hosts=set()
|
||
|
|
def issue(severity,code,wp_id,message,route=''):
|
||
|
|
counts['findings_'+severity]+=1
|
||
|
|
if len(issues)<MAX_FINDINGS:
|
||
|
|
issues.append({'severity':severity,'code':code,'wp_post_id':str(wp_id)[:50],
|
||
|
|
'route':route,'message':message[:240]})
|
||
|
|
from . import wordpress_compat
|
||
|
|
for node in items:
|
||
|
|
direct={n.tag.rsplit('}',1)[-1]:(n.text or '') for n in node if n.tag.rsplit('}',1)[-1] in ('post_type','status','post_id','link','title')}
|
||
|
|
kind=direct.get('post_type','post');status=direct.get('status','publish');wp_id=direct.get('post_id','')
|
||
|
|
counts['total_records']+=1;counts['type_'+kind[:40]]+=1
|
||
|
|
if kind in ('attachment','revision','nav_menu_item'):
|
||
|
|
if kind=='attachment':counts['attachments']+=1
|
||
|
|
continue
|
||
|
|
if status!='publish':
|
||
|
|
counts['non_public']+=1
|
||
|
|
continue
|
||
|
|
counts['published_content']+=1
|
||
|
|
link=direct.get('link','')
|
||
|
|
try:
|
||
|
|
p=urlsplit(link)
|
||
|
|
if p.scheme not in ('http','https') or not p.hostname or p.username or p.password or p.query or p.fragment or not p.path:
|
||
|
|
raise ValueError('缺少固定 HTTP(S) 路径或使用动态参数 URL')
|
||
|
|
route=route_path(p.path)
|
||
|
|
except (ValueError,TypeError):
|
||
|
|
issue('blocker','invalid_permalink',wp_id,'原链接无法无损映射为静态 URL；需要显式路径/301 映射')
|
||
|
|
continue
|
||
|
|
host=p.hostname.lower()
|
||
|
|
if hosts and host not in hosts:
|
||
|
|
issue('review','mixed_source_hosts',wp_id,'源站出现多个主机名，请核对 www/非 www、旧域和路径所属关系',route)
|
||
|
|
hosts.add(host)
|
||
|
|
if route in routes:
|
||
|
|
issue('blocker','duplicate_route',wp_id,'URL 路径与其他已发布内容冲突',route)
|
||
|
|
continue
|
||
|
|
routes[route]=wp_id
|
||
|
|
if len(example)<30:example.append({'route':route,'type':kind,'wp_post_id':wp_id})
|
||
|
|
meta,dup=postmeta(node)
|
||
|
|
if kind in ('product','product_variation') or '_product_attributes' in meta:
|
||
|
|
features['woocommerce']+=1
|
||
|
|
issue('review','woocommerce',wp_id,'WooCommerce 产品、变体、库存、定价与购物车行为需独立迁移验收',route)
|
||
|
|
for key in dup[:5]:issue('review','duplicate_postmeta',wp_id,'重复 WordPress 自定义字段：'+key,route)
|
||
|
|
acf=extract_acf(meta)
|
||
|
|
if acf['fields'] or acf['repeaters']:
|
||
|
|
features['acf']+=1
|
||
|
|
for warning in acf['warnings'][:5]:issue('review','acf_field',wp_id,warning,route)
|
||
|
|
for key,value in meta.items():
|
||
|
|
if not key.startswith('_') and isinstance(value,str) and SERIALIZED.match(value.strip()):
|
||
|
|
features['serialized_meta']+=1
|
||
|
|
issue('review','serialized_meta',wp_id,'含 PHP 序列化元数据：'+key[:60]+'，不会执行或自动解析',route)
|
||
|
|
if features['serialized_meta']>10:break
|
||
|
|
if meta.get('_elementor_data'):
|
||
|
|
features['elementor']+=1
|
||
|
|
result=wordpress_compat.elementor_to_html(meta['_elementor_data'])
|
||
|
|
if result['warnings']:
|
||
|
|
issue('review','elementor',wp_id,'Elementor 非无损转换：'+result['warnings'][0],route)
|
||
|
|
else:issue('review','elementor_layout',wp_id,'Elementor 样式、容器、响应式布局需要视觉对照验收',route)
|
||
|
|
body=next((e.text or '' for e in node if e.tag.startswith('{http://purl.org/rss/1.0/modules/content/}') and e.tag.endswith('encoded')),'')
|
||
|
|
s=BeautifulSoup(body,'html.parser')
|
||
|
|
if SHORTCODE.search(body):
|
||
|
|
features['dynamic_shortcodes']+=1
|
||
|
|
issue('blocker','dynamic_shortcode',wp_id,'存在动态短代码，纯静态导出不能提供同等业务行为',route)
|
||
|
|
if s.select('form,iframe'):
|
||
|
|
features['dynamic_html']+=1
|
||
|
|
issue('review','dynamic_html',wp_id,'表单或 iframe 需单独确认服务、第三方依赖与可用性',route)
|
||
|
|
if any(n.get('src','').startswith(('http://','https://')) for n in s.select('img[src]')):
|
||
|
|
features['external_images']+=1
|
||
|
|
issue('review','external_media',wp_id,'正文使用远程图片；需执行本地化及完整性对账',route)
|
||
|
|
if kind not in ('post','page','product','service','case','brand','vehicle'):
|
||
|
|
features['custom_post_types']+=1
|
||
|
|
issue('review','custom_post_type',wp_id,'自定义文章类型 '+kind[:60]+' 将映射 other；模板/字段关系需人工确认',route)
|
||
|
|
blocked=counts['findings_blocker']
|
||
|
|
from .version import VERSION
|
||
|
|
return {'version':VERSION,'input_bytes':len(data),'counts':dict(counts),
|
||
|
|
'unique_published_routes':len(routes),'features':dict(features),
|
||
|
|
'issues':issues,'findings_truncated':sum(v for k,v in counts.items() if k.startswith('findings_'))>len(issues),
|
||
|
|
'sample_routes':example,'source_hosts':sorted(hosts),'ready_for_automatic_publish':False,
|
||
|
|
'blocking_issues':blocked,
|
||
|
|
'note':'仅离线识别兼容性与风险；预检不改动站点、不执行 PHP，不构成 1:1 迁移验收。'}
|
||
|
|
|
||
|
|
|
||
|
|
@router.post('/{site_id}/wordpress/wxr-preflight')
|
||
|
|
async def wxr_preflight(request:Request,site_id:str,file:UploadFile=File(...)):
|
||
|
|
enterprise.require(request,site_id,'editor')
|
||
|
|
data=await file.read(MAX_XML+1)
|
||
|
|
try:return inspect_wxr(data)
|
||
|
|
except (ValueError,SafeET.ParseError) as exc:raise HTTPException(400,str(exc))
|