211 lines
7.0 KiB
Python
211 lines
7.0 KiB
Python
"""链接卡片 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
|
|
|
|
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_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,
|
|
)
|
|
|
|
|
|
@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")
|