531 lines
29 KiB
Python
531 lines
29 KiB
Python
import asyncio
|
|
import hashlib
|
|
import json
|
|
import time
|
|
import unicodedata
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
import psycopg
|
|
from fastapi import Depends, FastAPI, HTTPException, Request, Response, status
|
|
import httpx
|
|
from fastapi.responses import FileResponse, JSONResponse
|
|
from psycopg import Connection
|
|
from psycopg.types.json import Json
|
|
|
|
from .config import settings
|
|
from .db import get_connection
|
|
from .schemas import (
|
|
AuthResponse,
|
|
LoginRequest,
|
|
MembershipRequest,
|
|
PasswordChangeRequest,
|
|
QuotaAdjustmentRequest,
|
|
RegisterRequest,
|
|
StatusRequest,
|
|
UserPublic,
|
|
VerificationConfirmRequest,
|
|
VerificationSendRequest,
|
|
TtsTaskRequest,
|
|
TtsTaskPublic,
|
|
TtsVoicePublic,
|
|
normalize_email,
|
|
)
|
|
from .security import (
|
|
check_origin,
|
|
clear_session_cookie,
|
|
code_digest,
|
|
current_user,
|
|
hash_password,
|
|
new_csrf_token,
|
|
new_token,
|
|
password_hasher,
|
|
rate_limiter,
|
|
require_admin,
|
|
require_csrf,
|
|
set_csrf_cookie,
|
|
set_session_cookie,
|
|
token_digest,
|
|
utc_now,
|
|
verify_password,
|
|
)
|
|
|
|
app = FastAPI(title="Kaotings Business API", version="0.2.0", docs_url=None if settings.app_env == "production" else "/docs")
|
|
|
|
|
|
def error(code: str, message: str, http_status: int) -> HTTPException:
|
|
return HTTPException(status_code=http_status, detail={"code": code, "message": message})
|
|
|
|
|
|
def period_bounds(now: datetime | None = None) -> tuple[datetime, datetime]:
|
|
current = now or utc_now()
|
|
start = current.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
|
|
if start.month == 12:
|
|
end = start.replace(year=start.year + 1, month=1)
|
|
else:
|
|
end = start.replace(month=start.month + 1)
|
|
return start, end
|
|
|
|
|
|
def effective_plan(connection: Connection, user: dict) -> str:
|
|
if user["role"] == "admin":
|
|
return user["plan"]
|
|
grant = connection.execute(
|
|
"""
|
|
SELECT plan FROM membership_grants
|
|
WHERE user_id = %s AND revoked_at IS NULL AND starts_at <= now()
|
|
AND (expires_at IS NULL OR expires_at > now())
|
|
ORDER BY starts_at DESC LIMIT 1
|
|
""",
|
|
(user["id"],),
|
|
).fetchone()
|
|
return grant["plan"] if grant else "free"
|
|
|
|
|
|
def user_public(connection: Connection, user: dict) -> UserPublic:
|
|
return UserPublic(
|
|
id=user["id"], email=user["email"], phone=user["phone"], role=user["role"], plan=effective_plan(connection, user),
|
|
status=user["status"], email_verified=user["email_verified"], phone_verified=user["phone_verified"],
|
|
created_at=user["created_at"], last_login_at=user["last_login_at"],
|
|
)
|
|
|
|
|
|
def ensure_quota(connection: Connection, user_id: UUID, plan: str) -> dict:
|
|
start, end = period_bounds()
|
|
policy = connection.execute("SELECT period_limit FROM plan_policies WHERE code = %s", (plan,)).fetchone()
|
|
if not policy:
|
|
raise error("POLICY_MISSING", "权益策略未配置", 500)
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO quota_accounts(user_id, period_start, period_end, limit_snapshot)
|
|
VALUES (%s, %s, %s, %s)
|
|
ON CONFLICT (user_id, period_start, period_end) DO NOTHING
|
|
""",
|
|
(user_id, start, end, policy["period_limit"]),
|
|
)
|
|
row = connection.execute(
|
|
"SELECT * FROM quota_accounts WHERE user_id = %s AND period_start = %s AND period_end = %s",
|
|
(user_id, start, end),
|
|
).fetchone()
|
|
return row
|
|
|
|
|
|
def usage_view(quota: dict) -> dict[str, Any]:
|
|
available = max(0, quota["limit_snapshot"] + quota["adjustment"] - quota["used"] - quota["reserved"])
|
|
return {
|
|
"period_start": quota["period_start"], "period_end": quota["period_end"], "limit": quota["limit_snapshot"],
|
|
"adjustment": quota["adjustment"], "used": quota["used"], "reserved": quota["reserved"], "available": available,
|
|
}
|
|
|
|
|
|
def normalize_tts_text(value: str) -> str:
|
|
return unicodedata.normalize("NFC", value.replace("\r\n", "\n").replace("\r", "\n"))
|
|
|
|
|
|
def task_view(connection: Connection, task: dict) -> dict[str, Any]:
|
|
audio = connection.execute("SELECT expires_at, status FROM audio_files WHERE task_id = %s", (task["id"],)).fetchone()
|
|
return {
|
|
"id": task["id"], "status": task["status"], "text_length": task["text_length"],
|
|
"voice_id": task["provider_voice_id"], "parameters": task["parameters"],
|
|
"error_code": task["error_code"], "audio_available": bool(audio and audio["status"] == "available"),
|
|
"audio_expires_at": audio["expires_at"] if audio else None, "created_at": task["created_at"],
|
|
"started_at": task["started_at"], "finished_at": task["finished_at"],
|
|
}
|
|
|
|
|
|
def claim_next_task() -> dict | None:
|
|
with psycopg.connect(settings.database_url, row_factory=psycopg.rows.dict_row) as connection:
|
|
task = connection.execute("SELECT * FROM tts_tasks WHERE status = 'queued' ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1").fetchone()
|
|
if not task:
|
|
return None
|
|
lease_token = new_token()
|
|
updated = connection.execute("UPDATE tts_tasks SET status = 'running', lease_token = %s, lease_expires_at = now() + interval '5 minutes', started_at = now(), updated_at = now(), attempt_count = attempt_count + 1 WHERE id = %s RETURNING *", (lease_token, task["id"])).fetchone()
|
|
connection.commit()
|
|
return updated
|
|
|
|
|
|
def finish_task_failure(task: dict, code: str) -> None:
|
|
with psycopg.connect(settings.database_url, row_factory=psycopg.rows.dict_row) as connection:
|
|
updated = connection.execute("UPDATE tts_tasks SET status = 'failed', error_code = %s, finished_at = now(), updated_at = now(), lease_token = NULL, lease_expires_at = NULL WHERE id = %s AND status = 'running' AND lease_token = %s", (code, task["id"], task["lease_token"]))
|
|
if updated.rowcount != 1:
|
|
connection.rollback()
|
|
return
|
|
connection.execute("UPDATE quota_accounts SET reserved = GREATEST(0, reserved - %s), version = version + 1 WHERE id = %s", (task["reserved_amount"], task["quota_account_id"]))
|
|
connection.execute("INSERT INTO usage_records(user_id, quota_account_id, task_id, type, amount) VALUES (%s, %s, %s, 'release', %s)", (task["user_id"], task["quota_account_id"], task["id"], task["reserved_amount"]))
|
|
connection.commit()
|
|
|
|
|
|
def finish_task_success(task: dict, audio: bytes, mime_type: str) -> None:
|
|
storage_dir = Path(settings.audio_storage_dir)
|
|
storage_dir.mkdir(parents=True, exist_ok=True)
|
|
extension = "mp3" if mime_type == "audio/mpeg" else "wav"
|
|
storage_key = f"{task['user_id']}/{task['id']}.{extension}"
|
|
target = storage_dir / storage_key
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(audio)
|
|
checksum = hashlib.sha256(audio).hexdigest()
|
|
with psycopg.connect(settings.database_url, row_factory=psycopg.rows.dict_row) as connection:
|
|
current = connection.execute("SELECT status, lease_token FROM tts_tasks WHERE id = %s FOR UPDATE", (task["id"],)).fetchone()
|
|
if not current or current["status"] != "running" or current["lease_token"] != task["lease_token"]:
|
|
target.unlink(missing_ok=True)
|
|
connection.rollback()
|
|
return
|
|
connection.execute("INSERT INTO audio_files(task_id, owner_id, storage_key, mime_type, size_bytes, checksum) VALUES (%s, %s, %s, %s, %s, %s)", (task["id"], task["user_id"], storage_key, mime_type, len(audio), checksum))
|
|
connection.execute("UPDATE tts_tasks SET status = 'succeeded', finished_at = now(), updated_at = now(), lease_token = NULL, lease_expires_at = NULL WHERE id = %s AND status = 'running' AND lease_token = %s", (task["id"], task["lease_token"]))
|
|
connection.execute("UPDATE quota_accounts SET reserved = GREATEST(0, reserved - %s), used = used + %s, version = version + 1 WHERE id = %s", (task["reserved_amount"], task["reserved_amount"], task["quota_account_id"]))
|
|
connection.execute("INSERT INTO usage_records(user_id, quota_account_id, task_id, type, amount) VALUES (%s, %s, %s, 'consume', %s)", (task["user_id"], task["quota_account_id"], task["id"], task["reserved_amount"]))
|
|
connection.commit()
|
|
|
|
|
|
def recover_expired_tasks() -> None:
|
|
with psycopg.connect(settings.database_url, row_factory=psycopg.rows.dict_row) as connection:
|
|
tasks = connection.execute("SELECT * FROM tts_tasks WHERE status = 'running' AND lease_expires_at IS NOT NULL AND lease_expires_at < now() FOR UPDATE SKIP LOCKED").fetchall()
|
|
for task in tasks:
|
|
connection.execute("UPDATE tts_tasks SET status = 'failed', error_code = 'WORKER_LEASE_EXPIRED', finished_at = now(), updated_at = now(), lease_token = NULL, lease_expires_at = NULL WHERE id = %s AND status = 'running'", (task["id"],))
|
|
connection.execute("UPDATE quota_accounts SET reserved = GREATEST(0, reserved - %s), version = version + 1 WHERE id = %s", (task["reserved_amount"], task["quota_account_id"]))
|
|
connection.execute("INSERT INTO usage_records(user_id, quota_account_id, task_id, type, amount) VALUES (%s, %s, %s, 'release', %s)", (task["user_id"], task["quota_account_id"], task["id"], task["reserved_amount"]))
|
|
connection.commit()
|
|
|
|
|
|
def cleanup_orphan_audio() -> None:
|
|
storage_dir = Path(settings.audio_storage_dir)
|
|
if not storage_dir.exists():
|
|
return
|
|
with psycopg.connect(settings.database_url, row_factory=psycopg.rows.dict_row) as connection:
|
|
known = {row["storage_key"] for row in connection.execute("SELECT storage_key FROM audio_files WHERE status = 'available'").fetchall()}
|
|
now = time.time()
|
|
for path in storage_dir.rglob("*"):
|
|
if path.is_file() and path.relative_to(storage_dir).as_posix() not in known and now - path.stat().st_mtime > 3600:
|
|
path.unlink(missing_ok=True)
|
|
|
|
|
|
async def process_task(task: dict) -> None:
|
|
if not settings.tts_upstream_url:
|
|
await asyncio.to_thread(finish_task_failure, task, "UPSTREAM_NOT_CONFIGURED")
|
|
return
|
|
parameters = task["parameters"]
|
|
response_format = parameters.get("format", "wav")
|
|
payload = {"model": parameters.get("model", "qwen3-tts"), "input": task["text"], "voice": task["provider_voice_id"], "response_format": response_format, "speed": parameters.get("speed", 1.0)}
|
|
headers = {"Content-Type": "application/json"}
|
|
if settings.tts_api_key:
|
|
headers["Authorization"] = f"Bearer {settings.tts_api_key}"
|
|
try:
|
|
async with httpx.AsyncClient(timeout=settings.tts_timeout_seconds) as client:
|
|
response = await client.post(f"{settings.tts_upstream_url}/v1/audio/speech", headers=headers, json=payload)
|
|
if response.status_code == 401:
|
|
raise RuntimeError("UPSTREAM_UNAUTHORIZED")
|
|
if response.status_code >= 400:
|
|
raise RuntimeError(f"UPSTREAM_HTTP_{response.status_code}")
|
|
content_type = response.headers.get("content-type", "").split(";", 1)[0].lower()
|
|
if content_type not in {"audio/wav", "audio/x-wav"}:
|
|
raise RuntimeError("UPSTREAM_NOT_AUDIO")
|
|
declared_size = response.headers.get("content-length")
|
|
if declared_size and int(declared_size) > 50 * 1024 * 1024:
|
|
raise RuntimeError("UPSTREAM_RESPONSE_TOO_LARGE")
|
|
audio = response.content
|
|
if not audio or len(audio) > 50 * 1024 * 1024:
|
|
raise RuntimeError("UPSTREAM_AUDIO_INVALID")
|
|
if response_format == "wav" and (len(audio) < 12 or audio[:4] != b"RIFF" or audio[8:12] != b"WAVE"):
|
|
raise RuntimeError("UPSTREAM_AUDIO_CORRUPT")
|
|
await asyncio.to_thread(finish_task_success, task, audio, "audio/wav")
|
|
except httpx.TimeoutException:
|
|
await asyncio.to_thread(finish_task_failure, task, "UPSTREAM_TIMEOUT")
|
|
except Exception as exc:
|
|
await asyncio.to_thread(finish_task_failure, task, str(exc)[:80])
|
|
|
|
|
|
async def task_worker() -> None:
|
|
while True:
|
|
await asyncio.to_thread(recover_expired_tasks)
|
|
task = await asyncio.to_thread(claim_next_task)
|
|
if task:
|
|
await process_task(task)
|
|
else:
|
|
await asyncio.sleep(1)
|
|
|
|
|
|
worker_task: asyncio.Task | None = None
|
|
|
|
|
|
@app.on_event("startup")
|
|
async def start_worker():
|
|
global worker_task
|
|
await asyncio.to_thread(cleanup_orphan_audio)
|
|
worker_task = asyncio.create_task(task_worker())
|
|
|
|
|
|
@app.on_event("shutdown")
|
|
async def stop_worker():
|
|
if worker_task:
|
|
worker_task.cancel()
|
|
|
|
|
|
@app.exception_handler(HTTPException)
|
|
async def http_error_handler(_: Request, exc: HTTPException):
|
|
detail = exc.detail if isinstance(exc.detail, dict) else {"code": "REQUEST_FAILED", "message": str(exc.detail)}
|
|
return JSONResponse(status_code=exc.status_code, content={"error": detail})
|
|
|
|
|
|
@app.get("/healthz")
|
|
def health(connection: Connection = Depends(get_connection)):
|
|
connection.execute("SELECT 1")
|
|
return {"status": "ok", "database": "ok", "service": "api"}
|
|
|
|
|
|
@app.get("/api/v1/auth/csrf")
|
|
def csrf(response: Response, request: Request):
|
|
check_origin(request)
|
|
token = request.cookies.get("kaotings_csrf") or new_csrf_token()
|
|
set_csrf_cookie(response, token)
|
|
return {"csrf_token": token}
|
|
|
|
|
|
@app.post("/api/v1/auth/register", response_model=AuthResponse, status_code=201)
|
|
def register(payload: RegisterRequest, request: Request, connection: Connection = Depends(get_connection)):
|
|
check_origin(request)
|
|
key = f"register:{request.client.host if request.client else 'unknown'}"
|
|
if not rate_limiter.allow(key, 5, 3600):
|
|
raise error("RATE_LIMITED", "请求过于频繁", 429)
|
|
email = normalize_email(str(payload.email))
|
|
try:
|
|
user = connection.execute(
|
|
"""
|
|
INSERT INTO users(email, password_hash) VALUES (%s, %s)
|
|
RETURNING *
|
|
""",
|
|
(email, hash_password(payload.password)),
|
|
).fetchone()
|
|
ensure_quota(connection, user["id"], "free")
|
|
connection.commit()
|
|
except psycopg.errors.UniqueViolation:
|
|
connection.rollback()
|
|
raise error("REGISTRATION_FAILED", "注册信息不可用", 409)
|
|
return {"user": user_public(connection, user)}
|
|
|
|
|
|
@app.post("/api/v1/auth/login", response_model=AuthResponse)
|
|
def login(payload: LoginRequest, request: Request, response: Response, connection: Connection = Depends(get_connection)):
|
|
check_origin(request)
|
|
key = f"login:{request.client.host if request.client else 'unknown'}:{normalize_email(str(payload.email))}"
|
|
if not rate_limiter.allow(key, 10, 300):
|
|
raise error("RATE_LIMITED", "请求过于频繁", 429)
|
|
user = connection.execute("SELECT * FROM users WHERE email = %s", (normalize_email(str(payload.email)),)).fetchone()
|
|
if not user or not verify_password(payload.password, user["password_hash"]):
|
|
raise error("LOGIN_FAILED", "邮箱或密码错误", 401)
|
|
if user["status"] != "active":
|
|
raise error("ACCOUNT_DISABLED", "账户不可用", 403)
|
|
token = new_token()
|
|
connection.execute("INSERT INTO sessions(user_id, token_hash, expires_at) VALUES (%s, %s, %s)", (user["id"], token_digest(token), utc_now() + timedelta(seconds=settings.session_ttl_seconds)))
|
|
user = connection.execute("UPDATE users SET last_login_at = now(), updated_at = now() WHERE id = %s RETURNING *", (user["id"],)).fetchone()
|
|
ensure_quota(connection, user["id"], effective_plan(connection, user))
|
|
connection.commit()
|
|
set_session_cookie(response, token)
|
|
return {"user": user_public(connection, user)}
|
|
|
|
|
|
@app.post("/api/v1/auth/logout", status_code=204, dependencies=[Depends(require_csrf)])
|
|
def logout(response: Response, request: Request, connection: Connection = Depends(get_connection)):
|
|
token = request.cookies.get(settings.session_cookie_name)
|
|
if token:
|
|
connection.execute("UPDATE sessions SET revoked_at = now() WHERE token_hash = %s", (token_digest(token),))
|
|
connection.commit()
|
|
clear_session_cookie(response)
|
|
return Response(status_code=204)
|
|
|
|
|
|
@app.get("/api/v1/auth/me", response_model=UserPublic)
|
|
def me(user: dict = Depends(current_user), connection: Connection = Depends(get_connection)):
|
|
return user_public(connection, user)
|
|
|
|
|
|
@app.post("/api/v1/auth/password/change", dependencies=[Depends(require_csrf)])
|
|
def change_password(payload: PasswordChangeRequest, request: Request, connection: Connection = Depends(get_connection), user: dict = Depends(current_user)):
|
|
if not verify_password(payload.current_password, user["password_hash"]):
|
|
raise error("PASSWORD_INVALID", "当前密码错误", 400)
|
|
connection.execute("UPDATE users SET password_hash = %s, updated_at = now() WHERE id = %s", (hash_password(payload.new_password), user["id"]))
|
|
connection.execute("UPDATE sessions SET revoked_at = now() WHERE user_id = %s AND id <> %s", (user["id"], user["session_id"]))
|
|
connection.commit()
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.get("/api/v1/account/usage")
|
|
def account_usage(user: dict = Depends(current_user), connection: Connection = Depends(get_connection)):
|
|
quota = ensure_quota(connection, user["id"], effective_plan(connection, user))
|
|
connection.commit()
|
|
return {"plan": effective_plan(connection, user), **usage_view(quota)}
|
|
|
|
|
|
@app.post("/api/v1/auth/verification/send")
|
|
def verification_send(payload: VerificationSendRequest, request: Request, user: dict = Depends(current_user)):
|
|
check_origin(request)
|
|
enabled = settings.email_verification_enabled if payload.channel == "email" else settings.phone_verification_enabled
|
|
if not enabled:
|
|
raise error("VERIFICATION_NOT_ENABLED", "验证渠道尚未启用", 503)
|
|
raise error("VERIFICATION_PROVIDER_UNAVAILABLE", "验证渠道尚未配置", 503)
|
|
|
|
|
|
@app.post("/api/v1/auth/verification/confirm", dependencies=[Depends(require_csrf)])
|
|
def verification_confirm(payload: VerificationConfirmRequest, request: Request, connection: Connection = Depends(get_connection), user: dict = Depends(current_user)):
|
|
check_origin(request)
|
|
challenge = connection.execute("SELECT * FROM verification_challenges WHERE id = %s AND user_id = %s", (payload.challenge_id, user["id"])).fetchone()
|
|
if not challenge or challenge["consumed_at"] or challenge["expires_at"] <= utc_now() or challenge["attempt_count"] >= 5:
|
|
raise error("VERIFICATION_INVALID", "验证挑战无效或已过期", 400)
|
|
if not __import__("hmac").compare_digest(challenge["code_digest"], code_digest(payload.code)):
|
|
connection.execute("UPDATE verification_challenges SET attempt_count = attempt_count + 1 WHERE id = %s", (payload.challenge_id,))
|
|
connection.commit()
|
|
raise error("VERIFICATION_INVALID", "验证码错误", 400)
|
|
connection.execute("UPDATE verification_challenges SET consumed_at = now() WHERE id = %s", (payload.challenge_id,))
|
|
if challenge["channel"] == "email":
|
|
connection.execute("UPDATE users SET email_verified = true, updated_at = now() WHERE id = %s", (user["id"],))
|
|
else:
|
|
connection.execute("UPDATE users SET phone_verified = true, updated_at = now() WHERE id = %s", (user["id"],))
|
|
connection.commit()
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.get("/api/v1/tts/voices", response_model=list[TtsVoicePublic])
|
|
def tts_voices(connection: Connection = Depends(get_connection)):
|
|
return connection.execute("SELECT id, provider_voice_id, name, language, description, supported_parameters FROM tts_voices WHERE enabled = true ORDER BY name").fetchall()
|
|
|
|
|
|
@app.post("/api/v1/tts/tasks", status_code=202)
|
|
def create_tts_task(payload: TtsTaskRequest, request: Request, connection: Connection = Depends(get_connection), user: dict = Depends(current_user), _: None = Depends(require_csrf)):
|
|
idempotency_key = request.headers.get("idempotency-key", "").strip()
|
|
if len(idempotency_key) < 8 or len(idempotency_key) > 120:
|
|
raise error("IDEMPOTENCY_REQUIRED", "需要有效的 Idempotency-Key", 400)
|
|
plan = effective_plan(connection, user)
|
|
policy = connection.execute("SELECT * FROM plan_policies WHERE code = %s", (plan,)).fetchone()
|
|
text = normalize_tts_text(payload.text)
|
|
if not text.strip():
|
|
raise error("EMPTY_TEXT", "文本不能只有空白", 422)
|
|
text_length = len(text)
|
|
if not policy or text_length > policy["max_text_length"]:
|
|
raise error("TEXT_TOO_LONG", "文本超过当前计划限制", 422)
|
|
parameters = dict(payload.parameters)
|
|
response_format = str(parameters.get("format", "wav")).lower()
|
|
speed = float(parameters.get("speed", 1.0))
|
|
if response_format != "wav" or speed != 1:
|
|
raise error("UNVERIFIED_PARAMETERS", "V1 当前只支持 WAV 和 1.0x 语速", 422)
|
|
voice = connection.execute("SELECT * FROM tts_voices WHERE provider_voice_id = %s AND enabled = true", (payload.voice_id,)).fetchone()
|
|
if not voice or plan not in (voice["allowed_plans"] or ["free", "vip"]):
|
|
raise error("VOICE_NOT_ALLOWED", "音色不可用", 422)
|
|
request_hash = hashlib.sha256(json.dumps({"text": text, "voice": payload.voice_id, "parameters": parameters}, ensure_ascii=False, sort_keys=True).encode()).hexdigest()
|
|
connection.execute("SELECT pg_advisory_xact_lock(hashtext(%s))", (str(user["id"]),))
|
|
existing = connection.execute("SELECT * FROM tts_tasks WHERE user_id = %s AND idempotency_key = %s", (user["id"], idempotency_key)).fetchone()
|
|
if existing:
|
|
if existing["request_hash"] != request_hash:
|
|
raise error("IDEMPOTENCY_CONFLICT", "幂等键已用于其他请求", 409)
|
|
return task_view(connection, existing)
|
|
active_count = connection.execute("SELECT count(*) FROM tts_tasks WHERE user_id = %s AND status IN ('queued', 'running')", (user["id"],)).fetchone()["count"]
|
|
if active_count >= policy["max_concurrency"]:
|
|
raise error("CONCURRENCY_LIMIT", "并发任务数已达上限", 409)
|
|
quota = ensure_quota(connection, user["id"], plan)
|
|
quota = connection.execute("SELECT * FROM quota_accounts WHERE id = %s FOR UPDATE", (quota["id"],)).fetchone()
|
|
available = quota["limit_snapshot"] + quota["adjustment"] - quota["used"] - quota["reserved"]
|
|
if available < text_length:
|
|
raise error("QUOTA_EXCEEDED", "额度不足", 409)
|
|
task = connection.execute(
|
|
"""
|
|
INSERT INTO tts_tasks(user_id, text, text_length, voice_id, provider_voice_id, parameters, idempotency_key, request_hash, policy_version, quota_account_id, reserved_amount)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING *
|
|
""",
|
|
(user["id"], text, text_length, voice["id"], payload.voice_id, Json(parameters), idempotency_key, request_hash, policy["version"], quota["id"], text_length),
|
|
).fetchone()
|
|
connection.execute("UPDATE quota_accounts SET reserved = reserved + %s, version = version + 1 WHERE id = %s", (text_length, quota["id"]))
|
|
connection.execute("INSERT INTO usage_records(user_id, quota_account_id, task_id, type, amount, idempotency_key) VALUES (%s, %s, %s, 'reserve', %s, %s)", (user["id"], quota["id"], task["id"], text_length, f"reserve:{task['id']}"))
|
|
connection.commit()
|
|
return task_view(connection, task)
|
|
|
|
|
|
@app.get("/api/v1/tts/tasks", response_model=list[TtsTaskPublic])
|
|
def list_tts_tasks(user: dict = Depends(current_user), connection: Connection = Depends(get_connection)):
|
|
tasks = connection.execute("SELECT * FROM tts_tasks WHERE user_id = %s ORDER BY created_at DESC LIMIT 100", (user["id"],)).fetchall()
|
|
return [task_view(connection, task) for task in tasks]
|
|
|
|
|
|
def owned_task(task_id: UUID, user: dict, connection: Connection) -> dict:
|
|
task = connection.execute("SELECT * FROM tts_tasks WHERE id = %s", (task_id,)).fetchone()
|
|
if not task or (task["user_id"] != user["id"] and user["role"] != "admin"):
|
|
raise error("NOT_FOUND", "任务不存在", 404)
|
|
return task
|
|
|
|
|
|
@app.get("/api/v1/tts/tasks/{task_id}", response_model=TtsTaskPublic)
|
|
def get_tts_task(task_id: UUID, user: dict = Depends(current_user), connection: Connection = Depends(get_connection)):
|
|
return task_view(connection, owned_task(task_id, user, connection))
|
|
|
|
|
|
def audio_response(task_id: UUID, user: dict, connection: Connection, download: bool):
|
|
task = owned_task(task_id, user, connection)
|
|
audio = connection.execute("SELECT * FROM audio_files WHERE task_id = %s AND status = 'available'", (task_id,)).fetchone()
|
|
if task["status"] != "succeeded" or not audio:
|
|
raise error("AUDIO_NOT_AVAILABLE", "音频尚不可用", 404)
|
|
path = Path(settings.audio_storage_dir) / audio["storage_key"]
|
|
if not path.is_file():
|
|
raise error("AUDIO_NOT_AVAILABLE", "音频文件不可用", 404)
|
|
filename = f"kaotings-{task_id}.{'mp3' if audio['mime_type'] == 'audio/mpeg' else 'wav'}" if download else None
|
|
return FileResponse(path, media_type=audio["mime_type"], filename=filename)
|
|
|
|
|
|
@app.get("/api/v1/tts/tasks/{task_id}/audio")
|
|
def play_tts_audio(task_id: UUID, user: dict = Depends(current_user), connection: Connection = Depends(get_connection)):
|
|
return audio_response(task_id, user, connection, False)
|
|
|
|
|
|
@app.get("/api/v1/tts/tasks/{task_id}/download")
|
|
def download_tts_audio(task_id: UUID, user: dict = Depends(current_user), connection: Connection = Depends(get_connection)):
|
|
return audio_response(task_id, user, connection, True)
|
|
|
|
|
|
@app.get("/api/v1/admin/users")
|
|
def admin_users(user: dict = Depends(require_admin), connection: Connection = Depends(get_connection)):
|
|
rows = connection.execute("SELECT id, email, phone, role, plan, status, email_verified, phone_verified, created_at, last_login_at FROM users ORDER BY created_at DESC LIMIT 100").fetchall()
|
|
return {"items": rows}
|
|
|
|
|
|
@app.patch("/api/v1/admin/users/{user_id}/status", dependencies=[Depends(require_csrf)])
|
|
def admin_status(user_id: UUID, payload: StatusRequest, connection: Connection = Depends(get_connection), actor: dict = Depends(require_admin)):
|
|
target = connection.execute("SELECT * FROM users WHERE id = %s", (user_id,)).fetchone()
|
|
if not target:
|
|
raise error("NOT_FOUND", "用户不存在", 404)
|
|
if target["role"] == "admin" and payload.status == "disabled":
|
|
count = connection.execute("SELECT count(*) FROM users WHERE role = 'admin' AND status = 'active'").fetchone()["count"]
|
|
if count <= 1:
|
|
raise error("LAST_ADMIN_PROTECTED", "不能禁用最后一个可用管理员", 409)
|
|
updated = connection.execute("UPDATE users SET status = %s, updated_at = now() WHERE id = %s RETURNING id, status", (payload.status, user_id)).fetchone()
|
|
connection.execute("INSERT INTO admin_audit_logs(actor_id, action, target_type, target_id, before_value, after_value, reason) VALUES (%s, 'user_status', 'user', %s, %s, %s, %s)", (actor["id"], user_id, Json({"status": target["status"]}), Json({"status": updated["status"]}), payload.reason))
|
|
connection.commit()
|
|
return updated
|
|
|
|
|
|
@app.put("/api/v1/admin/users/{user_id}/membership", dependencies=[Depends(require_csrf)])
|
|
def admin_membership(user_id: UUID, payload: MembershipRequest, connection: Connection = Depends(get_connection), actor: dict = Depends(require_admin)):
|
|
target = connection.execute("SELECT * FROM users WHERE id = %s", (user_id,)).fetchone()
|
|
if not target:
|
|
raise error("NOT_FOUND", "用户不存在", 404)
|
|
starts_at = payload.starts_at or utc_now()
|
|
connection.execute("UPDATE membership_grants SET revoked_at = now() WHERE user_id = %s AND revoked_at IS NULL", (user_id,))
|
|
grant = connection.execute("INSERT INTO membership_grants(user_id, plan, starts_at, expires_at, created_by, reason) VALUES (%s, %s, %s, %s, %s, %s) RETURNING id, plan, starts_at, expires_at", (user_id, payload.plan, starts_at, payload.expires_at, actor["id"], payload.reason)).fetchone()
|
|
connection.execute("UPDATE users SET plan = %s, updated_at = now() WHERE id = %s", (payload.plan, user_id))
|
|
connection.execute("INSERT INTO admin_audit_logs(actor_id, action, target_type, target_id, before_value, after_value, reason) VALUES (%s, 'membership', 'user', %s, %s, %s, %s)", (actor["id"], user_id, Json({"plan": target["plan"]}), Json({"plan": payload.plan, "expires_at": payload.expires_at.isoformat() if payload.expires_at else None}), payload.reason))
|
|
connection.commit()
|
|
return grant
|
|
|
|
|
|
@app.post("/api/v1/admin/users/{user_id}/quota-adjustments", dependencies=[Depends(require_csrf)])
|
|
def admin_quota(user_id: UUID, payload: QuotaAdjustmentRequest, connection: Connection = Depends(get_connection), actor: dict = Depends(require_admin)):
|
|
target = connection.execute("SELECT * FROM users WHERE id = %s", (user_id,)).fetchone()
|
|
if not target:
|
|
raise error("NOT_FOUND", "用户不存在", 404)
|
|
quota = ensure_quota(connection, user_id, effective_plan(connection, target))
|
|
existing = connection.execute("SELECT id FROM usage_records WHERE user_id = %s AND idempotency_key = %s", (user_id, payload.idempotency_key)).fetchone()
|
|
if existing:
|
|
raise error("IDEMPOTENCY_CONFLICT", "该调整已提交", 409)
|
|
connection.execute("UPDATE quota_accounts SET adjustment = adjustment + %s, version = version + 1 WHERE id = %s", (payload.amount, quota["id"]))
|
|
connection.execute("INSERT INTO usage_records(user_id, quota_account_id, type, amount, idempotency_key) VALUES (%s, %s, 'adjust', %s, %s)", (user_id, quota["id"], payload.amount, payload.idempotency_key))
|
|
connection.execute("INSERT INTO admin_audit_logs(actor_id, action, target_type, target_id, after_value, reason) VALUES (%s, 'quota_adjustment', 'user', %s, %s, %s)", (actor["id"], user_id, Json({"amount": payload.amount}), payload.reason))
|
|
connection.commit()
|
|
return {"status": "ok", "amount": payload.amount}
|