from typing import List, Optional
from pydantic import BaseModel, Field
import json
import re

# ------------------------------------------------------------------------------
# 1. 状态机定义（你核心的“记忆+当前目标”）
# ------------------------------------------------------------------------------
class TaskState(BaseModel):
    goal: str = ""                 # 总目标
    current_step: str = ""         # 正在做哪一步
    steps_history: List[str] = []  # 已做步骤
    completed: bool = False
    failed_reason: str = ""
    env: dict = {}                 # 浏览器/文件/路径等环境信息

# ------------------------------------------------------------------------------
# 2. 上下文管理器（存储、压缩、拼接）
# ------------------------------------------------------------------------------
class ContextManager:
    def __init__(self, max_recent=5, max_rag_tokens=2048, total_budget=14336):
        self.system_prompt = ""
        self.state = TaskState()
        self.history = []          # (role, content)
        self.rag_chunks = []       # 所有文档块
        self.max_recent = max_recent
        self.max_rag_tokens = max_rag_tokens
        self.total_budget = total_budget  # 给16k留安全余量

    def set_system(self, s: str):
        self.system_prompt = s

    def add_user(self, msg: str):
        self.history.append(("user", msg))

    def add_assistant(self, msg: str):
        self.history.append(("assistant", msg))

    def update_state(self, **kwargs):
        for k, v in kwargs.items():
            setattr(self.state, k, v)

    # ------------------------------
    # 历史压缩：只保留最近 N 轮
    # ------------------------------
    def get_recent_history(self):
        return self.history[-self.max_recent:]

    # ------------------------------
    # 极简 RAG：BM25 你可以自己插，这里先给“伪检索”占位
    # ------------------------------
    def add_rag_chunk(self, chunk: str):
        self.rag_chunks.append(chunk)

    def retrieve_topk(self, query: str, k=2):
        # 这里替换成真实 BM25/embedding 检索
        return self.rag_chunks[-k:] if self.rag_chunks else []

    # ------------------------------
    # 最终拼给模型的上下文
    # ------------------------------
    def build_prompt(self):
        state_str = json.dumps(self.state.model_dump(), ensure_ascii=False, indent=2)
        recent = self.get_recent_history()
        rag = self.retrieve_topk(query=self.state.current_step)

        prompt_parts = [
            "[SYSTEM]",
            self.system_prompt,
            "\n[STATE]",
            state_str,
            "\n[HISTORY]",
        ]
        for role, msg in recent:
            prompt_parts.append(f"{role}: {msg}")

        if rag:
            prompt_parts.append("\n[RAG]")
            prompt_parts.extend(rag[:2])

        full = "\n".join(prompt_parts)
        return full[:self.total_budget]  # 硬截断防超16k

# ------------------------------------------------------------------------------
# 3. Agent 主循环（多步 + 状态机驱动）
# ------------------------------------------------------------------------------
class LocalAgent:
    def __init__(self, model_call_func):
        self.ctx = ContextManager()
        self.model_call = model_call_func  # 你填本地模型调用

    def step(self, user_input: str) -> str:
        self.ctx.add_user(user_input)
        prompt = self.ctx.build_prompt()
        # --------------------------
        # 本地模型推理
        # --------------------------
        resp = self.model_call(prompt)
        # --------------------------
        # 尝试解析模型输出里的状态更新（简单版）
        # --------------------------
        self._try_parse_state(resp)
        self.ctx.add_assistant(resp)
        return resp

    def _try_parse_state(self, resp: str):
        # 你可以让模型输出 JSON 块，这里简单示例
        # 实际可加 Pydantic 强制解析
        pass

# ------------------------------------------------------------------------------
# 4. 你只需要实现这个函数：本地模型调用
# ------------------------------------------------------------------------------
def local_16k_model(prompt: str) -> str:
    # 这里替换成 llama.cpp / transformers / vLLM 等
    return "模型回复：" + prompt[:100] + "..."

# ------------------------------------------------------------------------------
# 5. 使用示例
# ------------------------------------------------------------------------------
if __name__ == "__main__":
    agent = LocalAgent(model_call_func=local_16k_model)

    agent.ctx.set_system("""你是自动化助手，严格按状态机执行多步任务。""")
    agent.ctx.update_state(goal="备份豆包历史到本地文件")

    while not agent.ctx.state.completed:
        user_in = input("> ")
        resp = agent.step(user_in)
        print("→", resp)

# 清空剪贴板
subprocess.run(["gtkclipboard", "set", "-t", "text/plain"], input=b"", capture_output=True)

# 读取剪贴板
result = subprocess.run(["gtkclipboard", "get", "-t", "text/plain"], capture_output=True, text=True)
text = result.stdout.strip()

# 在 print_all_ocr_results 函数末尾加这段
# 保存OCR结果到文件
with open("/home/nick/workspace/RPA/chat_history.txt", "w", encoding="utf-8") as f:
    for item in non_script_results:
        f.write(f"{item['text']}\n")
print(f"✅ 对话列表已保存到：chat_history.txt")

import cv2
import easyocr
import numpy as np
import warnings
import subprocess
import time
import os
warnings.filterwarnings("ignore")

