cms-SiteForge/app/multilingual.py

"""Explicit multilingual relationships. No automatic fact translation.

One site per locale. Only approved, published and indexed content receives
alternate links. Pending translation records cannot accidentally enter a build.
"""
import json
import re
from urllib.parse import urlparse
from html import escape as html_escape
from xml.sax.saxutils import escape as xml_escape
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, Field
from . import db, enterprise
from .core import uid

router=APIRouter(prefix='/api/i18n', tags=['Multilingual'])
LOCALE_PATTERN=re.compile(r'^[a-z]{2,3}(?:-[A-Z][a-z]{3})?(?:-(?:[A-Z]{2}|[0-9]{3}))?$')
KEY_PATTERN=re.compile(r'^[A-Za-z0-9][A-Za-z0-9._:-]{0,119}$')

def locale_value(locale):
    value=str(locale or '').strip()
    if not LOCALE_PATTERN.fullmatch(value):
        raise ValueError('语言标识必须是 BCP47 格式，例如 zh-CN、en、fr-FR')
    return value

def _domain(site):
    value=(site.get('domain') or '').strip().rstrip('/')
    parsed=urlparse(value)
    if parsed.scheme!='https' or not parsed.hostname or parsed.username or parsed.password or parsed.path not in ('','/') or parsed.query or parsed.fragment:
        raise ValueError('多语言正式站点必须配置独立 HTTPS 域名（不允许路径、账号或查询参数）')
    return value

def setup():
    with db.connect() as c:
        c.executescript('''CREATE TABLE IF NOT EXISTS language_groups(
          id TEXT PRIMARY KEY, name TEXT NOT NULL, default_locale TEXT NOT NULL,
          created_at TEXT NOT NULL);
        CREATE TABLE IF NOT EXISTS language_sites(
          group_id TEXT NOT NULL REFERENCES language_groups(id) ON DELETE CASCADE,
          site_id TEXT NOT NULL REFERENCES sites(id) ON DELETE CASCADE,
          locale TEXT NOT NULL, PRIMARY KEY(group_id,site_id),
          UNIQUE(group_id,locale), UNIQUE(site_id));
        CREATE TABLE IF NOT EXISTS translation_units(
          id TEXT PRIMARY KEY, group_id TEXT NOT NULL REFERENCES language_groups(id) ON DELETE CASCADE,
          reference_key TEXT NOT NULL, created_at TEXT NOT NULL,
          UNIQUE(group_id,reference_key));
        CREATE TABLE IF NOT EXISTS translation_members(
          unit_id TEXT NOT NULL REFERENCES translation_units(id) ON DELETE CASCADE,
          site_id TEXT NOT NULL REFERENCES sites(id) ON DELETE CASCADE,
          entry_id TEXT NOT NULL REFERENCES entries(id) ON DELETE CASCADE,
          state TEXT NOT NULL CHECK(state IN ('pending','approved')),
          PRIMARY KEY(unit_id,site_id),UNIQUE(site_id,entry_id));
        CREATE INDEX IF NOT EXISTS idx_translations_site ON translation_members(site_id,state);
        ''')

def require_group(group_id):
    group=db.one('SELECT * FROM language_groups WHERE id=?',(group_id,))
    if not group:raise HTTPException(404,'语言组不存在')
    return group

def _authorized(request,group_id,role='viewer'):
    sites=db.rows('SELECT site_id FROM language_sites WHERE group_id=?',(group_id,))
    # At least one managed member is needed to read or edit the group.
    if not getattr(request.state,'is_superadmin',False):
        if not any(enterprise.level(request,s['site_id'])>=enterprise.ACCESS[role] for s in sites):
            raise HTTPException(403,'无权管理此语言组')
    return sites

class GroupIn(BaseModel):
    name:str=Field(min_length=2,max_length=120)
    default_locale:str='en'
class GroupSiteIn(BaseModel):
    site_id:str
    locale:str
class UnitIn(BaseModel):
    reference_key:str
class MemberIn(BaseModel):
    site_id:str
    entry_id:str
    state:str='pending'

