john
1aaef71f52
Document-driven MVP with FastAPI backend, Vue H5, WeChat mini shell, product demo, and Docker dev stack. Co-authored-by: Cursor <cursoragent@cursor.com>
135 lines
4.1 KiB
Python
135 lines
4.1 KiB
Python
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
from app.config import settings
|
|
|
|
_upload_tokens: dict[str, dict] = {}
|
|
|
|
|
|
def _upload_root() -> Path:
|
|
root = Path(settings.upload_local_dir)
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
return root
|
|
|
|
|
|
def generate_object_key(child_id: int, file_name: str) -> str:
|
|
safe_name = file_name.replace("..", "").replace("/", "_")
|
|
return f"videos/{child_id}/{safe_name}"
|
|
|
|
|
|
def _cleanup_expired_tokens() -> None:
|
|
now = datetime.now(timezone.utc)
|
|
expired = [k for k, v in _upload_tokens.items() if v["expire_at"] < now]
|
|
for key in expired:
|
|
_upload_tokens.pop(key, None)
|
|
|
|
|
|
def create_upload_token(object_key: str, content_type: str, size: int) -> dict:
|
|
_cleanup_expired_tokens()
|
|
expire_at = datetime.now(timezone.utc) + timedelta(minutes=settings.upload_token_expire_minutes)
|
|
token = secrets.token_urlsafe(32)
|
|
|
|
if settings.oss_enabled:
|
|
upload_url, method = _create_s3_presigned_url(object_key, content_type, expire_at)
|
|
extra = {
|
|
"storage": settings.oss_provider,
|
|
"cdnBaseUrl": settings.oss_cdn_base_url or None,
|
|
}
|
|
else:
|
|
upload_url = f"{settings.api_public_url.rstrip('/')}/api/videos/direct-upload"
|
|
method = "PUT"
|
|
extra = {"storage": "local", "uploadToken": token}
|
|
|
|
_upload_tokens[token] = {
|
|
"object_key": object_key,
|
|
"content_type": content_type,
|
|
"size": size,
|
|
"expire_at": expire_at,
|
|
}
|
|
|
|
return {
|
|
"uploadUrl": upload_url,
|
|
"objectKey": object_key,
|
|
"expireAt": expire_at.replace(microsecond=0).isoformat(),
|
|
"method": method,
|
|
**extra,
|
|
"uploadToken": token,
|
|
}
|
|
|
|
|
|
def _create_s3_presigned_url(object_key: str, content_type: str, expire_at: datetime) -> tuple[str, str]:
|
|
try:
|
|
import boto3
|
|
from botocore.client import Config
|
|
except ImportError as exc:
|
|
raise RuntimeError("boto3 required for OSS mode") from exc
|
|
|
|
client = boto3.client(
|
|
"s3",
|
|
endpoint_url=settings.oss_endpoint,
|
|
aws_access_key_id=settings.oss_access_key,
|
|
aws_secret_access_key=settings.oss_secret_key,
|
|
config=Config(signature_version="s3v4"),
|
|
region_name=settings.oss_region,
|
|
)
|
|
_ensure_bucket(client)
|
|
expires_in = max(60, int((expire_at - datetime.now(timezone.utc)).total_seconds()))
|
|
url = client.generate_presigned_url(
|
|
"put_object",
|
|
Params={
|
|
"Bucket": settings.oss_bucket,
|
|
"Key": object_key,
|
|
"ContentType": content_type,
|
|
},
|
|
ExpiresIn=expires_in,
|
|
)
|
|
return url, "PUT"
|
|
|
|
|
|
def _ensure_bucket(client) -> None:
|
|
if settings.oss_provider != "minio":
|
|
return
|
|
try:
|
|
client.head_bucket(Bucket=settings.oss_bucket)
|
|
except Exception:
|
|
client.create_bucket(Bucket=settings.oss_bucket)
|
|
|
|
|
|
def public_object_url(object_key: str) -> str | None:
|
|
if settings.oss_cdn_base_url:
|
|
return f"{settings.oss_cdn_base_url.rstrip('/')}/{object_key}"
|
|
if settings.oss_enabled and settings.oss_provider == "aliyun":
|
|
endpoint = settings.oss_endpoint.rstrip("/")
|
|
return f"{endpoint}/{settings.oss_bucket}/{object_key}"
|
|
return None
|
|
|
|
|
|
def validate_upload_token(token: str, object_key: str) -> bool:
|
|
_cleanup_expired_tokens()
|
|
meta = _upload_tokens.get(token)
|
|
if not meta or meta["object_key"] != object_key:
|
|
return False
|
|
if meta["expire_at"] < datetime.now(timezone.utc):
|
|
_upload_tokens.pop(token, None)
|
|
return False
|
|
return True
|
|
|
|
|
|
def consume_upload_token(token: str, object_key: str) -> None:
|
|
meta = _upload_tokens.pop(token, None)
|
|
if not meta or meta["object_key"] != object_key:
|
|
raise ValueError("invalid_upload_token")
|
|
|
|
|
|
def save_local_upload(object_key: str, data: bytes) -> Path:
|
|
path = _upload_root() / object_key
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(data)
|
|
return path
|
|
|
|
|
|
def resolve_local_path(object_key: str) -> Path | None:
|
|
path = _upload_root() / object_key
|
|
return path if path.exists() else None
|