commit 4bdbf31b3e088cfa2477fec30ebbb02419a92691 Author: songyuchao Date: Wed Aug 26 16:14:47 2026 +0800 first commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..e68e6ff --- /dev/null +++ b/.gitignore @@ -0,0 +1,32 @@ +# Python +__pycache__/ +*.py[cod] +*.egg-info/ +dist/ +build/ + +# Virtual environment +.venv/ +venv/ +env/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo + +# RPA Framework output +output/ +results/ + +# OS +.DS_Store +Thumbs.db + +# Data (sensitive) +data/*.xlsx +data/*.csv +data/*.json +!data/.gitkeep +output diff --git a/README.md b/README.md new file mode 100644 index 0000000..b9a778d --- /dev/null +++ b/README.md @@ -0,0 +1,49 @@ +# RPA-robot + +基于 [RPA Framework](https://rpaframework.org/) 的机器人流程自动化项目。 + +## 项目结构 + +``` +RPA-robot/ +├── tasks/ # Robot 任务文件 +│ ├── main.robot # 主任务入口 +│ └── example_task.robot # 示例任务 +├── resources/ # 可复用的资源文件 +│ ├── common.resource # 通用关键字 +│ └── variables.py # 变量定义 +├── libraries/ # 自定义 Python 库 +│ └── __init__.py +├── data/ # 测试/业务数据 +├── output/ # 运行输出(日志、报告) +├── requirements.txt # Python 依赖 +└── README.md +└── main.py #运行主文件 +└──config.py # 配置主文件 +``` + +## 环境准备 + +```bash +# 创建虚拟环境 +conda env create -f environment.yaml +conda activate rpa-robot +rfbrowser init + +## 运行任务 + +```bash +python .\main.py feishu-chat +``` + +## 输出说明 + +运行后会在 `output/` 目录生成: +- `report.html` - 测试报告 +- `log.html` - 详细日志 +- `output.xml` - 结构化输出 + +## 配置说明 +```bash +config.py 配置文件中有 FEISHU_GROUPS = [] #配置飞书群名 +``` \ No newline at end of file diff --git a/config.py b/config.py new file mode 100644 index 0000000..4455fd5 --- /dev/null +++ b/config.py @@ -0,0 +1,24 @@ +""" +项目全局配置 +""" +from dataclasses import Field +import os + +# ============ 飞书配置 ============ +FEISHU_DESKTOP_PATH = r"C:\Users\songy\AppData\Local\Feishu\app\Feishu.exe" # 飞书客户端路径(按实际修改) +FEISHU_WEB_URL = "https://www.feishu.cn/messenger/" # 飞书网页版 +FEISHU_GROUPS = ["资产管理组", "鲜生活信息支持沟通群(仅系统研发)","信息支持中台沟通群","参盘与总部人力费用沟通群"] + +# ============ 微信配置 ============ +WECHAT_DESKTOP_PATH = r"C:\Program Files (x86)\Tencent\WeChat\WeChat.exe" # 微信客户端路径(按实际修改) + +# ============ 常用网页 ============ +WEB_LINKS = { + "百度": "https://www.baidu.com", + "GitHub": "https://github.com", + "Google": "https://www.google.com", +} + +# ============ 输出配置 ============ +OUTPUT_DIR = os.path.join(os.path.dirname(__file__), "output") +os.makedirs(OUTPUT_DIR, exist_ok=True) diff --git a/environment.yaml b/environment.yaml new file mode 100644 index 0000000..6b82e74 --- /dev/null +++ b/environment.yaml @@ -0,0 +1,15 @@ +name: rpa-robot + +channels: + - conda-forge + - defaults + +dependencies: + - python=3.10 + - pip + - pip: + - rpaframework + - robotframework-browser + - pywinauto + - pyautogui + - pygetwindow diff --git a/main.py b/main.py new file mode 100644 index 0000000..29335a7 --- /dev/null +++ b/main.py @@ -0,0 +1,78 @@ +""" +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() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..f812398 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +rpaframework diff --git a/tasks/open_feishu_chat.py b/tasks/open_feishu_chat.py new file mode 100644 index 0000000..7f1d6c3 --- /dev/null +++ b/tasks/open_feishu_chat.py @@ -0,0 +1,229 @@ +""" +任务:打开飞书客户端,找到"资产管理组"群聊并截图 + +流程: +1. 启动飞书桌面客户端(或连接已运行的实例) +2. 等待窗口加载并激活 +3. 使用搜索功能找到"资产管理组"群聊 +4. 点击进入群聊 +5. 截取群聊页面 +6. 保存今日全部聊天记录 +""" +from ast import main +import os +import subprocess +import time +from datetime import datetime + +import pygetwindow as gw +from pywinauto import Application + +from config import FEISHU_DESKTOP_PATH, OUTPUT_DIR, FEISHU_GROUPS +from utils import get_logger +from utils.screenshot import take_window_screenshot + +log = get_logger("feishu_chat") + +# 飞书窗口标题可能的关键词 +FEISHU_WINDOW_TITLES = ["飞书", "Lark"] + + +def _find_feishu_window(): + """查找已运行的飞书窗口,返回窗口标题""" + for keyword in FEISHU_WINDOW_TITLES: + windows = gw.getWindowsWithTitle(keyword) + if windows: + return windows[0].title + return None + + +def _connect_feishu(max_retries: int = 6, interval: int = 5): + """ + 连接飞书窗口,带重试机制 + + Args: + max_retries: 最大重试次数 + interval: 每次重试间隔(秒) + """ + for attempt in range(1, max_retries + 1): + # 先检查飞书窗口是否已出现 + win_title = _find_feishu_window() + if win_title: + log.info("检测到飞书窗口: '%s'(第 %d 次尝试)", win_title, attempt) + app = Application(backend="uia").connect( + title=win_title, timeout=10 + ) + main_win = app.window(title=win_title) + main_win.wait("visible", timeout=10) + main_win.set_focus() + # 全屏显示 + main_win.maximize() + # // 全屏显示 + main_win.fullscreen = True + return app, main_win + + log.info("等待飞书窗口出现...(第 %d/%d 次)", attempt, max_retries) + time.sleep(interval) + + raise TimeoutError( + f"等待飞书窗口超时({max_retries * interval}秒)," + "请确认飞书客户端已正确安装且路径正确: " + FEISHU_DESKTOP_PATH + ) + + +def save_today_chat_history(app, chat_name: str) -> str: + """ + 通过滚动截图保存今日全部聊天记录 + + 从当前聊天位置向上滚动,逐屏截图, + 直到检测到今日日期分隔线或达到最大滚动次数。 + + Args: + app: pywinauto Application 对象 + chat_name: 群聊名称 + + Returns: + 截图保存目录路径 + """ + today_str = datetime.now().strftime("%Y-%m-%d") + # 按日期创建子目录,如 output/20260826/ + output_dir = os.path.join(OUTPUT_DIR, datetime.now().strftime("%Y%m%d")) + os.makedirs(output_dir, exist_ok=True) + + import pyautogui + + try: + # 直接通过 pygetwindow 查找聊天窗口(避免 pywinauto 多窗口匹配问题) + candidates = ( + gw.getWindowsWithTitle(chat_name) + or gw.getWindowsWithTitle("飞书") + or gw.getWindowsWithTitle("Lark") + ) + if not candidates: + raise ValueError(f"未找到飞书或群聊窗口: {chat_name}") + + win = candidates[0] + # 激活窗口到前台(最小化时先还原) + try: + if win.isMinimized: + win.restore() + win.activate() + except Exception: + log.warning("窗口激活失败,尝试继续截图") + time.sleep(0.5) + win_bbox = (win.left, win.top, win.width, win.height) + + max_scrolls = 30 # 最大滚动次数 + scroll_pause = 0.8 # 每次滚动后等待时间 + shot_index = 1 + + log.info("开始滚动截图,最多 %d 屏...", max_scrolls) + log.info("先滚动到聊天框最底部...") + scroll_to_bottom_times = 30 + for _ in range(scroll_to_bottom_times): + center_x = win.left + win.width - 100 # 聊天区域右侧中间位置 + center_y = win.top + win.height // 2 + pyautogui.moveTo(center_x, center_y) + pyautogui.scroll(-500000) # 向下滚动(负数表示向下,显示最新消息) + time.sleep(scroll_pause) + + for i in range(max_scrolls): + # 截图 + shot_name = f"{chat_name}_{today_str}_{shot_index:03d}" + shot_path = os.path.join(output_dir, f"{shot_name}.png") + screenshot = pyautogui.screenshot(region=win_bbox) + screenshot.save(shot_path) + log.info("截图 %d/%d: %s", shot_index, max_scrolls, shot_path) + shot_index += 1 + + # 向上滚动(鼠标滚轮向上滚动加载历史消息) + # 将鼠标移到聊天区域右侧 + center_x = win.left + win.width - 100 + center_y = win.top + win.height // 2 + # // 滚动 + pyautogui.moveTo(center_x, center_y) + pyautogui.scroll(500) # 向上滚动 + time.sleep(scroll_pause) + + # 检测是否已到达顶部(通过检查是否有日期分隔线) + # 简单策略:如果连续滚动后内容不再变化则停止 + # 这里用固定次数,实际可根据图像相似度判断 + + log.info("滚动截图完成,共 %d 张,保存在: %s", shot_index - 1, output_dir) + return output_dir + + except Exception as e: + log.error("滚动截图失败: %s", e) + return output_dir + +def feishu_groups_list(): + log.info("获取飞书群组") + # 任务开始前弹出桌面警告倒计时 + for chat_name in FEISHU_GROUPS: + open_feishu_and_find_chat(chat_name=chat_name) + log.info(f"获取群组列表完成{chat_name}") +def open_feishu_and_find_chat( + chat_name: str = "资产管理组", + screenshot_name: str = None, + save_chat_history: bool = True, +): + """ + 打开飞书客户端,找到指定群聊并截图 + + Args: + chat_name: 群聊名称,默认"资产管理组" + screenshot_name: 截图文件名(不含扩展名),默认为时间戳 + save_chat_history: 是否保存今日聊天记录 + """ + log.info("===== 开始任务:打开飞书并查找群聊 '%s' =====", chat_name) + + # 1. 检查飞书是否已在运行 + existing_title = _find_feishu_window() + if existing_title: + log.info("步骤 1: 飞书已在运行,直接连接") + else: + log.info("步骤 1: 启动飞书客户端") + try: + subprocess.Popen(FEISHU_DESKTOP_PATH) + except FileNotFoundError: + log.error("未找到飞书客户端,请检查路径: %s", FEISHU_DESKTOP_PATH) + raise + + # 2. 连接飞书窗口(带重试) + log.info("步骤 2: 等待飞书窗口加载...") + app, main_win = _connect_feishu() + log.info("飞书窗口已激活") + + # 3. 打开搜索(飞书支持 Ctrl+K 快捷键打开搜索) + log.info("步骤 3: 打开搜索框") + main_win.type_keys("^k") # Ctrl+K + time.sleep(1.5) + + # 4. 输入群聊名称 + log.info("步骤 4: 搜索 '%s'", chat_name) + # 先清空搜索框 + main_win.type_keys("^a") + time.sleep(0.3) + main_win.type_keys(chat_name, with_spaces=True) + time.sleep(2) # 等待搜索结果 + + # 5. 按回车选中第一个搜索结果 + log.info("步骤 5: 选中搜索结果") + main_win.type_keys("{ENTER}") + time.sleep(2) # 等待群聊页面加载 + + # 6. 保存今日聊天记录 + if save_chat_history: + log.info("步骤 6: 保存今日聊天记录") + chat_file = save_today_chat_history(app, chat_name) + log.info("聊天记录已保存: %s", chat_file) + + log.info("===== 任务完成 =====") + + +if __name__ == "__main__": + # 任务开始前弹出桌面警告倒计时 + from utils import show_warning_countdown + + show_warning_countdown(message="即将开始自动截图", seconds=5) + open_feishu_and_find_chat() diff --git a/utils/__init__.py b/utils/__init__.py new file mode 100644 index 0000000..c1568e1 --- /dev/null +++ b/utils/__init__.py @@ -0,0 +1,29 @@ +""" +通用工具模块 +""" +import logging +from datetime import datetime + +def get_logger(name: str) -> logging.Logger: + """创建仅输出到控制台的 logger""" + logger = logging.getLogger(name) + if not logger.handlers: + logger.setLevel(logging.INFO) + fmt = logging.Formatter("[%(asctime)s] %(name)s %(levelname)s - %(message)s") + + sh = logging.StreamHandler() + sh.setFormatter(fmt) + logger.addHandler(sh) + + return logger + + +def now_str(fmt: str = "%Y%m%d%H%M") -> str: + """返回当前时间字符串,用于文件命名""" + return datetime.now().strftime(fmt) + + +# 注意:该导入必须放在 get_logger/now_str 定义之后—— +# notify 模块内部通过 from . import get_logger 引用本包, +# 若放在定义之前会形成循环导入(utils → notify → utils) +from .notify import show_warning_countdown diff --git a/utils/notify.py b/utils/notify.py new file mode 100644 index 0000000..37d5ad0 --- /dev/null +++ b/utils/notify.py @@ -0,0 +1,99 @@ +""" +桌面提醒工具模块 + +在任务开始前于桌面弹出置顶警告窗口并倒计时, +倒计时结束后窗口自动关闭,程序继续执行 +""" +import tkinter as tk +from . import get_logger + + +log = get_logger("notify") + + +def show_warning_countdown( + message: str = "即将开始自动截图", + seconds: int = 5, + title: str = "警告", +): + """ + 在桌面弹出置顶警告窗口并倒计时,结束后自动关闭 + + 倒计时期间函数阻塞,结束后窗口销毁并返回,任务继续执行。 + + Args: + message: 警告正文 + seconds: 倒计时秒数(<=0 时直接返回) + title: 警告标题 + """ + if seconds <= 0: + return + + try: + root = tk.Tk() + except Exception as e: + # tkinter 初始化失败(如无显示环境)时仅记录日志,不阻塞任务 + log.warning("无法弹出桌面警告窗口: %s", e) + return + + root.title(title) + root.attributes("-topmost", True) # 始终置顶,确保用户能看到 + root.resizable(False, False) + + bg = "#c62828" # 红色警示背景 + root.configure(bg=bg) + + # 标题 + tk.Label( + root, + text=f"—— {title} ——", + font=("Microsoft YaHei UI", 32, "bold"), + fg="white", + bg=bg, + ).pack(padx=80, pady=(30, 10)) + + # 正文 + tk.Label( + root, + text=message, + font=("Microsoft YaHei UI", 20), + fg="white", + bg=bg, + ).pack(padx=80, pady=5) + + # 倒计时(每秒刷新) + countdown = tk.Label( + root, + text="", + font=("Microsoft YaHei UI", 48, "bold"), + fg="#ffeb3b", + bg=bg, + ) + countdown.pack(padx=80, pady=15) + + # 底部提示 + tk.Label( + root, + text="倒计时结束后任务自动开始,请勿操作鼠标键盘", + font=("Microsoft YaHei UI", 12), + fg="#ffebee", + bg=bg, + ).pack(padx=80, pady=(0, 30)) + + # 窗口居中显示 + root.update_idletasks() + w, h = root.winfo_width(), root.winfo_height() + sw, sh = root.winfo_screenwidth(), root.winfo_screenheight() + root.geometry(f"+{(sw - w) // 2}+{(sh - h) // 2}") + + def _tick(remaining: int): + if remaining > 0: + countdown.config(text=f"{remaining} 秒后开始") + root.after(1000, _tick, remaining - 1) + else: + root.destroy() + + log.info("桌面警告已弹出: '%s:%s',倒计时 %d 秒", title, message, seconds) + root.after(100, _tick, seconds) + root.mainloop() + log.info("倒计时结束,窗口已关闭,任务继续") diff --git a/utils/screenshot.py b/utils/screenshot.py new file mode 100644 index 0000000..1846d02 --- /dev/null +++ b/utils/screenshot.py @@ -0,0 +1,207 @@ +""" +截图工具模块 + +提供多种截图方式: +1. rpaframework 内置截图(浏览器/桌面) +2. pyautogui 全屏/区域截图 +3. pillow 窗口截图 +""" +import os +from typing import Optional, Tuple + +from RPA.Browser import Playwright +from config import OUTPUT_DIR +from utils import get_logger, now_str + +log = get_logger("screenshot") + +try: + import pyautogui + import pygetwindow as gw +except ImportError: + pyautogui = None + gw = None + + +def take_screenshot( + name: Optional[str] = None, + full_page: bool = False, + region: Optional[Tuple[int, int, int, int]] = None, +) -> str: + """ + 截图工具 + + Args: + name: 截图文件名(不含扩展名),默认为时间戳 + full_page: 是否截取整个页面(仅浏览器有效) + region: 区域截图 (left, top, width, height),None为全屏 + + Returns: + 截图文件完整路径 + """ + if name is None: + name = now_str() + + # 确保输出目录存在 + os.makedirs(OUTPUT_DIR, exist_ok=True) + filepath = os.path.join(OUTPUT_DIR, f"{name}.png") + + # 优先尝试 rpaframework 浏览器截图 + try: + browser = Playwright() + if browser.get_browser_count() > 0: + # 如果有浏览器实例,则截取当前页面 + if full_page: + browser.take_screenshot(filepath, selector="body") + else: + browser.take_screenshot(filepath) + log.info("已使用 rpaframework 截取浏览器页面: %s", filepath) + return filepath + except: + # 没有浏览器实例,继续尝试其他方式 + pass + + # 尝试 pyautogui 截图 + if pyautogui: + try: + if region: + # 区域截图 + screenshot = pyautogui.screenshot(region=region) + else: + # 全屏截图 + screenshot = pyautogui.screenshot() + + screenshot.save(filepath) + log.info("已使用 pyautogui 截取屏幕: %s", filepath) + return filepath + except Exception as e: + log.warning("pyautogui 截图失败: %s", e) + + # 尝试截取活动窗口 + if gw: + try: + active_window = gw.getActiveWindow() + if active_window: + bbox = ( + active_window.left, + active_window.top, + active_window.width, + active_window.height, + ) + screenshot = pyautogui.screenshot(region=bbox) + filepath = os.path.join(OUTPUT_DIR, f"{name}_window.png") + screenshot.save(filepath) + log.info("已截取活动窗口: %s", filepath) + return filepath + except Exception as e: + log.warning("窗口截图失败: %s", e) + + # 最后尝试 Pillow 截图(需要安装 Pillow) + try: + from PIL import ImageGrab + + if region: + bbox = (region[0], region[1], region[0] + region[2], region[1] + region[3]) + img = ImageGrab.grab(bbox=bbox) + else: + img = ImageGrab.grab() + + img.save(filepath) + log.info("已使用 Pillow 截取屏幕: %s", filepath) + return filepath + except ImportError: + log.error("未安装 Pillow,无法进行屏幕截图") + raise + except Exception as e: + log.error("所有截图方式均失败: %s", e) + raise + + +def take_element_screenshot(selector: str, name: Optional[str] = None) -> str: + """ + 截取浏览器中特定元素 + + Args: + selector: CSS 选择器或 XPath + name: 截图文件名 + + Returns: + 截图文件完整路径 + """ + if name is None: + name = f"element_{now_str()}" + + filepath = os.path.join(OUTPUT_DIR, f"{name}.png") + + try: + browser = Playwright() + if browser.get_browser_count() == 0: + raise RuntimeError("没有活动的浏览器实例") + + browser.take_screenshot(filepath, selector=selector) + log.info("已截取元素 '%s': %s", selector, filepath) + return filepath + except Exception as e: + log.error("元素截图失败: %s", e) + raise + + +def take_window_screenshot(window_title: str, name: Optional[str] = None) -> str: + """ + 截取指定窗口 + + Args: + window_title: 窗口标题(模糊匹配) + name: 截图文件名 + + Returns: + 截图文件完整路径 + """ + if name is None: + name = f"window_{now_str()}" + + if not gw: + raise RuntimeError("未安装 pygetwindow,无法截取窗口") + + try: + # 查找匹配的窗口 + windows = gw.getWindowsWithTitle(window_title) + if not windows: + raise ValueError(f"未找到标题包含 '{window_title}' 的窗口") + + # 使用第一个匹配的窗口 + target_window = windows[0] + bbox = ( + target_window.left, + target_window.top, + target_window.width, + target_window.height, + ) + + # 确保输出目录存在 + os.makedirs(OUTPUT_DIR, exist_ok=True) + filepath = os.path.join(OUTPUT_DIR, f"{name}.png") + + # 截取窗口 + screenshot = pyautogui.screenshot(region=bbox) + screenshot.save(filepath) + + log.info("已截取窗口 '%s': %s", window_title, filepath) + return filepath + except Exception as e: + log.error("窗口截图失败: %s", e) + raise + + +if __name__ == "__main__": + # 示例:全屏截图 + take_screenshot("full_screen_demo") + + # 示例:区域截图 (left, top, width, height) + # take_screenshot("region_demo", region=(100, 100, 500, 400)) + + # 示例:窗口截图 + # try: + # take_window_screenshot("微信", "wechat_window_demo") + # except: + # print("未找到微信窗口,跳过窗口截图示例")