Files
banban/talkingq-url/banban/service/message_audio_storage.py
2026-05-06 03:33:28 +08:00

145 lines
4.6 KiB
Python

import asyncio
from dataclasses import dataclass
from datetime import datetime
try:
from datetime import UTC # Python 3.11+
except ImportError:
from datetime import timezone
UTC = timezone.utc # Python 3.10 及更早版本
from urllib.parse import urlparse
from uuid import uuid4
try:
from qcloud_cos import CosConfig, CosS3Client
except ModuleNotFoundError: # pragma: no cover - exercised in runtime env
CosConfig = None
CosS3Client = None
from config import settings
class MessageAudioStorageError(Exception):
pass
@dataclass(frozen=True)
class StoredMessageAudio:
file_key: str
public_url: str
class MessageAudioStorageService:
def __init__(self) -> None:
self._client = None
def _assert_ready(self) -> None:
if CosConfig is None or CosS3Client is None:
raise MessageAudioStorageError("COS SDK is not installed")
required_pairs = {
"COS_SECRET_ID": settings.cos_secret_id,
"COS_SECRET_KEY": settings.cos_secret_key,
"COS_REGION": settings.cos_region,
"COS_BUCKET_MESSAGE": settings.cos_bucket_message,
}
missing = [key for key, value in required_pairs.items() if not value]
if missing:
raise MessageAudioStorageError(f"missing COS message config: {', '.join(missing)}")
def _get_client(self):
if self._client is None:
config = CosConfig(
Region=settings.cos_region,
SecretId=settings.cos_secret_id,
SecretKey=settings.cos_secret_key,
Scheme="https",
)
self._client = CosS3Client(config)
return self._client
def _build_key(self, *, device_id: str, extension: str) -> str:
prefix = settings.cos_message_prefix.strip("/") or "messages/audio"
now = datetime.now(UTC)
return (
f"{prefix}/{device_id}/{now.strftime('%Y/%m/%d')}/"
f"{uuid4().hex}.{extension}"
)
def _build_public_url(self, *, file_key: str) -> str:
base_url = settings.cos_public_base_url.strip().rstrip("/")
if not base_url:
base_url = f"https://{settings.cos_bucket_message}.cos.{settings.cos_region}.myqcloud.com"
return f"{base_url}/{file_key.lstrip('/')}"
def _normalize_file_key(self, file_key_or_url: str) -> str:
value = (file_key_or_url or "").strip()
if not value:
raise MessageAudioStorageError("audio file key is required")
if value.startswith("http://") or value.startswith("https://"):
parsed = urlparse(value)
path = parsed.path.lstrip("/")
if not path:
raise MessageAudioStorageError("audio file key is invalid")
return path
return value.lstrip("/")
async def upload_audio(
self,
*,
device_id: str,
content: bytes,
content_type: str = "audio/mpeg",
extension: str = "mp3",
) -> StoredMessageAudio:
self._assert_ready()
if not content:
raise MessageAudioStorageError("audio content is empty")
file_key = self._build_key(device_id=device_id, extension=extension)
await asyncio.to_thread(
self._upload_audio_sync,
file_key=file_key,
content=content,
content_type=content_type,
)
return StoredMessageAudio(
file_key=file_key,
public_url=self._build_public_url(file_key=file_key),
)
async def get_audio_url(self, file_key_or_url: str) -> str:
self._assert_ready()
normalized_key = self._normalize_file_key(file_key_or_url)
return await asyncio.to_thread(
self._get_client().get_presigned_url,
Bucket=settings.cos_bucket_message,
Key=normalized_key,
Method="GET",
Expired=settings.cos_avatar_url_expire_seconds,
)
async def delete_audio(self, file_key_or_url: str) -> None:
self._assert_ready()
normalized_key = self._normalize_file_key(file_key_or_url)
await asyncio.to_thread(
self._get_client().delete_object,
Bucket=settings.cos_bucket_message,
Key=normalized_key,
)
def _upload_audio_sync(
self,
*,
file_key: str,
content: bytes,
content_type: str,
) -> None:
self._get_client().put_object(
Bucket=settings.cos_bucket_message,
Body=content,
Key=file_key,
ContentType=content_type,
EnableMD5=False,
)