1

video-gen-api/app/utils/exceptions.py

@@ -0,0 +1,41 @@
+from fastapi import HTTPException, status
+
+
+class InsufficientCreditsError(HTTPException):
+    def __init__(self):
+        super().__init__(
+            status_code=status.HTTP_402_PAYMENT_REQUIRED,
+            detail="积分不足，请充值",
+        )
+
+
+class CaptchaFailedError(HTTPException):
+    def __init__(self):
+        super().__init__(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="验证码校验失败",
+        )
+
+
+class RecordNotFoundError(HTTPException):
+    def __init__(self):
+        super().__init__(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="记录不存在",
+        )
+
+
+class ProjectNotFoundError(HTTPException):
+    def __init__(self):
+        super().__init__(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="项目不存在",
+        )
+
+
+class InvalidStatusError(HTTPException):
+    def __init__(self, detail: str = "当前状态不允许此操作"):
+        super().__init__(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=detail,
+        )

video-gen-api/app/utils/id_gen.py

@@ -0,0 +1,16 @@
+import time
+import random
+
+
+def generate_id() -> str:
+    """Generate a time-sortable unique ID (simplified ULID-style)."""
+    timestamp = int(time.time() * 1000)
+    randomness = random.randint(0, 0xFFFFFF)
+    return f"{timestamp:013x}{randomness:06x}"
+
+
+def generate_order_no() -> str:
+    """Generate a human-readable order number."""
+    timestamp = int(time.time())
+    randomness = random.randint(1000, 9999)
+    return f"VG{timestamp}{randomness}"

video-gen-api/app/utils/redis.py

@@ -0,0 +1,30 @@
+from app.config import settings
+
+redis_client = None
+
+
+async def init_redis() -> None:
+    global redis_client
+    if not settings.REDIS_URL:
+        return
+    try:
+        from redis.asyncio import Redis
+        redis_client = Redis.from_url(settings.REDIS_URL, decode_responses=True)
+        await redis_client.ping()
+    except Exception:
+        redis_client = None
+
+
+async def close_redis() -> None:
+    global redis_client
+    if redis_client:
+        try:
+            await redis_client.close()
+        except Exception:
+            pass
+        redis_client = None
+
+
+def get_redis():
+    """Return Redis client or None if not available."""
+    return redis_client

video-gen-api/app/utils/security.py

@@ -0,0 +1,53 @@
+import base64
+import hashlib
+import hmac
+import json
+import time
+
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+
+from app.config import settings
+
+
+def get_aes_key() -> bytes:
+    """Derive a 32-byte AES key from the configured encryption key."""
+    return hashlib.sha256(settings.ENCRYPTION_KEY.encode()).digest()
+
+
+def encrypt_temp_token(record_id: str, expires_in: int = 3600) -> str:
+    """Create an encrypted token containing record_id and expiry timestamp."""
+    payload = json.dumps({"rid": record_id, "exp": int(time.time()) + expires_in})
+    aesgcm = AESGCM(get_aes_key())
+    nonce = AESGCM.generate_key(bit_length=96)
+    ciphertext = aesgcm.encrypt(nonce, payload.encode(), None)
+    token_bytes = nonce + ciphertext
+    return base64.urlsafe_b64encode(token_bytes).decode()
+
+
+def decrypt_temp_token(token: str) -> str | None:
+    """Decrypt token and return record_id if valid and not expired."""
+    try:
+        token_bytes = base64.urlsafe_b64decode(token)
+        nonce = token_bytes[:12]
+        ciphertext = token_bytes[12:]
+        aesgcm = AESGCM(get_aes_key())
+        payload = aesgcm.decrypt(nonce, ciphertext, None)
+        data = json.loads(payload)
+        if data["exp"] < time.time():
+            return None
+        return data["rid"]
+    except Exception:
+        return None
+
+
+def hmac_sign(data: str) -> str:
+    """Create HMAC-SHA256 signature."""
+    return hmac.new(
+        settings.SECRET_KEY.encode(), data.encode(), hashlib.sha256
+    ).hexdigest()
+
+
+def hmac_verify(data: str, signature: str) -> bool:
+    """Verify HMAC-SHA256 signature."""
+    expected = hmac_sign(data)
+    return hmac.compare_digest(expected, signature)