@router.post('/groups',status_code=201)
def create_group(data:GroupIn,request:Request):
    if not request.state.is_superadmin:raise HTTPException(403,'仅管理员可创建语言组')
    try:locale=locale_value(data.default_locale)
    except ValueError as exc:raise HTTPException(400,str(exc))
    gid=uid()
    db.insert('INSERT INTO language_groups VALUES (?,?,?,?)',(gid,data.name,locale,db.now()))
    return {'id':gid,'name':data.name,'default_locale':locale}

class LocaleIn(BaseModel):
    locale:str

@router.put('/sites/{site_id}/locale')
def set_site_locale(site_id:str,data:LocaleIn,request:Request):
    """Declare this site's content language (BCP47). Used for <html lang>, defaults zh-CN."""
    enterprise.require(request,site_id,'owner')
    try:locale=locale_value(data.locale)
    except ValueError as exc:raise HTTPException(400,str(exc))
    site=db.one('SELECT settings_json,theme FROM sites WHERE id=?',(site_id,))
    if not site:raise HTTPException(404,'网站不存在')
    settings=json.loads(site['settings_json'] or '{}')
    settings['locale']=locale
    db.execute('UPDATE sites SET settings_json=?,updated_at=? WHERE id=?',(json.dumps(settings,ensure_ascii=False),db.now(),site_id))
    return {'site_id':site_id,'locale':locale}

@router.get('/groups')
def groups(request:Request):
    gs=db.rows('SELECT * FROM language_groups ORDER BY created_at DESC')
    if not request.state.is_superadmin:
        gs=[g for g in gs if any(enterprise.level(request,s['site_id'])>=enterprise.ACCESS['viewer'] for s in db.rows('SELECT site_id FROM language_sites WHERE group_id=?',(g['id'],)))]
    return gs

@router.get('/sites/{site_id}')
def site_language(site_id:str,request:Request):
    enterprise.require(request,site_id,'viewer')
    membership=db.one('''SELECT s.locale,s.group_id,g.name,g.default_locale FROM language_sites s
                          JOIN language_groups g ON g.id=s.group_id WHERE s.site_id=?''',(site_id,))
    if not membership:return {'group':None,'sites':[],'units':[]}
    gid=membership['group_id']
    _authorized(request,gid)
    sites=db.rows('''SELECT s.locale,s.site_id,t.name,t.domain FROM language_sites s
        JOIN sites t ON t.id=s.site_id WHERE s.group_id=? ORDER BY s.locale''',(gid,))
    units=db.rows('''SELECT u.id,u.reference_key,m.site_id,m.entry_id,m.state,e.title,e.route,s.locale
       FROM translation_units u JOIN translation_members m ON m.unit_id=u.id
       JOIN entries e ON e.id=m.entry_id JOIN language_sites s ON s.site_id=m.site_id
       WHERE u.group_id=? ORDER BY u.reference_key,s.locale''',(gid,))
    return {'group':membership,'sites':sites,'units':units}

@router.post('/groups/{group_id}/sites',status_code=201)
def join_site(group_id:str,data:GroupSiteIn,request:Request):
    require_group(group_id)
    _authorized(request,group_id,'owner')
    enterprise.require(request,data.site_id,'owner')
    site=db.one('SELECT * FROM sites WHERE id=?',(data.site_id,))
    if not site:raise HTTPException(404,'网站不存在')
    try:
        locale=locale_value(data.locale)
        domain=_domain(site)
    except ValueError as exc:raise HTTPException(400,str(exc))
    existing=db.rows('''SELECT x.site_id,t.domain FROM language_sites x JOIN sites t ON t.id=x.site_id
                        WHERE x.group_id=?''',(group_id,))
    if any(_domain(s)==domain and s['site_id']!=data.site_id for s in existing):
        raise HTTPException(409,'语言组内两个网站不能使用相同域名')
    if db.one('SELECT site_id FROM language_sites WHERE site_id=?',(data.site_id,)):
        raise HTTPException(409,'该站点已经属于语言组')
    try:db.insert('INSERT INTO language_sites VALUES (?,?,?)',(group_id,data.site_id,locale))
    except Exception as exc:
        if 'UNIQUE' in str(exc).upper():raise HTTPException(409,'组内语言标识已存在')
        raise
    return {'group_id':group_id,'site_id':data.site_id,'locale':locale}

