import re
from html import escape
from io import BytesIO
from urllib.parse import quote
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from fastapi.responses import StreamingResponse
from sqlalchemy.orm import Session
from typing import Optional
from database import get_db
from models.animation import Animation
from models.resource import Resource
from models.user import User
from schemas.animation import AnimationGenerate, AnimationCreate, AnimationUpdate, AnimationOut
from services.credits import credits_payload, spend_credits
from services.limiter import limiter
from services.auth import get_current_user, get_optional_current_user
from services.ai_service import AIService
router = APIRouter(prefix="/api/animations", tags=["教学动画"])
ai_service = AIService()
def _safe_filename(name: str, suffix: str) -> str:
stem = re.sub(r"[\\/:*?\"<>|\r\n]+", "_", name).strip(" .") or "教学动画"
return f"{stem[:80]}{suffix}"
def _animation_or_404(anim_id: int, current_user: User | None, db: Session) -> Animation:
anim = db.query(Animation).filter(Animation.id == anim_id).first()
if not anim:
raise HTTPException(status_code=404, detail="动画不存在")
if (not current_user or anim.user_id != current_user.id) and not has_public_resource(db, anim_id):
raise HTTPException(status_code=403, detail="无权访问该动画")
return anim
def _build_animation_html(anim: Animation) -> str:
config = anim.config or {}
body = config.get("html") or config.get("content") or ""
if not isinstance(body, str) or not body.strip():
raise HTTPException(status_code=400, detail="当前动画没有可导出的 HTML")
title = escape(anim.title or config.get("title") or "教学动画")
return f"""
{title}
{body}
"""
def has_public_resource(db: Session, anim_id: int) -> bool:
return db.query(Resource).filter(
Resource.content_ref == f"animation:{anim_id}",
Resource.status == "active",
Resource.is_public == 1,
).first() is not None
@limiter.limit("10/minute")
@router.post("/ai-generate", response_model=dict)
async def ai_generate_animation(
request: Request,
data: AnimationGenerate,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
):
account = spend_credits(db, current_user, "animation_generate", f"生成教学动画：{data.prompt[:80]}")
db.commit()
result = await ai_service.generate_animation(prompt=data.prompt, anim_type=data.anim_type)
return {"success": True, "data": result, "credits": credits_payload(account, "animation_generate")}
@router.post("/", response_model=AnimationOut, status_code=201)
def create_animation(data: AnimationCreate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
anim = Animation(user_id=current_user.id, **data.model_dump())
db.add(anim)
db.commit()
db.refresh(anim)
return anim
@router.get("/", response_model=list[AnimationOut])
def list_animations(
anim_type: Optional[str] = None,
skip: int = Query(0, ge=0),
limit: int = Query(20, ge=1, le=100),
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
):
query = db.query(Animation).filter(Animation.user_id == current_user.id)
if anim_type:
query = query.filter(Animation.anim_type == anim_type)
return query.order_by(Animation.updated_at.desc()).offset(skip).limit(limit).all()
@router.get("/{anim_id}", response_model=AnimationOut)
def get_animation(
anim_id: int,
current_user: User | None = Depends(get_optional_current_user),
db: Session = Depends(get_db),
):
return _animation_or_404(anim_id, current_user, db)
@router.get("/{anim_id}/export-html")
def export_animation_html(
anim_id: int,
current_user: User | None = Depends(get_optional_current_user),
db: Session = Depends(get_db),
):
anim = _animation_or_404(anim_id, current_user, db)
content = _build_animation_html(anim)
filename = _safe_filename(f"{anim.title or '教学动画'}-动画", ".html")
return StreamingResponse(
BytesIO(content.encode("utf-8")),
media_type="text/html; charset=utf-8",
headers={"Content-Disposition": f"attachment; filename*=UTF-8''{quote(filename)}"},
)
@router.put("/{anim_id}", response_model=AnimationOut)
def update_animation(
anim_id: int, data: AnimationUpdate,
current_user: User = Depends(get_current_user), db: Session = Depends(get_db),
):
anim = db.query(Animation).filter(Animation.id == anim_id, Animation.user_id == current_user.id).first()
if not anim:
raise HTTPException(status_code=404, detail="动画不存在")
for key, value in data.model_dump(exclude_unset=True).items():
setattr(anim, key, value)
db.commit()
db.refresh(anim)
return anim
@router.post("/{anim_id}/remix", response_model=AnimationOut, status_code=201)
def remix_animation(
anim_id: int,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
):
source = db.query(Animation).filter(Animation.id == anim_id).first()
if not source:
raise HTTPException(status_code=404, detail="动画不存在")
if source.user_id != current_user.id and not has_public_resource(db, anim_id):
raise HTTPException(status_code=403, detail="无权改编该动画")
clone = Animation(
user_id=current_user.id,
title=f"{source.title}（改编）",
anim_type=source.anim_type or "general",
description=source.description or "",
config=source.config or {},
thumbnail=source.thumbnail or "",
status="draft",
)
db.add(clone)
db.commit()
db.refresh(clone)
return clone
@router.delete("/{anim_id}", status_code=204)
def delete_animation(anim_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
anim = db.query(Animation).filter(Animation.id == anim_id, Animation.user_id == current_user.id).first()
if not anim:
raise HTTPException(status_code=404, detail="动画不存在")
db.query(Resource).filter(
Resource.user_id == current_user.id,
Resource.content_ref == f"animation:{anim_id}",
Resource.status == "active",
).update({"status": "archived"}, synchronize_session=False)
db.delete(anim)
db.commit()