# ========== 全局配置（根据你的屏幕/列表调整） ==========
# 对话列表的大致区域（不用精准，OCR会自动识别）
LIST_X1, LIST_X2 = 30, 400    # 列表X范围（缩小到左侧对话区）
LIST_Y1, LIST_Y2 = 80, 1000   # 列表Y范围
OUTPUT_IMG_PATH = "/home/nick/workspace/RPA/real_time_ocr_verify.png"
SCREENSHOT_PATH = "/home/nick/workspace/RPA/real_time_screenshot.png"
DRAG_OFFSET = 50   # 修复：缩小拖拽范围（精准选中单行文字）
COPY_DELAY = 1     # 修复：缩短等待时间（不需要3秒）

# ========== 初始化EasyOCR ==========
reader = easyocr.Reader(
    lang_list=['ch_sim', 'en'],
    gpu=False,
    verbose=False,
    model_storage_directory="/home/nick/.EasyOCR/model"
)

# ========== 修复点1：替代wmctrl（解决找不到命令的问题） ==========
def focus_window_fallback():
    """没有wmctrl时，用xdotool模拟用户切窗口（提示用户配合）"""
    print("\n⚠️  请在3秒内点击Firefox窗口使其获得焦点！")
    time.sleep(3)

# ========== 修复点2：每次拷贝前清空剪贴板 ==========
def clear_clipboard():
    """清空剪贴板，避免旧内容干扰"""
    try:
        subprocess.run(
            ["xclip", "-selection", "clipboard", "-i"],
            input=b"",
            capture_output=True,
            check=True
        )
        time.sleep(0.1)
    except Exception as e:
        print(f"⚠️  清空剪贴板失败：{e}")

# ========== 修复点3：精准拷贝列表项文字（核心修复） ==========
def copy_list_item_text(center_coords):
    """修复拖拽范围+清空剪贴板+精准选中"""
    copied_results = []
    
    # 优先用wmctrl，没有则提示用户
    try:
        subprocess.run(["wmctrl", "-a", "Firefox"], check=True, capture_output=True)
        time.sleep(0.5)
    except FileNotFoundError:
        focus_window_fallback()

    for idx, (x, y) in enumerate(center_coords, 1):
        print(f"\n🔹 处理列表项 {idx}（坐标：{x},{y}）：")
        clear_clipboard()
        
        # 1. 移动到文字左侧（精准定位，避免选到终端）
        subprocess.run(["xdotool", "mousemove", str(x - DRAG_OFFSET), str(y)])
        time.sleep(0.2)
        
        # 2. 拖拽选中（只选单行，范围缩小到50px）
        subprocess.run(["xdotool", "mousedown", "1"])
        time.sleep(0.2)
        subprocess.run(["xdotool", "mousemove", str(x + DRAG_OFFSET), str(y)])
        time.sleep(0.2)
        subprocess.run(["xdotool", "mouseup", "1"])
        time.sleep(0.2)
        
        # 3. Ctrl+C拷贝（确保按键触发）
        subprocess.run(["xdotool", "key", "ctrl+c"])
        time.sleep(0.3)
        
        # 4. 读取剪贴板（增加容错）
        text = get_clipboard_text()
        copied_results.append({
            "index": idx,
            "coords": (x, y),
            "text": text
        })
        
        # 输出结果（区分成功/失败）
        if text and len(text) > 2 and "import cv2" not in text:
            print(f"✅ 拷贝成功：{text[:50]}...")  # 只显示前50字，避免刷屏
        else:
            print(f"❌ 拷贝失败（内容为空/错误）")
    
    return copied_results

# ========== 函数1：实时截屏 ==========
def capture_screen():
    """截取整个屏幕并返回图片"""
    subprocess.run(["scrot", SCREENSHOT_PATH], check=True, capture_output=True)
    img = cv2.imread(SCREENSHOT_PATH)
    if img is None:
        raise Exception("❌ 无法读取截屏图片")
    print(f"✅ 已截取屏幕：{SCREENSHOT_PATH}")
    return img

# ========== 函数2：OCR识别列表坐标（优化过滤规则） ==========
def ocr_list_coordinates(img):
    """识别列表项中心坐标，优化过滤规则"""
    # 裁剪列表区域
    crop_img = img[LIST_Y1:LIST_Y2, LIST_X1:LIST_X2]
    gray = cv2.cvtColor(crop_img, cv2.COLOR_BGR2GRAY)
    
    # OCR识别
    result = reader.readtext(gray, detail=1)
    
    # 过滤无效结果，提取坐标
    list_coords = []
    img_verify = img.copy()
    for (bbox, text, conf) in result:
        text = text.strip()
        # 优化过滤：排除脚本代码关键词+无效文本
        if (all(c == '?' for c in text) 
            or len(text) < 2 
            or text.count('?')/len(text) > 0.5
            or "import cv2" in text
            or "subprocess" in text):
            continue
        # 计算屏幕绝对坐标
        center_x = int((bbox[0][0] + bbox[2][0])/2 + LIST_X1)
        center_y = int((bbox[0][1] + bbox[2][1])/2 + LIST_Y1)
        list_coords.append((center_x, center_y))
        # 可视化标记
        cv2.circle(img_verify, (center_x, center_y), 5, (0,0,255), -1)
        cv2.putText(img_verify, f"({center_x},{center_y})", 
                    (center_x+10, center_y), cv2.FONT_HERSHEY_SIMPLEX, 
                    0.4, (0,0,255), 1)
    
    # 保存验证图
    cv2.imwrite(OUTPUT_IMG_PATH, img_verify)
    print(f"✅ 验证图已保存：{OUTPUT_IMG_PATH}")
    print(f"✅ 识别到 {len(list_coords)} 个有效列表项")
    return list_coords