@router.post('/groups/{group_id}/units',status_code=201)
def create_unit(group_id:str,data:UnitIn,request:Request):
    require_group(group_id);_authorized(request,group_id,'editor')
    key=data.reference_key.strip()
    if not KEY_PATTERN.fullmatch(key):raise HTTPException(400,'翻译组标识只能包含字母、数字、短横线、下划线、点和冒号')
    found=db.one('SELECT id FROM translation_units WHERE group_id=? AND reference_key=?',(group_id,key))
    if found:return {'id':found['id'],'reference_key':key,'existing':True}
    id=uid();db.insert('INSERT INTO translation_units VALUES (?,?,?,?)',(id,group_id,key,db.now()))
    return {'id':id,'reference_key':key,'existing':False}

@router.post('/groups/{group_id}/units/{unit_id}/members')
def link_member(group_id:str,unit_id:str,data:MemberIn,request:Request):
    require_group(group_id);_authorized(request,group_id,'editor')
    enterprise.require(request,data.site_id,'editor')
    unit=db.one('SELECT * FROM translation_units WHERE id=? AND group_id=?',(unit_id,group_id))
    if not unit:raise HTTPException(404,'翻译内容组不存在')
    if not db.one('SELECT site_id FROM language_sites WHERE site_id=? AND group_id=?',(data.site_id,group_id)):
        raise HTTPException(400,'网站不属于该语言组')
    entry=db.one('SELECT * FROM entries WHERE id=? AND site_id=?',(data.entry_id,data.site_id))
    if not entry:raise HTTPException(404,'内容不存在或不属于该网站')
    if data.state not in ('pending','approved'):raise HTTPException(400,'无效的翻译状态')
    if data.state=='pending' and entry['status']=='published':
        raise HTTPException(409,'待翻译内容必须先保存为草稿，避免泄露未完成翻译')
    if data.state=='approved' and entry['status']!='published':
        raise HTTPException(409,'只有已发布的正式内容才能审核通过翻译关系')
    prior=db.one('SELECT unit_id FROM translation_members WHERE site_id=? AND entry_id=?',(data.site_id,data.entry_id))
    if prior and prior['unit_id']!=unit_id:raise HTTPException(409,'内容已经属于另一翻译组')
    # Atomic upsert: one locale / site has at most one entry in a semantic group.
    with db.connect() as c:
        c.execute('''INSERT INTO translation_members(unit_id,site_id,entry_id,state) VALUES (?,?,?,?)
            ON CONFLICT(unit_id,site_id) DO UPDATE SET entry_id=excluded.entry_id,state=excluded.state''',
            (unit_id,data.site_id,data.entry_id,data.state))
    return {'unit_id':unit_id,'site_id':data.site_id,'entry_id':data.entry_id,'state':data.state}

@router.delete('/groups/{group_id}/units/{unit_id}/members/{site_id}')
def unlink_member(group_id:str,unit_id:str,site_id:str,request:Request):
    require_group(group_id);_authorized(request,group_id,'editor');enterprise.require(request,site_id,'editor')
    deleted=db.execute('DELETE FROM translation_members WHERE unit_id=? AND site_id=?',(unit_id,site_id))
    if not deleted:raise HTTPException(404,'没有匹配的翻译关联')
    return {'removed':True}

def site_locale(site_id):
    row=db.one('SELECT locale FROM language_sites WHERE site_id=?',(site_id,))
    if row:return row['locale']
    # Per-site locale override (e.g. standalone English sites outside any language group)
    site=db.one('SELECT settings_json,theme FROM sites WHERE id=?',(site_id,))
    try:override=json.loads(site['settings_json'] or '{}').get('locale') if site else None
    except ValueError:override=None
    if override:
        try:return locale_value(override)
        except ValueError:pass
    if site and site['theme']=='trade-pro':return 'en'
    return 'zh-CN'

