154 lines
5.0 KiB
Python
154 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
腾讯云COS文件上传工具
|
||
用法:
|
||
python3 cos_upload.py <local_file> <cos_key> [--content-type <mime>]
|
||
"""
|
||
import os
|
||
import sys
|
||
import re
|
||
import mimetypes
|
||
import argparse
|
||
from qcloud_cos import CosConfig, CosS3Client
|
||
|
||
# ============ 配置（从 secrets.md 读取，不再硬编码） ============
|
||
SECRETS_PATH = "/root/.openclaw/workspace-xiaokui/secrets.md"
|
||
|
||
def _load_cos_config():
|
||
"""从 secrets.md 加载COS配置"""
|
||
if not os.path.exists(SECRETS_PATH):
|
||
raise RuntimeError(f"密钥文件不存在: {SECRETS_PATH}")
|
||
|
||
with open(SECRETS_PATH, 'r', encoding='utf-8') as f:
|
||
content = f.read()
|
||
|
||
# 提取腾讯云COS配置
|
||
patterns = {
|
||
'secret_id': r'TENCENT_SECRET_ID.*?`([^`]+)`',
|
||
'secret_key': r'TENCENT_SECRET_KEY.*?`([^`]+)`',
|
||
'region': r'TENCENT_COS_REGION.*?`([^`]+)`',
|
||
'bucket': r'TENCENT_COS_BUCKET.*?`([^`]+)`',
|
||
'domain': r'TENCENT_COS_DOWNLOAD_PATH.*?`([^`]+)`'
|
||
}
|
||
|
||
config = {}
|
||
for key, pattern in patterns.items():
|
||
match = re.search(pattern, content, re.MULTILINE)
|
||
if not match:
|
||
raise RuntimeError(f"secrets.md 中未找到 COS 配置项: {key}")
|
||
config[key] = match.group(1)
|
||
|
||
return config
|
||
|
||
_cos_config = _load_cos_config()
|
||
COS_SECRET_ID = _cos_config['secret_id']
|
||
COS_SECRET_KEY = _cos_config['secret_key']
|
||
COS_REGION = _cos_config['region']
|
||
COS_BUCKET = _cos_config['bucket']
|
||
COS_DOWNLOAD_DOMAIN = _cos_config['domain']
|
||
|
||
|
||
class CosUploader:
|
||
"""腾讯云COS上传器"""
|
||
|
||
def __init__(self, secret_id=None, secret_key=None, region=None, bucket=None, domain=None):
|
||
self.bucket = bucket or COS_BUCKET
|
||
self.domain = domain or COS_DOWNLOAD_DOMAIN
|
||
config = CosConfig(
|
||
Region=region or COS_REGION,
|
||
SecretId=secret_id or COS_SECRET_ID,
|
||
SecretKey=secret_key or COS_SECRET_KEY,
|
||
Scheme='https'
|
||
)
|
||
self.client = CosS3Client(config)
|
||
|
||
def upload(self, local_path: str, cos_key: str, content_type: str = None) -> str:
|
||
"""
|
||
上传文件到COS
|
||
Args:
|
||
local_path: 本地文件路径
|
||
cos_key: COS存储路径
|
||
content_type: MIME类型，不指定则自动检测
|
||
Returns:
|
||
可访问的URL
|
||
"""
|
||
if not os.path.exists(local_path):
|
||
raise FileNotFoundError(f"文件不存在: {local_path}")
|
||
|
||
if not content_type:
|
||
content_type = self._detect_content_type(local_path)
|
||
|
||
kwargs = {
|
||
'Bucket': self.bucket,
|
||
'Key': cos_key,
|
||
'LocalFilePath': local_path,
|
||
}
|
||
if content_type:
|
||
kwargs['ContentType'] = content_type
|
||
|
||
self.client.upload_file(**kwargs)
|
||
return f"https://{self.domain}/{cos_key}"
|
||
|
||
def upload_bytes(self, data: bytes, cos_key: str, content_type: str = None) -> str:
|
||
"""上传字节数据到COS"""
|
||
kwargs = {
|
||
'Bucket': self.bucket,
|
||
'Key': cos_key,
|
||
'Body': data,
|
||
}
|
||
if content_type:
|
||
kwargs['ContentType'] = content_type
|
||
self.client.put_object(**kwargs)
|
||
return f"https://{self.domain}/{cos_key}"
|
||
|
||
def upload_batch(self, items: list) -> list:
|
||
"""
|
||
批量上传
|
||
Args:
|
||
items: [(local_path, cos_key), ...] 或 [(local_path, cos_key, content_type), ...]
|
||
Returns:
|
||
[(cos_key, url), ...]
|
||
"""
|
||
results = []
|
||
for item in items:
|
||
local_path = item[0]
|
||
cos_key = item[1]
|
||
content_type = item[2] if len(item) > 2 else None
|
||
try:
|
||
url = self.upload(local_path, cos_key, content_type)
|
||
results.append((cos_key, url))
|
||
except Exception as e:
|
||
print(f"[ERROR] 上传失败 {cos_key}: {e}", file=sys.stderr)
|
||
results.append((cos_key, None))
|
||
return results
|
||
|
||
def delete(self, cos_key: str):
|
||
"""删除COS上的文件"""
|
||
self.client.delete_object(Bucket=self.bucket, Key=cos_key)
|
||
|
||
def list_objects(self, prefix: str, max_keys: int = 100) -> list:
|
||
"""列出COS上的文件"""
|
||
resp = self.client.list_objects(Bucket=self.bucket, Prefix=prefix, MaxKeys=max_keys)
|
||
return [item['Key'] for item in resp.get('Contents', []) if not item['Key'].endswith('/')]
|
||
|
||
@staticmethod
|
||
def _detect_content_type(filepath: str) -> str:
|
||
mime, _ = mimetypes.guess_type(filepath)
|
||
return mime or 'application/octet-stream'
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description='上传文件到腾讯云COS')
|
||
parser.add_argument('local_file', help='本地文件路径')
|
||
parser.add_argument('cos_key', help='COS存储路径')
|
||
parser.add_argument('--content-type', help='MIME类型（自动检测）', default=None)
|
||
args = parser.parse_args()
|
||
|
||
uploader = CosUploader()
|
||
url = uploader.upload(args.local_file, args.cos_key, args.content_type)
|
||
print(url)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|