digitalhumantalk/thirdparty/lingda/filestorage.py
2024-12-10 17:05:37 +08:00

67 lines
1.8 KiB
Python

# -*- coding: UTF-8 -*-
# Python 2.x引入httplib模块。
# import httplib
# Python 3.x引入http.client模块。
import http.client
# Python 2.x引入urllib模块。
# import urllib
# Python 3.x引入urllib.parse模块。
import urllib.parse
import json
import os
import asyncio
import requests
from urllib.parse import urljoin
# 全局配置
host=os.getenv('LINGDA_HOST')
api=os.getenv('FILE_UPLOAD')
url = urljoin(host, api)
def upload_common_files(files, path):
result = []
if len(files) == 0:
return result
files_to_upload = []
for file_data in files:
files_to_upload.append(('files', (file_data.get('name'), file_data.get('data'))))
if len(files_to_upload) == 0:
# No files to upload
return result
response = requests.post(
path,
files=files_to_upload,
data={'returnType': 'url'},
)
# Handle response
if response.status_code == 200:
return eval(response.text)['data']
else:
print('File upload failed')
raise RuntimeError("File upload failed")
def upload_files(files):
print(f"Uploading files... {url}")
return upload_common_files(files, url)
if __name__ == "__main__":
base_dir = os.path.dirname(os.path.abspath(__file__))
# 本地音频文件路径
audio_file_path = os.path.join(base_dir, "中文女.mp3")
# 打开音频文件并读取数据
with open(audio_file_path, "rb") as f:
file_data = {
'name': os.path.basename(audio_file_path),
'data': f.read()
}
# 调用函数上传音频文件
try:
result = upload_files([file_data])
print("Upload succeeded. Return data:", result)
except Exception as e:
print("Upload failed:", str(e))