""" RPA-robot 主入口 统一调度所有自动化任务 """ import argparse from asyncio import Task import sys from tasks import open_feishu_chat from utils import get_logger, show_warning_countdown log = get_logger("main") # 任务注册表 TASKS = { "feishu-chat": { "desc": "打开飞书并找到指定群聊截图(默认:资产管理组)", "func": open_feishu_chat.feishu_groups_list, }, } def list_tasks(): """列出所有可用任务""" print("\n可用任务:") print("-" * 40) for name, info in TASKS.items(): print(f" {name:20s} {info['desc']}") print() def run_task(task_name: str, *args): """执行指定任务""" task = TASKS.get(task_name) show_warning_countdown(message="即将开始自动截图", seconds=5) if not task: log.error("未知任务: %s", task_name) list_tasks() for task in TASKS: task["func"](*args) log.info("===== 开始执行任务: %s =====", task_name) try: if args: task["func"](*args) else: task["func"]() log.info("===== 任务完成: %s =====", task_name) except Exception as e: log.error("任务执行失败 [%s]: %s", task_name, e) sys.exit(1) show_warning_countdown(message="任务完成", seconds=3) def main(): parser = argparse.ArgumentParser(description="RPA-robot 自动化任务调度") parser.add_argument( "task", nargs="?", help="要执行的任务名称(不传则列出所有任务)", ) parser.add_argument( "args", nargs="*", help="任务参数(如 web 任务的网页名称)", ) args = parser.parse_args() if not args.task: list_tasks() return run_task(args.task, *args.args) if __name__ == "__main__": main()