cms-SiteForge/app/theme_store.py

"""Versioned *declarative* theme packages. Never execute user uploaded templates.

Uploads can customize styles and built-in page blueprint presets. The trusted
Jinja template is copied from the built-in base. Arbitrary HTML/Jinja/JS in
uploaded archives is deliberately not supported.
"""
from __future__ import annotations
import io, json, re, zipfile, shutil
from pathlib import Path
from . import db

BASE=Path(__file__).resolve().parent.parent/'themes'
THEME_ROOT=db.DATA/'theme_packages'
from .core import THEMES as BUILTIN
THEME_ID=re.compile(r'^[a-z][a-z0-9-]{2,39}$')
VERSION=re.compile(r'^\d+\.\d+\.\d+$')

class ThemeError(ValueError):pass

def setup():
    with db.connect() as c:
        c.executescript('''CREATE TABLE IF NOT EXISTS theme_versions (
          theme_id TEXT NOT NULL, version TEXT NOT NULL, name TEXT NOT NULL,
          description TEXT NOT NULL DEFAULT '', installed_at TEXT NOT NULL,
          PRIMARY KEY(theme_id,version));
          CREATE TABLE IF NOT EXISTS site_theme_pins (
          site_id TEXT PRIMARY KEY REFERENCES sites(id) ON DELETE CASCADE,
          theme_id TEXT NOT NULL, version TEXT NOT NULL);
        ''')

def catalog():
    builtin=[]
    for tid in BUILTIN:
        obj=json.loads((BASE/tid/'manifest.json').read_text(encoding='utf-8'))
        builtin.append({**obj,'builtin':True,'installed':True})
    custom=[dict(r,builtin=False,installed=True) for r in db.rows('SELECT * FROM theme_versions ORDER BY installed_at DESC')]
    return builtin+custom

def ids():
    return set(BUILTIN)|{r['theme_id'] for r in db.rows('SELECT DISTINCT theme_id FROM theme_versions')}

def latest(tid):
    if tid in BUILTIN:
        return json.loads((BASE/tid/'manifest.json').read_text(encoding='utf-8')).get('version','1.0.0')
    items=db.rows('SELECT version FROM theme_versions WHERE theme_id=?',(tid,))
    if not items:raise ThemeError('主题不存在')
    return sorted((r['version'] for r in items),key=lambda s:tuple(map(int,s.split('.'))))[-1]

def resolve(tid,version=None):
    if tid in BUILTIN:
        current=latest(tid)
        if not version or version==current:return BASE/tid
        if not VERSION.fullmatch(version):raise ThemeError('非法主题版本')
        previous=(BASE/tid/'versions'/version).resolve()
        if not previous.is_relative_to((BASE/tid/'versions').resolve()) or not previous.is_dir():
            raise ThemeError('内置主题的历史版本不存在')
        return previous
    ver=version or latest(tid)
    if not db.one('SELECT theme_id FROM theme_versions WHERE theme_id=? AND version=?',(tid,ver)):
        raise ThemeError('主题版本不存在')
    result=(THEME_ROOT/tid/ver).resolve()
    if not result.is_relative_to(THEME_ROOT.resolve()) or not result.is_dir():raise ThemeError('主题目录丢失')
    return result

def pin(site_id,tid,ver=None):
    target=resolve(tid,ver)
    ver=ver or latest(tid)
    with db.connect() as c:
        c.execute('INSERT INTO site_theme_pins VALUES (?,?,?) ON CONFLICT(site_id) DO UPDATE SET theme_id=excluded.theme_id,version=excluded.version',(site_id,tid,ver))
        c.execute('UPDATE sites SET theme=?,updated_at=? WHERE id=?',(tid,db.now(),site_id))
    return {'theme':tid,'version':ver,'path':str(target)}

def pinned(site_id):
    return db.one('SELECT theme_id,version FROM site_theme_pins WHERE site_id=?',(site_id,))

def compatibility(theme_id,version=None):
    """Theme build/renderer compatibility, not a subjective visual quality score."""
    from . import blueprints
    try:folder=resolve(theme_id,version)
    except ThemeError as exc:return {'valid':False,'errors':[str(exc)],'warnings':[]}
    issues=[];warnings=[]
    for resource in ('manifest.json','theme.css','page-presets.json','pro.html','pro.css'):
        if not (folder/resource).is_file():issues.append('缺少 '+resource)
    if not issues:
        try:manifest=json.loads((folder/'manifest.json').read_text(encoding='utf-8'))
        except (ValueError,UnicodeError) as exc:issues.append('主题清单解析失败：'+str(exc));manifest={}
        if manifest.get('id')!=theme_id:issues.append('清单 ID 与文件夹不匹配')
        try:presets=json.loads((folder/'page-presets.json').read_text(encoding='utf-8'))
        except (ValueError,UnicodeError):presets={};issues.append('页面预设 JSON 不合法')
        missing=[kind for kind in blueprints.PAGE_TEMPLATES if kind not in presets]
        if missing:warnings.append('缺少可编辑页面预设：'+','.join(missing))
        layout_family=manifest.get('layout_family',theme_id)
        family=layout_family if layout_family in BUILTIN else 'corporate'
        page_root=(folder/'pages') if (folder/'pages').exists() else BASE/family/'pages'
        for kind in ('home','about','contact','listing','product','brand','article','case','service','faq','category','search','standard','notfound'):
            if not (page_root/(kind+'.html')).is_file():issues.append('缺少实际页面布局：'+kind)
    return {'valid':not issues,'theme':theme_id,'version':version or (latest(theme_id) if not issues else version),
            'checked_page_types':14,'errors':issues,'warnings':warnings,
            'note':'技术兼容性检查不会代替企业内容准确性和真实设备视觉验收'}

