478 lines
24 KiB
Python
478 lines
24 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""后台工作线程：执行单个/批量上传任务，与 main.Pdd 交互。"""
|
||
|
||
import os
|
||
import time
|
||
from pathlib import Path
|
||
|
||
from PyQt5.QtCore import QThread, pyqtSignal
|
||
|
||
from main import Pdd
|
||
from loguru import logger
|
||
|
||
from gui_constants import get_default_folder_path, VIDEO_EXTENSIONS
|
||
|
||
|
||
class WorkerThread(QThread):
|
||
"""工作线程，用于执行自动化任务"""
|
||
finished = pyqtSignal(bool, str)
|
||
log_message = pyqtSignal(str)
|
||
progress = pyqtSignal(int)
|
||
item_result = pyqtSignal(object) # 单条发布结果（dict）
|
||
|
||
def __init__(self, config_data, is_batch_mode, prepared_files=None, parent=None):
|
||
super().__init__(parent)
|
||
self.config_data = config_data
|
||
self.is_batch_mode = is_batch_mode
|
||
self.prepared_files = prepared_files # 预查找的文件列表
|
||
self.is_running = True
|
||
|
||
def run(self):
|
||
sink_id = None
|
||
start_ts = time.time()
|
||
try:
|
||
sink_id = logger.add(lambda msg: self.log_message.emit(str(msg).strip()))
|
||
if self.is_batch_mode:
|
||
self.run_batch_mode()
|
||
else:
|
||
self.run_single_mode()
|
||
except Exception as e:
|
||
error_msg = f"执行失败: {str(e)}"
|
||
self.finished.emit(False, error_msg)
|
||
self.log_message.emit(error_msg)
|
||
logger.error(f"执行失败: {e}")
|
||
import traceback
|
||
traceback_str = traceback.format_exc()
|
||
logger.error(traceback_str)
|
||
self.log_message.emit(f"错误详情: {traceback_str}")
|
||
finally:
|
||
try:
|
||
cost = time.time() - start_ts
|
||
self.log_message.emit(f"[线程] 任务结束，耗时 {cost:.2f}s")
|
||
except Exception:
|
||
pass
|
||
if sink_id is not None:
|
||
try:
|
||
logger.remove(sink_id)
|
||
except Exception:
|
||
pass
|
||
|
||
def run_single_mode(self):
|
||
"""执行单个上传模式（action方法）"""
|
||
try:
|
||
config = self.config_data
|
||
self.log_message.emit(f"开始处理单个任务（逐个上传模式）...")
|
||
|
||
pdd = Pdd(
|
||
url=config.get('达人链接', ''),
|
||
user_id=config.get('多多id', ''),
|
||
time_start=config.get('定时发布', '') if config.get('定时发布') else None,
|
||
ht=config.get('话题', ''),
|
||
index=config.get('序号', ''),
|
||
title=config.get('标题', None)
|
||
)
|
||
|
||
if self.prepared_files:
|
||
self.log_message.emit(f"使用预查找的文件列表，共 {len(self.prepared_files)} 个文件")
|
||
self.progress.emit(30)
|
||
|
||
# 将预查找的文件按类型分类，直接传递给 action()，跳过文件夹扫描
|
||
prepared_video_files = [f['path'] for f in self.prepared_files
|
||
if f['path'].is_file() and any(
|
||
f['path'].suffix.lower() == ext for ext in VIDEO_EXTENSIONS)]
|
||
prepared_image_folders = [f['path'] for f in self.prepared_files
|
||
if f['path'].is_dir()]
|
||
|
||
self.log_message.emit(f"预查找文件: {len(prepared_video_files)} 个视频, {len(prepared_image_folders)} 个图片文件夹")
|
||
try:
|
||
result = pdd.action(
|
||
prepared_video_files=prepared_video_files,
|
||
prepared_image_folders=prepared_image_folders,
|
||
collect_all_videos=False
|
||
)
|
||
ok = bool(result.get("ok")) if isinstance(result, dict) else True
|
||
if not ok:
|
||
self.log_message.emit(f"失败原因: {result.get('reason') if isinstance(result, dict) else ''}")
|
||
if isinstance(result, dict):
|
||
self.item_result.emit({
|
||
"user_id": config.get("多多id", ""),
|
||
"index": config.get("序号", ""),
|
||
"name": config.get("标题", "") or "",
|
||
"ok": ok,
|
||
"reason": result.get("reason", ""),
|
||
})
|
||
self.progress.emit(100)
|
||
self.finished.emit(ok, f"单个任务执行完成（ok={ok}）")
|
||
except Exception as e:
|
||
error_msg = f"执行action方法失败: {str(e)}"
|
||
self.log_message.emit(error_msg)
|
||
logger.error(error_msg)
|
||
import traceback
|
||
traceback_str = traceback.format_exc()
|
||
logger.error(traceback_str)
|
||
self.log_message.emit(f"错误详情: {traceback_str}")
|
||
self.finished.emit(False, error_msg)
|
||
return
|
||
else:
|
||
folder_path = config.get('文件夹路径', '').strip()
|
||
if not folder_path:
|
||
folder_path = get_default_folder_path()
|
||
|
||
if not os.path.exists(folder_path):
|
||
error_msg = f"文件夹路径不存在: {folder_path}"
|
||
self.finished.emit(False, error_msg)
|
||
self.log_message.emit(error_msg)
|
||
return
|
||
|
||
self.log_message.emit(f"使用文件夹路径: {folder_path}")
|
||
self.progress.emit(30)
|
||
is_batch_mode = self.is_batch_mode
|
||
|
||
try:
|
||
result = pdd.action(folder_path=folder_path, collect_all_videos=is_batch_mode)
|
||
ok = bool(result.get("ok")) if isinstance(result, dict) else True
|
||
if not ok:
|
||
self.log_message.emit(f"失败原因: {result.get('reason') if isinstance(result, dict) else ''}")
|
||
if isinstance(result, dict):
|
||
self.item_result.emit({
|
||
"user_id": config.get("多多id", ""),
|
||
"index": config.get("序号", ""),
|
||
"name": config.get("标题", "") or "",
|
||
"ok": ok,
|
||
"reason": result.get("reason", ""),
|
||
})
|
||
except Exception as e:
|
||
error_msg = f"执行action方法失败: {str(e)}"
|
||
self.log_message.emit(error_msg)
|
||
logger.error(error_msg)
|
||
import traceback
|
||
traceback_str = traceback.format_exc()
|
||
logger.error(traceback_str)
|
||
self.log_message.emit(f"错误详情: {traceback_str}")
|
||
self.finished.emit(False, error_msg)
|
||
return
|
||
|
||
self.progress.emit(100)
|
||
self.finished.emit(ok, f"单个任务执行完成（ok={ok}）")
|
||
except Exception as e:
|
||
error_msg = f"执行失败: {str(e)}"
|
||
self.finished.emit(False, error_msg)
|
||
self.log_message.emit(error_msg)
|
||
logger.error(f"执行失败: {e}")
|
||
import traceback
|
||
traceback_str = traceback.format_exc()
|
||
logger.error(traceback_str)
|
||
self.log_message.emit(f"错误详情: {traceback_str}")
|
||
|
||
def run_batch_mode(self):
|
||
"""执行批量上传模式：先批量上传所有视频，然后逐个上传图片"""
|
||
try:
|
||
config = self.config_data
|
||
self.log_message.emit(f"开始处理批量上传任务...")
|
||
index = config.get('序号', '')
|
||
|
||
pdd = Pdd(
|
||
url=config.get('达人链接', ''),
|
||
user_id=config.get('多多id', ''),
|
||
time_start=config.get('定时发布', '') if config.get('定时发布') else None,
|
||
ht=config.get('话题', ''),
|
||
index=index,
|
||
title=config.get('标题', None)
|
||
)
|
||
|
||
if self.prepared_files:
|
||
self.log_message.emit(f"使用预查找的文件列表，共 {len(self.prepared_files)} 个文件")
|
||
self.progress.emit(30)
|
||
|
||
video_file_paths = [f for f in self.prepared_files if f['path'].is_file() and any(
|
||
f['path'].suffix.lower() == ext for ext in VIDEO_EXTENSIONS)]
|
||
image_folders = [f for f in self.prepared_files if f['path'].is_dir()]
|
||
|
||
if video_file_paths:
|
||
self.log_message.emit(f"找到 {len(video_file_paths)} 个视频文件，开始批量上传...")
|
||
self.progress.emit(30)
|
||
|
||
def on_video_done(result_item):
|
||
try:
|
||
payload = dict(result_item)
|
||
payload["user_id"] = config.get("多多id", "")
|
||
if "index" not in payload or not payload.get("index"):
|
||
for vid_file in video_file_paths:
|
||
if str(vid_file.get('path')) == str(payload.get('path')):
|
||
payload["index"] = vid_file.get("index", config.get("序号", ""))
|
||
break
|
||
if "index" not in payload or not payload.get("index"):
|
||
payload["index"] = config.get("序号", "")
|
||
self.item_result.emit(payload)
|
||
self.log_message.emit(f" {'✓' if payload.get('ok') else '✗'} {payload.get('name', '')} 已更新状态")
|
||
except Exception as e:
|
||
self.log_message.emit(f" 回调处理异常: {e}")
|
||
|
||
result = pdd.action1(folder_path=video_file_paths, on_item_done=on_video_done)
|
||
ok = bool(result.get("ok")) if isinstance(result, dict) else True
|
||
if not ok:
|
||
fails = [r for r in (result.get("results") or []) if not r.get("ok")]
|
||
self.log_message.emit(f"发布校验失败条数: {len(fails)}")
|
||
for r in fails[:20]:
|
||
self.log_message.emit(f" ✗ {r.get('name')} | {r.get('reason')}")
|
||
self.log_message.emit(f"批量上传视频完成，共处理 {len(video_file_paths)} 个视频")
|
||
else:
|
||
self.log_message.emit("未找到视频文件，跳过视频上传")
|
||
ok = True
|
||
|
||
if image_folders:
|
||
self.log_message.emit(f"找到 {len(image_folders)} 个图片文件夹，开始逐个上传...")
|
||
self.progress.emit(70)
|
||
|
||
for idx, image_folder_info in enumerate(image_folders):
|
||
if not self.is_running:
|
||
break
|
||
image_folder_path = image_folder_info['path']
|
||
self.log_message.emit(f"处理第 {idx + 1}/{len(image_folders)} 个图片文件夹: {image_folder_path}")
|
||
# 直接传递图片文件夹路径，跳过文件夹扫描
|
||
pdd.action(prepared_image_folders=[image_folder_path])
|
||
self.log_message.emit(f"图片文件夹 {idx + 1} 上传完成")
|
||
|
||
self.log_message.emit(f"所有图片文件夹上传完成，共处理 {len(image_folders)} 个文件夹")
|
||
else:
|
||
self.log_message.emit("未找到图片文件夹，跳过图片上传")
|
||
|
||
self.progress.emit(100)
|
||
self.finished.emit(ok,
|
||
f"批量任务执行完成，共处理 {len(video_file_paths)} 个视频和 {len(image_folders)} 个图片文件夹（ok={ok}）")
|
||
else:
|
||
folder_path = config.get('文件夹路径', '').strip()
|
||
if not folder_path:
|
||
folder_path = get_default_folder_path()
|
||
|
||
if not os.path.exists(folder_path):
|
||
self.finished.emit(False, f"文件夹路径不存在: {folder_path}")
|
||
return
|
||
|
||
self.log_message.emit("第一步：收集所有视频文件...")
|
||
video_file_paths = self.prepare_batch_files(folder_path, config)
|
||
self.log_message.emit("第二步：收集所有图片文件夹...")
|
||
image_folders = self.prepare_image_folders(folder_path, index)
|
||
|
||
if video_file_paths:
|
||
self.log_message.emit(f"找到 {len(video_file_paths)} 个视频文件，开始批量上传...")
|
||
self.progress.emit(30)
|
||
pdd.action1(folder_path=video_file_paths)
|
||
self.log_message.emit(f"批量上传视频完成，共处理 {len(video_file_paths)} 个视频")
|
||
else:
|
||
self.log_message.emit("未找到视频文件，跳过视频上传")
|
||
|
||
if image_folders:
|
||
self.log_message.emit(f"找到 {len(image_folders)} 个图片文件夹，开始逐个上传...")
|
||
self.progress.emit(70)
|
||
|
||
for idx, image_folder in enumerate(image_folders):
|
||
if not self.is_running:
|
||
break
|
||
self.log_message.emit(f"处理第 {idx + 1}/{len(image_folders)} 个图片文件夹: {image_folder}")
|
||
# 直接传递图片文件夹路径，跳过文件夹扫描
|
||
pdd.action(prepared_image_folders=[Path(image_folder)])
|
||
self.log_message.emit(f"图片文件夹 {idx + 1} 上传完成")
|
||
|
||
self.log_message.emit(f"所有图片文件夹上传完成，共处理 {len(image_folders)} 个文件夹")
|
||
else:
|
||
self.log_message.emit("未找到图片文件夹，跳过图片上传")
|
||
|
||
self.progress.emit(100)
|
||
total_count = len(video_file_paths) + len(image_folders)
|
||
self.finished.emit(True,
|
||
f"批量任务执行完成，共处理 {len(video_file_paths)} 个视频和 {len(image_folders)} 个图片文件夹")
|
||
|
||
except Exception as e:
|
||
self.finished.emit(False, f"批量执行失败: {str(e)}")
|
||
logger.error(f"批量执行失败: {e}")
|
||
|
||
def prepare_image_folders(self, folder_path, index):
|
||
"""收集所有匹配序号的图片文件夹（与main.py中action方法的逻辑一致）"""
|
||
image_folders = []
|
||
try:
|
||
if not folder_path or not os.path.exists(folder_path):
|
||
logger.warning(f"文件夹路径无效或不存在: {folder_path}")
|
||
return image_folders
|
||
if not os.access(folder_path, os.R_OK):
|
||
logger.error(f"没有权限读取文件夹: {folder_path}")
|
||
return image_folders
|
||
try:
|
||
subdirs = os.listdir(folder_path)
|
||
except PermissionError:
|
||
logger.error(f"没有权限访问文件夹: {folder_path}")
|
||
return image_folders
|
||
except Exception as e:
|
||
logger.error(f"读取文件夹失败: {e}")
|
||
return image_folders
|
||
|
||
for file in subdirs:
|
||
try:
|
||
file_path = os.path.join(folder_path, file)
|
||
if not os.path.isdir(file_path):
|
||
continue
|
||
try:
|
||
files = os.listdir(file_path)
|
||
except (PermissionError, Exception) as e:
|
||
logger.warning(f"读取子文件夹失败: {file_path}, 错误: {e}")
|
||
continue
|
||
|
||
for file_name in files:
|
||
try:
|
||
if file_name.startswith('.'):
|
||
continue
|
||
file_names = file_name.split("-")
|
||
if len(file_names) > 0 and file_names[0] == str(index):
|
||
path = Path(os.path.join(file_path, file_name))
|
||
if path.exists() and path.is_dir():
|
||
image_folders.append(str(path))
|
||
except Exception as e:
|
||
logger.warning(f"处理文件 {file_name} 时出错: {e}")
|
||
continue
|
||
except Exception as e:
|
||
logger.warning(f"处理子文件夹 {file} 时出错: {e}")
|
||
continue
|
||
except Exception as e:
|
||
logger.error(f"收集图片文件夹失败: {e}", exc_info=True)
|
||
return image_folders
|
||
|
||
def count_videos_in_folder(self, folder_path, index):
|
||
"""统计文件夹中匹配序号的视频文件数量（与main.py逻辑一致）"""
|
||
count = 0
|
||
try:
|
||
for file in os.listdir(folder_path):
|
||
file_path = os.path.join(folder_path, file)
|
||
if not os.path.isdir(file_path):
|
||
continue
|
||
files = os.listdir(file_path)
|
||
for file_name in files:
|
||
if ".mp4" in file_name:
|
||
file_names = file_name.split("-")
|
||
if len(file_names) > 0 and file_names[0] == str(index):
|
||
path = Path(os.path.join(file_path, file_name))
|
||
if path.is_file():
|
||
count += 1
|
||
else:
|
||
for sub_file in os.listdir(path):
|
||
sub_path = Path(os.path.join(path, sub_file))
|
||
if sub_path.is_file():
|
||
count += 1
|
||
except Exception as e:
|
||
logger.error(f"统计视频文件失败: {e}")
|
||
return count
|
||
|
||
def prepare_batch_files(self, folder_path, config):
|
||
"""准备批量上传的文件列表（与main.py中的action1方法逻辑一致）"""
|
||
file_paths = []
|
||
if not config:
|
||
logger.error("配置为空，无法准备批量上传文件")
|
||
return file_paths
|
||
index = str(config.get('序号', '')).strip()
|
||
if not index:
|
||
logger.warning("序号为空，无法准备批量上传文件")
|
||
return file_paths
|
||
if not folder_path or not os.path.exists(folder_path):
|
||
logger.error(f"文件夹路径无效或不存在: {folder_path}")
|
||
return file_paths
|
||
if not os.access(folder_path, os.R_OK):
|
||
logger.error(f"没有权限读取文件夹: {folder_path}")
|
||
return file_paths
|
||
|
||
logger.info("=" * 50)
|
||
logger.info("开始准备批量上传文件列表...")
|
||
logger.info(f"文件夹路径: {folder_path}")
|
||
logger.info(f"查找序号: {index}")
|
||
|
||
try:
|
||
try:
|
||
all_items = os.listdir(folder_path)
|
||
subdirs = [f for f in all_items if os.path.isdir(os.path.join(folder_path, f))]
|
||
except PermissionError:
|
||
logger.error(f"没有权限访问文件夹: {folder_path}")
|
||
return file_paths
|
||
except Exception as e:
|
||
logger.error(f"读取文件夹失败: {e}")
|
||
return file_paths
|
||
|
||
logger.info(f"在最外层文件夹下找到 {len(subdirs)} 个子文件夹（多多ID文件夹）")
|
||
|
||
for subdir_name in subdirs:
|
||
file_path = os.path.join(folder_path, subdir_name)
|
||
logger.info(f" 正在扫描子文件夹: {subdir_name}")
|
||
if not os.path.isdir(file_path):
|
||
logger.info(f" 跳过（不是文件夹）")
|
||
continue
|
||
files = os.listdir(file_path)
|
||
logger.info(f" 该文件夹下有 {len(files)} 个项目")
|
||
|
||
for file_name in files:
|
||
logger.info(f" 检查项目: {file_name}")
|
||
if ".mp4" in file_name:
|
||
logger.info(f" ✓ 是视频文件（包含.mp4）")
|
||
file_names = file_name.split("-")
|
||
logger.info(f" 文件名分割结果: {file_names}")
|
||
|
||
if len(file_names) > 0 and file_names[0] == str(index):
|
||
logger.info(f" ✓ 序号匹配！序号: {file_names[0]}, 目标序号: {index}")
|
||
path = Path(os.path.join(file_path, file_name))
|
||
|
||
if path.is_file():
|
||
logger.info(f" ✓ 是文件，添加到列表: {path.name}")
|
||
file_paths.append({
|
||
"url": config.get('达人链接', ''),
|
||
"user_id": config.get('多多id', ''),
|
||
"time_start": config.get('定时发布', '') if config.get('定时发布') else None,
|
||
"ht": config.get('话题', ''),
|
||
"index": str(index),
|
||
"path": path
|
||
})
|
||
logger.info(f" 当前视频总数: {len(file_paths)}")
|
||
else:
|
||
logger.info(f" 是文件夹，扫描其中的文件...")
|
||
try:
|
||
sub_files = os.listdir(path)
|
||
logger.info(f" 文件夹中有 {len(sub_files)} 个文件")
|
||
for sub_file in sub_files:
|
||
sub_path = Path(os.path.join(path, sub_file))
|
||
if sub_path.is_file():
|
||
logger.info(f" ✓ 添加文件: {sub_file}")
|
||
file_paths.append({
|
||
"url": config.get('达人链接', ''),
|
||
"user_id": config.get('多多id', ''),
|
||
"time_start": config.get('定时发布', '') if config.get('定时发布') else None,
|
||
"ht": config.get('话题', ''),
|
||
"index": str(index),
|
||
"path": sub_path
|
||
})
|
||
logger.info(f" 当前视频总数: {len(file_paths)}")
|
||
except Exception as e:
|
||
logger.error(f" 扫描子文件夹失败: {e}")
|
||
else:
|
||
if len(file_names) > 0:
|
||
logger.info(f" ✗ 序号不匹配: {file_names[0]} != {index}")
|
||
else:
|
||
logger.info(f" ✗ 文件名格式不正确，无法提取序号")
|
||
else:
|
||
logger.info(f" ✗ 不是视频文件（不包含.mp4），跳过")
|
||
except Exception as e:
|
||
logger.error(f"准备批量文件失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
logger.info("=" * 50)
|
||
logger.info(f"文件收集完成！共找到 {len(file_paths)} 个视频文件")
|
||
if file_paths:
|
||
logger.info("视频文件列表:")
|
||
for idx, video_info in enumerate(file_paths, 1):
|
||
logger.info(f" {idx}. {video_info['path'].name} ({video_info['path']})")
|
||
return file_paths
|
||
|
||
def stop(self):
|
||
"""停止执行"""
|
||
self.is_running = False
|
||
|
||
def requestInterruption(self):
|
||
"""请求中断线程（重写QThread方法）"""
|
||
super().requestInterruption()
|
||
self.is_running = False
|