# ========== 函数3：读取剪贴板（增加异常处理） ==========
def get_clipboard_text():
    """读取系统剪贴板内容（容错版）"""
    try:
        result = subprocess.run(
            ["xclip", "-o", "-selection", "clipboard"],
            capture_output=True,
            text=True,
            check=True,
            timeout=2
        )
        return result.stdout.strip()
    except Exception as e:
        print(f"⚠️  读取剪贴板失败：{e}")
        return ""

# ========== 主函数：串联所有操作 ==========
def main():
    try:
        # 步骤1：安装依赖提示（首次运行用）
        print("========================================")
        print("📌 先安装缺失工具（可选）：")
        print("   sudo apt install wmctrl scrot xclip")
        print("========================================")
        
        # 步骤2：等待用户触发截屏
        print("\n========================================")
        print("📸 按回车键触发实时截屏（确保Firefox在前台）")
        print("========================================")
        input()
        
        # 步骤3：实时截屏
        screen_img = capture_screen()
        
        # 步骤4：OCR识别列表坐标
        list_coords = ocr_list_coordinates(screen_img)
        if not list_coords:
            print("❌ 未识别到任何有效列表项")
            return
        
        # 步骤5：自动拷贝所有列表项文字
        copy_list_item_text(list_coords)
        
        print("\n🎉 所有操作完成！验证图可查看坐标标记。")
        
    except Exception as e:
        print(f"\n❌ 执行失败：{str(e)}")

if __name__ == "__main__":
    main()

import cv2
import easyocr
import numpy as np
import warnings
import subprocess
import time
import os
warnings.filterwarnings("ignore")

# ========== 全局配置（关键：精准框选左侧对话列表区域） ==========
# 请根据你的屏幕调整这4个值（参考验证图里的红色框）
CHAT_LIST_X1 = 10    # 左侧对话列表最左X坐标
CHAT_LIST_X2 = 350   # 左侧对话列表最右X坐标
CHAT_LIST_Y1 = 80    # 左侧对话列表最上Y坐标
CHAT_LIST_Y2 = 950   # 左侧对话列表最下Y坐标

# 文件路径
SCREENSHOT_PATH = "/home/nick/workspace/RPA/chat_screenshot.png"
VERIFY_IMG_PATH = "/home/nick/workspace/RPA/chat_verify.png"
OUTPUT_TXT_PATH = "/home/nick/workspace/RPA/chat_history.txt"

# ========== 初始化EasyOCR（确保中文识别） ==========
reader = easyocr.Reader(
    lang_list=['ch_sim', 'en'],  # 优先识别中文+英文
    gpu=False,
    verbose=False,
    model_storage_directory="/home/nick/.EasyOCR/model"
)

# ========== 步骤1：截取Firefox窗口（而非全屏，减少干扰） ==========
def capture_firefox_window():
    """只截取Firefox窗口，而非全屏，减少终端文本干扰"""
    try:
        # 找到Firefox窗口ID
        firefox_win_id = subprocess.check_output(
            ["xdotool", "search", "--name", "Firefox"],
            text=True, timeout=2
        ).strip().split("\n")[0]
        
        # 截取指定窗口（而非全屏）
        subprocess.run([
            "scrot", "-u", "-b",  # -u：截取活动窗口，-b：包含边框
            SCREENSHOT_PATH,
            "--window", firefox_win_id
        ], check=True, capture_output=True, timeout=5)
        
        img = cv2.imread(SCREENSHOT_PATH)
        if img is None:
            raise Exception("❌ 无法读取Firefox窗口截图")
        print(f"✅ 已截取Firefox窗口：{SCREENSHOT_PATH}")
        return img
    except Exception as e:
        print(f"⚠️  截取Firefox窗口失败，改用全屏截图：{e}")
        # 备用：全屏截图
        subprocess.run(["scrot", SCREENSHOT_PATH], check=True, capture_output=True, timeout=5)
        img = cv2.imread(SCREENSHOT_PATH)
        return img

# ========== 步骤2：裁剪左侧对话列表+OCR识别 ==========
def ocr_chat_list(img):
    """裁剪左侧对话区，OCR识别并过滤有效文字"""
    # 1. 裁剪左侧对话列表区域
    chat_region = img[CHAT_LIST_Y1:CHAT_LIST_Y2, CHAT_LIST_X1:CHAT_LIST_X2]
    gray = cv2.cvtColor(chat_region, cv2.COLOR_BGR2GRAY)
    
    # 2. OCR识别（保留所有细节）
    ocr_results = reader.readtext(gray, detail=1)
    
    # 3. 过滤有效对话文本（核心规则）
    valid_chat_texts = []
    verify_img = img.copy()
    
    for (bbox, text, conf) in ocr_results:
        text = text.strip()
        # 过滤规则：只保留对话标题（排除脚本/乱码/短文本）
        if (len(text) < 3                          # 太短的跳过
            or text.count('?') / len(text) > 0.3   # 乱码问号多的跳过
            or any(keyword in text for keyword in [  # 排除脚本相关文本
                "import", "subprocess", "def ", "cv2", 
                "bash", "xdotool", "xclip", "#!/bin"
            ])):
            continue
        
        # 计算屏幕绝对坐标（用于验证图标记）
        x1 = int(bbox[0][0]) + CHAT_LIST_X1
        y1 = int(bbox[0][1]) + CHAT_LIST_Y1
        x2 = int(bbox[2][0]) + CHAT_LIST_X1
        y2 = int(bbox[2][1]) + CHAT_LIST_Y1
        
        # 保存有效文本
        valid_chat_texts.append(text)
        
        # 验证图标记（红色框+文字）
        cv2.rectangle(verify_img, (x1, y1), (x2, y2), (0, 0, 255), 2)
        cv2.putText(verify_img, text[:10] + "...", (x1, y1-5),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.4, (0, 0, 255), 1)
    
    # 保存验证图
    cv2.imwrite(VERIFY_IMG_PATH, verify_img)
    print(f"✅ 验证图已保存：{VERIFY_IMG_PATH}（红色框为识别的对话）")
    
    return valid_chat_texts

