import os
|
import threading
|
from auto_media_publisher.utils.uploader_helper import create_and_run_uploader
|
from auto_media_publisher.config.conf_base import COOKIE_DIR,VIDEO_DIR_FINAL,MAX_VIDEO_FILES
|
from auto_media_publisher.utils.logger import PLATFORM_LOGGER
|
from auto_media_publisher.utils.utils_file import load_json
|
from auto_media_publisher.proxy.proxy_manager import get_user_proxy
|
from auto_media_publisher.task.task_generator import UserManager
|
import asyncio
|
user_manager = UserManager()
|
|
def find_pending_medias(user_dir):
|
video_json_paths = []
|
for root, dirs, files in os.walk(user_dir):
|
for file in files:
|
if file.endswith(".json") and not file.endswith(".json.ok"):
|
json_path = os.path.join(root, file)
|
if os.path.exists(json_path):
|
video_json_paths.append(json_path)
|
if len(video_json_paths) >= MAX_VIDEO_FILES:
|
return video_json_paths
|
return video_json_paths
|
|
async def upload_user_medias(user):
|
PLATFORM_LOGGER.info(f"[THREAD] Starting upload for user: {user}")
|
cookie_path = os.path.join(COOKIE_DIR, user)
|
video_dir_path = os.path.join(VIDEO_DIR_FINAL, user)
|
|
#platforms = [f.split(".")[0] for f in os.listdir(cookie_path) if f.endswith(".json")]
|
platforms = user_manager.get_user_platforms(user)
|
PLATFORM_LOGGER.info(f"[THREAD] user {user} has {len(platforms)} platforms . which is/are:{platforms}")
|
video_jsons = find_pending_medias(video_dir_path)
|
proxy = get_user_proxy(user)
|
|
for video_json_path in video_jsons:
|
video_json_data = load_json(video_json_path)
|
for platform in platforms:
|
PLATFORM_LOGGER.info(f"[INFO] User {user} -> Platform {platform}, uploading {len(video_jsons)} medias")
|
await create_and_run_uploader(platform=platform,json_data=video_json_data,proxy=proxy)
|
video_json_data["platform"] = platform
|
user_manager.insert_upload_logs_from_json(video_json_data)
|
os.rename(video_json_path, video_json_path + ".ok")
|
|
|
|
def start_user_threads():
|
users = user_manager.get_enabled_users()
|
PLATFORM_LOGGER.info(f"[INFO] Found users: {users}")
|
|
threads = []
|
|
def run_async_upload(user):
|
loop = asyncio.new_event_loop()
|
asyncio.set_event_loop(loop)
|
try:
|
loop.run_until_complete(upload_user_medias(user))
|
finally:
|
loop.close()
|
|
for user in users:
|
thread = threading.Thread(target=run_async_upload, args=(user,))
|
thread.start()
|
threads.append(thread)
|
|
# 🔴 等待所有线程完成再退出主进程
|
for thread in threads:
|
thread.join()
|
|
|
if __name__ == "__main__":
|
start_user_threads()
|
