upload_to_oss.py
4.79 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
上传文件到阿里云 OSS
"""
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider
import os
import sys
def get_credentials():
"""获取 OSS 凭证"""
# 1. 首先尝试环境变量
access_key_id = os.getenv('OSS_ACCESS_KEY_ID')
access_key_secret = os.getenv('OSS_ACCESS_KEY_SECRET')
if access_key_id and access_key_secret:
print("✅ 从环境变量读取凭证")
return access_key_id, access_key_secret
# 2. 尝试 .env 文件
env_file = '/Users/yibo/Documents/LemonAI/multi-agent-system/.env'
if os.path.exists(env_file):
with open(env_file, 'r') as f:
for line in f:
line = line.strip()
if line.startswith('export OSS_ACCESS_KEY_ID='):
access_key_id = line.split('=', 1)[1].strip().strip("'")
elif line.startswith('export OSS_ACCESS_KEY_SECRET='):
access_key_secret = line.split('=', 1)[1].strip().strip("'")
if access_key_id and access_key_secret:
print("✅ 从 .env 文件读取凭证")
return access_key_id, access_key_secret
return None, None
def main():
# 获取凭证
access_key_id, access_key_secret = get_credentials()
if not access_key_id or not access_key_secret:
print("❌ 未找到 OSS 凭证")
print("\n请通过以下任一方式提供凭证：")
print("\n方式 1 - 环境变量（推荐）:")
print(" export OSS_ACCESS_KEY_ID='your-access-key-id'")
print(" export OSS_ACCESS_KEY_SECRET='your-access-key-secret'")
print("\n方式 2 - 添加到 .env 文件:")
print(" export OSS_ACCESS_KEY_ID='your-access-key-id'")
print(" export OSS_ACCESS_KEY_SECRET='your-access-key-secret'")
print("\n方式 3 - 直接输入（临时）:")
try:
access_key_id = input("\n请输入 AccessKey ID: ").strip()
access_key_secret = input("请输入 AccessKey Secret: ").strip()
except EOFError:
print("\n❌ 无法交互式输入，请使用环境变量方式")
sys.exit(1)
if not access_key_id or not access_key_secret:
print("❌ 凭证不能为空")
sys.exit(1)
# 配置
bucket_name = 'lemondrop-oss-test'
region = 'cn-hangzhou'
endpoint = f'https://oss-{region}.aliyuncs.com'
local_file = '/Users/yibo/Documents/LemonAI/multi-agent-system/skills/yipin_buyi_chapter_1_output.md'
object_key = 'novels/yipin_buyi_chapter_1_output.md'
# 检查本地文件
if not os.path.exists(local_file):
print(f"❌ 错误: 本地文件不存在: {local_file}")
sys.exit(1)
file_size = os.path.getsize(local_file)
print(f"\n📄 准备上传文件: {local_file}")
print(f" 文件大小: {file_size} bytes ({file_size/1024:.2f} KB)")
print(f" 目标 Bucket: {bucket_name}")
print(f" 目标路径: {object_key}")
print()
try:
# 初始化认证和 Bucket
print("🔐 正在初始化 OSS 客户端...")
auth = oss2.Auth(access_key_id, access_key_secret)
bucket = oss2.Bucket(
auth,
endpoint,
bucket_name,
region=region,
connect_timeout=30,
read_timeout=60
)
# 上传文件
print("⬆️ 正在上传文件...")
result = bucket.put_object_from_file(object_key, local_file)
print(f"✅ 文件上传成功!")
print(f" ETag: {result.etag}")
print()
# 设置公共读权限
print("🔓 设置 ACL 为公共读...")
bucket.put_object_acl(object_key, oss2.OBJECT_ACL_PUBLIC_READ)
print("✅ ACL 设置成功!")
print()
# 生成访问 URL
url = f'https://{bucket_name}.oss-{region}.aliyuncs.com/{object_key}'
print("="*60)
print("📦 上传完成!")
print("="*60)
print(f"🌐 访问 URL: {url}")
print(f"📁 Object Key: {object_key}")
print(f"📊 文件大小: {file_size} bytes")
print("="*60)
return url
except oss2.exceptions.NoSuchBucket as e:
print(f"❌ 错误: Bucket 不存在 - {bucket_name}")
print(f" Request ID: {e.request_id}")
sys.exit(1)
except oss2.exceptions.AccessDenied as e:
print(f"❌ 错误: 访问被拒绝，请检查凭证权限")
print(f" Request ID: {e.request_id}")
sys.exit(1)
except oss2.exceptions.OssError as e:
print(f"❌ OSS 错误: {e}")
sys.exit(1)
except Exception as e:
print(f"❌ 错误: {e}")
sys.exit(1)
if __name__ == '__main__':
main()