import asyncio
|
import os
|
from datetime import datetime
|
from pathlib import Path
|
from auto_media_publisher.config.conf_base import TASK_CONFIG_DIR,COOKIE_DIR,PLATFORM_UPLOAD_CONCURRENT_COUNT,DATABASE_PATH,VIDEO_DIR_FINAL
|
from auto_media_publisher.utils.logger import get_logger
|
from auto_media_publisher.utils.uploader_helper import create_media_uploader
|
from auto_media_publisher.utils.utils_network import check_proxy_valid
|
from auto_media_publisher.utils.utils_file import load_json
|
from auto_media_publisher.utils.utils_database import platform_operation_fail_insert,update_task_upload_done,get_platform_upload_tasks
|
|
# ----------------- 任务生成器 -----------------
|
|
class PlatformUploaderChecker:
|
def __init__(self):
|
self.config_dir_path = Path(TASK_CONFIG_DIR)
|
self.cookie_dir_path = Path(COOKIE_DIR)
|
self.logger = get_logger(self.__class__.__name__,"admin")
|
self.logger.info("PlatformUploaderChecker started.")
|
|
async def _async_run(self):
|
need_upload_users = get_platform_upload_tasks()
|
self.logger.info(f"[PlatformUploaderChecker] 待上传媒体的用户数量: {len(need_upload_users)}")
|
|
if not need_upload_users:
|
self.logger.info("[PlatformUploaderChecker] 无需要上传媒体的用户，自动退出。")
|
return
|
|
upload_task_queue = asyncio.Queue()
|
for row in need_upload_users:
|
await upload_task_queue.put(row)
|
|
semaphore = asyncio.Semaphore(PLATFORM_UPLOAD_CONCURRENT_COUNT)
|
|
async def worker_wrapper(task_row):
|
async with semaphore:
|
try:
|
user =task_row["user"]
|
platform = task_row["platform"]
|
proxy = task_row["proxy"]
|
print(f"using proxy :{proxy}")
|
proxy_valid, ip, city = check_proxy_valid(proxy)
|
if proxy_valid:
|
handler = await create_media_uploader(user,platform,proxy)
|
scheduled_times = task_row["scheduled_times"]
|
for index,scheduled_time in enumerate(scheduled_times):
|
task_id = scheduled_time["id"]
|
scheduled_time = scheduled_time["scheduled_time"]
|
json_data = load_json(os.path.join(VIDEO_DIR_FINAL,user,f"{scheduled_time}.json"))
|
if not json_data:
|
platform_operation_fail_insert(user,platform,proxy,f"{scheduled_time}.json_not_exist","")
|
continue
|
if index ==len(scheduled_times) -1:
|
tmp_result = await handler.upload_and_close(json_data)
|
else:
|
tmp_result = await handler.upload(json_data)
|
|
if tmp_result:
|
update_task_upload_done(task_id,datetime.now().strftime("%Y-%m-%d %H:%M:%S"),f"{ip}#{city}")
|
else:
|
platform_operation_fail_insert(user,platform,proxy,f"{scheduled_time} upload fail",f"{ip}#{city}")
|
else:
|
platform_operation_fail_insert(user,platform,proxy,"proxy fail","")
|
self.logger.error(f"代理端口不可用，Proxy Port NOT valid。 {user} : {proxy}")
|
|
except Exception as e:
|
self.logger.error(f"[PlatformUploaderChecker] 任务处理异常: {e}")
|
|
# 创建所有任务
|
tasks = []
|
while not upload_task_queue.empty():
|
row = await upload_task_queue.get()
|
tasks.append(asyncio.create_task(worker_wrapper(row)))
|
|
# 等待所有任务完成
|
await asyncio.gather(*tasks)
|
|
def run(self):
|
asyncio.run(self._async_run())
|
