50 lines
1.6 KiB
Python
50 lines
1.6 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
OSS 上传工具 - 用于阿里听悟音频上传
|
||
"""
|
||
|
||
import os
|
||
import uuid
|
||
from datetime import datetime, timedelta
|
||
import oss2
|
||
|
||
class OSSUploader:
|
||
def __init__(self):
|
||
self.access_key = os.getenv("ALI_ACCESS_KEY")
|
||
self.secret_key = os.getenv("ALI_SECRET_KEY")
|
||
# 使用杭州区域，听悟服务在杭州
|
||
self.endpoint = "oss-cn-hangzhou.aliyuncs.com"
|
||
self.bucket_name = os.getenv("ALI_OSS_BUCKET", "insightflow-audio")
|
||
|
||
if not self.access_key or not self.secret_key:
|
||
raise ValueError("ALI_ACCESS_KEY and ALI_SECRET_KEY must be set")
|
||
|
||
self.auth = oss2.Auth(self.access_key, self.secret_key)
|
||
self.bucket = oss2.Bucket(self.auth, self.endpoint, self.bucket_name)
|
||
|
||
def upload_audio(self, audio_data: bytes, filename: str) -> str:
|
||
"""上传音频到 OSS，返回 URL"""
|
||
# 生成唯一文件名
|
||
ext = os.path.splitext(filename)[1] or ".wav"
|
||
object_name = f"audio/{datetime.now().strftime('%Y%m%d')}/{uuid.uuid4().hex}{ext}"
|
||
|
||
# 上传文件
|
||
self.bucket.put_object(object_name, audio_data)
|
||
|
||
# 生成临时访问 URL (1小时有效)
|
||
url = self.bucket.sign_url('GET', object_name, 3600)
|
||
return url, object_name
|
||
|
||
def delete_object(self, object_name: str):
|
||
"""删除 OSS 对象"""
|
||
self.bucket.delete_object(object_name)
|
||
|
||
# 单例
|
||
_oss_uploader = None
|
||
|
||
def get_oss_uploader() -> OSSUploader:
|
||
global _oss_uploader
|
||
if _oss_uploader is None:
|
||
_oss_uploader = OSSUploader()
|
||
return _oss_uploader
|