"""Earth asset management APIs.""" from __future__ import annotations from pathlib import Path from typing import Any from uuid import uuid4 from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from pydantic import BaseModel, Field from sqlalchemy import delete, select, text from sqlalchemy.ext.asyncio import AsyncSession from app.core.security import decode_token, get_current_user, redis_client from app.db.session import get_db from app.models.system_setting import SystemSetting from app.models.user import User from app.services.earth_boundaries import ( EarthBoundaryBuildError, get_boundary_build_status, get_boundary_status, save_boundary_config, start_boundary_build_job, ) router = APIRouter() optional_bearer = HTTPBearer(auto_error=False) REPO_ROOT = Path(__file__).resolve().parents[4] EARTH_BRAND_ASSET_DIR = REPO_ROOT / "data" / "earth-brand" EARTH_BRAND_ASSET_URL_PREFIX = "/earth-brand-assets" EARTH_BRAND_CATEGORY = "earth_brand" MAX_EARTH_BRAND_ASSET_BYTES = 3 * 1024 * 1024 ALLOWED_EARTH_BRAND_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp", ".svg"} DEFAULT_EARTH_BRAND = { "logo_src": "/earth/assets/brand/earth-logo.png", "title_src": "/earth/assets/brand/title-zh.png", "title_text": "智能星球计划", "subtitle": "现实层宇宙全息感知系统", "description": "卫星 · 海底光缆 · 算力基础设施", "aria_label": "智能星球计划品牌标识", "title_alt": "智能星球计划", } class EarthBoundaryConfigPayload(BaseModel): config: dict[str, Any] = Field(default_factory=dict) class EarthBrandPayload(BaseModel): logo_src: str = Field(default=DEFAULT_EARTH_BRAND["logo_src"], max_length=1000) title_src: str = Field(default=DEFAULT_EARTH_BRAND["title_src"], max_length=1000) title_text: str = Field(default=DEFAULT_EARTH_BRAND["title_text"], max_length=120) subtitle: str = Field(default=DEFAULT_EARTH_BRAND["subtitle"], max_length=160) description: str = Field(default=DEFAULT_EARTH_BRAND["description"], max_length=200) aria_label: str = Field(default=DEFAULT_EARTH_BRAND["aria_label"], max_length=200) title_alt: str = Field(default=DEFAULT_EARTH_BRAND["title_alt"], max_length=200) def _normalize_earth_brand_payload(payload: dict[str, Any] | None) -> dict[str, str]: merged = DEFAULT_EARTH_BRAND.copy() if payload: for key in DEFAULT_EARTH_BRAND: value = payload.get(key) if value is not None: merged[key] = str(value).strip() if not merged["title_text"]: merged["title_text"] = DEFAULT_EARTH_BRAND["title_text"] if not merged["aria_label"]: merged["aria_label"] = merged["title_text"] if not merged["title_alt"]: merged["title_alt"] = merged["title_text"] return merged async def _get_earth_brand_record(db: AsyncSession) -> SystemSetting | None: result = await db.execute( select(SystemSetting).where(SystemSetting.category == EARTH_BRAND_CATEGORY) ) return result.scalar_one_or_none() async def _get_earth_brand_payload(db: AsyncSession) -> dict[str, Any]: record = await _get_earth_brand_record(db) return { "brand": _normalize_earth_brand_payload(record.payload if record else None), "is_default": record is None, } @router.get("/brand") async def get_earth_brand(db: AsyncSession = Depends(get_db)): return await _get_earth_brand_payload(db) @router.put("/brand") async def update_earth_brand( payload: EarthBrandPayload, _current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): normalized = _normalize_earth_brand_payload(payload.model_dump()) record = await _get_earth_brand_record(db) if record is None: record = SystemSetting(category=EARTH_BRAND_CATEGORY, payload=normalized) db.add(record) else: record.payload = normalized await db.commit() await db.refresh(record) return {"status": "updated", "brand": _normalize_earth_brand_payload(record.payload), "is_default": False} @router.delete("/brand") @router.post("/brand/reset") async def reset_earth_brand( _current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): await db.execute(delete(SystemSetting).where(SystemSetting.category == EARTH_BRAND_CATEGORY)) await db.commit() return {"status": "reset", "brand": DEFAULT_EARTH_BRAND.copy(), "is_default": True} @router.post("/brand/assets") async def upload_earth_brand_asset( file: UploadFile = File(...), _current_user: User = Depends(get_current_user), ): original_name = file.filename or "" extension = Path(original_name).suffix.lower() if extension not in ALLOWED_EARTH_BRAND_EXTENSIONS: raise HTTPException( status_code=400, detail={ "code": "unsupported_file_type", "message": "Only png, jpg, jpeg, webp, and svg brand assets are supported.", }, ) content = await file.read(MAX_EARTH_BRAND_ASSET_BYTES + 1) if len(content) > MAX_EARTH_BRAND_ASSET_BYTES: raise HTTPException( status_code=400, detail={ "code": "file_too_large", "message": "Brand asset must be 3 MB or smaller.", }, ) EARTH_BRAND_ASSET_DIR.mkdir(parents=True, exist_ok=True) safe_name = f"{uuid4().hex}{extension}" destination = EARTH_BRAND_ASSET_DIR / safe_name destination.write_bytes(content) asset_url = f"{EARTH_BRAND_ASSET_URL_PREFIX}/{safe_name}" return {"url": asset_url, "filename": safe_name, "content_type": file.content_type} @router.get("/boundaries/status") async def get_earth_boundary_status(): return get_boundary_status() async def _get_optional_current_user( credentials: HTTPAuthorizationCredentials | None = Depends(optional_bearer), db: AsyncSession = Depends(get_db), ) -> User | None: if credentials is None: return None token = credentials.credentials if redis_client.sismember("blacklisted_tokens", token): return None payload = decode_token(token) if payload is None or payload.get("type") != "access": return None user_id = payload.get("sub") if user_id is None: return None result = await db.execute( text( "SELECT id, username, email, password_hash, role, is_active, gatekeeper_groups FROM users WHERE id = :id" ), {"id": int(user_id)}, ) row = result.fetchone() if row is None or not row[5]: return None user = User() user.id = row[0] user.username = row[1] user.email = row[2] user.password_hash = row[3] user.role = row[4] user.is_active = row[5] user.gatekeeper_groups = row[6] or [] return user def _is_loopback_request(request: Request) -> bool: host = request.client.host if request.client else "" return host in {"127.0.0.1", "::1", "localhost"} or host.startswith("127.") def _require_local_or_user(request: Request, user: User | None) -> None: if user is not None or _is_loopback_request(request): return raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required outside localhost", ) @router.put("/boundaries/config") async def update_earth_boundary_config( payload: EarthBoundaryConfigPayload, _current_user: User = Depends(get_current_user), ): try: return save_boundary_config(payload.config) except EarthBoundaryBuildError as exc: raise HTTPException( status_code=400, detail={"code": exc.code, "message": str(exc), "details": exc.details}, ) from exc @router.post("/boundaries/build") async def build_earth_boundary_assets( request: Request, current_user: User | None = Depends(_get_optional_current_user), ): _require_local_or_user(request, current_user) try: return await start_boundary_build_job() except EarthBoundaryBuildError as exc: raise HTTPException( status_code=400, detail={"code": exc.code, "message": str(exc), "details": exc.details}, ) from exc @router.get("/boundaries/build/status") async def get_earth_boundary_build_status(): return get_boundary_build_status()