You've already forked pruchase_jhw
113 lines
4.6 KiB
Python
113 lines
4.6 KiB
Python
import asyncio
|
|
import signal
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# src 为代码 root：项目不打包安装（tool.uv package=false），
|
|
# 此处显式将 src 加入 sys.path，使 `uv run python -m src.main` 与 IDE source-root 行为一致
|
|
_SRC_DIR = str(Path(__file__).resolve().parent)
|
|
if _SRC_DIR not in sys.path:
|
|
sys.path.insert(0, _SRC_DIR)
|
|
|
|
from loguru import logger
|
|
|
|
from src.browser.manager import get_browser_manager
|
|
from src.config.settings import get_settings
|
|
from src.config.accounts import load_accounts_config
|
|
from src.notifications.service import NotificationService
|
|
from src.utils.logger import setup_logger
|
|
from src.task_queue.dispatcher import TaskDispatcher
|
|
from src.task_queue.delay_zset_listener import RedisZSetDelayListener
|
|
from src.task_queue.listener import RedisStreamListener
|
|
from src.task_queue.trade_fetch_listener import TradeFetchStreamListener
|
|
from src.task_queue.redis_manager import get_redis_manager
|
|
from src.api.server import start_api_server
|
|
|
|
async def main() -> None:
|
|
"""
|
|
应用程序主入口函数。
|
|
- 初始化定制的 Loguru 结构化日志系统。
|
|
- 校验 Pydantic 配置及多站点账号 YAML 配置文件是否正确。
|
|
- 针对不同操作系统平台注册优雅退出信号处理器。
|
|
- 启动 Redis Stream 消息队列持续消费监听循环。
|
|
"""
|
|
# 1. 初始化按天分割日志系统
|
|
setup_logger()
|
|
logger.info("正在初始化日本采购自动化守护进程...")
|
|
|
|
# 2. 启动前快速校验：验证全局环境变量设置与账号配置文件
|
|
try:
|
|
settings = get_settings()
|
|
logger.info(f"应用配置加载成功: APP_ENV={settings.app_env.upper()}")
|
|
NotificationService(settings).validate_configuration()
|
|
|
|
# 解析并校验 accounts.yaml
|
|
load_accounts_config()
|
|
await get_redis_manager().initialize()
|
|
except Exception as e:
|
|
logger.critical(f"由于配置错误，守护进程启动失败: {e}")
|
|
sys.exit(1)
|
|
|
|
# 3. 预热浏览器：为所有已配置账号提前启动常驻浏览器上下文，任务到达时即开即用
|
|
await get_browser_manager().prewarm()
|
|
|
|
# 4. 实例化核心任务分发器与 Stream 监听器
|
|
dispatcher = TaskDispatcher()
|
|
stream_listener = RedisStreamListener(dispatcher)
|
|
delay_listener = RedisZSetDelayListener(dispatcher)
|
|
trade_fetch_listener = TradeFetchStreamListener(dispatcher)
|
|
|
|
# 4.1 启动截图查看 API 服务
|
|
api_runner = await start_api_server()
|
|
|
|
# 5. 优雅停机信号处理 (跨平台支持)
|
|
loop = asyncio.get_running_loop()
|
|
|
|
def shutdown_handler(sig_name: str) -> None:
|
|
logger.warning(f"接收到系统退出信号 '{sig_name}'。正在启动优雅停机流程...")
|
|
# 触发监听器停止的协程任务
|
|
asyncio.create_task(stream_listener.stop())
|
|
asyncio.create_task(delay_listener.stop())
|
|
asyncio.create_task(trade_fetch_listener.stop())
|
|
|
|
# Windows 默认的 Proactor 事件循环不支持 add_signal_handler。
|
|
if sys.platform != "win32":
|
|
logger.debug("正在注册 Unix 系统信号监听器 (SIGINT / SIGTERM)。")
|
|
for sig in (signal.SIGINT, signal.SIGTERM):
|
|
try:
|
|
loop.add_signal_handler(sig, lambda s=sig: shutdown_handler(s.name))
|
|
except NotImplementedError:
|
|
pass
|
|
else:
|
|
logger.debug("运行于 Windows 平台。跳过系统信号注册，将直接捕获 KeyboardInterrupt。")
|
|
|
|
# 6. 启动消息监听消费主循环
|
|
try:
|
|
stream_task = asyncio.create_task(stream_listener.listen())
|
|
delay_task = asyncio.create_task(delay_listener.listen())
|
|
trade_fetch_task = asyncio.create_task(trade_fetch_listener.listen())
|
|
await asyncio.gather(stream_task, delay_task, trade_fetch_task)
|
|
except asyncio.CancelledError:
|
|
logger.warning("监听器轮询执行被事件循环取消。")
|
|
except Exception as e:
|
|
logger.exception(f"主守护进程轮询中发生致命崩溃: {e}")
|
|
finally:
|
|
# 保证安全回收与连接断开
|
|
await asyncio.gather(
|
|
stream_listener.stop(),
|
|
delay_listener.stop(),
|
|
trade_fetch_listener.stop(),
|
|
get_browser_manager().close_all(),
|
|
get_redis_manager().close(),
|
|
api_runner.cleanup(),
|
|
return_exceptions=True,
|
|
)
|
|
logger.success("日本采购自动化守护进程已安全优雅退出。")
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
asyncio.run(main())
|
|
except KeyboardInterrupt:
|
|
logger.warning("捕获到键盘手动中断。立即退出程序...")
|
|
sys.exit(0)
|