# ========== 步骤3：保存结果到文件 ==========
def save_chat_history(chat_texts):
    """将识别的对话列表保存到文本文件"""
    with open(OUTPUT_TXT_PATH, "w", encoding="utf-8") as f:
        f.write("=== 豆包对话列表 ===\n")
        f.write(f"识别时间：{time.strftime('%Y-%m-%d %H:%M:%S')}\n")
        f.write("====================\n\n")
        for idx, text in enumerate(chat_texts, 1):
            f.write(f"{idx}. {text}\n")
    print(f"✅ 对话列表已保存到：{OUTPUT_TXT_PATH}")
    # 打印结果到终端
    print("\n📋 识别到的对话列表：")
    for idx, text in enumerate(chat_texts, 1):
        print(f"  {idx}. {text}")

# ========== 主函数 ==========
def main():
    try:
        print("========================================")
        print("📌 请确保Firefox窗口在前台（左侧对话列表可见）")
        print("========================================")
        input("按回车键开始识别...")
        
        # 步骤1：截图
        firefox_img = capture_firefox_window()
        if firefox_img is None:
            print("❌ 截图失败")
            return
        
        # 步骤2：OCR识别对话列表
        chat_texts = ocr_chat_list(firefox_img)
        if not chat_texts:
            print("❌ 未识别到有效对话文本，请调整CHAT_LIST_X1/X2/Y1/Y2")
            return
        
        # 步骤3：保存结果
        save_chat_history(chat_texts)
        
        print("\n🎉 识别完成！")
        
    except Exception as e:
        print(f"\n❌ 执行失败：{str(e)}")

if __name__ == "__main__":
    main()

import cv2
import easyocr
import numpy as np
import warnings
import subprocess
import time
import os
warnings.filterwarnings("ignore")

# ========== 全局配置（严格限定左侧对话区） ==========
LIST_X1, LIST_X2 = 20, 320    # 左侧对话列表X范围
LIST_Y1, LIST_Y2 = 120, 900   # 左侧对话列表Y范围
OUTPUT_IMG_PATH = "/home/nick/workspace/RPA/real_time_ocr_verify.png"
SCREENSHOT_PATH = "/home/nick/workspace/RPA/real_time_screenshot.png"

# ========== 初始化EasyOCR ==========
reader = easyocr.Reader(
    lang_list=['ch_sim', 'en'],
    gpu=False,
    verbose=False,
    model_storage_directory="/home/nick/.EasyOCR/model"
)

# ========== 函数1：截屏 ==========
def capture_screen():
    """截取整个屏幕并返回图片"""
    try:
        subprocess.run(["scrot", SCREENSHOT_PATH], check=True, capture_output=True, timeout=5)
        img = cv2.imread(SCREENSHOT_PATH)
        if img is None:
            raise Exception("❌ 无法读取截屏图片")
        print(f"✅ 已截取屏幕：{SCREENSHOT_PATH}")
        return img
    except Exception as e:
        print(f"❌ 截屏失败：{e}")
        return None

# ========== 函数2：打印所有OCR原始结果（核心实验） ==========
def print_all_ocr_results(img):
    """打印OCR识别的所有文字+坐标，验证是否可用"""
    if img is None:
        return []
    
    # 裁剪左侧对话区
    crop_img = img[LIST_Y1:LIST_Y2, LIST_X1:LIST_X2]
    gray = cv2.cvtColor(crop_img, cv2.COLOR_BGR2GRAY)
    
    # OCR识别（保留所有结果，不过滤）
    result = reader.readtext(gray, detail=1)
    
    # 打印所有OCR原始结果（实验关键）
    print("\n========================================")
    print("📝 所有OCR原始识别结果（文字+坐标+置信度）：")
    print("========================================")
    valid_results = []
    for idx, (bbox, text, conf) in enumerate(result, 1):
        # 计算屏幕绝对坐标
        center_x = int((bbox[0][0] + bbox[2][0])/2 + LIST_X1)
        center_y = int((bbox[0][1] + bbox[2][1])/2 + LIST_Y1)
        # 打印完整信息
        print(f"\n【{idx}】")
        print(f"  文字：{text.strip()}")
        print(f"  坐标：({center_x}, {center_y})")
        print(f"  置信度：{conf:.2f}")
        # 收集有效结果（仅过滤空文本）
        if text.strip():
            valid_results.append({
                "index": idx,
                "text": text.strip(),
                "coords": (center_x, center_y),
                "confidence": conf
            })
    
    # 保存验证图
    img_verify = img.copy()
    for item in valid_results:
        x, y = item["coords"]
        cv2.circle(img_verify, (x, y), 5, (0,0,255), -1)
        cv2.putText(img_verify, f"{item['index']}", (x+10, y), 
                    cv2.FONT_HERSHEY_SIMPLEX, 0.4, (0,0,255), 1)
    cv2.imwrite(OUTPUT_IMG_PATH, img_verify)
    print(f"\n✅ 验证图已保存：{OUTPUT_IMG_PATH}（红色数字对应上面的序号）")
    
    return valid_results

