video-compliance-ai/backend/app/services/oss.py

"""
腾讯云 COS 服务 — 表单直传签名
"""
import time
import hmac
import base64
import hashlib
import json
from typing import Optional
from datetime import datetime
from app.config import settings


def generate_upload_policy(
    max_size_mb: int = 500,
    expire_seconds: int = 3600,
    upload_dir: Optional[str] = None,
) -> dict:
    """
    生成前端直传 COS 所需的 Policy 和签名

    COS 表单直传签名流程:
    1. key_time = "{start_time};{end_time}"
    2. sign_key = HMAC-SHA1(secret_key, key_time)
    3. policy JSON → Base64 编码
    4. string_to_sign = SHA1(policy_base64)
    5. signature = HMAC-SHA1(sign_key, string_to_sign)

    Returns:
        {
            "q_sign_algorithm": "sha1",
            "q_ak": "SecretId",
            "q_key_time": "{start};{end}",
            "q_signature": "...",
            "policy": "base64 encoded policy",
            "host": "https://bucket.cos.region.myqcloud.com",
            "dir": "uploads/2026/02/",
            "expire": 1234567890,
        }
    """
    if not settings.COS_SECRET_ID or not settings.COS_SECRET_KEY:
        raise ValueError("COS 配置未设置")

    # 计算时间范围
    start_time = int(time.time())
    end_time = start_time + expire_seconds

    # key_time: "{start};{end}"
    key_time = f"{start_time};{end_time}"

    # 默认上传目录：uploads/年/月/
    if upload_dir is None:
        now = datetime.now()
        upload_dir = f"uploads/{now.year}/{now.month:02d}/"

    # 1. sign_key = HMAC-SHA1(secret_key, key_time)
    sign_key = hmac.new(
        settings.COS_SECRET_KEY.encode(),
        key_time.encode(),
        hashlib.sha1,
    ).hexdigest()

    # 2. 构建 Policy（COS 表单上传 Policy 格式）
    policy_dict = {
        "expiration": datetime.utcfromtimestamp(end_time).strftime(
            "%Y-%m-%dT%H:%M:%S.000Z"
        ),
        "conditions": [
            {"bucket": settings.COS_BUCKET_NAME},
            ["starts-with", "$key", upload_dir],
            {"q-sign-algorithm": "sha1"},
            {"q-ak": settings.COS_SECRET_ID},
            {"q-sign-time": key_time},
            ["content-length-range", 0, max_size_mb * 1024 * 1024],
        ],
    }

    # 3. Base64 编码 Policy
    policy_json = json.dumps(policy_dict)
    policy_base64 = base64.b64encode(policy_json.encode()).decode()

    # 4. string_to_sign = SHA1(policy_base64)
    string_to_sign = hashlib.sha1(policy_base64.encode()).hexdigest()

    # 5. signature = HMAC-SHA1(sign_key, string_to_sign)
    signature = hmac.new(
        sign_key.encode(),
        string_to_sign.encode(),
        hashlib.sha1,
    ).hexdigest()

    # 构建 Host
    host = f"https://{settings.COS_BUCKET_NAME}.cos.{settings.COS_REGION}.myqcloud.com"

    return {
        "q_sign_algorithm": "sha1",
        "q_ak": settings.COS_SECRET_ID,
        "q_key_time": key_time,
        "q_signature": signature,
        "policy": policy_base64,
        "host": host,
        "dir": upload_dir,
        "expire": end_time,
    }


def get_file_url(file_key: str) -> str:
    """
    获取文件的访问 URL

    优先使用 CDN 域名，否则用 COS 源站域名。

    Args:
        file_key: 文件在 COS 中的 key，如 "uploads/2026/02/video.mp4"

    Returns:
        完整的访问 URL
    """
    if settings.COS_CDN_DOMAIN:
        host = settings.COS_CDN_DOMAIN
    else:
        host = f"https://{settings.COS_BUCKET_NAME}.cos.{settings.COS_REGION}.myqcloud.com"

    # 确保 host 以 https:// 开头
    if not host.startswith("http"):
        host = f"https://{host}"

    # 确保 host 不以 / 结尾
    host = host.rstrip("/")

    # 确保 file_key 不以 / 开头
    file_key = file_key.lstrip("/")

    return f"{host}/{file_key}"


def parse_file_key_from_url(url: str) -> str:
    """
    从完整 URL 解析出文件 key

    Args:
        url: 完整的 COS URL

    Returns:
        文件 key
    """
    # 尝试移除 CDN 域名
    if settings.COS_CDN_DOMAIN:
        cdn = settings.COS_CDN_DOMAIN.rstrip("/")
        if not cdn.startswith("http"):
            cdn = f"https://{cdn}"
        if url.startswith(cdn):
            return url[len(cdn):].lstrip("/")

    # 尝试移除 COS 源站域名
    cos_host = f"https://{settings.COS_BUCKET_NAME}.cos.{settings.COS_REGION}.myqcloud.com"
    if url.startswith(cos_host):
        return url[len(cos_host):].lstrip("/")

    return url