916 lines
39 KiB
Python
916 lines
39 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""微信值守 Agent：盯屏 → 识别新消息 → GLM-5.3-Flash 决策回复 → GUI 发送。
|
||
模式: observe=只记录 | draft=拟回复不发送 | auto=自动发送(仅白名单)
|
||
无 API key 时自动降级为 observe。触摸 /opt/wechat-agent/PAUSE 可暂停。
|
||
"""
|
||
import json, os, sys, time, base64, logging, subprocess, re
|
||
from datetime import datetime, timedelta
|
||
from PIL import Image
|
||
|
||
BASE = os.path.dirname(os.path.abspath(__file__))
|
||
CFG = json.load(open(os.path.join(BASE, "config.json"), encoding="utf-8"))
|
||
ENV = dict(os.environ, DISPLAY=CFG["display"], HOME=os.environ.get("HOME", "/root"))
|
||
# HOME 必须有:systemd 服务环境默认不带 HOME,拉起的微信会因写不了 ~/.xwechat 秒退
|
||
PAUSE = os.path.join(BASE, "PAUSE")
|
||
INBOX = os.path.join(BASE, "inbox.log")
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s %(levelname)s %(message)s",
|
||
handlers=[logging.FileHandler(os.path.join(BASE, "agent.log"), encoding="utf-8"),
|
||
logging.StreamHandler()])
|
||
log = logging.getLogger("agent")
|
||
|
||
def sh(cmd, timeout=20):
|
||
return subprocess.run(cmd, shell=True, capture_output=True, timeout=timeout, env=ENV)
|
||
|
||
def scrot(path="/tmp/agent_shot.png"):
|
||
sh(f"scrot -o {path}")
|
||
return Image.open(path).convert("RGB")
|
||
|
||
LAST_RESTART = 0.0
|
||
|
||
def _weixin_width():
|
||
out = sh("wmctrl -lG").stdout.decode(errors="ignore")
|
||
for line in out.splitlines():
|
||
if "Weixin" in line:
|
||
try:
|
||
return int(line.split()[4]) # lG 格式: id desk x y W H name
|
||
except (IndexError, ValueError):
|
||
return 0
|
||
return 0
|
||
|
||
def _center_bright(img):
|
||
"""窗口中心区亮度占比。亮=有内容渲染(主窗口或登录窗),暗=透明壳/无窗口"""
|
||
region = img.crop((550, 300, 900, 620))
|
||
n = tot = 0
|
||
for px in region.getdata():
|
||
r, g, b = px[:3]
|
||
tot += 1
|
||
if (r + g + b) / 3 > 110:
|
||
n += 1
|
||
return n / max(tot, 1)
|
||
|
||
def _login_button(img):
|
||
"""登录窗的实心 Log In 按钮(~170x33px)。判据:单行连续绿像素>50。
|
||
'Scan to log in'绿字每行只有零散笔画(≤30),实心按钮中心行约75+,可靠区分。"""
|
||
px = img.load()
|
||
max_cnt = 0
|
||
for yy in range(470, 580, 2):
|
||
cnt = 0
|
||
for xx in range(620, 820, 2):
|
||
r, g, b = px[xx, yy][:3]
|
||
if r < 80 and g > 160 and 60 < b < 140:
|
||
cnt += 1
|
||
max_cnt = max(max_cnt, cnt)
|
||
return max_cnt > 50
|
||
|
||
def _launch_app():
|
||
"""硬重启:只用于进程死亡或窗口透明壳。进程活着但窗口未映射是登录/启动
|
||
过渡态,绝不能走到这里——杀进程会毁掉刚登录的会话(实战踩坑)。"""
|
||
global LAST_RESTART
|
||
if time.time() - LAST_RESTART < 90: # 冷却,防止连环重启
|
||
return False
|
||
LAST_RESTART = time.time()
|
||
log.warning("重启微信进程(渲染失效或进程消失)")
|
||
sh("pkill -x wechat")
|
||
time.sleep(3)
|
||
subprocess.Popen("nohup /usr/bin/wechat >/dev/null 2>&1 &",
|
||
shell=True, env=ENV,
|
||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
||
start_new_session=True)
|
||
time.sleep(12)
|
||
_activate_all()
|
||
return True
|
||
|
||
def _activate_all():
|
||
ids = sh("xdotool search --class wechat").stdout.decode().split()
|
||
for line in ids:
|
||
sh(f"wmctrl -i -a {hex(int(line))}") # 激活=自动取消最小化/托盘隐藏
|
||
time.sleep(0.3)
|
||
if "Weixin" in sh("wmctrl -l").stdout.decode(errors="ignore"):
|
||
break
|
||
|
||
def ensure_window():
|
||
"""确保微信主窗口可用。返回 True=主窗口就绪(已最大化)。
|
||
自愈链:show-desktop清除 → 无条件激活 → 透明壳检测重启进程 → 登录窗自动点
|
||
Log In → 手机确认期记日志等待。注意:微信隐藏到托盘后 wmctrl -l 仍列出它,
|
||
不能用'列表里没有'当恢复条件。"""
|
||
sh("wmctrl -k off")
|
||
if not sh("xdotool search --class wechat").stdout.decode().split():
|
||
_launch_app()
|
||
_activate_all()
|
||
for attempt in range(3):
|
||
w = _weixin_width()
|
||
img = scrot()
|
||
bright = _center_bright(img)
|
||
if w >= 700: # 主窗口(最大化或默认尺寸)
|
||
if bright > 0.4:
|
||
sh("wmctrl -r Weixin -b add,maximized_vert,maximized_horz")
|
||
return True
|
||
if not _launch_app(): # 透明壳:渲染死了,重启进程
|
||
return False
|
||
elif w > 0:
|
||
# 小窗=登录/二维码/手机确认。只要还亮着就绝不重启进程——
|
||
# 重启会刷新二维码,用户永远扫不上(实测踩坑)。
|
||
if _login_button(img):
|
||
log.info("检测到登录窗,自动点击 Log In")
|
||
click(720, 540, 8.0)
|
||
_activate_all()
|
||
elif bright > 0.3:
|
||
log_inbox("微信等待扫码或手机确认登录")
|
||
log.warning("等待扫码/手机确认登录")
|
||
return False
|
||
elif not _launch_app():
|
||
return False
|
||
else:
|
||
# 无窗口:进程活着=启动/登录过渡态,只等不杀(杀=毁掉登录会话)
|
||
if sh("pgrep -x wechat").stdout.decode().split():
|
||
log.info("微信窗口未映射,等待过渡完成")
|
||
time.sleep(8)
|
||
_activate_all()
|
||
elif not _launch_app():
|
||
return False
|
||
return False
|
||
|
||
def click(x, y, wait=0.6):
|
||
sh(f"xdotool mousemove {x} {y} click 1")
|
||
time.sleep(wait)
|
||
|
||
def press(key, wait=0.5):
|
||
sh(f"xdotool key {key}")
|
||
time.sleep(wait)
|
||
|
||
def paste(text):
|
||
with open("/tmp/agent_clip.txt", "w", encoding="utf-8") as f:
|
||
f.write(text)
|
||
subprocess.Popen("xclip -selection clipboard < /tmp/agent_clip.txt",
|
||
shell=True, env=ENV,
|
||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
||
start_new_session=True)
|
||
time.sleep(0.5)
|
||
press("ctrl+v", 0.6)
|
||
|
||
def red_score(img, slot_i):
|
||
"""某个聊天位号的红点像素数(新消息指示)"""
|
||
c = CFG["coords"]
|
||
y = c["slot_y0"] + slot_i * c["slot_dy"]
|
||
box = (c["badge_x0"], y - 25, c["badge_x1"], y + 25)
|
||
region = img.crop(box)
|
||
n = 0
|
||
for px in region.getdata():
|
||
r, g, b = px[:3]
|
||
if r > 195 and g < 95 and b < 95:
|
||
n += 1
|
||
return n
|
||
|
||
def list_region(img):
|
||
c = CFG["coords"]
|
||
return img.crop((c["list_x0"], c["list_y0"], c["list_x1"], c["list_y1"]))
|
||
|
||
def img_hash(img):
|
||
small = img.resize((24, 64))
|
||
return hash(tuple(v // 24 for p in small.getdata() for v in p[:3]))
|
||
|
||
def to_b64(img):
|
||
img.save("/tmp/agent_b64.png")
|
||
return base64.b64encode(open("/tmp/agent_b64.png", "rb").read()).decode()
|
||
|
||
# ---------------- GLM 大脑 ----------------
|
||
def glm_chat(payload, timeout=90):
|
||
import requests
|
||
api = CFG["api"]
|
||
r = requests.post(f"{api['base']}/chat/completions",
|
||
headers={"Authorization": f"Bearer {api['key']}"},
|
||
json={"model": payload.get("model", api["text_model"]),
|
||
"messages": payload["messages"], "temperature": payload.get("temperature", 0.6)},
|
||
timeout=timeout)
|
||
r.raise_for_status()
|
||
return r.json()["choices"][0]["message"]["content"]
|
||
|
||
def brain_ready():
|
||
return bool(CFG["api"].get("key"))
|
||
|
||
def vision(prompt, img, model=None):
|
||
return glm_chat({"model": model or CFG["api"]["vision_model"], "temperature": 0.2,
|
||
"messages": [{"role": "user", "content": [
|
||
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{to_b64(img)}"}},
|
||
{"type": "text", "text": prompt}]}]})
|
||
|
||
def think(prompt):
|
||
return glm_chat({"messages": [{"role": "user", "content": prompt}]})
|
||
|
||
# ---------------- 业务动作 ----------------
|
||
def chat_header_name():
|
||
"""读取聊天窗口顶部标题,用于发送前核验,防止发错人"""
|
||
shot = scrot()
|
||
try:
|
||
ans = vision("这是微信聊天窗口顶部标题栏截图。如果里面显示了聊天对象的名字,只输出名字本身;"
|
||
"看不到名字就只输出NONE。不要输出任何解释。",
|
||
shot.crop((300, 60, 900, 105)))
|
||
ans = ans.strip().strip('"')
|
||
return "" if ("NONE" in ans.upper() or "微信" == ans or len(ans) > 20) else ans
|
||
except Exception as e:
|
||
log.warning("读标题失败: %s" % e)
|
||
return ""
|
||
|
||
def header_matches(header, name):
|
||
"""标题核验。header 为空一律不通过——空字符串是任何名字的子串,曾导致
|
||
在登录界面也'核验通过'的漏洞。"""
|
||
if not header:
|
||
return False
|
||
h, n = header.lower(), name.lower()
|
||
return n in h or h in n
|
||
|
||
def open_chat_via_search(name):
|
||
c = CFG["coords"]
|
||
# 注意:此处绝不能按 Escape——实测按 Esc 会打断搜索交互并让渲染崩成透明壳,
|
||
# 进而被自愈逻辑误判死亡而重启进程,毁掉登录会话(坑#18)。
|
||
click(*c["search_box"])
|
||
press("ctrl+a", 0.2)
|
||
paste(name)
|
||
time.sleep(1.6)
|
||
shot = scrot()
|
||
try:
|
||
ans = vision("这是微信搜索下拉结果截图。找到「Contacts」区域下名字等于「%s」的联系人条目,"
|
||
"返回中心坐标,格式 JSON: {\"x\":整数,\"y\":整数,\"found\":true/false}。"
|
||
"坐标以截图左上角为原点,宽1440高900。没有本地联系人结果时 found=false。" % name, shot)
|
||
m = re.search(r"\{[^{}]*\"x\"[^{}]*\}", ans, re.S)
|
||
if m:
|
||
pos = json.loads(m.group(0))
|
||
if pos.get("found"):
|
||
click(int(pos["x"]), int(pos["y"]), 1.5)
|
||
header = chat_header_name()
|
||
if header_matches(header, name):
|
||
return True
|
||
log.warning("打开的会话标题(%s)与目标(%s)不符,中止" % (header, name))
|
||
return False
|
||
except Exception as e:
|
||
log.warning("视觉定位失败(%s),退回默认坐标" % e)
|
||
click(*c["search_first_result"], 1.5) # 兜底:第一个结果位
|
||
header = chat_header_name()
|
||
ok = header_matches(header, name)
|
||
if not ok:
|
||
log.warning("兜底打开的会话标题(%s)与目标(%s)不符,中止" % (header, name))
|
||
return ok
|
||
|
||
def read_conversation():
|
||
shot = scrot()
|
||
try:
|
||
ans = vision("Read the WeChat chat window in this screenshot. Output JSON with the ACTUAL "
|
||
"window title and the ACTUAL text of the most recent messages, transcribed "
|
||
"character-for-character from the image. Never output placeholder text like "
|
||
"... or generic words - only real content you can see. If the screenshot shows "
|
||
"no chat window, output {\"chat_name\":\"\",\"messages\":[]}. "
|
||
"Format: {\"chat_name\":\"<real title>\",\"messages\":[{\"from\":\"them|me\","
|
||
"\"text\":\"<real message>\"}]} (newest last, max 6 messages).", shot)
|
||
m = re.search(r"\{.*\}", ans, re.S)
|
||
info = json.loads(m.group(0)) if m else None
|
||
# 防幻觉:全是占位符/空文本视为读取失败
|
||
if info and info.get("messages"):
|
||
real = [x for x in info["messages"]
|
||
if len(x.get("text", "")) >= 2 and x["text"] not in ("...", "…")]
|
||
if not real:
|
||
log.warning("视觉读取返回疑似占位符,丢弃")
|
||
return None
|
||
return info
|
||
except Exception as e:
|
||
log.warning("读会话失败: %s" % e)
|
||
return None
|
||
|
||
def send_text(text):
|
||
c = CFG["coords"]
|
||
click(*c["input_box"])
|
||
paste(text)
|
||
press("Return", 0.8)
|
||
if ensure_window(): # 发送动作可能触发窗口隐藏,救回来再审计
|
||
time.sleep(0.5)
|
||
scrot().save(os.path.join(BASE, "last_sent.png"))
|
||
log.info("已发送: %s" % text[:40].replace("\n", " "))
|
||
|
||
SAFE_ACK = "这条我记下来了,回头我详细答复你"
|
||
DANGER = re.compile(r"转账|红包|密码|验证码|汇款|付款")
|
||
URL_RE = re.compile(r"https?://[^\s\"'<>]+")
|
||
|
||
def fetch_url_text(url):
|
||
"""链接阅读(学自 WXAUTO_SE):抓网页正文,失败返回 None 走兜底"""
|
||
try:
|
||
import requests
|
||
r = requests.get(url, timeout=10, headers={"User-Agent": "Mozilla/5.0"})
|
||
if not r.encoding or r.encoding.lower() in ("iso-8859-1", "latin-1"):
|
||
r.encoding = r.apparent_encoding # 同上,防中文乱码
|
||
r.raise_for_status()
|
||
text = re.sub(r"(?is)<(script|style).*?</\1>", " ", r.text)
|
||
text = re.sub(r"<[^>]+>", " ", text)
|
||
return re.sub(r"\s+", " ", text)[:3000]
|
||
except Exception as e:
|
||
log.warning("链接抓取失败 %s: %s" % (url, e))
|
||
return None
|
||
|
||
def memory_file(chat):
|
||
safe = re.sub(r"[^\w\u4e00-\u9fff-]", "_", chat) or "unknown"
|
||
return os.path.join(BASE, "memory", safe + ".md")
|
||
|
||
def memory_context(chat):
|
||
mf = memory_file(chat)
|
||
if not os.path.exists(mf):
|
||
return ""
|
||
lines = open(mf, encoding="utf-8").readlines()
|
||
return "".join(lines[-40:])
|
||
|
||
def remember_exchange(chat, last_them, reply):
|
||
"""对话落记忆文件;每 10 次交互让 GLM 压缩一份摘要(学自 WXAUTO_SE 记忆功能)"""
|
||
mf = memory_file(chat)
|
||
os.makedirs(os.path.dirname(mf), exist_ok=True)
|
||
stamp = datetime.now().strftime("%m-%d %H:%M")
|
||
with open(mf, "a", encoding="utf-8") as f:
|
||
f.write("[%s] 对方: %s\n[%s] 我: %s\n" % (
|
||
stamp, (last_them or "")[:80], stamp, reply[:80].replace("\n", " ")))
|
||
try:
|
||
n = sum(1 for _ in open(mf, encoding="utf-8"))
|
||
if n >= 20 and n % 20 < 2:
|
||
head = "".join(open(mf, encoding="utf-8").readlines()[-40:])
|
||
summary = think("把以下微信对话记录压缩成不超过300字的记忆要点(事实/承诺/偏好/待办),直接输出要点:\n" + head)
|
||
sf = os.path.join(os.path.dirname(mf), "summary-" + os.path.basename(mf))
|
||
open(sf, "w", encoding="utf-8").write(summary.strip())
|
||
log.info("记忆摘要已更新: %s" % os.path.basename(sf))
|
||
except Exception:
|
||
log.warning("记忆压缩失败", exc_info=True)
|
||
|
||
KB_DIR = os.path.join(BASE, "kb")
|
||
|
||
def _ima_headers():
|
||
ima = CFG.get("ima", {})
|
||
return {"ima-openapi-clientid": ima.get("client_id", ""),
|
||
"ima-openapi-apikey": ima.get("api_key", ""),
|
||
"Content-Type": "application/json"}
|
||
|
||
def ima_kb_id():
|
||
"""解析知识库名→ID(内存缓存)。凭证只存服务器 config,不落日志不进仓库。"""
|
||
ima = CFG.get("ima", {})
|
||
if ima.get("kb_id"):
|
||
return ima["kb_id"]
|
||
if not (ima.get("enabled") and ima.get("client_id") and ima.get("api_key") and ima.get("kb_name")):
|
||
return ""
|
||
try:
|
||
import requests
|
||
r = requests.post("https://ima.qq.com/openapi/wiki/v1/search_knowledge_base",
|
||
headers=_ima_headers(),
|
||
json={"query": ima["kb_name"], "cursor": "", "limit": 10}, timeout=15)
|
||
r.raise_for_status()
|
||
for kb in (r.json().get("data") or {}).get("info_list") or []:
|
||
# 实测 API 返回 kb_id/kb_name(文档写 id/name,不一致——兼容两者)
|
||
name = kb.get("kb_name") or kb.get("name")
|
||
kid = kb.get("kb_id") or kb.get("id")
|
||
if name == ima["kb_name"] and kid:
|
||
ima["kb_id"] = kid
|
||
log.info("ima 知识库已解析: %s -> %s" % (name, kid))
|
||
return kid
|
||
log.warning("ima 未找到知识库: %s" % ima["kb_name"])
|
||
except Exception as e:
|
||
log.warning("ima 知识库解析失败: %s" % e)
|
||
return ""
|
||
|
||
def _keywords(query):
|
||
"""中文分词提关键词(jieba 优先),过滤虚词,按词长降序(信息量大的优先),最多6个"""
|
||
words = []
|
||
try:
|
||
import jieba
|
||
jieba.setLogLevel(logging.WARNING) # 静音 jieba 的建词典 DEBUG 输出
|
||
words = [w for w in jieba.lcut(query) if len(w) >= 2]
|
||
except Exception:
|
||
words = re.findall(r"[A-Za-z0-9]{2,}", query)
|
||
stop = {"怎么办", "怎么", "什么", "为什么", "可以", "应该", "如何", "一下", "看看",
|
||
"介绍", "需要", "还是", "能把", "告诉", "我们", "你们", "知道", "出来",
|
||
"相关", "信息", "没有", "这个", "那个", "时候", "开始", "内容", "资料",
|
||
"文章", "重点", "整理"}
|
||
seen, out = set(), []
|
||
for w in sorted(set(words) - stop, key=len, reverse=True):
|
||
if w not in seen:
|
||
seen.add(w)
|
||
out.append(w)
|
||
# 中文前缀候选:长词的前两字也作为关键词(「四会开」→「四会」,专名常见形态)
|
||
for w in list(out):
|
||
if len(w) >= 3 and re.match(r"^[\u4e00-\u9fff]{2}", w):
|
||
pre = w[:2]
|
||
if pre not in seen:
|
||
seen.add(pre)
|
||
out.append(pre)
|
||
return out[:6]
|
||
|
||
def ima_search(query):
|
||
"""ima 开放平台 search_knowledge。长问题分词分别查,合并去重;
|
||
返回 [{title, highlight, media_id}, ...] 最多6条。"""
|
||
kb_id = ima_kb_id()
|
||
if not kb_id:
|
||
return []
|
||
queries, seen_q = [], set()
|
||
first = query.strip()[:20]
|
||
if first:
|
||
queries.append(first)
|
||
for t in _keywords(query):
|
||
if t not in seen_q:
|
||
queries.append(t)
|
||
out, seen = [], set()
|
||
try:
|
||
import requests
|
||
for q in queries[:6]: # 控延迟:最多6次查询
|
||
r = requests.post("https://ima.qq.com/openapi/wiki/v1/search_knowledge",
|
||
headers=_ima_headers(),
|
||
json={"query": q, "cursor": "", "knowledge_base_id": kb_id},
|
||
timeout=15)
|
||
r.raise_for_status()
|
||
for x in ((r.json() or {}).get("data") or {}).get("info_list") or []:
|
||
key = x.get("media_id") or x.get("title")
|
||
if not key or key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append({"title": x.get("title", ""),
|
||
"highlight": x.get("highlight_content", ""),
|
||
"media_id": x.get("media_id", "")})
|
||
if len(out) >= 6:
|
||
break
|
||
except Exception as e:
|
||
log.warning("ima 搜索失败: %s" % e)
|
||
return out[:6]
|
||
|
||
def ima_media_text(media_id):
|
||
"""网页型收藏取原文(get_media_info → url_info.url → 抓正文),失败返回 None"""
|
||
try:
|
||
import requests
|
||
r = requests.post("https://ima.qq.com/openapi/wiki/v1/get_media_info",
|
||
headers=_ima_headers(), json={"media_id": media_id}, timeout=15)
|
||
url = (((r.json() or {}).get("data") or {}).get("url_info") or {}).get("url")
|
||
if not url:
|
||
return None
|
||
rr = requests.get(url, timeout=15, headers={"User-Agent": "Mozilla/5.0"})
|
||
if not rr.encoding or rr.encoding.lower() in ("iso-8859-1", "latin-1"):
|
||
rr.encoding = rr.apparent_encoding # 防 UTF-8 被按 Latin-1 解成乱码
|
||
rr.raise_for_status()
|
||
text = re.sub(r"(?is)<(script|style).*?</\1>", " ", rr.text)
|
||
text = re.sub(r"<[^>]+>", " ", text)
|
||
return re.sub(r"\s+", " ", text)[:1500]
|
||
except Exception as e:
|
||
log.warning("ima 原文获取失败 %s: %s" % (media_id, e))
|
||
return None
|
||
|
||
def kb_context(question):
|
||
"""知识库检索(轻量RAG):扫 kb/ 下 md/txt 的段落,按问题关键词重合度取最优段。
|
||
知识来源是 ima 知识库「AI教育杨老师」的导出文档;MCP 直连见 Loong 工单 #13。"""
|
||
try:
|
||
if not os.path.isdir(KB_DIR):
|
||
return ""
|
||
q = set(question)
|
||
best, best_score = [], 0
|
||
for fn in os.listdir(KB_DIR):
|
||
if not fn.lower().endswith((".md", ".txt")):
|
||
continue
|
||
try:
|
||
content = open(os.path.join(KB_DIR, fn), encoding="utf-8").read()
|
||
except Exception:
|
||
continue
|
||
for para in re.split(r"\n\s*\n", content):
|
||
para = para.strip()
|
||
if len(para) < 30:
|
||
continue
|
||
score = sum(1 for ch in q if ch in para and not ch.isspace())
|
||
if score > best_score and score >= 3:
|
||
best, best_score = ["[%s] %s" % (fn, para[:600])], score
|
||
elif score == best_score and score >= 3 and len(best) < 3:
|
||
best.append("[%s] %s" % (fn, para[:600]))
|
||
return ("\n\n".join(best))[:2500] if best else ""
|
||
except Exception:
|
||
log.warning("知识库检索失败", exc_info=True)
|
||
return ""
|
||
|
||
def compose_reply(info):
|
||
chat = info.get("chat_name", "未知")
|
||
last_them = next((m["text"] for m in reversed(info.get("messages", []))
|
||
if m["from"] == "them"), "")
|
||
history = "\n".join(("对方: " if m["from"] == "them" else "我: ") + m["text"]
|
||
for m in info["messages"])
|
||
persona = CFG.get("personas", {}).get(chat, CFG["persona"]) # 每会话独立人设(学自 WXAUTO_SE)
|
||
memory = memory_context(chat)
|
||
link_note = ""
|
||
urls = URL_RE.findall(last_them)
|
||
if urls:
|
||
content = fetch_url_text(urls[0])
|
||
link_note = ("\n对方消息含链接 %s,网页正文如下:\n%s\n请结合网页内容回复。" % (urls[0], content)
|
||
if content else
|
||
"\n对方消息含链接 %s(抓取失败,可建议对方粘贴正文)。" % urls[0])
|
||
prompt = (persona +
|
||
"\n\n当前时间: " + datetime.now().strftime("%Y-%m-%d %H:%M %A") + # AI时间感知(学自 WXAUTO_SE)
|
||
"\n当前聊天对象: " + chat +
|
||
(("\n相关历史记忆:\n" + memory) if memory else "") +
|
||
link_note +
|
||
"\n最近对话:\n" + history +
|
||
"\n\n请以「我」的身份回复对方最后一条消息。只输出回复正文,不要任何解释、引号或表情符号堆砌。")
|
||
if chat in CFG.get("kb_chats", []):
|
||
q = (last_them or history[-200:])
|
||
kb_parts = []
|
||
for h in ima_search(q)[:3]: # ima OpenAPI 优先
|
||
line = "《%s》%s" % (h["title"], h["highlight"])
|
||
if len(kb_parts) < 2 and h.get("media_id"): # 最多取2篇原文,控延迟
|
||
text = ima_media_text(h["media_id"])
|
||
if text:
|
||
line += " 正文:" + text
|
||
kb_parts.append(line)
|
||
if not kb_parts:
|
||
kb_parts.append(kb_context(q)) # 文件检索兜底
|
||
kb = "\n".join(p for p in kb_parts if p)
|
||
if kb:
|
||
prompt += "\n\n知识库参考(「AI教育杨老师」知识库,回答以下列内容为准,没有的如实说需要查证):\n" + kb
|
||
reply = think(prompt).strip().strip('"')
|
||
if DANGER.search(reply):
|
||
return SAFE_ACK
|
||
remember_exchange(chat, last_them, reply)
|
||
return reply
|
||
|
||
def in_quiet_hours():
|
||
q = CFG.get("quiet_hours")
|
||
if not q:
|
||
return False
|
||
h = datetime.now().hour
|
||
a, b = q
|
||
return h >= a or h < b
|
||
|
||
RATE = {"hour": None, "count": 0}
|
||
SCHED_FILE = os.path.join(BASE, "scheduled.json")
|
||
|
||
def load_tasks():
|
||
try:
|
||
return json.load(open(SCHED_FILE, encoding="utf-8"))
|
||
except Exception as e:
|
||
if os.path.exists(SCHED_FILE):
|
||
log.warning("load_tasks 解析失败: %s (内容前80字节: %r)" % (
|
||
e, open(SCHED_FILE, "rb").read(80)))
|
||
return []
|
||
|
||
def save_tasks(tasks):
|
||
json.dump(tasks, open(SCHED_FILE, "w", encoding="utf-8"), ensure_ascii=False, indent=1)
|
||
|
||
def remove_task(task):
|
||
"""手术式移除已发送的任务(按 run_at+message+chat 精确匹配),保留其他任务"""
|
||
try:
|
||
kept = [x for x in load_tasks()
|
||
if not (x.get("run_at") == task.get("run_at")
|
||
and x.get("message") == task.get("message")
|
||
and x.get("chat") == task.get("chat"))]
|
||
save_tasks(kept)
|
||
except Exception:
|
||
log.warning("任务移除失败", exc_info=True)
|
||
|
||
def run_due_tasks():
|
||
"""到期定时任务经核验路径发送(学自 WXAUTO_SE 定时任务);daily 自动顺延。
|
||
发送成功后手术式移除该条;失败重排队(+2分钟,3次放弃)。
|
||
⚠️ 需 config scheduled_enabled=true;实测曾出现重复触发循环,修复前保持关闭。"""
|
||
if not CFG.get("scheduled_enabled"):
|
||
return
|
||
now = datetime.now()
|
||
tasks = load_tasks()
|
||
if not tasks:
|
||
return
|
||
changed = False
|
||
fired = 0
|
||
for t in tasks:
|
||
try:
|
||
run_at = datetime.strptime(t["run_at"], "%Y-%m-%d %H:%M")
|
||
except Exception:
|
||
continue
|
||
if now < run_at:
|
||
continue
|
||
chat, msg = t.get("chat", ""), t.get("message", "")
|
||
log.info("定时任务触发: %s -> %s" % (chat, msg[:30]))
|
||
sent = False
|
||
try:
|
||
ensure_window()
|
||
if open_chat_from_list(chat) or open_chat_via_search(chat):
|
||
send_text(msg)
|
||
log_inbox("定时任务已发送 %s: %s" % (chat, msg[:40]))
|
||
sent = True
|
||
fired += 1
|
||
except Exception:
|
||
log.exception("定时任务发送失败")
|
||
if sent:
|
||
if t.get("repeat") == "daily":
|
||
t["run_at"] = (run_at + timedelta(days=1)).strftime("%Y-%m-%d %H:%M")
|
||
changed = True
|
||
else:
|
||
remove_task(t)
|
||
changed = True
|
||
else:
|
||
t["attempts"] = t.get("attempts", 0) + 1
|
||
if t["attempts"] >= 3:
|
||
log_inbox("定时任务连续3次失败已放弃: %s -> %s" % (chat, msg[:30]))
|
||
remove_task(t)
|
||
changed = True
|
||
else:
|
||
t["run_at"] = (now + timedelta(minutes=2)).strftime("%Y-%m-%d %H:%M")
|
||
changed = True
|
||
if changed:
|
||
ensure_window()
|
||
save_tasks(load_tasks()) # 重读后保存,合并并发改动
|
||
|
||
def maybe_schedule(text, chat):
|
||
"""白名单消息含定时意图时 GLM 抽取结构化任务(学自 WXAUTO_SE);返回确认语或 None"""
|
||
if not re.search(r"\d+\s*分钟|\d+\s*小时|每天.*点|明天.*点|提醒我", text):
|
||
return None
|
||
prompt = ("从这条微信消息抽取定时任务。只输出JSON:"
|
||
'{"has": true/false, "run_at": "YYYY-MM-DD HH:MM", "message": "提醒内容", "repeat": "none或daily"}。'
|
||
"没有定时意图则 has=false。当前时间: %s。消息: %s"
|
||
% (datetime.now().strftime("%Y-%m-%d %H:%M"), text[:200]))
|
||
try:
|
||
m = re.search(r"\{[^{}]*\}", think(prompt), re.S)
|
||
if not m:
|
||
return None
|
||
d = json.loads(m.group(0))
|
||
if not d.get("has") or not d.get("run_at"):
|
||
return None
|
||
tasks = load_tasks()
|
||
tasks.append({"run_at": d["run_at"], "chat": chat,
|
||
"message": d.get("message", ""), "repeat": d.get("repeat", "none")})
|
||
save_tasks(tasks)
|
||
log.info("定时任务已登记: %s %s" % (d["run_at"], d.get("message", "")[:30]))
|
||
return "好的，已定好：%s（%s）" % (d.get("message", "")[:30], d["run_at"])
|
||
except Exception:
|
||
log.warning("定时任务抽取失败", exc_info=True)
|
||
return None
|
||
|
||
def group_needs_reply(chat_name, messages):
|
||
"""群聊仅在被 @ 本号昵称时响应(学自 WXAUTO_SE)。群标题带成员数括号;
|
||
未配置 self_nick 或非群会话保持原行为。"""
|
||
if "(" not in (chat_name or ""):
|
||
return True
|
||
nick = CFG.get("self_nick", "")
|
||
if not nick:
|
||
return True
|
||
them = " ".join(m.get("text", "") for m in messages if m.get("from") == "them")
|
||
return nick in them
|
||
|
||
def rate_ok():
|
||
h = datetime.now().hour
|
||
if RATE["hour"] != h:
|
||
RATE["hour"], RATE["count"] = h, 0
|
||
return RATE["count"] < CFG["max_replies_per_hour"]
|
||
|
||
def log_inbox(line):
|
||
with open(INBOX, "a", encoding="utf-8") as f:
|
||
f.write("[%s] %s\n" % (datetime.now().strftime("%m-%d %H:%M:%S"), line))
|
||
|
||
def handle_trigger(slots, img):
|
||
c = CFG["coords"]
|
||
ensure_window()
|
||
# 借视觉把槽位映射成联系人名(有大脑时)
|
||
slot_map = {}
|
||
if brain_ready():
|
||
try:
|
||
names = vision("这是微信聊天列表竖条截图。从上到下列出每个会话名字,"
|
||
"只输出JSON数组:[{\"slot\":0,\"name\":\"...\"}]", list_region(img))
|
||
m = re.search(r"\[.*\]", names, re.S)
|
||
slot_map = {x["slot"]: x["name"] for x in json.loads(m.group(0))} if m else {}
|
||
except Exception as e:
|
||
log.warning("解析聊天列表失败: %s" % e)
|
||
for s in slots:
|
||
name = slot_map.get(s, "槽位%d" % s)
|
||
log_inbox("新消息 来自 %s" % name)
|
||
if not any(w and (w in name or name in w) for w in CFG["whitelist"]):
|
||
log.info("跳过非白名单: %s" % name)
|
||
continue
|
||
if not brain_ready():
|
||
log.info("观察模式,无大脑,不操作: %s" % name)
|
||
continue
|
||
# 消息合并去抖(学自 WXAUTO_SE):等突发的一串消息到齐再读,避免逐条回复
|
||
time.sleep(CFG.get("merge_wait", 8))
|
||
y = c["slot_y0"] + s * c["slot_dy"]
|
||
click(c["slot_click_x"], y, 1.5) # 点开该会话
|
||
info = read_conversation()
|
||
if not info or not info.get("messages"):
|
||
continue
|
||
real_name = info.get("chat_name", "")
|
||
if real_name and CFG.get("verify_chat_name", True) and \
|
||
real_name not in name and name not in real_name:
|
||
log.warning("窗口标题(%s)与预期(%s)不符,放弃回复" % (real_name, name))
|
||
continue
|
||
if not group_needs_reply(info.get("chat_name"), info.get("messages", [])):
|
||
log.info("群内未被@本号,跳过: %s" % info.get("chat_name"))
|
||
continue
|
||
last_them = next((m["text"] for m in reversed(info.get("messages", []))
|
||
if m["from"] == "them"), "")
|
||
sched = maybe_schedule(last_them, info.get("chat_name") or "")
|
||
if sched:
|
||
send_text(sched)
|
||
log_inbox("定时任务已设定: %s" % sched)
|
||
continue
|
||
if not rate_ok():
|
||
log.warning("超时速率限制,跳过回复 %s" % name)
|
||
continue
|
||
reply = compose_reply(info)
|
||
mode = CFG["mode"]
|
||
if mode == "draft" or in_quiet_hours():
|
||
log_inbox("草稿(%s%s): %s" % (name, " 静默时段" if in_quiet_hours() else "", reply))
|
||
log.info("草稿模式,未发送")
|
||
continue
|
||
send_text(reply)
|
||
RATE["count"] += 1
|
||
log_inbox("已回复 %s: %s" % (name, reply))
|
||
time.sleep(1)
|
||
|
||
def expand_pinned_if_collapsed(img):
|
||
"""列表被折叠(显示 N pinned chats)时点开展开,否则找不到 File Transfer"""
|
||
try:
|
||
ans = vision("这是微信聊天列表顶部一小块截图。如果里面有「pinned chats」字样,"
|
||
"只输出FOUND,否则只输出NO。", img.crop((60, 240, 300, 285)))
|
||
if "FOUND" in ans.upper():
|
||
click(160, 260, 1.0)
|
||
time.sleep(0.8)
|
||
log.info("已展开置顶聊天")
|
||
return True
|
||
except Exception as e:
|
||
log.warning("检查折叠条失败: %s" % e)
|
||
return False
|
||
|
||
def _ft_icon_ref():
|
||
return os.path.join(BASE, "ft_icon_ref.png")
|
||
|
||
def save_ft_template(img, y_top):
|
||
"""标题核验成功后把该图标存为参考模板(自学习),供后续模板匹配择优"""
|
||
try:
|
||
img.crop((74, y_top, 110, y_top + 36)).save(_ft_icon_ref())
|
||
log.info("FT 图标参考模板已更新")
|
||
except Exception:
|
||
log.warning("FT 模板保存失败", exc_info=True)
|
||
|
||
def find_green_icon_y(img):
|
||
"""聊天列表图标列找 File Transfer 图标。候选=绿圈+白箭头双特征(green>250,white>250);
|
||
存在参考模板时按 9x9 缩略图最小距离择优(修颜色质量误配,坑#2),无模板取最底部候选。"""
|
||
c = CFG["coords"]
|
||
px = img.load()
|
||
cands = []
|
||
y_cap = min(c["list_y1"] - 44, 826) # 避开底部 pinned chats 开关区
|
||
for y in range(c["list_y0"], y_cap, 2):
|
||
green = white = 0
|
||
for yy in range(y, y + 36):
|
||
for xx in range(74, 110):
|
||
r, g, b = px[xx, yy][:3]
|
||
if r < 80 and g > 160 and 60 < b < 140:
|
||
green += 1
|
||
elif r > 220 and g > 220 and b > 220:
|
||
white += 1
|
||
if green > 250 and white > 250:
|
||
cands.append(y)
|
||
if not cands:
|
||
return None
|
||
ref_path = _ft_icon_ref()
|
||
if os.path.exists(ref_path):
|
||
try:
|
||
ref = Image.open(ref_path).convert("RGB").resize((9, 9))
|
||
ref_data = list(ref.getdata())
|
||
best, best_d = None, None
|
||
for y in cands:
|
||
for dy in (0, -2, 2): # 容忍1-2px行偏移
|
||
yy = y + dy
|
||
if yy < c["list_y0"] or yy + 36 > 870:
|
||
continue
|
||
patch = img.crop((74, yy, 110, yy + 36)).convert("RGB").resize((9, 9))
|
||
d = sum((a[0]-b[0])**2 + (a[1]-b[1])**2 + (a[2]-b[2])**2
|
||
for a, b in zip(patch.getdata(), ref_data))
|
||
if best_d is None or d < best_d:
|
||
best, best_d = yy, d
|
||
if best_d is None or best_d > 100000: # 校准:真身<10万(含微移),误配270万+
|
||
log.warning("FT 模板匹配距离过大(%s),疑似列表异常,跳过" % best_d)
|
||
return None
|
||
log.info("FT 图标模板匹配: 候选%s 取 y=%s (距离%s)" % (cands[:5], best, best_d))
|
||
return best + 18
|
||
except Exception:
|
||
log.warning("模板匹配异常,回退颜色规则", exc_info=True)
|
||
return cands[-1] + 18
|
||
|
||
def open_chat_from_list(name):
|
||
"""在聊天列表里用视觉找到目标会话并点开(适用于搜索索引搜不到的内置会话如 File Transfer)"""
|
||
c = CFG["coords"]
|
||
img = scrot()
|
||
try:
|
||
ans = vision("这是微信聊天列表截图。找到名字为「%s」的会话条目,返回它的头像中心坐标,"
|
||
"格式 JSON: {\"x\":整数,\"y\":整数,\"found\":true/false}。"
|
||
"坐标以图片左上角为原点,图片宽240高840。" % name, list_region(img))
|
||
m = re.search(r"\{[^{}]*\"x\"[^{}]*\}", ans, re.S)
|
||
if not m:
|
||
return False
|
||
pos = json.loads(m.group(0))
|
||
if not pos.get("found"):
|
||
log.warning("列表中未找到 %s" % name)
|
||
return False
|
||
click(int(pos["x"]) + c["list_x0"], int(pos["y"]) + c["list_y0"], 1.5)
|
||
except Exception as e:
|
||
log.warning("列表定位异常: %s" % e)
|
||
return False
|
||
header = chat_header_name()
|
||
ok = header_matches(header, name)
|
||
if not ok:
|
||
log.warning("点开的会话标题(%s)与目标(%s)不符,中止" % (header, name))
|
||
return ok
|
||
|
||
def heartbeat():
|
||
hb = CFG.get("heartbeat_chat")
|
||
if not hb or not brain_ready():
|
||
return # 没有大脑时无法核验目标,不发,防误发
|
||
hb_file = os.path.join(BASE, "last_hb")
|
||
try:
|
||
since = time.time() - os.path.getmtime(hb_file)
|
||
except OSError:
|
||
since = CFG["heartbeat_interval_min"] * 60
|
||
if since < CFG["heartbeat_interval_min"] * 60 - 120:
|
||
return # 未到期。上次心跳时间持久化到文件,
|
||
# 服务重启不再重发(实测重启曾连发3条骚扰)
|
||
if not ensure_window():
|
||
return
|
||
img = scrot()
|
||
expand_pinned_if_collapsed(img)
|
||
img = scrot()
|
||
if hb.lower() in ("file transfer", "文件传输助手"):
|
||
gy = find_green_icon_y(img)
|
||
if gy is None:
|
||
log.warning("未找到 File Transfer 绿色图标,跳过心跳")
|
||
return
|
||
click(91, gy, 2.0) # 点图标中心,避开下方折叠开关
|
||
click_y_top = gy - 18
|
||
elif not (open_chat_from_list(hb) or open_chat_via_search(hb)):
|
||
click_y_top = None
|
||
return
|
||
header = chat_header_name()
|
||
if hb.lower() not in header.lower():
|
||
log.warning("心跳目标核验失败(标题:%s),不发送" % header)
|
||
return
|
||
try:
|
||
if click_y_top is not None:
|
||
# 必须用点击前的那张 img:点击后列表会重排(FT 跳到顶部),旧坐标处已不是 FT
|
||
save_ft_template(img, click_y_top)
|
||
except Exception:
|
||
pass
|
||
free = sh("free -m | awk '/Mem:/{print $7}'").stdout.decode().strip()
|
||
send_text("〔自动心跳〕Agent在线 ✓ 可用内存%sMB 时间%s" % (free, datetime.now().strftime("%m-%d %H:%M")))
|
||
open(hb_file, "w").write(datetime.now().isoformat())
|
||
log.info("心跳已发")
|
||
|
||
# ---------------- 主循环 ----------------
|
||
def main():
|
||
ensure_window()
|
||
prev = None
|
||
red_state = {}
|
||
log.info("Agent启动 mode=%s brain=%s" % (CFG["mode"], "GLM" if brain_ready() else "无(观察)"))
|
||
while True:
|
||
try:
|
||
if os.path.exists(PAUSE):
|
||
prev = None
|
||
time.sleep(3)
|
||
continue
|
||
if CFG["mode"] == "observe" and not brain_ready():
|
||
pass # 观察模式仍然盯屏记录
|
||
if not ensure_window():
|
||
prev = None # 未登录/渲染坏:不监控,等自愈
|
||
time.sleep(8)
|
||
continue
|
||
img = scrot()
|
||
cur = img_hash(list_region(img))
|
||
if prev is not None and cur != prev:
|
||
slots = [i for i in range(CFG["slot_count"]) if red_score(img, i) > CFG["red_threshold"]]
|
||
fresh = [s for s in slots if red_state.get(s, 0) == 0]
|
||
if fresh:
|
||
log.info("检测到新消息槽位: %s" % fresh)
|
||
handle_trigger(fresh, img)
|
||
for s in slots:
|
||
red_state[s] = 1
|
||
for k in [s for s in red_state if s not in slots]:
|
||
red_state[k] = 0
|
||
prev = cur
|
||
elif prev is None:
|
||
prev = cur
|
||
heartbeat() # 内部按 last_hb 文件到期判定,重启不重发
|
||
run_due_tasks() # 定时任务到期即发
|
||
except Exception:
|
||
log.exception("循环异常")
|
||
time.sleep(5)
|
||
time.sleep(CFG["poll_interval"])
|
||
|
||
def cli_send(name, text):
|
||
if not ensure_window():
|
||
log.error("微信窗口未就绪(登录/渲染),未发送")
|
||
return
|
||
if not (open_chat_from_list(name) or open_chat_via_search(name)):
|
||
log.error("无法核验目标会话 %s,未发送" % name)
|
||
return
|
||
send_text(text)
|
||
log_inbox("CLI发送 %s: %s" % (name, text))
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) >= 4 and sys.argv[1] == "send":
|
||
cli_send(sys.argv[2], sys.argv[3])
|
||
elif len(sys.argv) >= 2 and sys.argv[1] == "once":
|
||
img = scrot(); ensure_window()
|
||
slots = [i for i in range(CFG["slot_count"]) if red_score(img, i) > CFG["red_threshold"]]
|
||
print("红点槽位:", slots)
|
||
handle_trigger(slots, img)
|
||
else:
|
||
main()
|