def install(blob:bytes):
    if len(blob)>8_000_000:raise ThemeError('主题包最大 8 MB')
    try:archive=zipfile.ZipFile(io.BytesIO(blob))
    except zipfile.BadZipFile:raise ThemeError('不是 ZIP 压缩包')
    names={n.filename for n in archive.infolist()}
    permitted={'manifest.json','theme.css','page-presets.json','preview.png','preview.jpg','preview.webp'}
    if not {'manifest.json','theme.css','page-presets.json'}.issubset(names):raise ThemeError('缺少主题清单、样式或页面预设')
    if len(names)>6 or any(name not in permitted for name in names):raise ThemeError('主题包只能包含受支持的声明式文件，不允许模板脚本')
    for f in archive.infolist():
        if f.file_size>1_500_000 or f.is_dir():raise ThemeError('主题资源体积或结构不符合要求')
    manifest=json.loads(archive.read('manifest.json'))
    tid=manifest.get('id',''); ver=manifest.get('version','')
    if not THEME_ID.fullmatch(tid) or not VERSION.fullmatch(ver) or tid in BUILTIN:raise ThemeError('主题 ID 或 SemVer 不合法，内置主题不能覆盖')
    if db.one('SELECT theme_id FROM theme_versions WHERE theme_id=? AND version=?',(tid,ver)):
        raise ThemeError('同版本主题已存在，请升级版本号')
    if str(manifest.get('layout_family') or 'corporate') not in BUILTIN:
        raise ThemeError('layout_family 只能选择已审查的基础布局')
    title=str(manifest.get('name',''))[:100]; description=str(manifest.get('description',''))[:400]
    if not title:raise ThemeError('主题名称不能为空')
    presets=json.loads(archive.read('page-presets.json'))
    if not isinstance(presets,dict):raise ThemeError('页面模板预设必须为对象')
    from . import blueprints
    for kind,blocks in presets.items():
        if kind not in blueprints.PAGE_TEMPLATES:raise ThemeError('未知的页面模板：'+kind)
        if not isinstance(blocks,list):raise ThemeError('页面预设内容必须是区块数组')
        for b in blocks:
            if not isinstance(b,dict):raise ThemeError('页面预设区块格式不正确')
        blueprints.validate_blocks([blueprints.new_block(b['type'],b) for b in blocks])
    css=archive.read('theme.css').decode('utf-8')
    # Style packages are declarative: no network requests or executable CSS extensions.
    if len(css)>180000 or re.search(r'@import\b|url\s*\(|expression\s*\(|behavior\s*:|</style|<script',css,re.I):
        raise ThemeError('主题样式不允许引用外部地址或脚本行为')
    temp=THEME_ROOT/tid/(ver+'-upload')
    path=THEME_ROOT/tid/ver
    if path.exists():raise ThemeError('主题版本文件已存在')
    temp.mkdir(parents=True,exist_ok=False)
    try:
        (temp/'manifest.json').write_text(json.dumps(manifest,ensure_ascii=False),encoding='utf-8')
        (temp/'theme.css').write_text(css,encoding='utf-8')
        (temp/'page-presets.json').write_text(json.dumps(presets,ensure_ascii=False),encoding='utf-8')
        # This Jinja template is shipped and reviewed with SiteForge itself.
        family=str(manifest.get('layout_family') or 'corporate')
        if family not in BUILTIN:raise ThemeError('不支持的专业主题版式')
        shutil.copy2(BASE/family/'site.html',temp/'site.html')
        shutil.copy2(BASE/family/'pro.html',temp/'pro.html')
        shutil.copy2(BASE/family/'pro.css',temp/'pro.css')
        # Custom declarative themes inherit the trusted family's local visual assets.
        if (BASE/family/'visual').is_dir():
            shutil.copytree(BASE/family/'visual',temp/'visual')
        for n in ('preview.png','preview.jpg','preview.webp'):
            if n in names:(temp/n).write_bytes(archive.read(n))
        temp.rename(path)
        db.insert('INSERT INTO theme_versions VALUES (?,?,?,?,?)',(tid,ver,title,description,db.now()))
    except Exception:
        shutil.rmtree(temp,ignore_errors=True)
        if path.exists() and not db.one('SELECT theme_id FROM theme_versions WHERE theme_id=? AND version=?',(tid,ver)):
            shutil.rmtree(path,ignore_errors=True)
        raise
    return {'id':tid,'version':ver,'name':title,'page_templates':list(presets)}