# ========== 主函数：只做OCR验证，不碰xdotool/xclip ==========
def main():
    try:
        print("========================================")
        print("📸 按回车键触发实时截屏（确保Firefox窗口在前台）")
        print("========================================")
        input()
        
        # 步骤1：截屏
        screen_img = capture_screen()
        if screen_img is None:
            return
        
        # 步骤2：打印所有OCR结果（核心实验）
        valid_results = print_all_ocr_results(screen_img)
        
        # 步骤3：统计有效结果
        print(f"\n========================================")
        print(f"📊 统计：共识别到 {len(valid_results)} 个文本块")
        # 过滤掉明显是脚本的内容
        non_script_results = [
            item for item in valid_results 
            if not any(keyword in item["text"] for keyword in ["import", "subprocess", "def ", "cv2"])
        ]
        print(f"📊 过滤后有效对话文本：{len(non_script_results)} 个")
        
        # 打印过滤后的对话文本
        if non_script_results:
            print(f"\n✅ 过滤后的有效对话文本：")
            for item in non_script_results:
                print(f"  [{item['index']}] {item['text']}")
        else:
            print(f"\n❌ 未识别到有效对话文本（全是脚本内容）")
        
    except Exception as e:
        print(f"\n❌ 执行失败：{str(e)}")

if __name__ == "__main__":
    main()

import cv2
import easyocr
import numpy as np
import warnings
import subprocess
import time
import os
warnings.filterwarnings("ignore")

# ========== 全局配置 ==========
LIST_X1, LIST_X2 = 20, 320    
LIST_Y1, LIST_Y2 = 120, 900   
OUTPUT_IMG_PATH = "/home/nick/workspace/RPA/real_time_ocr_verify.png"
SCREENSHOT_PATH = "/home/nick/workspace/RPA/real_time_screenshot.png"
DRAG_OFFSET = 40              

# ========== 初始化EasyOCR ==========
reader = easyocr.Reader(
    lang_list=['ch_sim', 'en'],
    gpu=False,
    verbose=False,
    model_storage_directory="/home/nick/.EasyOCR/model"
)

# ========== 新增：获取Firefox窗口ID ==========
def get_firefox_window_id():
    """获取Firefox窗口ID，失败则退出"""
    try:
        result = subprocess.run(
            ["xdotool", "search", "--name", "Firefox"],
            capture_output=True, text=True, check=True, timeout=2
        )
        win_ids = result.stdout.strip().split("\n")
        if win_ids:
            return win_ids[0]  # 取第一个Firefox窗口
        else:
            raise Exception("未找到Firefox窗口")
    except Exception as e:
        print(f"❌ 获取Firefox窗口ID失败：{e}")
        return None

# ========== 带超时的子进程调用 ==========
def run_with_timeout(cmd, timeout=1):
    try:
        return subprocess.run(cmd, capture_output=True, timeout=timeout, check=False)
    except subprocess.TimeoutExpired:
        print(f"⚠️  命令超时：{' '.join(cmd)}")
        return None

# ========== 核心拷贝逻辑（指定窗口ID操作） ==========
def copy_list_item_text(center_coords, firefox_win_id):
    copied_results = []
    if not firefox_win_id:
        return copied_results

    # 激活Firefox窗口
    run_with_timeout(["xdotool", "windowactivate", "--sync", firefox_win_id], timeout=1)
    time.sleep(0.5)

    for idx, (x, y) in enumerate(center_coords, 1):
        print(f"\n🔹 处理列表项 {idx}（坐标：{x},{y}）：")
        
        # 1. 清空剪贴板
        run_with_timeout(["xclip", "-selection", "clipboard", "-i"], input=b"", timeout=0.5)
        time.sleep(0.1)
        
        # 2. 移动鼠标到目标位置（指定Firefox窗口）
        run_with_timeout([
            "xdotool", "mousemove", "--window", firefox_win_id,
            str(x - DRAG_OFFSET), str(y)
        ], timeout=0.5)
        time.sleep(0.1)
        
        # 3. 拖拽选中（指定窗口）
        run_with_timeout([
            "xdotool", "mousedown", "--window", firefox_win_id, "1"
        ], timeout=0.3)
        time.sleep(0.1)
        run_with_timeout([
            "xdotool", "mousemove", "--window", firefox_win_id,
            str(x + DRAG_OFFSET), str(y)
        ], timeout=0.3)
        time.sleep(0.1)
        run_with_timeout([
            "xdotool", "mouseup", "--window", firefox_win_id, "1"
        ], timeout=0.3)
        time.sleep(0.1)
        
        # 4. 按Ctrl+C（指定窗口）
        run_with_timeout([
            "xdotool", "key", "--window", firefox_win_id, "ctrl+c"
        ], timeout=0.5)
        time.sleep(0.2)
        
        # 5. 读取剪贴板
        text = ""
        try:
            result = subprocess.run(
                ["xclip", "-o", "-selection", "clipboard"],
                capture_output=True, text=True, timeout=1
            )
            text = result.stdout.strip()
        except:
            pass
        
        # 6. 输出结果
        if text and len(text) > 3 and "import" not in text:
            print(f"✅ 拷贝成功：{text[:40]}...")
            copied_results.append({"index": idx, "coords": (x,y), "text": text})
        else:
            print(f"❌ 拷贝失败/无效内容")
    
    return copied_results

