"""链接卡片 API 与公开落地页。""" from __future__ import annotations import os import uuid from datetime import datetime from typing import Optional from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile from fastapi.responses import HTMLResponse from pydantic import BaseModel, Field from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from auth.dependencies import get_current_user, require_write from link_cards import ( absolute_media_url, build_keywords, favicon_path_for_cover, generate_slug, is_valid_slug, media_path_to_disk, normalize_target_url, process_card_upload, public_base_url, render_link_card_page, ) from models.database import get_db from models.models import LinkCardPage, User router = APIRouter(tags=["link-cards"]) UPLOAD_DIR = os.path.join(os.path.dirname(__file__), "uploads", "link-cards") os.makedirs(UPLOAD_DIR, exist_ok=True) class LinkCardUpsert(BaseModel): id: Optional[int] = None title: str = Field(..., min_length=1, max_length=200) content: str = Field(default="", max_length=2000) target_url: str = Field(..., min_length=1, max_length=2000) image_path: str = Field(..., min_length=1, max_length=512) class LinkCardResponse(BaseModel): id: int slug: str title: str content: str target_url: str image_path: str page_url: str cover_url: str warning: Optional[str] = None class Config: from_attributes = True class UploadImageResponse(BaseModel): image_path: str cover_url: str async def _get_owned_card( db: AsyncSession, user: User, card_id: int, *, write: bool = False ) -> LinkCardPage: stmt = select(LinkCardPage).where(LinkCardPage.id == card_id) card = (await db.execute(stmt)).scalar_one_or_none() if not card or card.owner_id != user.id: raise HTTPException(status_code=404, detail="卡片不存在") if write and user.role == "viewer": raise HTTPException(status_code=403, detail="无写入权限") return card def _card_public_url_warning(page_url: str, cover_url: str) -> Optional[str]: from rpa_engine.douyin_im.reply_payload import is_internal_http_url if is_internal_http_url(page_url) or is_internal_http_url(cover_url): return ( "落地页或封面使用了 localhost/内网地址，抖音客户端无法展示链接卡片。" "请设置 KEFU_PUBLIC_BASE_URL 为公网 HTTPS 域名后重新保存规则。" ) return None def _card_response(card: LinkCardPage, request: Request) -> LinkCardResponse: base = public_base_url(request) page_url = f"{base}/p/{card.slug}" if base else f"/p/{card.slug}" cover_url = absolute_media_url(card.image_path, request) return LinkCardResponse( id=card.id, slug=card.slug, title=card.title, content=card.content or "", target_url=card.target_url, image_path=card.image_path, page_url=page_url, cover_url=cover_url, warning=_card_public_url_warning(page_url, cover_url), ) @router.get("/api/link-cards", response_model=list[LinkCardResponse]) async def list_link_cards( request: Request, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), ): stmt = ( select(LinkCardPage) .where(LinkCardPage.owner_id == user.id) .order_by(LinkCardPage.updated_at.desc()) ) cards = (await db.execute(stmt)).scalars().all() return [_card_response(card, request) for card in cards] @router.post("/api/link-cards/upload-image", response_model=UploadImageResponse) async def upload_link_card_image( request: Request, file: UploadFile = File(...), user: User = Depends(require_write), ): if not file.content_type or not file.content_type.startswith("image/"): raise HTTPException(status_code=400, detail="仅支持上传图片文件") if file.content_type == "image/svg+xml": raise HTTPException(status_code=400, detail="请上传 JPG/PNG 等位图，系统将自动转为 favicon PNG") raw = await file.read() if not raw: raise HTTPException(status_code=400, detail="图片为空") if len(raw) > 4 * 1024 * 1024: raise HTTPException(status_code=400, detail="图片大小不能超过 4MB") try: cover_bytes, favicon_bytes = process_card_upload(raw) except RuntimeError as exc: raise HTTPException(status_code=500, detail=str(exc)) from exc except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc owner_dir = os.path.join(UPLOAD_DIR, str(user.id)) os.makedirs(owner_dir, exist_ok=True) file_id = uuid.uuid4().hex cover_filename = f"{file_id}.png" favicon_filename = f"{file_id}.favicon.png" cover_path = os.path.join(owner_dir, cover_filename) favicon_path = os.path.join(owner_dir, favicon_filename) with open(cover_path, "wb") as f: f.write(cover_bytes) with open(favicon_path, "wb") as f: f.write(favicon_bytes) image_path = f"/api/media/link-cards/{user.id}/{cover_filename}" return UploadImageResponse( image_path=image_path, cover_url=absolute_media_url(image_path, request), ) @router.post("/api/link-cards", response_model=LinkCardResponse) async def upsert_link_card( body: LinkCardUpsert, request: Request, db: AsyncSession = Depends(get_db), user: User = Depends(require_write), ): title = body.title.strip() content = (body.content or "").strip() target_url = normalize_target_url(body.target_url) image_path = (body.image_path or "").strip() if not title: raise HTTPException(status_code=400, detail="请填写卡片标题") if not target_url: raise HTTPException(status_code=400, detail="请填写跳转链接") if not image_path.startswith("/api/media/link-cards/"): raise HTTPException(status_code=400, detail="请先上传卡片封面图") if body.id: card = await _get_owned_card(db, user, body.id, write=True) card.title = title card.content = content card.target_url = target_url card.image_path = image_path card.updated_at = datetime.utcnow() else: slug = generate_slug() for _ in range(5): exists = ( await db.execute(select(LinkCardPage.id).where(LinkCardPage.slug == slug)) ).scalar_one_or_none() if not exists: break slug = generate_slug() else: raise HTTPException(status_code=500, detail="生成页面标识失败，请重试") card = LinkCardPage( owner_id=user.id, slug=slug, title=title, content=content, target_url=target_url, image_path=image_path, ) db.add(card) await db.commit() await db.refresh(card) return _card_response(card, request) @router.get("/p/{slug}", response_class=HTMLResponse, include_in_schema=False) async def serve_link_card_page(slug: str, request: Request, db: AsyncSession = Depends(get_db)): if not is_valid_slug(slug): raise HTTPException(status_code=404, detail="页面不存在") stmt = select(LinkCardPage).where(LinkCardPage.slug == slug) card = (await db.execute(stmt)).scalar_one_or_none() if not card: raise HTTPException(status_code=404, detail="页面不存在") cover_url = absolute_media_url(card.image_path, request) favicon_path = favicon_path_for_cover(card.image_path) favicon_disk = media_path_to_disk(UPLOAD_DIR, favicon_path) favicon_url = absolute_media_url(favicon_path, request) if favicon_disk and os.path.isfile(favicon_disk) else cover_url html = render_link_card_page( title=card.title, content=card.content or "", keywords=build_keywords(card.title, card.content or ""), cover_url=cover_url, favicon_url=favicon_url, target_url=card.target_url, ) return HTMLResponse(content=html, media_type="text/html; charset=utf-8")