def assert_build_safe(site_id):
    invalid=db.rows('''SELECT e.route,e.title FROM translation_members m JOIN entries e ON e.id=m.entry_id
       WHERE m.site_id=? AND m.state='pending' AND e.status='published' ''',(site_id,))
    if invalid:
        raise ValueError('有待翻译内容被设为已发布，禁止构建：'+', '.join(x['route'] for x in invalid[:8]))

def alternates(site_id):
    """Get {route: [(locale,absolute_url)]} with approved cross-site peers only.

    Return zero for unlinked/draft/noindex/invalid domain pages, including for
    pending translation records. All members use the same symmetric set.
    """
    row=db.one('SELECT group_id FROM language_sites WHERE site_id=?',(site_id,))
    if not row:return {}
    gid=row['group_id'];group=db.one('SELECT * FROM language_groups WHERE id=?',(gid,))
    members=db.rows('''SELECT m.unit_id,s.locale,e.route,e.status,e.robots,e.canonical,t.domain,t.id AS site_id
       FROM translation_members m JOIN entries e ON e.id=m.entry_id AND e.site_id=m.site_id
       JOIN language_sites s ON s.site_id=m.site_id JOIN sites t ON t.id=m.site_id
       JOIN translation_units u ON u.id=m.unit_id
       WHERE u.group_id=? AND m.state='approved' ''',(gid,))
    by_unit={}
    for item in members:
        if item['status']!='published' or 'noindex' in item['robots'].lower():continue
        try:domain=_domain(item)
        except ValueError:continue
        if item['canonical'] and item['canonical'].rstrip('/') != (domain+item['route']).rstrip('/'):
            continue  # a canonical elsewhere is not an indexable translation target
        by_unit.setdefault(item['unit_id'],[]).append((item['site_id'],item['route'],item['locale'],domain+item['route']))
    result={}
    for linked in by_unit.values():
        if len(linked)<2:continue
        peers=sorted([(locale,url) for _,_,locale,url in linked],key=lambda x:x[0])
        # x-default is explicit only if an approved default-language peer exists.
        default=next((url for locale,url in peers if locale==group['default_locale']),None)
        if default:peers.append(('x-default',default))
        for sid,route,_,_ in linked:
            if sid==site_id:result[route]=peers
    return result

def inject_seo(html,locale,links):
    """Lightweight escape-safe HTML head augmentation without reparsing 5k pages."""
    html=re.sub(r"(<html\b[^>]*?\blang=)[\"'\"][^\"']*[\"'\"]",lambda m:m.group(1)+'"'+html_escape(locale,quote=True)+'"',html,count=1,flags=re.I)
    if not re.search(r'<html\b[^>]*\blang=',html,flags=re.I):
        html=re.sub(r'<html\b','<html lang="'+html_escape(locale,quote=True)+'"',html,count=1,flags=re.I)
    if links and '</head>' in html.lower():
        tags=''.join('<link rel="alternate" hreflang="'+html_escape(language,quote=True)+'" href="'+html_escape(url,quote=True)+'">' for language,url in links)
        html=re.sub(r'</head>',tags+'</head>',html,count=1,flags=re.I)
        # Visible accessible language switch; never fabricate links for missing translations.
        labels={'zh-CN':'简体中文','en':'English','fr-FR':'Français','ar':'العربية',
                'es':'Español','ru':'Русский','pt':'Português'}
        choices=[(language,url) for language,url in links if language!='x-default']
        nav='<nav class="sf-language-switch" aria-label="选择网站语言">'+''.join(
            '<a lang="'+html_escape(lang,quote=True)+'" hreflang="'+html_escape(lang,quote=True)+
            '" href="'+html_escape(url,quote=True)+'"'+(' aria-current="page"' if lang==locale else '')+
            '>'+html_escape(labels.get(lang,lang))+'</a>' for lang,url in choices)+'</nav>'
        html=re.sub(r'</header>',lambda _m:nav+'</header>',html,count=1,flags=re.I)
    return html

def sitemap_alternates(links):
    return ''.join('<xhtml:link rel="alternate" hreflang="'+xml_escape(locale)+'" href="'+xml_escape(url)+'"/>' for locale,url in links)