# ========== 截屏+OCR ==========
def capture_screen():
    run_with_timeout(["scrot", SCREENSHOT_PATH], timeout=2)
    img = cv2.imread(SCREENSHOT_PATH)
    if img is None:
        raise Exception("❌ 截屏失败")
    print("✅ 已截取屏幕")
    return img

def ocr_list_coordinates(img):
    crop_img = img[LIST_Y1:LIST_Y2, LIST_X1:LIST_X2]
    gray = cv2.cvtColor(crop_img, cv2.COLOR_BGR2GRAY)
    result = reader.readtext(gray, detail=1)

    list_coords = []
    img_verify = img.copy()
    for (bbox, text, conf) in result:
        text = text.strip()
        if (len(text) < 3
            or "import" in text
            or "subprocess" in text
            or "def " in text):
            continue
        center_x = int((bbox[0][0] + bbox[2][0])/2 + LIST_X1)
        center_y = int((bbox[0][1] + bbox[2][1])/2 + LIST_Y1)
        list_coords.append((center_x, center_y))
        cv2.circle(img_verify, (center_x, center_y), 5, (0,0,255), -1)

    cv2.imwrite(OUTPUT_IMG_PATH, img_verify)
    print(f"✅ 识别到 {len(list_coords)} 个有效列表项")
    return list_coords

# ========== 主函数 ==========
def main():
    try:
        print("========================================")
        print("📸 按回车键开始（确保Firefox窗口已打开）")
        print("========================================")
        input()

        # 步骤1：获取Firefox窗口ID
        firefox_win_id = get_firefox_window_id()
        if not firefox_win_id:
            return

        # 步骤2：截屏+OCR
        screen_img = capture_screen()
        list_coords = ocr_list_coordinates(screen_img)
        if not list_coords:
            print("❌ 没识别到列表项")
            return

        # 步骤3：精准拷贝
        copy_list_item_text(list_coords, firefox_win_id)
        print("\n🎉 流程结束！")

    except Exception as e:
        print(f"\n❌ 错误：{e}")

if __name__ == "__main__":
    main()

import subprocess
import time
import re

def get_firefox_window_by_title(keyword="豆包"):
    """按标题关键词找Firefox窗口，返回最匹配的ID"""
    try:
        # 获取所有Firefox窗口ID
        result = subprocess.run(
            ["xdotool", "search", "--name", "Firefox"],
            capture_output=True, text=True, timeout=2
        )
        firefox_ids = result.stdout.strip().split("\n")
        
        # 遍历找含关键词的主窗口
        target_id = ""
        for win_id in firefox_ids:
            if not win_id.strip():
                continue
            # 获取窗口标题
            title_result = subprocess.run(
                ["xdotool", "getwindowname", win_id],
                capture_output=True, text=True, timeout=1
            )
            win_title = title_result.stdout.strip()
            # 筛选：标题含关键词 + 长度>10（排除小弹窗）
            if keyword in win_title and len(win_title) > 10:
                target_id = win_id
                print(f"✅ 找到目标窗口：ID={win_id}，标题={win_title}")
                break
        
        return target_id if target_id else None
    except Exception as e:
        print(f"❌ 查找窗口失败：{e}")
        return None

def get_mouse_coords():
    """获取当前鼠标坐标（用于校准书签位置）"""
    result = subprocess.run(
        ["xdotool", "getmouselocation"],
        capture_output=True, text=True
    )
    coords = re.findall(r'x:(\d+) y:(\d+)', result.stdout)
    if coords:
        return int(coords[0][0]), int(coords[0][1])
    return None, None

def execute_bookmark():
    # 1. 校准坐标（先提示用户把鼠标放到目标位置）
    print("📌 请把鼠标放到Firefox右上角菜单按钮，按回车...")
    input()
    menu_x, menu_y = get_mouse_coords()
    if not menu_x:
        print("❌ 无法获取鼠标坐标")
        return
    
    print("📌 请把鼠标放到'豆包备份带进度条'书签，按回车...")
    input()
    bookmark_x, bookmark_y = get_mouse_coords()
    if not bookmark_x:
        print("❌ 无法获取鼠标坐标")
        return
    
    # 2. 找豆包窗口
    target_win_id = get_firefox_window_by_title("豆包")
    if not target_win_id:
        print("❌ 没找到豆包窗口")
        return
    
    # 3. 执行操作
    subprocess.run(["xdotool", "windowactivate", "--sync", target_win_id], timeout=2)
    time.sleep(0.8)
    
    # 点击菜单
    subprocess.run(["xdotool", "mousemove", "--window", target_win_id, str(menu_x), str(menu_y)], timeout=1)
    time.sleep(0.2)
    subprocess.run(["xdotool", "click", "--window", target_win_id, "1"], timeout=1)
    time.sleep(0.5)
    
    # 点击书签
    subprocess.run(["xdotool", "mousemove", "--window", target_win_id, str(bookmark_x), str(bookmark_y)], timeout=1)
    time.sleep(0.2)
    subprocess.run(["xdotool", "click", "--window", target_win_id, "1"], timeout=1)
    time.sleep(2)
    
    print("🎉 操作完成！JS书签已执行")

if __name__ == "__main__":
    execute_bookmark()

from playwright.sync_api import sync_playwright
import time

