230 lines
8.0 KiB
Python
230 lines
8.0 KiB
Python
"""
|
|
任务：打开飞书客户端，找到"资产管理组"群聊并截图
|
|
|
|
流程：
|
|
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()
|