85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
#!/usr/bin/env python
|
||
#coding=utf-8
|
||
|
||
import os
|
||
import oss2
|
||
from oss2 import Auth
|
||
|
||
def upload_file_to_oss(local_file_path, bucket_name, object_name, access_key_id, access_key_secret, endpoint='https://oss-cn-beijing.aliyuncs.com'):
|
||
"""
|
||
上传文件到OSS
|
||
:param local_file_path: 本地文件路径
|
||
:param bucket_name: OSS桶名称
|
||
:param object_name: OSS对象名称
|
||
:param access_key_id: 阿里云AccessKey ID
|
||
:param access_key_secret: 阿里云AccessKey Secret
|
||
:param endpoint: OSS端点，默认为北京区域
|
||
:return: OSS文件URL
|
||
"""
|
||
print(f"开始上传文件到OSS: {local_file_path}")
|
||
|
||
try:
|
||
# 创建OSS客户端
|
||
auth = Auth(access_key_id, access_key_secret)
|
||
bucket = oss2.Bucket(auth, endpoint, bucket_name)
|
||
|
||
# 上传文件
|
||
bucket.put_object_from_file(object_name, local_file_path)
|
||
print(f"文件上传成功: {object_name}")
|
||
|
||
# 生成可访问的URL
|
||
# 注意：这里生成的是临时URL，有效期为3600秒
|
||
# 如果需要永久URL，需要在OSS桶中设置文件为公共读
|
||
url = bucket.sign_url('GET', object_name, 3600)
|
||
print(f"生成OSS文件URL: {url}")
|
||
|
||
return url
|
||
|
||
except Exception as e:
|
||
print(f"文件上传失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return None
|
||
|
||
def make_bucket_public(bucket_name, access_key_id, access_key_secret, endpoint='https://oss-cn-beijing.aliyuncs.com'):
|
||
"""
|
||
设置OSS桶为公共读
|
||
:param bucket_name: OSS桶名称
|
||
:param access_key_id: 阿里云AccessKey ID
|
||
:param access_key_secret: 阿里云AccessKey Secret
|
||
:param endpoint: OSS端点，默认为北京区域
|
||
:return: 是否设置成功
|
||
"""
|
||
print(f"开始设置OSS桶为公共读: {bucket_name}")
|
||
|
||
try:
|
||
# 创建OSS客户端
|
||
auth = Auth(access_key_id, access_key_secret)
|
||
bucket = oss2.Bucket(auth, endpoint, bucket_name)
|
||
|
||
# 设置桶的访问策略为公共读
|
||
policy = """
|
||
{
|
||
"Version": "1",
|
||
"Statement": [
|
||
{
|
||
"Action": ["oss:GetObject"],
|
||
"Effect": "Allow",
|
||
"Principal": ["*"],
|
||
"Resource": ["acs:oss:*:*:{bucket_name}/*"],
|
||
"Condition": {}
|
||
}
|
||
]
|
||
}
|
||
""".format(bucket_name=bucket_name)
|
||
|
||
bucket.put_bucket_policy(policy)
|
||
print(f"OSS桶设置为公共读成功: {bucket_name}")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"OSS桶设置失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|