d6625762cf
modified: app/ali_oss.py modified: app/app.py new file: app/blueprints/__pycache__/communication.cpython-39.pyc modified: app/blueprints/communication.py modified: app/brand_spider/__pycache__/main.cpython-39.pyc modified: app/config.py modified: app/main.py modified: app/static/bg.jpg modified: app/tool/__pycache__/devices.cpython-39.pyc modified: app/web_source/brand.html modified: app/web_source/templates_backup/brand.html
72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
import argparse
|
||
import base64
|
||
import re
|
||
import time
|
||
import alibabacloud_oss_v2 as oss
|
||
import requests
|
||
|
||
from config import region, endpoint, bucket, file_url_pre, bucket_path
|
||
|
||
|
||
def upload_file(file_content: bytes, key: str):
|
||
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
|
||
cfg = oss.config.load_default()
|
||
cfg.credentials_provider = credentials_provider
|
||
cfg.region = region
|
||
cfg.endpoint = endpoint
|
||
client = oss.Client(cfg)
|
||
|
||
result = client.put_object(
|
||
oss.PutObjectRequest(
|
||
bucket=bucket, # 存储空间名称
|
||
key=key, # 对象名称
|
||
body=file_content # 读取文件内容
|
||
)
|
||
)
|
||
# print(result)
|
||
return file_url_pre + key
|
||
|
||
|
||
def upload_data_url(data_url: str, prefix: str = "history", key_hint: str = "") -> str:
|
||
"""
|
||
将 base64 data URL 上传到 OSS，返回图片链接
|
||
data_url: data:image/png;base64,xxxx 或 data:image/jpeg;base64,xxxx
|
||
prefix: OSS key 前缀
|
||
key_hint: 可选后缀避免重名，如 "_0", "_1"
|
||
"""
|
||
match = re.match(r'data:image/(\w+);base64,(.+)', data_url)
|
||
if not match:
|
||
raise ValueError('无效的 data URL 格式')
|
||
ext = 'png' if match.group(1).lower() in ('png', 'webp') else 'jpg'
|
||
file_content = base64.b64decode(match.group(2))
|
||
ts = int(time.time() * 1000)
|
||
key = f"{bucket_path}{prefix}/{ts}{key_hint}.{ext}"
|
||
return upload_file(file_content, key)
|
||
|
||
|
||
def upload_data_urls(data_urls: list, prefix: str = "history") -> list:
|
||
"""批量上传 base64 图片到 OSS，返回图片链接列表"""
|
||
urls = []
|
||
ts = int(time.time() * 1000)
|
||
for i, data_url in enumerate(data_urls or []):
|
||
if not data_url or not isinstance(data_url, str):
|
||
continue
|
||
if not data_url.startswith("http"):
|
||
match = re.match(r'data:image/(\w+);base64,(.+)', data_url)
|
||
if not match:
|
||
continue
|
||
ext = 'png' if match.group(1).lower() in ('png', 'webp') else 'jpg'
|
||
file_content = base64.b64decode(match.group(2))
|
||
else:
|
||
file_content = requests.get(data_url).content
|
||
ext = "png"
|
||
key = f"{bucket_path}{prefix}/{ts}_{i}.{ext}"
|
||
urls.append(upload_file(file_content, key))
|
||
return urls
|
||
|
||
|
||
# 脚本入口，当文件被直接运行时调用main函数
|
||
if __name__ == "__main__":
|
||
with open("测试图片数据/IMG_2685.JPG", "rb") as f:
|
||
file_content = f.read()
|
||
upload_file(file_content,key=bucket_path+"test.png") |