124 lines
4.0 KiB
Python
124 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
腾讯云COS文件上传工具
|
||
用法:
|
||
python3 cos_upload.py <local_file> <cos_key> [--content-type <mime>]
|
||
"""
|
||
import os
|
||
import sys
|
||
import mimetypes
|
||
import argparse
|
||
from qcloud_cos import CosConfig, CosS3Client
|
||
|
||
# ============ 配置（修改这里切换桶/域名）============
|
||
COS_SECRET_ID = "AKIDpbWMzOYaP2UltHWccS4TJJDxX1Pfkvzt"
|
||
COS_SECRET_KEY = "ythOwofOxvtklL3XpxsaChEbntqhOrFi"
|
||
COS_REGION = "ap-beijing"
|
||
COS_BUCKET = "static-1317843270"
|
||
COS_DOWNLOAD_DOMAIN = "static.valavala.com"
|
||
|
||
|
||
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()
|