"""Whole-site header/footer/navigation/responsive drafts and publications. Only site-design `publish` changes the source of static builds. A draft is versioned and fenced against edits to the underlying site settings. """ from __future__ import annotations import hashlib import json import os import re import uuid from fastapi import APIRouter, Body, HTTPException, Request from . import db, enterprise from .core import route_path router=APIRouter(prefix='/api/sites',tags=['V8.7 site design']) DEFAULTS={ 'header':{'announcement':'','show_announcement':False,'show_search':True, 'show_cta':True,'cta_text':'联系我们','cta_href':'/contact/'}, 'footer':{'title':'','description':'','email':'','show_nav':True}, 'nav':[], 'mobile_tabs':{'enabled':False,'items':[]}, 'tokens':{'font':'system','heading_scale':1.0,'radius':12,'section_spacing':72, 'container_width':1200,'button_style':'rounded','header_style':'standard'}, 'responsive':{'tablet_gap':64,'mobile_gap':40,'mobile_hide_announcement':True, 'mobile_hide_footer_links':False,'mobile_container_padding':20}, } def require_legacy_enabled(): """Block historical direct-to-live design writes unless explicitly enabled. Operators migrating an old automation may temporarily enable this escape hatch; the V8.7 editor never uses it. """ if os.environ.get('SITEFORGE_ALLOW_LEGACY_DIRECT_DESIGN_WRITE') != '1': raise HTTPException(409,'旧版直接修改正式设计的接口已停用；请使用 site-design 或 studio-draft 草稿发布流程') def setup(): with db.connect() as c: c.execute('''CREATE TABLE IF NOT EXISTS site_design_drafts ( site_id TEXT PRIMARY KEY, revision INTEGER NOT NULL, base_hash TEXT NOT NULL, config_json TEXT NOT NULL, updated_at TEXT NOT NULL)''') c.execute('''CREATE TABLE IF NOT EXISTS site_design_publications ( id TEXT PRIMARY KEY, site_id TEXT NOT NULL, actor TEXT NOT NULL, revision INTEGER NOT NULL, old_config_json TEXT NOT NULL, new_config_json TEXT NOT NULL, published_at TEXT NOT NULL)''') def _site(site_id): row=db.one('SELECT * FROM sites WHERE id=?',(site_id,)) if not row:raise HTTPException(404,'网站不存在') return row def _settings(row): return json.loads(row['settings_json'] or '{}') def _hash(settings): return hashlib.sha256(json.dumps(settings,sort_keys=True,ensure_ascii=False).encode()).hexdigest() def _url(value): if not isinstance(value,str) or len(value)>350 or any(x in value for x in ('<','>','"',"'",'\\','\n','\r','\t')): raise HTTPException(400,'不安全的网址') if value.startswith('/') and not value.startswith('//'): try:return route_path(value) except ValueError:raise HTTPException(400,'不合法的站内链接') # External links are intentionally not allowed inside site chrome; the # enterprise's external CTA can be built with a vetted form-link block. raise HTTPException(400,'全站导航只允许以 / 开头的站内路径') def _nav(rows,depth=0,counter=None): if counter is None:counter=[0] if not isinstance(rows,list) or len(rows)>18:raise HTTPException(400,'导航项过多') result=[] for x in rows: counter[0]+=1 if counter[0]>48 or not isinstance(x,dict):raise HTTPException(400,'导航结构不合法') if set(x)-{'title','route','children'}:raise HTTPException(400,'导航出现未支持的字段') title=x.get('title') if not isinstance(title,str) or not title.strip() or len(title)>48:raise HTTPException(400,'导航名称不合法') children=x.get('children',[]) if depth==2 and children:raise HTTPException(400,'最多支持三级导航') result.append({'title':title.strip(),'route':_url(x.get('route','/')), 'children':_nav(children,depth+1,counter) if depth<2 else []}) return result def _mobile_tab_url(value): """Allow site-internal routes plus safe, explicit user-facing contact links.""" from urllib.parse import urlsplit if not isinstance(value,str) or not value or len(value)>350 or any(c in value for c in ('<','>','"',"'",'\\','\n','\r','\t')): raise HTTPException(400,'底部导航网址不安全') if value.startswith('/') and not value.startswith('//'): return _url(value) if re.fullmatch(r'tel:\+?[0-9()\- ]{5,30}',value): return value if re.fullmatch(r'mailto:[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}',value): return value parsed=urlsplit(value) if parsed.scheme=='https' and parsed.hostname and not parsed.username and not parsed.password and not any(c.isspace() for c in value): return value raise HTTPException(400,'仅支持站内路径、HTTPS、电话或邮箱链接') def validate(raw): if not isinstance(raw,dict) or set(raw)-set(DEFAULTS):raise HTTPException(400,'未知的全站设计设置') normalized={} for group in ('header','footer','responsive'): val=raw.get(group,{}) if not isinstance(val,dict) or set(val)-set(DEFAULTS[group]):raise HTTPException(400,'无效的全站设置组: '+group) normalized[group]={**DEFAULTS[group],**val} header=normalized['header'];footer=normalized['footer'];mobile=normalized['responsive'] for key in ('show_announcement','show_search','show_cta'): if type(header[key]) is not bool:raise HTTPException(400,'无效的页眉开关') for key in ('show_nav',): if type(footer[key]) is not bool:raise HTTPException(400,'无效的页脚开关') for key in ('mobile_hide_announcement','mobile_hide_footer_links'): if type(mobile[key]) is not bool:raise HTTPException(400,'无效的移动端开关') for field,cap in [('announcement',140),('cta_text',40)]: if not isinstance(header[field],str) or len(header[field])>cap:raise HTTPException(400,'页眉文字超限') for field,cap in [('title',120),('description',650),('email',150)]: if not isinstance(footer[field],str) or len(footer[field])>cap:raise HTTPException(400,'页脚文字超限') header['cta_href']=_url(header['cta_href']) for field,lo,hi in [('tablet_gap',20,120),('mobile_gap',16,100),('mobile_container_padding',8,48)]: n=mobile[field] if type(n) is not int or not lo<=n<=hi:raise HTTPException(400,'设备间距超出范围: '+field) tokens=raw.get('tokens',{}) if not isinstance(tokens,dict) or set(tokens)-set(DEFAULTS['tokens']): raise HTTPException(400,'未知设计规范字段') tokens={**DEFAULTS['tokens'],**tokens} for field,options in [('font',('system','serif','modern')), ('button_style',('rounded','square','pill')),('header_style',('standard','compact'))]: if tokens[field] not in options:raise HTTPException(400,'不支持的设计规范: '+field) for field,lo,hi in [('heading_scale',.75,1.5),('radius',0,30), ('section_spacing',32,140),('container_width',900,1600)]: v=tokens[field] if type(v) not in (float,int) or not lo<=v<=hi:raise HTTPException(400,'设计变量超出范围: '+field) normalized['tokens']=tokens normalized['nav']=_nav(raw.get('nav',[])) bar=raw.get('mobile_tabs',{}) if not isinstance(bar,dict) or set(bar)-{'enabled','items'}: raise HTTPException(400,'底部导航配置格式不正确') enabled=bar.get('enabled',False) items=bar.get('items',[]) if type(enabled) is not bool or not isinstance(items,list) or len(items)>5: raise HTTPException(400,'底部导航仅支持 0 至 5 项') allowed_icons={'home','grid','products','news','contact','phone','info','search','menu','star'} result=[] for item in items: if not isinstance(item,dict) or set(item)-{'title','route','icon','action'}: raise HTTPException(400,'底部导航项目格式不正确') title=item.get('title') icon=item.get('icon','grid') action=item.get('action','link') if not isinstance(title,str) or not 1<=len(title.strip())<=12: raise HTTPException(400,'底部导航标题须为 1 至 12 字') if icon not in allowed_icons or action not in ('link','menu'): raise HTTPException(400,'底部导航图标或操作不受支持') if action=='menu': route='/' else: route=_mobile_tab_url(item.get('route','/')) result.append({'title':title.strip(),'icon':icon,'route':route,'action':action}) if enabled and not result: raise HTTPException(400,'启用底部导航时至少添加一项；不需要可直接关闭') normalized['mobile_tabs']={'enabled':enabled,'items':result} if len(json.dumps(normalized,ensure_ascii=False))>35_000:raise HTTPException(400,'全站设计过大') return normalized def published(row_or_settings): settings=_settings(row_or_settings) if 'settings_json' in row_or_settings else row_or_settings merged=dict(settings.get('site_design') or {}) if 'tokens' not in merged:merged['tokens']=settings.get('design_tokens') or {} if 'nav' not in merged:merged['nav']=settings.get('nav') or [] if 'footer' not in merged:merged['footer']={'description':settings.get('footer_text') or '', 'email':settings.get('email') or ''} return validate(merged) def read(site_id): site=_site(site_id) config=published(site) draft=db.one('SELECT * FROM site_design_drafts WHERE site_id=?',(site_id,)) return {'config':json.loads(draft['config_json']) if draft else config, 'revision':int(draft['revision']) if draft else 0, 'state':'draft' if draft else 'published', 'conflict':bool(draft and draft['base_hash']!=_hash(_settings(site)))} def write(site_id,payload): plan=validate(payload.get('config')) site=_site(site_id) with db.connect() as c: old=c.execute('SELECT revision,base_hash FROM site_design_drafts WHERE site_id=?',(site_id,)).fetchone() revision=old[0] if old else 0 if type(payload.get('expected_revision')) is not int or payload['expected_revision']!=revision: raise HTTPException(409,'全站草稿已更新，请重新加载') if old and old[1]!=_hash(_settings(site)): raise HTTPException(409,'网站信息已在其他窗口修改') base=old[1] if old else _hash(_settings(site)) c.execute('''INSERT INTO site_design_drafts VALUES(?,?,?,?,?) ON CONFLICT(site_id) DO UPDATE SET revision=excluded.revision, config_json=excluded.config_json, updated_at=excluded.updated_at''', (site_id,revision+1,base,json.dumps(plan,ensure_ascii=False),db.now())) return {'saved':True,'state':'draft','revision':revision+1,'build_unchanged':True} def publish(site_id,expected_revision,actor,expected_plan_token=None): site=_site(site_id) with db.connect() as c: c.execute('BEGIN IMMEDIATE') draft=c.execute('SELECT * FROM site_design_drafts WHERE site_id=?',(site_id,)).fetchone() if not draft:raise HTTPException(409,'尚无全站设计草稿') if type(expected_revision) is not int or expected_revision!=draft['revision']: raise HTTPException(409,'草稿版本不匹配') live=c.execute('SELECT settings_json FROM sites WHERE id=?',(site_id,)).fetchone() settings=json.loads(live['settings_json']) if draft['base_hash']!=_hash(settings):raise HTTPException(409,'网站信息已变更，请重新审核') before=published(settings) config=validate(json.loads(draft['config_json'])) if expected_plan_token is not None: proof=_review_token(site_id,draft['revision'],settings,config) if not isinstance(expected_plan_token,str) or not __import__('hmac').compare_digest(expected_plan_token,proof): raise HTTPException(409,'审核预览已过期，请重新查看差异后确认') settings['site_design']=config settings['design_tokens']=config['tokens'] # Keep old nav/settings for non-professional legacy templates too. settings['nav']=config['nav'] c.execute('UPDATE sites SET settings_json=?,updated_at=? WHERE id=?', (json.dumps(settings,ensure_ascii=False),db.now(),site_id)) publication_id=uuid.uuid4().hex c.execute('INSERT INTO site_design_publications VALUES(?,?,?,?,?,?,?)', (publication_id,site_id,actor,expected_revision,json.dumps(before,ensure_ascii=False), json.dumps(config,ensure_ascii=False),db.now())) c.execute('DELETE FROM site_design_drafts WHERE site_id=?',(site_id,)) return {'published':True,'publication_id':publication_id,'build_required':True} # V8.26 - reviewable, immutable full-site design changes. This workflow uses # the existing site_design_drafts/publications tables: no destructive migration. REVIEW_SECTIONS = ('header', 'nav', 'footer', 'mobile_tabs', 'tokens', 'responsive') def _changes(before, after): """Return a bounded, deterministic, safe structured diff for human review.""" old, new = validate(before), validate(after) rows = [] def inspect(path, a, b): if a == b: return if isinstance(a, dict) and isinstance(b, dict): for key in sorted(set(a) | set(b)): inspect(path + '.' + key, a.get(key), b.get(key)) elif isinstance(a, list) and isinstance(b, list): # Lists (menu order) are atomic to avoid obscuring a move as deletes. rows.append({'field': path, 'before': a, 'after': b}) else: rows.append({'field': path, 'before': a, 'after': b}) for section in REVIEW_SECTIONS: inspect(section, old[section], new[section]) return rows def _review_token(site_id, revision, live_settings, draft_config): payload={'site_id':site_id,'revision':revision, 'settings_hash':_hash(live_settings), 'draft':validate(draft_config)} return hashlib.sha256(json.dumps(payload,ensure_ascii=False, sort_keys=True,separators=(',',':')).encode()).hexdigest() def review(site_id): with db.connect() as c: site=c.execute('SELECT settings_json FROM sites WHERE id=?',(site_id,)).fetchone() if not site:raise HTTPException(404,'网站不存在') draft=c.execute('SELECT * FROM site_design_drafts WHERE site_id=?',(site_id,)).fetchone() if not draft:raise HTTPException(409,'没有待审核的全站设计草稿') settings=json.loads(site['settings_json'] or '{}') if draft['base_hash']!=_hash(settings):raise HTTPException(409,'网站设置已经修改，请先处理草稿冲突') proposed=validate(json.loads(draft['config_json'])) delta=_changes(published(settings),proposed) sections=[name for name in REVIEW_SECTIONS if any( r['field']==name or r['field'].startswith(name+'.') for r in delta)] return {'revision':draft['revision'], 'changed_fields':delta, 'changed_sections':sections, 'change_count':len(delta), 'affected_scope':'entire_site_next_build', 'published_site_unchanged':True, 'plan_token':_review_token(site_id,draft['revision'],settings,proposed)} def restore_publication(site_id,publication_id,which,expected_revision,expected_live_hash): """Restore a historical snapshot as a NEW draft, never touch live settings.""" if which not in ('before','after'):raise HTTPException(400,'恢复来源仅支持 before 或 after') if type(expected_revision) is not int:raise HTTPException(400,'需要草稿版本号') if not isinstance(expected_live_hash,str) or len(expected_live_hash)!=64: raise HTTPException(400,'需要当前全站发布版本指纹') with db.connect() as c: c.execute('BEGIN IMMEDIATE') site=c.execute('SELECT settings_json FROM sites WHERE id=?',(site_id,)).fetchone() if not site:raise HTTPException(404,'网站不存在') historic=c.execute('SELECT * FROM site_design_publications WHERE site_id=? AND id=?', (site_id,publication_id)).fetchone() if not historic:raise HTTPException(404,'发布记录不存在') settings=json.loads(site['settings_json'] or '{}') live_hash=_hash(settings) if live_hash!=expected_live_hash:raise HTTPException(409,'全站发布版本已变化，请重新载入') raw=historic['old_config_json'] if which=='before' else historic['new_config_json'] plan=validate(json.loads(raw)) old=c.execute('SELECT revision FROM site_design_drafts WHERE site_id=?',(site_id,)).fetchone() current=old['revision'] if old else 0 if expected_revision!=current:raise HTTPException(409,'草稿版本已变化，请重新加载') new_rev=current+1 c.execute('INSERT INTO site_design_drafts(site_id,revision,base_hash,config_json,updated_at) ' 'VALUES(?,?,?,?,?) ON CONFLICT(site_id) DO UPDATE SET ' 'revision=excluded.revision,base_hash=excluded.base_hash, ' 'config_json=excluded.config_json,updated_at=excluded.updated_at', (site_id,new_rev,live_hash,json.dumps(plan,ensure_ascii=False),db.now())) return {'restored_as_draft':True,'revision':new_rev,'published_unchanged':True} def responsive_css(config): r=validate(config)['responsive'] tablet=r['tablet_gap'];mobile=r['mobile_gap'];padding=r['mobile_container_padding'] rules=(f'@media(max-width:1023px){{.sf-canvas .sf-section{{padding-block:{tablet}px}}}}' f'@media(max-width:767px){{.sf-canvas .sf-section{{padding-block:{mobile}px}}' f'.pro-wrap,.sf-container{{padding-left:{padding}px;padding-right:{padding}px}}') if r['mobile_hide_announcement']:rules+='body .pro-utility{display:none!important}' if r['mobile_hide_footer_links']:rules+='body .pro-footer-links{display:none!important}' return 'body .pro-utility[hidden]{display:none!important}'+rules+'}' def token_css(tokens): safe=validate({'tokens':tokens})['tokens'] fonts={'system':'system-ui,-apple-system,BlinkMacSystemFont,sans-serif', 'serif':'Georgia,Times New Roman,serif', 'modern':'Arial,Helvetica,sans-serif'} radius=safe['radius'];width=safe['container_width'];space=safe['section_spacing'] scale=safe['heading_scale'] button={'rounded':f'{radius}px','square':'0px','pill':'999px'}[safe['button_style']] return (f':root{{--sf-font:{fonts[safe["font"]]};--sf-radius:{radius}px;' f'--sf-space:{space}px;--sf-width:{width}px;--sf-heading-scale:{scale};}}' f'body{{font-family:var(--sf-font)}}' f'.sf-canvas .sf-section,.sf-canvas .section{{padding-block:var(--sf-space)}}' f'.container,.sf-container{{max-width:var(--sf-width)}}' f'.btn,.primary,.request,.button,.nav-btn{{border-radius:{button}}}') @router.get('/{site_id}/site-design') def api_read(request:Request,site_id:str): enterprise.require(request,site_id,'viewer') return read(site_id) @router.put('/{site_id}/site-design') def api_write(request:Request,site_id:str,payload:dict=Body(...)): enterprise.require(request,site_id,'editor') return write(site_id,payload) @router.post('/{site_id}/site-design/publish') def api_publish(request:Request,site_id:str,payload:dict=Body(...)): enterprise.require(request,site_id,'owner') return publish(site_id,payload.get('expected_revision'),enterprise.actor(request)) @router.delete('/{site_id}/site-design') def api_discard(request:Request,site_id:str): enterprise.require(request,site_id,'editor') _site(site_id) db.execute('DELETE FROM site_design_drafts WHERE site_id=?',(site_id,)) return {'discarded':True} @router.get('/{site_id}/site-design/publications') def api_history(request:Request,site_id:str): enterprise.require(request,site_id,'viewer') _site(site_id) return db.rows('''SELECT id,site_id,actor,revision,published_at FROM site_design_publications WHERE site_id=? ORDER BY published_at DESC LIMIT 50''',(site_id,)) @router.get('/{site_id}/site-design/review') def api_review(request:Request,site_id:str): enterprise.require(request,site_id,'viewer') return review(site_id) @router.post('/{site_id}/site-design/publish-reviewed') def api_publish_reviewed(request:Request,site_id:str,payload:dict=Body(...)): enterprise.require(request,site_id,'owner') return publish(site_id,payload.get('expected_revision'),enterprise.actor(request), payload.get('plan_token','')) @router.get('/{site_id}/site-design/publications/{publication_id}') def api_publication(request:Request,site_id:str,publication_id:str): enterprise.require(request,site_id,'viewer') _site(site_id) row=db.one('SELECT * FROM site_design_publications WHERE site_id=? AND id=?', (site_id,publication_id)) if not row:raise HTTPException(404,'发布版本不存在') old=validate(json.loads(row['old_config_json'])) new=validate(json.loads(row['new_config_json'])) return {'id':row['id'],'published_at':row['published_at'],'actor':row['actor'], 'before':old,'after':new,'changes':_changes(old,new)} @router.get('/{site_id}/site-design/live-hash') def api_live_hash(request:Request,site_id:str): enterprise.require(request,site_id,'viewer') return {'live_hash':_hash(_settings(_site(site_id)))} @router.post('/{site_id}/site-design/publications/{publication_id}/restore') def api_restore_publication(request:Request,site_id:str,publication_id:str,payload:dict=Body(...)): enterprise.require(request,site_id,'editor') return restore_publication(site_id,publication_id,payload.get('snapshot'), payload.get('expected_revision'),payload.get('expected_live_hash'))