def extract_doubao_history():
    with sync_playwright() as p:
        browser = p.firefox.launch(headless=False)  # 打开可见的Firefox
        page = browser.new_page()
        page.goto("https://www.doubao.com/chat/38416444134119938")  # 你的对话页
        page.wait_for_load_state("networkidle")
        time.sleep(2)  # 等列表加载

        # 定位左侧历史对话列表（按实际DOM调整选择器）
        history_items = page.query_selector_all("div[class*='history-item'] a")
        results = []
        for item in history_items:
            text = item.text_content().strip()
            link = item.get_attribute("href")
            results.append({"text": text, "link": link})
            print(f"✅ 提取：{text}")

        browser.close()
        return results

if __name__ == "__main__":
    extract_doubao_history()

import subprocess
import time

# ========== 修复点1：自动激活Firefox窗口，不需要用户手动切回 ==========
def focus_firefox():
    """自动将Firefox窗口切到前台并获得焦点"""
    subprocess.run(["wmctrl", "-a", "Firefox"], check=True)
    time.sleep(0.5)

# ========== 修复点2：每次拷贝前清空剪贴板 ==========
def clear_clipboard():
    subprocess.run(["xclip", "-selection", "clipboard", "-i"], input=b"", check=True)
    time.sleep(0.1)

# ========== 修复点3：缩小拖拽范围，精准选中对话列表文字 ==========
def copy_list_item_text(center_coords):
    copied_results = []
    focus_firefox()  # 自动切窗口
    for idx, (x, y) in enumerate(center_coords, 1):
        print(f"\n🔹 处理列表项 {idx}（坐标：{x},{y}）：")
        clear_clipboard()
        # 1. 移动到文字左侧（缩小偏移，只覆盖当前行）
        subprocess.run(["xdotool", "mousemove", str(x - 30), str(y)])
        time.sleep(0.1)
        # 2. 拖拽选中（范围缩小，避免跨行吗）
        subprocess.run(["xdotool", "mousedown", "1"])
        time.sleep(0.1)
        subprocess.run(["xdotool", "mousemove", str(x + 80), str(y)])
        time.sleep(0.1)
        subprocess.run(["xdotool", "mouseup", "1"])
        time.sleep(0.1)
        # 3. Ctrl+C 拷贝
        subprocess.run(["xdotool", "key", "ctrl+c"])
        time.sleep(0.2)
        # 4. 读取结果
        text = get_clipboard_text()
        copied_results.append({"index": idx, "coords": (x,y), "text": text})
        if text:
            print(f"✅ 拷贝成功：{text}")
        else:
            print(f"❌ 拷贝失败")
    return copied_results

import cv2
import easyocr
import numpy as np
import warnings
import subprocess
import time
import os
warnings.filterwarnings("ignore")

# ========== 全局配置（严格限定左侧对话区，避免识别到终端） ==========
LIST_X1, LIST_X2 = 20, 320    # 更窄的X范围，只覆盖左侧对话列表
LIST_Y1, LIST_Y2 = 120, 900   # 避开顶部标题栏和底部输入框
OUTPUT_IMG_PATH = "/home/nick/workspace/RPA/real_time_ocr_verify.png"
SCREENSHOT_PATH = "/home/nick/workspace/RPA/real_time_screenshot.png"
DRAG_OFFSET = 40              # 进一步缩小拖拽范围，避免跨区
COPY_DELAY = 0.5              # 极短等待，减少阻塞

# ========== 初始化EasyOCR ==========
reader = easyocr.Reader(
    lang_list=['ch_sim', 'en'],
    gpu=False,
    verbose=False,
    model_storage_directory="/home/nick/.EasyOCR/model"
)

# ========== 修复卡死：带超时的 subprocess 调用 ==========
def run_with_timeout(cmd, timeout=1):
    """带超时的子进程调用，避免卡死"""
    try:
        return subprocess.run(cmd, capture_output=True, timeout=timeout, check=False)
    except subprocess.TimeoutExpired:
        print(f"⚠️  命令超时：{' '.join(cmd)}")
        return None

# ========== 修复卡死：简化窗口聚焦 ==========
def focus_firefox_simple():
    """极简版聚焦：只提示用户，不阻塞"""
    print("\n⚠️  请确保 Firefox 窗口在前台，2秒后开始操作...")
    time.sleep(2)

# ========== 修复卡死：清空剪贴板（极简版） ==========
def clear_clipboard():
    run_with_timeout(["xclip", "-selection", "clipboard", "-i"], timeout=0.5)
    time.sleep(0.1)

# ========== 修复卡死：核心拷贝逻辑（去掉冗余 sleep） ==========
def copy_list_item_text(center_coords):
    copied_results = []
    focus_firefox_simple()

    for idx, (x, y) in enumerate(center_coords, 3):  # 跳过前2个脚本识别项
        print(f"\n🔹 处理列表项 {idx}（坐标：{x},{y}）：")
        clear_clipboard()

        # 1. 移动鼠标（带超时）
        run_with_timeout(["xdotool", "mousemove", str(x - DRAG_OFFSET), str(y)], timeout=0.5)
        time.sleep(0.1)

        # 2. 拖拽选中（极短时间，避免阻塞）
        run_with_timeout(["xdotool", "mousedown", "1"], timeout=0.3)
        time.sleep(0.1)
        run_with_timeout(["xdotool", "mousemove", str(x + DRAG_OFFSET), str(y)], timeout=0.3)
        time.sleep(0.1)
        run_with_timeout(["xdotool", "mouseup", "1"], timeout=0.3)
        time.sleep(0.1)

        # 3. Ctrl+C 拷贝（带超时）
        run_with_timeout(["xdotool", "key", "ctrl+c"], timeout=0.5)
        time.sleep(0.2)

        # 4. 读剪贴板（带超时）
        text = ""
        try:
            result = subprocess.run(
                ["xclip", "-o", "-selection", "clipboard"],
                capture_output=True, text=True, timeout=1
            )
            text = result.stdout.strip()
        except:
            pass

        copied_results.append({"index": idx, "coords": (x,y), "text": text})

        # 只打印有效结果
        if text and len(text) > 3 and "import" not in text:
            print(f"✅ 成功：{text[:40]}...")
        else:
            print(f"❌ 跳过（无效内容）")
    return copied_results

