117 lines
4.2 KiB
Python
117 lines
4.2 KiB
Python
import os
|
||
import requests
|
||
from tkinter import Tk, filedialog
|
||
from typing import List, Dict, Optional
|
||
|
||
|
||
class AudioBatchUploader:
|
||
def __init__(self):
|
||
self.base_url = "http://pms.dtyx.net:9158"
|
||
self.upload_endpoint = "/audio/upload-batch/{imageId}"
|
||
self.image_id = "1f042a710e0a6367f22faedf5efd4116"
|
||
|
||
def select_audios(self) -> List[str]:
|
||
"""打开文件选择对话框，让用户选择多个音频文件"""
|
||
root = Tk()
|
||
root.withdraw() # 隐藏主窗口
|
||
|
||
file_paths = filedialog.askopenfilenames(
|
||
title="选择要上传的音频文件",
|
||
filetypes=[("音频文件", "*.mp3 *.wav *.ogg *.m4a"), ("所有文件", "*.*")]
|
||
)
|
||
|
||
return file_paths
|
||
|
||
def upload_audios(self, file_paths: List[str]) -> Dict:
|
||
"""批量上传音频文件"""
|
||
if not file_paths:
|
||
print("没有选择任何文件!")
|
||
return {"success": False, "msg": "没有选择任何文件"}
|
||
|
||
successful_uploads = 0
|
||
failed_uploads = 0
|
||
results = []
|
||
|
||
for file_path in file_paths:
|
||
try:
|
||
# 准备请求URL和headers
|
||
url = f"{self.base_url}{self.upload_endpoint.format(imageId=self.image_id)}"
|
||
headers = {"Authorization": None} # token为null
|
||
|
||
# 读取文件内容
|
||
with open(file_path, 'rb') as f:
|
||
files = {'file': (os.path.basename(file_path), f)}
|
||
|
||
# 发送请求
|
||
response = requests.post(
|
||
url=url,
|
||
files=files,
|
||
headers=headers
|
||
)
|
||
|
||
# 处理响应
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get("success"):
|
||
print(f"文件 {os.path.basename(file_path)} 上传成功!")
|
||
successful_uploads += 1
|
||
results.extend(result.get("data", []))
|
||
else:
|
||
print(f"文件 {os.path.basename(file_path)} 上传失败: {result.get('msg', '未知错误')}")
|
||
failed_uploads += 1
|
||
else:
|
||
print(f"文件 {os.path.basename(file_path)} 上传失败，HTTP状态码: {response.status_code}")
|
||
failed_uploads += 1
|
||
|
||
except Exception as e:
|
||
print(f"文件 {os.path.basename(file_path)} 上传过程中发生异常: {str(e)}")
|
||
failed_uploads += 1
|
||
|
||
print("\n上传完成!")
|
||
print(f"成功: {successful_uploads} 个, 失败: {failed_uploads} 个")
|
||
|
||
return {
|
||
"success": True,
|
||
"success_count": successful_uploads,
|
||
"fail_count": failed_uploads,
|
||
"results": results
|
||
}
|
||
|
||
def run(self):
|
||
"""运行上传程序"""
|
||
print("=== 音频批量上传工具 ===")
|
||
print(f"目标服务器: {self.base_url}")
|
||
print(f"图片ID: {self.image_id}")
|
||
print("\n请选择要上传的音频文件...")
|
||
|
||
# 选择文件
|
||
audio_files = self.select_audios()
|
||
|
||
if not audio_files:
|
||
print("未选择任何文件，程序退出。")
|
||
return
|
||
|
||
print(f"\n已选择 {len(audio_files)} 个音频文件:")
|
||
for file in audio_files:
|
||
print(f"- {os.path.basename(file)}")
|
||
|
||
# 确认上传
|
||
confirm = input("\n确认上传这些文件吗? (y/n): ").strip().lower()
|
||
if confirm != 'y':
|
||
print("上传已取消。")
|
||
return
|
||
|
||
# 开始上传
|
||
print("\n开始上传...")
|
||
upload_result = self.upload_audios(audio_files)
|
||
|
||
# 显示上传结果详情
|
||
if upload_result.get("success") and upload_result.get("results"):
|
||
print("\n上传结果详情:")
|
||
for result in upload_result["results"]:
|
||
print(f"文件ID: {result.get('id')}, 保存路径: {result.get('filePath')}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
uploader = AudioBatchUploader()
|
||
uploader.run() |