"""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)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))