# ========== 函数1：截屏 ==========
def capture_screen():
    run_with_timeout(["scrot", SCREENSHOT_PATH], timeout=2)
    img = cv2.imread(SCREENSHOT_PATH)
    if img is None:
        raise Exception("❌ 截屏失败")
    print("✅ 已截取屏幕")
    return img

# ========== 函数2：OCR识别（过滤掉前2个脚本识别项） ==========
def ocr_list_coordinates(img):
    crop_img = img[LIST_Y1:LIST_Y2, LIST_X1:LIST_X2]
    gray = cv2.cvtColor(crop_img, cv2.COLOR_BGR2GRAY)
    result = reader.readtext(gray, detail=1)

    list_coords = []
    img_verify = img.copy()
    for (bbox, text, conf) in result:
        text = text.strip()
        # 强过滤：只保留看起来像对话标题的文本
        if (len(text) < 3
            or text.count('?')/len(text) > 0.3
            or "import" in text
            or "subprocess" in text
            or "def " in text):
            continue
        center_x = int((bbox[0][0] + bbox[2][0])/2 + LIST_X1)
        center_y = int((bbox[0][1] + bbox[2][1])/2 + LIST_Y1)
        list_coords.append((center_x, center_y))
        cv2.circle(img_verify, (center_x, center_y), 5, (0,0,255), -1)

    cv2.imwrite(OUTPUT_IMG_PATH, img_verify)
    print(f"✅ 识别到 {len(list_coords)} 个有效列表项")
    return list_coords

# ========== 主函数 ==========
def main():
    try:
        print("========================================")
        print("📸 按回车键开始（确保 Firefox 窗口在最前面）")
        print("========================================")
        input()

        screen_img = capture_screen()
        list_coords = ocr_list_coordinates(screen_img)
        if not list_coords:
            print("❌ 没识别到列表项")
            return

        copy_list_item_text(list_coords)
        print("\n🎉 流程结束！")

    except Exception as e:
        print(f"\n❌ 错误：{e}")

if __name__ == "__main__":
    main()

import subprocess
import time

# ========== 配置（根据你的屏幕调整） ==========
# Firefox 右上角菜单按钮坐标（打开书签面板用）
MENU_X = 930
MENU_Y = 40
# 目标书签坐标（比如"豆包备份带进度条"）
BOOKMARK_X = 250
BOOKMARK_Y = 200

def get_firefox_window_id():
    try:
        result = subprocess.run(
            ["xdotool", "search", "--name", "Firefox"],
            capture_output=True, text=True, check=True, timeout=2
        )
        return result.stdout.strip().split("\n")[0]
    except:
        return None

def run_xdotool(cmd, timeout=1):
    try:
        return subprocess.run(cmd, capture_output=True, timeout=timeout, check=False)
    except subprocess.TimeoutExpired:
        print(f"⚠️  命令超时：{' '.join(cmd)}")
        return None

def execute_bookmark_js():
    # 1. 获取 Firefox 窗口ID
    firefox_win_id = get_firefox_window_id()
    if not firefox_win_id:
        print("❌ 没找到 Firefox 窗口")
        return False

    # 2. 激活窗口
    run_xdotool(["xdotool", "windowactivate", "--sync", firefox_win_id], timeout=2)
    time.sleep(0.8)
    print("✅ 已激活 Firefox 窗口")

    # 3. 点击右上角菜单按钮，打开书签面板
    run_xdotool([
        "xdotool", "mousemove", "--window", firefox_win_id, str(MENU_X), str(MENU_Y)
    ], timeout=0.5)
    time.sleep(0.2)
    run_xdotool(["xdotool", "click", "--window", firefox_win_id, "1"], timeout=0.5)
    time.sleep(0.5)
    print("✅ 已打开书签菜单")

    # 4. 点击目标书签（执行 JS）
    run_xdotool([
        "xdotool", "mousemove", "--window", firefox_win_id, str(BOOKMARK_X), str(BOOKMARK_Y)
    ], timeout=0.5)
    time.sleep(0.2)
    run_xdotool(["xdotool", "click", "--window", firefox_win_id, "1"], timeout=0.5)
    time.sleep(2)
    print("✅ 已点击书签，JS 执行中...")
    return True

if __name__ == "__main__":
    print("========================================")
    print("📌 确保 Firefox 窗口在前台，书签面板可展开")
    print("========================================")
    input("按回车键开始执行...")
    
    if execute_bookmark_js():
        print("\n🎉 操作完成！请检查对话是否已下载。")
    else:
        print("\n❌ 执行失败，请调整坐标或检查 Firefox 窗口。")