6a2632da70
将 Clover 从上层产品包旧仓库中独立出来,建立专属版本控制。 当前状态=纵切片端到端已打通(登录→选品→出文出图→审核→下载包), M1文案质量去套路化已验收。此提交作为后续按核销清单逐条修复的基线。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
98 lines
3.3 KiB
Python
98 lines
3.3 KiB
Python
"""
|
||
app/api/v1/stream.py — SSE ticket 签发 + SSE 流路由
|
||
安全红线（倩倩姐2026-06-08拍板）：SSE 认证用一次性短票 ticket，禁止 ?token= 直传 JWT。
|
||
|
||
流程：
|
||
1. FE 带 JWT 调 POST /tasks/{id}/sse-ticket → 拿 ticket（60s有效，一次性）
|
||
2. FE 用 ticket 建 EventSource: GET /tasks/{id}/stream?ticket=<ticket>
|
||
3. BE 验 ticket（原子 getdel），用后即失效，换不了其他接口
|
||
"""
|
||
|
||
import logging
|
||
from typing import Annotated
|
||
|
||
import redis as sync_redis
|
||
import redis.asyncio as aioredis
|
||
from fastapi import APIRouter, Depends, Query
|
||
from fastapi.responses import StreamingResponse
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.config import get_settings
|
||
from app.core.database import get_db
|
||
from app.core.response import ok, raise_not_found, raise_unauthorized
|
||
from app.core.sse_ticket import consume_ticket, issue_ticket
|
||
from app.middleware.workspace_guard import CurrentUser, get_current_user
|
||
from app.models.task import GenerationTask
|
||
from app.utils.sse_utils import stream_events
|
||
|
||
logger = logging.getLogger(__name__)
|
||
router = APIRouter()
|
||
settings = get_settings()
|
||
|
||
|
||
# ── 签发接口（用 JWT 换 ticket）────────────────────────────
|
||
@router.post("/tasks/{task_id}/sse-ticket")
|
||
def create_sse_ticket(
|
||
task_id: int,
|
||
current_user: Annotated[CurrentUser, Depends(get_current_user)],
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""
|
||
签发一次性 SSE ticket（60s 有效，仅对该 task 有效）。
|
||
FE 拿到 ticket 后立即建 EventSource，不要存起来复用。
|
||
"""
|
||
task = (
|
||
db.query(GenerationTask)
|
||
.filter(
|
||
GenerationTask.id == task_id,
|
||
GenerationTask.workspace_id == current_user.workspace_id,
|
||
)
|
||
.first()
|
||
)
|
||
if not task:
|
||
raise_not_found("任务不存在")
|
||
|
||
r = sync_redis.from_url(settings.REDIS_URL, decode_responses=True)
|
||
ticket = issue_ticket(r, task_id, current_user.workspace_id)
|
||
return ok({"ticket": ticket, "expires_in": 60})
|
||
|
||
|
||
# ── SSE 流（ticket 认证，绝不传 JWT）─────────────────────
|
||
@router.get("/tasks/{task_id}/stream")
|
||
async def task_stream(
|
||
task_id: int,
|
||
ticket: str = Query(...),
|
||
last_seq: int = Query(default=0),
|
||
):
|
||
"""
|
||
SSE 实时进度推送。
|
||
ticket 验证：原子 getdel，用后即失效，无法重放、无法访问其他接口。
|
||
支持断线重连补发（?last_seq=<上次收到的seq>）。
|
||
断线重连时 FE 须重新调 sse-ticket 换新 ticket。
|
||
"""
|
||
r_sync = sync_redis.from_url(settings.REDIS_URL, decode_responses=True)
|
||
result = consume_ticket(r_sync, ticket)
|
||
if not result:
|
||
raise_unauthorized("ticket 无效或已过期")
|
||
|
||
verified_task_id, workspace_id = result
|
||
if verified_task_id != task_id:
|
||
raise_unauthorized("ticket 与任务不匹配")
|
||
|
||
redis_async = aioredis.from_url(settings.REDIS_URL, decode_responses=True)
|
||
|
||
return StreamingResponse(
|
||
stream_events(
|
||
redis_client=redis_async,
|
||
task_id=task_id,
|
||
workspace_id=workspace_id,
|
||
last_seq=last_seq,
|
||
),
|
||
media_type="text/event-stream",
|
||
headers={
|
||
"Cache-Control": "no-cache",
|
||
"X-Accel-Buffering": "no",
|
||
"Connection": "keep-alive",
|
||
},
|
||
)
|