import asyncio
|
import json
|
from pathlib import Path
|
from datetime import datetime, timedelta
|
from auto_media_publisher.config.conf_base import TASK_CONFIG_DIR,COOKIE_DIR,PLATFORM_LOGIN_CONCURRENT_COUNT,DATABASE_PATH
|
from auto_media_publisher.utils.logger import get_logger
|
from auto_media_publisher.utils.login_helper import create_and_run_login
|
from auto_media_publisher.utils.utils_network import check_proxy_valid
|
from auto_media_publisher.utils.utils_database import platform_operation_fail_insert
|
# ----------------- 任务生成器 -----------------
|
|
class PlatformLoginChecker:
|
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("PlatformLoginChecker started.")
|
|
def _check_need_login(self, user,platform,max_session_hour):
|
|
cookie_root = Path(COOKIE_DIR) # 把 COOKIE_DIR 变成 Path 对象，如果它是字符串的话
|
user_cookie_path = cookie_root / user / f"{platform}.json"
|
|
if not user_cookie_path.exists():
|
return False
|
|
mtime = user_cookie_path.stat().st_mtime
|
last_modified = datetime.fromtimestamp(mtime)
|
|
now = datetime.now()
|
# 计算时间差
|
delta = now - last_modified
|
|
# 如果超过指定小时数，返回 True
|
return delta > timedelta(hours=max_session_hour)
|
|
def _get_avalible_users(self):
|
all_need_logins = []
|
for config_file in self.config_dir_path.rglob("*.json"):
|
try:
|
user_config = json.loads(config_file.read_text(encoding="utf-8"))
|
if not user_config.get("enabled", True):
|
self.logger.info(f"跳过未启用用户：{config_file.name}")
|
continue
|
|
user = user_config.get("user")
|
proxy = user_config.get("proxy")
|
|
platforms = user_config.get("platforms", [])
|
|
for platform in platforms:
|
if not platform.get("enabled", True):
|
continue
|
|
platform_name = platform["name"]
|
max_session_hour = platform["max_session_hour"]
|
|
if self._check_need_login(user,platform_name,max_session_hour):
|
if check_proxy_valid(proxy):
|
all_need_logins.append({
|
"user":user,
|
"proxy":proxy,
|
"platform":platform_name
|
})
|
else:
|
platform_operation_fail_insert(user,platform,proxy,"proxy","")
|
self.logger.error(f"代理端口不可用，Proxy Port NOT valid。 {user} : {proxy}")
|
except Exception as e:
|
self.logger.warning(f"无法读取配置文件 {config_file.name}：{e}")
|
continue
|
return all_need_logins
|
|
async def _async_run(self):
|
need_logins = self._get_avalible_users()
|
self.logger.info(f"[PlatformLoginChecker] 待登录的用户/平台数量: {len(need_logins)}")
|
|
if not need_logins:
|
self.logger.info("[PlatformLoginChecker] 无任务，自动退出。")
|
return
|
|
login_task_queue = asyncio.Queue()
|
for row in need_logins:
|
await login_task_queue.put(row)
|
|
semaphore = asyncio.Semaphore(PLATFORM_LOGIN_CONCURRENT_COUNT)
|
|
async def worker_wrapper(task_row):
|
async with semaphore:
|
try:
|
user =task_row["user"]
|
platform = task_row["platform"]
|
proxy = task_row["proxy"]
|
|
if check_proxy_valid(proxy):
|
login_ok = await create_and_run_login(user,platform,proxy)
|
if not login_ok:
|
platform_operation_fail_insert(user,platform,proxy,"login_auth","")
|
else:
|
platform_operation_fail_insert(user,platform,proxy,"proxy fail","")
|
|
except Exception as e:
|
self.logger.error(f"[PlatformLoginChecker] 任务处理异常: {e}")
|
|
# 创建所有任务
|
tasks = []
|
while not login_task_queue.empty():
|
row = await login_task_queue.get()
|
tasks.append(asyncio.create_task(worker_wrapper(row)))
|
|
# 等待所有任务完成
|
await asyncio.gather(*tasks)
|
|
def run(self):
|
asyncio.run(self._async_run())
|
