"""WeChat Viewer Flask 主应用""" import subprocess import json import time import os import threading from datetime import datetime, timedelta from pathlib import Path from flask import Flask, jsonify, request, render_template, send_file, redirect, url_for, Response import config import wechat_db import local_db def _su_exists(path: str) -> bool: try: r = subprocess.run( ["su", "-c", f"test -f {path} && echo yes || echo no"], capture_output=True, text=True, timeout=5 ) return "yes" in r.stdout except Exception: return False def _safe_exists(path: str) -> bool: try: if Path(path).exists(): return True except PermissionError: return _su_exists(path) # WeChat data dir: Path.exists() returns False (not PermissionError) on Android # when the process lacks read permission on parent directories. if path.startswith("/data/data/") and _su_exists(path): return True return False import re import hashlib import html as html_module app = Flask(__name__) local_db.init_db() # ── 统一 Hermes 子进程启动 ────────────────────────── HERMES_JOBS_DIR = Path("/data/data/com.termux/files/home/tmp/hermes_jobs") def _spawn_hermes_job(job_type: str, job_name: str, cmd: list[str], timeout: int = 300, on_done=None, task_id: str = "") -> str: """启动 Hermes CLI 子进程并追踪。 返回 task_id（12 位 hex）。完成后调用 on_done(task_id, success)。 如果传入 task_id 则使用，否则自动生成。 全局并发控制：使用 _hermes_semaphore（上限 3）限制同时运行的 hermes chat 进程数， 避免 Pixel 内存溢出导致 Termux 崩溃。 """ # 防御：所有 hermes 子进程必须指定 skill if "-s" not in cmd and "--skills" not in cmd: raise ValueError(f"_spawn_hermes_job({job_type}): 缺少 -s/--skills，拒绝执行。所有 hermes 子进程必须指定 skill。") import uuid, threading if not hasattr(_spawn_hermes_job, "_hermes_semaphore"): _spawn_hermes_job._hermes_semaphore = threading.Semaphore(3) herm_sem = _spawn_hermes_job._hermes_semaphore if not task_id: task_id = uuid.uuid4().hex[:12] HERMES_JOBS_DIR.mkdir(parents=True, exist_ok=True) log_file = str(HERMES_JOBS_DIR / f"{task_id}.log") local_db.job_add(task_id, job_type, job_name, log_file) # ── Harness Trace: 结构化错误码 ── # 对齐 Harness Trace 四步诊断（Model Input / Tool Selection / Runtime / Context Assembly） # 每个失败原因映射到 reason_code，方便离线归因而非盲目重试 # HydraMind Gate trace quad: [node=spawn,attempt=1,gate=,action=BLOCK] # reason codes 统一在 VerifyReason 类中定义 def _run(): success = False reason_code = None # ── 全局并发控制：获取 semaphore 后才能启动 hermes chat ── acquired = herm_sem.acquire(blocking=False) if not acquired: # 不能阻塞等待 → 快速失败，让 cron 下一轮重试 reason_code = "CONCURRENCY_LIMIT" try: with open(log_file, "w") as lf: lf.write(f"=== Hermes Job SKIPPED: concurrency limit reached ===\n" f"=== Task ID: {task_id} ===\n" f"=== Type: {job_type} - {job_name} ===\n") except: pass else: try: with open(log_file, "w") as lf: lf.write(f"=== Hermes Job Started: {job_type} - {job_name} ===\n") lf.write(f"=== Task ID: {task_id} ===\n") lf.write(f"=== Command: {' '.join(cmd)} ===\n") lf.write(f"=== Timeout: {timeout}s ===\n") lf.write(f"=== Started at: {time.strftime('%Y-%m-%d %H:%M:%S')} ===\n\n") lf.flush() proc = subprocess.Popen(cmd, stdout=lf, stderr=subprocess.STDOUT, text=True, cwd=str(Path(__file__).parent)) try: proc.wait(timeout=timeout) if proc.returncode == 0: success = True # ── HydraMind Gate: post-execution content thin check ── try: log_size = Path(log_file).stat().st_size if log_size < 200: lf.write( f"\n=== [node=spawn,attempt=1,gate={VerifyReason.CONTENT_THIN}," f"action=CONTINUE] " f"log file only {log_size} bytes, job may have " f"produced no meaningful output ===\n") except Exception: pass else: reason_code = VerifyReason.SPAWN_NONZERO lf.write(f"\n=== [node=spawn,attempt=1,gate={reason_code},action=BLOCK] " f"Exit code: {proc.returncode} ===\n") except subprocess.TimeoutExpired: proc.kill() proc.wait() reason_code = VerifyReason.SPAWN_TIMEOUT lf.write(f"\n=== [node=spawn,attempt=1,gate={reason_code},action=BLOCK] " f"job killed after {timeout}s timeout ===\n") lf.write(f"=== Finished at: {time.strftime('%Y-%m-%d %H:%M:%S')} ===\n") except Exception as e: reason_code = VerifyReason.SPAWN_FAILED try: with open(log_file, "a") as lf: lf.write(f"\n=== [node=spawn,attempt=1,gate={reason_code},action=BLOCK] {e} ===\n") except Exception: pass finally: herm_sem.release() status = "completed" if success else "failed" local_db.job_update_status(task_id, status) if on_done: try: on_done(task_id, success, reason_code) except TypeError: # Backward-compat: old on_done with 2 params try: on_done(task_id, success) except Exception as e2: try: with open(log_file, "a") as lf: lf.write(f"\n=== [node=spawn,attempt=1,gate={VerifyReason.SPAWN_ON_DONE_ERR},action=BLOCK] " f"on_done callback failed: {e2} ===\n") except Exception: pass except Exception as e: # Harness Trace: callback 失败也记录，不静默吞 try: with open(log_file, "a") as lf: lf.write(f"\n=== [node=spawn,attempt=1,gate={VerifyReason.SPAWN_ON_DONE_ERR},action=BLOCK] " f"on_done callback failed: {e} ===\n") except Exception: pass threading.Thread(target=_run, daemon=True).start() return task_id def _extract_dedup_key(url: str) -> str: if not url: return "" if "mp.weixin.qq.com" in url: biz = re.search(r"__biz=([^&]+)", url) mid = re.search(r"mid=(\d+)", url) idx = re.search(r"idx=(\d+)", url) sn = re.search(r"sn=([a-f0-9]+)", url) if biz and mid and idx and sn: return f"{biz.group(1)}|{mid.group(1)}|{idx.group(1)}|{sn.group(1)}" return url # ── Page: 文章列表 ────────────────────────────────── @app.route("/articles") def articles_page(): return render_template("articles.html") @app.route("/api/articles") def api_articles(): try: page = max(1, int(request.args.get("page", "1"))) per_page = max(1, min(500, int(request.args.get("per_page", "20")))) date_from = request.args.get("date_from", "") date_to = request.args.get("date_to", "") total, articles = local_db.article_list_new(page, per_page, date_from, date_to) return jsonify({"ok": True, "total": total, "page": page, "per_page": per_page, "data": articles}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/articles/extract", methods=["POST"]) def api_article_extract(): """Extract articles from WeChat DB into articles + article_shares (two-table schema).""" try: sql = """SELECT m.msgId, m.talker, m.content, m.createTime FROM message m WHERE m.type = 49 ORDER BY m.createTime DESC LIMIT 1000""" rows = wechat_db.query(sql) new_articles, new_shares, total = 0, 0, 0 for row in rows: content = row.get("content", "") or "" url_match = re.search(r"(.*?)", content) if not url_match: continue url = url_match.group(1) url = re.sub(r"", r"\1", url, flags=re.DOTALL) url = html_module.unescape(url) if not url or url.startswith("https://support.weixin.qq.com"): continue total += 1 title_match = re.search(r"(.*?)", content) title = title_match.group(1) if title_match else "" title = html_module.unescape(re.sub(r"", r"\1", title, flags=re.DOTALL)) # Extract actual sender from content prefix (group msgs: "sender:\n") raw_talker = row.get("talker", "") sender_match = re.match(r"^([^:\n<]+):\n", content) actual_sender = sender_match.group(1) if sender_match else raw_talker chatroom_id = raw_talker if "@chatroom" in raw_talker else "" msg_create_time = row.get("createTime", 0) source_msg_id = row.get("msgId", 0) # Upsert article (dedup by URL → biz|mid|idx|sn) was_new = local_db.article_upsert(url, title, return_was_new=True) article_id = was_new[0] if isinstance(was_new, tuple) else was_new if isinstance(was_new, tuple) and was_new[1]: new_articles += 1 # Record share (dedup by source_msg_id) if not local_db.share_exists(source_msg_id): local_db.article_share_add(article_id, chatroom_id or raw_talker, actual_sender, source_msg_id, msg_create_time) new_shares += 1 return jsonify({"ok": True, "data": {"new_articles": new_articles, "new_shares": new_shares, "total": total}}) except Exception as e: import traceback traceback.print_exc() return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/articles/refresh", methods=["POST"]) def api_article_refresh(): """Manual refresh: run scan_articles.py via Hermes CLI to extract new articles.""" import uuid task_id = uuid.uuid4().hex[:12] scan_script = "/data/data/com.termux/files/home/.hermes/skills/wechat-group-digest/scripts/scan_articles.py" prompt = f"执行 python3 {scan_script} --extract-only。只做提取不进摘要。完成后报告结果。" _spawn_hermes_job( "article_refresh", "手动刷新文章列表", ["hermes", "chat", "-q", prompt, "-s", "wechat-group-digest", "--yolo", "-t", "terminal"], timeout=120, task_id=task_id ) return jsonify({"ok": True, "data": {"task_id": task_id, "message": "提取任务已启动"}}) @app.route("/api/articles/unsummarized") def api_article_unsummarized(): """Get articles without summaries for cron job.""" try: limit = min(int(request.args.get("limit", "20")), 200) rows = local_db.article_get_unsummarized_new(limit) return jsonify({"ok": True, "data": list(rows)}) except Exception as e: import traceback traceback.print_exc() return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/articles/summarize", methods=["POST"]) def api_article_summarize(): data = request.json or {} url = data.get("url", "") title = data.get("title", "") if not url: return jsonify({"ok": False, "error": "需要 url"}), 400 dedup_key = _extract_dedup_key(url) if not dedup_key: return jsonify({"ok": False, "error": "无法提取去重标识"}), 400 existing = local_db.find_article_summary(dedup_key, "summary") if existing: return jsonify({"ok": True, "data": {"status": "cached", "report_id": existing["id"], "content": existing["content"], "summary": (existing.get("content", "") or "")[:200], "created_at": existing.get("created_at", "")}}) import uuid task_id = uuid.uuid4().hex[:12] result_file = f"/data/data/com.termux/files/home/tmp/article_summary_{task_id}.md" prompt = (f"请抓取并阅读以下文章，生成一份核心观点摘要。\n\n" f"文章标题: {title}\n文章链接: {url}\n\n" f"要求:\n1. 用 terminal curl 或 web 抓取文章全文\n" f"2. 提取 3-5 个核心观点，每个观点用一句话概括\n" f"3. 列出文章中的关键数据或案例（如有）\n" f"4. 格式为 Markdown，标题: # {title} 摘要\n" f"5. 将结果写入文件: {result_file}\n6. 确认文件已写入\n\n" f"重要: 只做抓取和摘要，完成后不要做其他操作。") def _on_summary_done(tid, ok): try: if Path(result_file).exists(): md = Path(result_file).read_text(encoding="utf-8", errors="replace") # 检查是否生成成功：内容足够长且不含失败标记 if not md or len(md.strip()) < 50 or "生成失败" in md or "重试" in md: try: Path(result_file).unlink() except: pass return # 失败不写 DB，用户可重试 else: return # 没有结果文件，不写 DB local_db.report_add("article", title, "summary", md, 0, 0, 0, dedup_key=dedup_key, article_url=url) try: Path(result_file).unlink() except: pass except Exception: pass _spawn_hermes_job("article_summary", title[:50], ["hermes", "chat", "-q", prompt, "-s", "article-summarizer", "--yolo", "-t", "terminal,web,file"], timeout=180, on_done=_on_summary_done, task_id=task_id) return jsonify({"ok": True, "data": {"status": "started", "task_id": task_id}}) @app.route("/api/articles/learn", methods=["POST"]) def api_article_learn(): data = request.json or {} url = data.get("url", "") title = data.get("title", "") if not url: return jsonify({"ok": False, "error": "需要 url"}), 400 dedup_key = _extract_dedup_key(url) if not dedup_key: return jsonify({"ok": False, "error": "无法提取去重标识"}), 400 existing = local_db.find_article_summary(dedup_key, "learn") if existing: return jsonify({"ok": True, "data": {"status": "cached", "report_id": existing["id"], "content": existing["content"], "summary": (existing.get("content", "") or "")[:200], "created_at": existing.get("created_at", "")}}) import uuid task_id = uuid.uuid4().hex[:12] result_file = f"/data/data/com.termux/files/home/tmp/article_learn_{task_id}.md" prompt = (f"请学习以下文章，按照 learn-and-build skill 的流程处理。\n\n" f"文章标题: {title}\n文章链接: {url}\n\n" f"要求:\n1. 抓取文章全文\n2. 判断文章类型（开源代码/论文/方法论）\n" f"3. 按类型执行对应的学习+构建流程\n" f"4. 将学习结果和构建内容写入文件: {result_file}\n5. 确认文件已写入\n\n" f"完成后不要做其他操作。") def _on_learn_done(tid, ok): try: if Path(result_file).exists(): md = Path(result_file).read_text(encoding="utf-8", errors="replace") if not md or len(md.strip()) < 50 or "生成失败" in md or "重试" in md: try: Path(result_file).unlink() except: pass return else: return local_db.report_add("article", title, "learn", md, 0, 0, 0, dedup_key=dedup_key, article_url=url) try: Path(result_file).unlink() except: pass except Exception: pass _spawn_hermes_job("article_learn", title[:50], ["hermes", "chat", "-q", prompt, "-s", "learn-and-build", "--yolo", "-t", "terminal,web,file"], timeout=300, on_done=_on_learn_done, task_id=task_id) return jsonify({"ok": True, "data": {"status": "started", "task_id": task_id}}) @app.route("/api/articles/check-summaries", methods=["POST"]) def api_article_check_summaries(): data = request.json or {} urls = data.get("urls", []) report_type = data.get("type", "summary") if not urls: return jsonify({"ok": False, "error": "需要 urls 数组"}), 400 url_map = {} for item in urls: url = item.get("url", "") if isinstance(item, dict) else str(item) title = item.get("title", "") if isinstance(item, dict) else "" dk = _extract_dedup_key(url) if dk: url_map[dk] = {"url": url, "title": title} if not url_map: return jsonify({"ok": True, "data": {"results": {}, "missing": urls}}) found = local_db.find_article_summaries_batch(list(url_map.keys()), report_type) missing = [] for dk, info in url_map.items(): if dk not in found: missing.append({"url": info["url"], "title": info["title"], "dedup_key": dk}) return jsonify({"ok": True, "data": {"results": found, "missing": missing}}) @app.route("/api/articles/generate-image", methods=["POST"]) def api_article_generate_image(): data = request.json or {} title = data.get("title","") url = data.get("url","") if not title: return jsonify({"ok":False,"error":"需要 title"}), 400 import uuid task_id = uuid.uuid4().hex[:12] img_file = f"/data/data/com.termux/files/home/tmp/article_img_{task_id}.png" prompt = f"为以下文章内容生成一张 19 世纪复古铜版画风格图片。\n\n文章标题: {title}\n\n要求:\n1. 生成中文 prompt: 复古老旧报纸版面，19世纪黑白铜版画蚀刻风格，泛黄做旧牛皮纸，折痕毛边污渍，顶部大号粗黑体报纸刊头={title[:20]}，主标题加粗醒目，油墨印刷颗粒质感，黑白单色调，高细节，8K\n2. 用 GPT Image 2 API 生成图片(model=GPT Image 2, n=1, size=1024x1024)\n3. 将 base64 解码保存到 {img_file}\n4. 确认文件已写入\n\n重要: 完成后不要做其他操作。" def _on_img_done(tid, ok): try: if Path(img_file).exists(): local_db.report_image_add(0, img_file, "wsj") try: Path(img_file).unlink() except: pass except: pass _spawn_hermes_job("article_image", title[:50], ["hermes","chat","-q",prompt,"-s","gpt-image-gen","--yolo","-t","terminal,web,file"], timeout=300, on_done=_on_img_done, task_id=task_id) return jsonify({"ok":True,"data":{"status":"started","task_id":task_id}}) # ── 账号切换 ────────────────────────────────────── @app.route("/api/accounts") def api_accounts(): """返回所有已发现账号""" return jsonify({ "ok": True, "data": { "accounts": config.ACCOUNTS, "current": config.get_current_account_id() } }) @app.route("/api/accounts/switch", methods=["POST"]) def api_accounts_switch(): """切换当前账号""" data = request.json or {} aid = data.get("account_id", "") if not aid: return jsonify({"ok": False, "error": "需要 account_id"}), 400 if not any(a["id"] == aid for a in config.ACCOUNTS): return jsonify({"ok": False, "error": "无效的账号"}), 400 config.set_current_account_id(aid) return jsonify({"ok": True, "data": {"current": aid}}) # ── Page 1: 消息浏览器 ────────────────────────────── @app.route("/") def index(): return render_template("index.html") @app.route("/chat/") def chat_page(talker): return render_template("chat.html", talker=talker) @app.route("/messages") def messages_page(): return render_template("index.html") @app.route("/message/") def message_redirect(talker): return redirect(url_for("chat_page", talker=talker)) @app.route("/api/sessions") def api_sessions(): try: page = max(1, int(request.args.get("page", "1"))) per_page = max(1, min(500, int(request.args.get("per_page", "10")))) sort = request.args.get("sort", "time_desc") if sort not in ("time_desc", "time_asc", "name"): sort = "time_desc" total = wechat_db.count_sessions() rows = wechat_db.get_session_list(page=page, per_page=per_page, sort=sort) for r in rows: # 从 content 提取纯文本预览 raw = (r.get("content", "") or "") # 群聊消息去掉 "sender:\n" 前缀 if ":\n" in raw and raw.split(":\n")[0]: raw = raw.split(":\n", 1)[1] if len(raw.split(":\n", 1)) > 1 else raw # 去掉 XML 标签 + HTML 实体 clean = re.sub(r"<[^>]+>", "", raw).strip() clean = html_module.unescape(clean) r["preview"] = clean[:100] return jsonify({"ok": True, "data": rows, "total": total, "page": page, "per_page": per_page}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/talker/") def api_talker_info(talker): try: info = wechat_db.get_talker_info(talker) if info: return jsonify({"ok": True, "data": info}) return jsonify({"ok": False, "error": "not found"}), 404 except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/messages") def api_messages(): talker = request.args.get("talker", "") limit = int(request.args.get("limit", "40")) offset = int(request.args.get("offset", "0")) if not talker: return jsonify({"ok": False, "error": "需要 talker"}), 400 try: rows = wechat_db.get_messages(talker, limit=limit, offset=offset) # Batch lookup sender nicknames sender_usernames = set() for row in rows: su = row.get("sender_username") if su: sender_usernames.add(su) nickname_map = wechat_db.get_talker_infos_batch(list(sender_usernames)) if sender_usernames else {} parsed = [] for row in rows: p = _parse_message_content(row.get("content", "") or "", row.get("imgPath", "") or "", row.get("type", 0)) p["msgId"] = row.get("msgId"); p["type"] = row.get("type") p["isSend"] = row.get("isSend"); p["createTime"] = row.get("createTime") p["talker"] = row.get("talker") su = row.get("sender_username") p["sender_username"] = su p["sender_nickname"] = nickname_map.get(su) if su else None parsed.append(p) return jsonify({"ok": True, "data": parsed}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 # ── 消息内容解析 ──────────────────────────────────── def _parse_message_content(content: str, img_path: str, msg_type: int) -> dict: result = {"text": "", "images": [], "voice": None, "articles": []} clean_content = content or "" clean_content = re.sub(r"", r"\1", clean_content, flags=re.DOTALL) display_text = clean_content if clean_content and "\n" in clean_content and ":" in clean_content.split("\n")[0]: parts = clean_content.split("\n", 1) if len(parts) == 2: display_text = parts[1] if msg_type == 1: result["text"] = re.sub(r"<[^>]+>", "", display_text).strip() if "" in display_text else display_text.strip() elif msg_type == 3: if img_path: result["images"].append({"src": img_path}) result["text"] = "[图片]" elif msg_type == 34: if img_path: result["voice"] = {"path": img_path} elif msg_type == 49 or msg_type == 285212721: url_match = re.search(r"(.*?)", clean_content) title_match = re.search(r"(.*?)", clean_content) desc_match = re.search(r"(.*?)", clean_content) thumb_match = re.search(r"(.*?)", clean_content) if url_match or title_match: result["articles"].append({"url": url_match.group(1) if url_match else "", "title": html_module.unescape(title_match.group(1)) if title_match else "", "description": html_module.unescape(desc_match.group(1)) if desc_match else "", "thumbnail": thumb_match.group(1) if thumb_match else ""}) txt = re.sub(r"<[^>]+>", "", display_text).strip() if txt: result["text"] = txt elif msg_type == 47: result["text"] = "[表情]" elif msg_type == 43: result["text"] = "[视频]" elif msg_type == 754974769 or "" in clean_content: cdata_clean = re.sub(r"", r"\1", clean_content) fnick = re.search(r"(.*?)", cdata_clean) favatar = re.search(r"(.*?)", cdata_clean) fdesc = re.search(r"(.*?)", cdata_clean) fcover = re.search(r"(.*?)", cdata_clean) fthumb = re.search(r"(.*?)", cdata_clean) fvideo = re.search(r".*?(.*?)", cdata_clean, re.DOTALL) fdur = re.search(r"(\d+)", cdata_clean) result["finder"] = {"nickname": html_module.unescape(fnick.group(1).strip()) if fnick else "", "avatar": favatar.group(1).strip() if favatar else "", "desc": html_module.unescape(fdesc.group(1).strip()) if fdesc else "", "cover": (fcover.group(1) or fthumb.group(1)).strip() if (fcover or fthumb) else "", "video_url": fvideo.group(1).strip() if fvideo else "", "duration": fdur.group(1) if fdur else "0"} elif msg_type == 822083633: display_name = re.search(r"(.*?)", clean_content) refer_title = re.search(r"(.*?)", clean_content) result["text"] = f"[引用 {display_name.group(1)}: {refer_title.group(1)}]" if display_name and refer_title else "[引用消息]" elif msg_type == 10000: result["text"] = "[系统消息]" return result # ── 媒体文件代理 ──────────────────────────────────── @app.route("/media/image") def serve_image(): path = request.args.get("path", "") if not path: return "no path", 400 real_path = _resolve_image_path(path) if not real_path: return "image not found", 404 tmp_path = f"/tmp/img_{abs(hash(real_path))}" if not _safe_exists(tmp_path): subprocess.run(["su", "-c", f"cp {real_path} {tmp_path} && chmod 644 {tmp_path}"], capture_output=True, text=True, timeout=10) if _safe_exists(tmp_path): return send_file(tmp_path) return "image not found", 404 @app.route("/media/image/full") def serve_image_full(): path = request.args.get("path", "") if not path: return "no path", 400 hash_part = path.replace("THUMBNAIL_DIRPATH://", "").replace("THUMBNAIL_DIR_PATH://", "") clean = hash_part.replace("th_", "") full_path = _resolve_image_path(f"THUMBNAIL_DIRPATH://{clean}") if full_path: tmp_path = f"/tmp/img_full_{abs(hash(full_path))}" if not _safe_exists(tmp_path): subprocess.run(["su", "-c", f"cp {full_path} {tmp_path} && chmod 644 {tmp_path}"], capture_output=True, text=True, timeout=10) if _safe_exists(tmp_path): return send_file(tmp_path) return serve_image() @app.route("/media/voice") def serve_voice(): path = request.args.get("path", "") if not path: return "no path", 400 real_path = _resolve_voice_path(path) if not real_path: return "voice not found", 404 tmp_amr = f"/tmp/voice_{abs(hash(real_path))}.amr" if not _safe_exists(tmp_amr): subprocess.run(["su", "-c", f"cp {real_path} {tmp_amr} && chmod 644 {tmp_amr}"], capture_output=True, text=True, timeout=10) if not _safe_exists(tmp_amr): return "voice not found", 404 mp3_path = tmp_amr.replace(".amr", ".mp3") if not _safe_exists(mp3_path): subprocess.run(["ffmpeg", "-y", "-i", tmp_amr, "-acodec", "libmp3lame", "-ab", "32k", mp3_path], capture_output=True, text=True, timeout=30) if _safe_exists(mp3_path): return send_file(mp3_path, mimetype="audio/mpeg") return send_file(tmp_amr) def _resolve_image_path(path: str) -> str | None: """THUMBNAIL_DIRPATH://th_ → image2/XX/YY/.jpg (原图优先) 或 th_ (缩略图)""" if not path: return None if path.startswith("/") and _safe_exists(path): return path path2 = path.replace("THUMBNAIL_DIRPATH://", "").replace("THUMBNAIL_DIR_PATH://", "") if not path2: return None hash_part = path2.replace("th_", "") if len(hash_part) >= 4: sub1, sub2 = hash_part[:2], hash_part[2:4] base = f"{config.get_current_account()['data_root']}/image2/{sub1}/{sub2}" for c in [f"{base}/{hash_part}.jpg", f"{base}/{hash_part}.png", f"{base}/{hash_part}", f"{base}/{path2}"]: if _safe_exists(c): return c for ext in [".jpg", ".png", ""]: c = f"{config.get_current_account()['data_root']}/image2/{hash_part}{ext}" if _safe_exists(c): return c return None def _resolve_voice_path(path: str) -> str | None: if not path: return None if _safe_exists(path): return path clean = path.replace("THUMBNAIL_DIRPATH://", "").replace("THUMBNAIL_DIR_PATH://", "") try: r = subprocess.run(["su", "-c", f"find {config.get_current_account()['data_root']}/voice2 -name '*{clean}*' 2>/dev/null | head -1"], capture_output=True, text=True, timeout=10) found = r.stdout.strip() if found: return found except Exception: pass return None # ── Avatar 服务 ────────────────────────────────────── def _resolve_avatar_path(username: str) -> str | None: """username → MD5 → avatar/{hash[:2]}/{hash[2:4]}/user_hd_{hash}.png""" if not username: return None h = hashlib.md5(username.encode()).hexdigest() base = f"{config.get_current_account()['data_root']}/avatar/{h[:2]}/{h[2:4]}" # Try HD first, then regular, then inside hash directory for fname in [f"user_hd_{h}.png", f"user_{h}.png"]: path = f"{base}/{fname}" if _safe_exists(path): return path # Also try inside the full hash subdirectory for fname in [f"user_hd_{h}.png", f"user_{h}.png"]: path = f"{base}/{h}/{fname}" if _safe_exists(path): return path return None @app.route("/media/avatar/") def serve_avatar(username): real_path = _resolve_avatar_path(username) if not real_path: return "avatar not found", 404 tmp_path = f"/tmp/avatar_{abs(hash(real_path))}.png" if not _safe_exists(tmp_path): subprocess.run(["su", "-c", f"cp {real_path} {tmp_path} && chmod 644 {tmp_path}"], capture_output=True, text=True, timeout=10) if _safe_exists(tmp_path): return send_file(tmp_path, mimetype="image/png") return "avatar not accessible", 404 # ── Page 2: 数据中心 ──────────────────────────────── @app.route("/circle") def circle_page(): return render_template("circle.html") @app.route("/api/circle/stats") def api_circle_stats(): try: rows = wechat_db.query("SELECT COUNT(*) as cnt FROM message") messages_total = rows[0]["cnt"] if rows else 0 rows = wechat_db.query("SELECT COUNT(DISTINCT talker) as cnt FROM message WHERE talker LIKE '%@chatroom'") groups_total = rows[0]["cnt"] if rows else 0 conn = local_db.get_conn() images_total = conn.execute("SELECT COUNT(*) FROM image_analysis").fetchone()[0] reports_total = conn.execute("SELECT COUNT(*) FROM daily_reports WHERE report_date NOT IN ('summary','learn')").fetchone()[0] conn.close() return jsonify({"ok": True, "data": {"messages_total": messages_total, "groups_total": groups_total, "images_total": images_total, "reports_total": reports_total}}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 # ── Page 3: 日报系统 ───────────────────────────────── @app.route("/reports") def reports_page(): return render_template("reports.html") @app.route("/summaries") def summaries_page(): return render_template("reports.html", page_type="summary") @app.route("/learn-records") def learn_records_page(): return render_template("reports.html", page_type="learn") @app.route("/report/") def report_detail(report_id): return render_template("report_detail.html", report_id=report_id) @app.route("/api/reports") def api_reports(): try: report_type = request.args.get("type", "digest") rows = local_db.report_list(report_type=report_type) for r in rows: r["summary"] = (r.get("content", "") or "")[:200] r.pop("content", None) return jsonify({"ok": True, "data": rows}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/report/") def api_report_detail(report_id): try: row = local_db.report_get(report_id) if row: return jsonify({"ok": True, "data": row}) return jsonify({"ok": False, "error": "not found"}), 404 except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/reports/generate", methods=["POST"]) def api_report_generate(): data = request.json or {} chatroom_id = data.get("chatroom_id", "") date_str = data.get("date", "") if not chatroom_id or not date_str: return jsonify({"ok": False, "error": "需要 chatroom_id 和 date"}), 400 try: import uuid talker_info = wechat_db.get_talker_info(chatroom_id) chatroom_name = str(talker_info.get("displayname", chatroom_id)) if talker_info else chatroom_id if isinstance(chatroom_name, dict): chatroom_name = chatroom_id task_id = uuid.uuid4().hex[:12] task_file = f"/data/data/com.termux/files/home/tmp/report_task_{task_id}.json" result_file = f"/data/data/com.termux/files/home/tmp/report_result_{task_id}.md" with open(task_file, "w") as f: json.dump({"chatroom_id": chatroom_id, "date": date_str, "result_file": result_file}, f) prompt = (f"你的任务是生成一份微信群的日报。\n\n群聊 ID: {chatroom_id}\n日期: {date_str}\n\n" f"## 步骤（简洁高效，不要过度解析）\n\n" f"### 1. 查消息列表\n" f"用 terminal curl 调 http://127.0.0.1:5001/api/messages?talker={chatroom_id}&date={date_str}\n" f"拿到消息列表即可，**不要逐条解析 finder/voice/image 内容**。\n" f"消息类型标注为文本/图片/语音/文章/视频号即可，不需要提取具体内容。\n\n" f"### 2. 提取文章并查摘要\n" f"从消息中提取所有 type=49（文章）的 URL 和标题。\n" f"用 terminal curl 调 POST http://127.0.0.1:5001/api/articles/check-summaries\n" f"批量查哪些已有摘要（JSON: {{\"urls\":[{{\"url\":\"...\",\"title\":\"...\"}}], \"type\":\"summary\"}}）\n\n" f"### 3. 生成日报\n" f"- 已有摘要的文章：直接引用摘要内容\n" f"- **没有摘要的文章：只列标题和链接，标注「⏳ 待摘要」，不要调 summarize API，不要等待**\n" f"- 按主题分组 + 讨论亮点\n" f"- 写入 {result_file}\n\n" f"**重要：不要解析图片/视频号/语音/文件的具体内容。不要逐条写 Python 脚本解析消息。用 API 就够了。**") def _on_report_done(tid, ok): try: if Path(result_file).exists(): md_content = Path(result_file).read_text(encoding="utf-8", errors="replace") else: log_path = str(HERMES_JOBS_DIR / f"{tid}.log") md_content = "# 生成失败\n\n请查看日志了解详情。" local_db.report_add(chatroom_id, chatroom_name, date_str, md_content, 0, 0, 0) except Exception: pass try: Path(task_file).unlink() except: pass try: Path(result_file).unlink() except: pass _spawn_hermes_job("daily_report", f"{chatroom_name} {date_str}", ["hermes", "chat", "-q", prompt, "-s", "wechat-group-digest", "--yolo", "-t", "terminal,file,web", "-m", "deepseek-v4-pro", "--provider", "deepseek"], timeout=600, on_done=_on_report_done, task_id=task_id) return jsonify({"ok": True, "data": {"status": "started", "task_id": task_id}}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/report//delete", methods=["POST"]) def api_report_delete(report_id): try: local_db.report_delete(report_id) return jsonify({"ok": True}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/reports/generate/status") def api_report_status(): try: rows = local_db.report_list(1) if rows: return jsonify({"ok": True, "data": rows[0]}) return jsonify({"ok": True, "data": None}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 # ── Page 4: 日报图片 ───────────────────────────────── @app.route("/report-images") def report_images_page(): return render_template("report_images.html") @app.route("/api/report-images") def api_report_images(): try: rows = local_db.report_image_list() return jsonify({"ok": True, "data": rows}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/media/report-image/") def serve_report_image(image_id): conn = local_db.get_conn() row = conn.execute("SELECT image_path FROM report_images WHERE id = ?", (image_id,)).fetchone() conn.close() if row and row["image_path"] and _safe_exists(row["image_path"]): return send_file(row["image_path"]) return "image not found", 404 @app.route("/api/reports//generate-image", methods=["POST"]) def api_report_generate_image(report_id): try: report = local_db.report_get(report_id) if not report: return jsonify({"ok": False, "error": "日报不存在"}), 404 import uuid task_id = uuid.uuid4().hex[:12] content_file = f"/data/data/com.termux/files/home/tmp/report_content_{task_id}.md" img_file = f"/data/data/com.termux/files/home/tmp/report_img_{task_id}.png" Path(content_file).write_text(report["content"] or "", encoding="utf-8") chatroom_name = report.get("chatroom_name", "群聊") report_date = report.get("report_date", "") prompt = ( f"按 gpt-image-gen skill 的日报图片 Prompt 生成报纸图片。\n" f"日报文件: {content_file}\n输出图片: {img_file}\n" f"群名: {chatroom_name}\n日期: {report_date}\n\n" f"步骤:\n" f"1. read_file 读完整日报内容\n" f"2. 按 skill 提炼: 群名+日期(报头)、选定头条主题+配大图、所有文章的标题+摘要(多栏)、群友金句(原文)、供稿名单\n" f"3. 按 skill 的 Prompt 限制规则生成图片 prompt: 英文描述结构，关键内容(标题/金句/人名)用中文\n" f"4. GPT Image 2 API (model='GPT Image 2', n=1, size='1024x1024')\n" f"5. base64解码保存到 {img_file}, ls 确认\n\n" f"禁止: 第X版、详见、参见、X版——这是单页群日报" ) STORAGE_DIR = Path("/data/data/com.termux/files/home/wechat-viewer/static/report_images") STORAGE_DIR.mkdir(parents=True, exist_ok=True) def _on_img_done(tid, ok): try: if Path(img_file).exists(): # 移到永久存储，不在 DB 入库后删除 dest = STORAGE_DIR / f"report_{report_id}_{tid}.png" import shutil shutil.move(img_file, str(dest)) local_db.report_image_add(report_id, str(dest), "wsj") try: Path(content_file).unlink() except: pass except Exception: pass _spawn_hermes_job("report_image", f"{chatroom_name} {report_date}", ["hermes", "chat", "-q", prompt, "-s", "gpt-image-gen", "--yolo", "-t", "terminal,web,file"], timeout=300, on_done=_on_img_done, task_id=task_id) return jsonify({"ok": True, "data": {"status": "started", "task_id": task_id}}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/report-image//delete", methods=["POST"]) def api_report_image_delete(image_id): try: conn = local_db.get_conn() row = conn.execute("SELECT image_path FROM report_images WHERE id = ?", (image_id,)).fetchone() if row and row["image_path"]: try: Path(row["image_path"]).unlink() except: pass conn.close() local_db.report_image_delete(image_id) return jsonify({"ok": True}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/report-image//send", methods=["POST"]) def api_report_image_send(image_id): data = request.json or {} wxid = data.get("to", "") if not wxid: return jsonify({"ok": False, "error": "需要 to (wxid)"}), 400 try: conn = local_db.get_conn() row = conn.execute("SELECT image_path FROM report_images WHERE id = ?", (image_id,)).fetchone() conn.close() if not row or not row["image_path"]: return jsonify({"ok": False, "error": "图片不存在"}), 404 import shutil dest = f"/sdcard/Pictures/report_{image_id}.png" shutil.copy2(row["image_path"], dest) import sys def _send(): try: sys.path.insert(0, "/data/data/com.termux/files/home/.hermes/event_bus") from wechat_hook_send import send_image_via_hook send_image_via_hook(wxid, dest) except Exception: pass threading.Thread(target=_send, daemon=True).start() return jsonify({"ok": True, "data": {"status": "sent"}}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 # ── Page 5: 任务监控 ───────────────────────────────── @app.route("/jobs") def jobs_page(): return render_template("jobs.html") @app.route("/job/") def job_detail_page(task_id): return render_template("job_detail.html", task_id=task_id) @app.route("/api/jobs") def api_jobs(): try: rows = local_db.job_list(200) return jsonify({"ok": True, "data": rows}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/job/") def api_job_detail(task_id): try: row = local_db.job_get(task_id) if row: return jsonify({"ok": True, "data": row}) return jsonify({"ok": False, "error": "not found"}), 404 except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/job//log") def api_job_log(task_id): try: row = local_db.job_get(task_id) if not row or not row.get("log_file"): return jsonify({"ok": False, "error": "not found"}), 404 log_path = row["log_file"] offset_val = request.args.get("offset", "0") # offset can be line number or byte position (for polling) try: with open(log_path, "r") as f: content = f.read() except FileNotFoundError: return jsonify({"ok": True, "data": {"content": "", "size": 0, "status": row.get("status", "unknown")}}) return jsonify({"ok": True, "data": { "content": content, "size": len(content), "status": row.get("status", "unknown"), "started_at": row.get("started_at", ""), "completed_at": row.get("completed_at", "") }}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/job//delete", methods=["POST"]) def api_job_delete(task_id): try: row = local_db.job_get(task_id) if row and row.get("log_file"): try: Path(row["log_file"]).unlink() except: pass local_db.job_delete(task_id) return jsonify({"ok": True}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 # ── Page 6: 学习审核 ───────────────────────────────── @app.route("/learn") def learn_page(): return render_template("learn.html") @app.route("/learn/") def learn_detail_page(item_id): """学习项详情页 — 展示 Step 4 吸收报告 + Step 6 应用审计（如有）""" item = local_db.learn_get(item_id) if not item: return "Not Found", 404 # Step 4: 取最近的 非apply_audit 类型结果（吸收报告） result = local_db.learn_result_get_step4(item_id) # Step 6: 取 apply_audit 类型结果（知识应用审计） apply_result = local_db.learn_result_get_apply_audit(item_id) return render_template("learn_detail.html", item=item, result=result, apply_result=apply_result) @app.route("/api/learn-queue") def api_learn_queue(): try: status = request.args.get("status", "pending_review") rows = local_db.learn_list(status) return jsonify({"ok": True, "data": rows}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/learn/classify", methods=["POST"]) def api_learn_classify(): """对未分类的文章摘要运行分类器""" try: summaries = local_db.learn_get_unclassified_summaries(20) if not summaries: return jsonify({"ok": True, "data": {"classified": 0, "message": "无未分类摘要"}}) import uuid task_id = uuid.uuid4().hex[:12] summaries_file = f"/data/data/com.termux/files/home/tmp/classify_input_{task_id}.json" output_file = f"/data/data/com.termux/files/home/tmp/classify_output_{task_id}.json" input_data = [{"dedup_key": s["dedup_key"], "title": s.get("chatroom_name", ""), "article_url": s.get("article_url", ""), "summary": (s.get("content", "") or "")[:1000]} for s in summaries if s.get("dedup_key")] import json Path(summaries_file).write_text(json.dumps(input_data, ensure_ascii=False, indent=2)) prompt = f"""你是文章评分分类器。阅读文件 {summaries_file} 中的文章摘要，对每篇判断学习类型并写入 {output_file}。 ## 评分标准 (score: 1-10) 评估这篇文章对 Hermes AI Agent 的价值： - 9-10: 直接可用的工具/API/代码，显著提升 Hermes 能力 - 7-8: 高质量方法论/架构，值得深入学习后构建 - 5-6: 行业动态/趋势，有助于决策上下文 - 3-4: 一般资讯，可能有参考价值 - 1-2: 低价值，不值得花时间 ## 分类规则 - skill: 有 GitHub/PyPI 链接、可安装运行的开源工具/框架 - memory: 行业趋势、市场数据、竞品动态等需要记住的事实 - todo: 待跟进的项目、产品发布、需要采取行动的事项 - reference: 技术架构、设计模式、方法论等值得存档的知识 - skip: 泛泛资讯，不值得深入学习 ## 输出格式 (JSON) {{"results":[{{"dedup_key":"...", "learn_type":"skill", "score":8, "confidence":"high", "learn_target":"skill-name", "learn_content":"# 要学习的内容\\n\\n核心要点..."}}, ...]}} ## learn_target 规则 - skill: 建议的 skill 名称 (小写, 用连字符) - memory: 记忆键名 (描述性) - todo: 待办标题 - reference: 目标 skill 名或 "new:topic-name" - skip: 空字符串 完成后不要做其他操作。""" def _on_classify_done(tid, ok, spawn_reason=None): try: if not ok: # ── HydraMind Gate: 分类器失败 trace ── # 对齐四元组: node/attempt/gate/reason_code # 与 spawn gate 不同：分类器失败需要 REPLAN（人工判断是否重试），而非 BLOCK classify_trace = ( f"[node=classify,attempt=1,gate=RUNTIME_CLASSIFY_FAILED," f"action={GateAction.REPLAN}] " f"分类子进程异常退出 task_id={tid}") # 写入数据库供 UI 展示（learn_queue 中最后一条相关记录） try: conn = local_db.get_conn() latest = conn.execute( "SELECT id FROM learn_queue WHERE status='unclassified' OR status='pending_review' " "ORDER BY id DESC LIMIT 1").fetchone() if latest: conn.execute("UPDATE learn_queue SET error_msg=? WHERE id=?", (classify_trace, latest["id"])) conn.commit() conn.close() except Exception: pass # 同时保留文件日志用于离线排查 try: with open(f"/data/data/com.termux/files/home/tmp/classify_error_{tid}.log", "w") as ef: ef.write(f"=== {classify_trace} ===\n") except Exception: pass if Path(output_file).exists(): results = json.loads(Path(output_file).read_text()) for r in results.get("results", []): dedup = r.get("dedup_key", "") ltype = r.get("learn_type", "skip") if ltype == "skip": continue # Find original summary orig = next((s for s in summaries if s.get("dedup_key") == dedup), None) if not orig: continue # Dedup: skip if already in learn_queue for this type existing = local_db.learn_find_by_dedup(dedup, ltype) if existing: continue # Score-based routing: ≥7 cron自动学, 4-6 人工审, 1-3 丢弃 score = int(r.get("score", 0) or 0) if score >= 7: status = "pending_learn" confidence = r.get("confidence", "high") elif score >= 4: status = "pending_review" confidence = r.get("confidence", "medium") else: continue # 低分不写入 # ── HydraMind Gate: 分类器输出质量门控 ── # 依据可验证事实（learn_content 长度）决策，非 LLM 评分 learn_content = r.get("learn_content", "") or "" if not learn_content or len(learn_content.strip()) < 20: # 空内容不能进学习队列——写 REPLAN gate 让人工判断 classify_bad_trace = ( f"[node=classify,attempt=1,gate=CONTENT_EMPTY," f"action={GateAction.REPLAN}] " f"type={ltype} target={r.get('learn_target','?')} score={score}" f" — learn_content 为空或过短，分类器可能幻觉") try: with open(f"/data/data/com.termux/files/home/tmp/classify_content_empty_{tid}.log", "a") as ef: ef.write(f"=== {classify_bad_trace} ===\n dedup={dedup}\n content={learn_content[:100]}\n") except Exception: pass continue # 不入库，等下一次分类器重试\n # ── HydraMind Gate: 分类路由 trace ── # 补齐四元组：分类成功也有 gate trace，不只是失败路径 route_reason = "pending_learn" if score >= 7 else "pending_review" route_trace = ( f"[node=classify,attempt=1,gate=ROUTED," f"action={GateAction.CONTINUE}] " f"score={score} type={ltype} → {route_reason}") new_id = local_db.learn_add( article_url=orig.get("article_url", ""), title=orig.get("chatroom_name", ""), dedup_key=dedup, learn_type=ltype, learn_target=r.get("learn_target", ""), learn_content=r.get("learn_content", ""), confidence=confidence, score=score, status=status ) local_db.learn_update_status(new_id, status, route_trace) try: Path(summaries_file).unlink() except: pass try: Path(output_file).unlink() except: pass except Exception as e: # ── HydraMind Gate: 分类回调异常 ── classify_cb_trace = ( f"[node=classify,attempt=1,gate=CLASSIFY_CALLBACK_ERR," f"action={GateAction.REPLAN}] " f"分类回调异常: {str(e)[:200]}") try: with open(f"/data/data/com.termux/files/home/tmp/classify_error_{tid}.log", "a") as ef: ef.write(f"=== {classify_cb_trace} ===\n") except Exception: pass _spawn_hermes_job("learn_classify", f"分类 {len(input_data)} 篇", ["hermes", "chat", "-q", prompt, "-s", "learn-and-build", "--yolo", "-t", "terminal,file,web"], timeout=180, on_done=_on_classify_done, task_id=task_id) return jsonify({"ok": True, "data": {"classified": len(input_data), "task_id": task_id}}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 def _run_absorb(item_id: int): """统一吸收函数：6 步 Loop — Step 1-4: 学习摘要 → 消化吸收 → 验证测试 → 生成报告; Step 5: 吸收验证; Step 6: 知识应用审计""" try: item = local_db.learn_get(item_id) if not item: return import uuid tid = uuid.uuid4().hex[:12] content_file = f"/data/data/com.termux/files/home/tmp/learn_absorb_{tid}.md" Path(content_file).write_text(item.get("learn_content", "") or "", encoding="utf-8") ltype = item["learn_type"] target = item.get("learn_target", "") prompt = f"""你是 Hermes 的知识吸收引擎。按 extract-learning-outline skill 的四步 Loop 执行吸收。 **铁律：没有系统影响的吸收等于垃圾。吸收完成后，必须能回答：Hermes 下次遇到什么场景会有什么不同的行为？如果答不上来 → 标记 rejected，不吸收。** 文章学习内容: {content_file} 吸收类型: {ltype} 吸收目标: {target} ## Step 1: 制作学习摘要 1. 读 {content_file} 的原文内容 2. 扫描 Hermes 当前状态 — skill_view 看相关 skill、search_files 搜代码、memory 查记忆 3. 写一份「学习摘要」到 {content_file}.summary.md： - 这篇文章的核心知识点（1-5 条，每条一句话） - Hermes 已有多少覆盖？新知识占比？ - **关键判断：这篇文章能让 Hermes 做什么之前做不到的事？如果不能 → 直接 rejected** **覆盖度判断铁律：文档里有 ≠ 实现了。用以下量表强制打分：** | 判据 | 加分 | |------|:--:| | SKILL.md 提到概念 | +1 | | 代码有对应模块 | +3 | | memory 有结构化记录 | +2 | | 有可执行测试 | +2 | | 近7天实际使用过 | +2 | **总分 ≤4 → 必须吸收，不能 reject。** 常见误判：SKILL.md 写了「Memory五类分类」→ 只得1分，应吸收。 **吸收深度分级**（不是吸/不吸二值）： - L1: 文档追加（skill patch / memory add）— 当前工具能做 - L2: 结构化（需改代码/memory schema）— 标「部分吸收:需L2」传给Step6 - L3: 代码重构（改core逻辑）— 标「部分吸收:需L3」传给Step6 ## Step 2: 消化吸收 基于学习摘要，制定 Planning 到 {content_file}.plan.md： (a) 吸收目标 — 具体学什么、达到什么效果 (b) 测试用例 — 至少 2 个场景，描述吸收后 Hermes 应有的不同反应 (c) 关键要点 — 需要记住的 API/方法/结论 (d) 引用路径 — 代码路径、skill 名、外部链接 然后**实际调用工具**执行吸收（不能用文字替代）： ### 所有类型统一规则：必须产出可验证的系统变更 - **skill**: skill_manage(action='create'/'patch')，完整 SKILL.md。创建后用 skill_view 确认。 - **memory**: memory(action='add', target='memory')。写入后验证。 - **reference**: 不是写孤立 MD 文件！必须： 1. 扫描所有可用 skill，找到最相关的一个 2. skill_manage(action='patch') 把核心知识作为新章节加入目标 skill 的 SKILL.md 3. 如果涉及跨 skill 的通用认知 → memory add 4. **如果完全无法匹配任何 skill 或 memory → 标记 rejected（reason: "无可内化目标"），不吸收** ## Step 3: 验证测试 执行 Benchmark： 1. 对 Step 2 的测试用例，逐一验证 2. skill 创建/更新后用 skill_view 确认 3. memory 写入后确认存在 4. 如果吸收后的 Hermes 能通过测试用例 → 成功 5. 如果失败 → 记录失败原因，标记 failed **Step 级 trace**：每一步结束时写入 checkpoint 文件供外部 Gate 审计： - Step 1 完成 → 写 `[node=absorb_step1,gate=STEP_DONE,action=CONTINUE]` 到 `{content_file}.step1.done` - Step 2 完成 → 写 `[node=absorb_step2,gate=STEP_DONE,action=CONTINUE]` 到 `{content_file}.step2.done` - Step 3 完成 → 写 `[node=absorb_step3,gate=STEP_DONE,action=CONTINUE]` 到 `{content_file}.step3.done` - 任一步失败 → 写 `[node=absorb_stepN,gate=STEP_FAILED,action=REVISE] reason:...` 到 `{content_file}.stepN.failed` 这样即使吸收中途崩溃，也能通过 checkpoint 文件定位失败步骤。 ## Step 4: 生成报告 写入 {content_file}.result.md，严格按此结构（前端依赖这些字段渲染卡片）： ``` # 学习完成报告 ## 学习内容 - 标题: [文章标题] - 来源: [URL 或文件路径] - 类型: {ltype} - 日期: [YYYY-MM-DD] ## 吸收产出 ### 创建/更新 Skill - **[skill-name]**: create/patch +N行 | 新增「章节名」 （如无则写「无」） ### 写入 Memory - "关键认知" → memory.json （如无则写「无」） ### 代码变更 - [path/to/file.py]: +N行 -M行 | 改了什么 （如无则写「无」） ### 安装/集成 - pip install xxx → 成功/失败 （如无则写「无」） ## 验证结果 - TC1: [PASS/FAIL] — [场景 → 验证了什么] - TC2: [PASS/FAIL] — [场景 → 验证了什么] ## 行为变化 - 下次遇到 [X 场景] → Hermes 会 [Y 不同反应] ## 未完成/待跟进 - [原因] （如无则写「无」） ``` 如果没有产出任何 skill/memory → 报告标题写「REJECTED: 无可内化内容」。 注意：每一步都要实际执行工具调用，不能用文字描述代替。用 --yolo 模式。""" def _on_absorb_done(tid2, ok, spawn_reason=None): try: result_file = f"{content_file}.result.md" plan_file = f"{content_file}.plan.md" summary_file = f"{content_file}.summary.md" is_rejected = False # default: not rejected content_thin = False # Gate: content quality check if ok and Path(result_file).exists(): result_text = Path(result_file).read_text()[:1200] # ── Gate 密度加强: content quality gate ── # 依据可验证事实（文本长度）决策，非 LLM 评分 if len(result_text.strip()) < 50: content_thin = True # HydraMind REVISE: content thin → auto-retry absorb (not BLOCK) # Must set status='pending_learn' so cron picks it up for retry local_db.learn_update_status(item_id, "pending_learn", f"[node=absorb,attempt=1,gate={VerifyReason.CONTENT_THIN}," f"action={GateAction.REVISE}] " f"{VERIFY_REASON_CODE[VerifyReason.CONTENT_THIN]}") # Gate: 行为变化缺失 — 吸收没有产生系统影响 elif "## 行为变化" not in result_text and "REJECTED" not in result_text: # 宽松处理：如果「吸收产出」章节有内容，仍放行但加 gate trace if "## 吸收产出" in result_text: # Soft-pass: 有产出但缺行为变化声明 → CONTINUE with trace soft_trace = ( f"[node=absorb,attempt=1,gate=NO_BEHAVIOR_CHANGE_SOFT," f"action={GateAction.CONTINUE}] " f"报告有吸收产出但缺「行为变化」章节，放行但需人工复核") local_db.learn_update_status(item_id, "applied", soft_trace + " | " + result_text[:300]) # Build outline + reflection for learning_results (needed by Step 5) outline_parts = [] for f in [summary_file, plan_file]: if Path(f).exists(): outline_parts.append(Path(f).read_text()[:500]) outline = "\n\n".join(outline_parts) local_db.learn_result_add( learn_queue_id=item_id, article_url=item.get("article_url", ""), article_title=item.get("title", ""), outline=outline, result_type=ltype, result_summary=result_text[:500], created_skill=target if ltype == "skill" else "", auto_applied=1 ) else: local_db.learn_update_status(item_id, "failed", f"[node=absorb,attempt=1,gate={VerifyReason.NO_BEHAVIOR_CHANGE}," f"action={GateAction.REPLAN}] " f"吸收报告缺少「行为变化」章节，无法确认系统影响（建议人工审核后 /relearn）") # 检测 REJECTED：无系统影响的吸收不应该标 applied elif "REJECTED" in result_text[:100]: is_rejected = True local_db.learn_update_status(item_id, "rejected", f"[node=absorb,attempt=1,gate=REJECTED," f"action={GateAction.BLOCK}] " f"{result_text[:300] or '无可内化内容'}") else: # 仅在正常成功路径（非 soft-pass）执行 # ── Gate 密度加强: Step 1 摘要质量 gate ── # 吸收子进程不可见，事后检查 Step 1 产物是否存在且有实质内容 summary_ok = (Path(summary_file).exists() and len(Path(summary_file).read_text()[:100].strip()) >= 30) absorb_trace = ( f"[node=absorb,attempt=1,gate=PASS," f"action={GateAction.CONTINUE}] " f"summary={'ok' if summary_ok else 'thin'}") # Prepend gate trace to error_msg (replaces the result_text-only msg) local_db.learn_update_status(item_id, "applied", absorb_trace + " | " + result_text[:300]) # Build outline from summary + plan + reflection outline_parts = [] for f in [summary_file, plan_file]: if Path(f).exists(): outline_parts.append(Path(f).read_text()[:500]) outline = "\n\n".join(outline_parts) reflection = result_text[:500] local_db.learn_result_add( learn_queue_id=item_id, article_url=item.get("article_url", ""), article_title=item.get("title", ""), outline=outline, result_type=ltype, result_summary=reflection, created_skill=target if ltype == "skill" else "", auto_applied=1 ) else: # ── HydraMind Gate: spawn failure → check Step-level checkpoints ── # Step trace files let us pinpoint which step the LLM failed at step_checks = [] for step_n in [1, 2, 3]: sf = Path(f"{content_file}.step{step_n}.failed") df = Path(f"{content_file}.step{step_n}.done") if sf.exists(): step_checks.append(sf.read_text()[:80]) try: sf.unlink() except: pass elif df.exists(): try: df.unlink() except: pass if step_checks: rc = VerifyReason.NO_OUTPUT local_db.learn_update_status(item_id, "failed", f"[node=spawn,attempt=1,gate={rc}," f"action={GateAction.REPLAN}] " f"Step-level failures: {'; '.join(step_checks)[:300]}") else: # No step checkpoints → spawn-layer diagnostic fallback rc = spawn_reason if spawn_reason else VerifyReason.NO_OUTPUT ga = GateAction.REVISE if rc in (VerifyReason.SPAWN_TIMEOUT,) else GateAction.BLOCK local_db.learn_update_status(item_id, "failed", f"[node=spawn,attempt=1,gate={rc}," f"action={ga}] " f"{VERIFY_REASON_CODE.get(rc, '子进程执行失败')}") # ── Step 5+6: 验证吸收 → 知识应用（必须在 cleanup 之前！）── should_apply = ok and not is_rejected and not content_thin # Cleanup（先保留 result_text 引用，cleanup 后再用） for f in [content_file, plan_file, summary_file]: try: Path(f).unlink() except: pass # Cleanup step checkpoint files for step_n in [1, 2, 3]: for suffix in [".done", ".failed"]: try: Path(f"{content_file}.step{step_n}{suffix}").unlink() except: pass # result_file 最后删——_verify_and_apply 不需要它 if should_apply: # ── HydraMind Gate: absorb→step5 transition gate trace ── # Emit explicit gate at the between-node boundary for auditability gate_rc = VerifyReason.PASS if not ("BEHAVIOR_CHANGE_SOFT" in (item.get("error_msg","") or "")) else "NO_BEHAVIOR_CHANGE_SOFT" local_db.learn_update_status(item_id, "applied", f"{item.get('error_msg','') or ''} | " f"[node=absorb_to_step5,attempt=1,gate={gate_rc}," f"action={GateAction.CONTINUE}] 进入 Step5 验证") _verify_and_apply(item_id, ltype, target) try: Path(result_file).unlink() except: pass except Exception as e: local_db.learn_update_status(item_id, "failed", f"[node=absorb,attempt=1,gate={VerifyReason.NO_OUTPUT}," f"action={GateAction.BLOCK}] " f"on_done 异常: {str(e)[:150]}") _spawn_hermes_job("learn_absorb", f"{ltype}: {target[:40]}", ["hermes", "chat", "-q", prompt, "--yolo", "-t", "terminal,file,web", "-s", "extract-learning-outline,learn-and-build"], timeout=600, on_done=_on_absorb_done, task_id=tid) except Exception: local_db.learn_update_status(item_id, "failed", "吸收启动失败") # ── Step 5+6: 吸收验证 + 知识应用 ────────────────────── # ── HydraMind Gate Audit: Gate 动作枚举 ── # 每个 Gate 不只 binary pass/fail，应该产出控制动作 # 对应 Gate 动作审计第二层：continue/revise/replan/block class GateAction: CONTINUE = "CONTINUE" # 放行下一步 REVISE = "REVISE" # 可自动修复 → 重试当前步（带修正参数） REPLAN = "REPLAN" # 需人工判断 → 暂停当前，写 Feishu task BLOCK = "BLOCK" # 不可恢复 → 标记 failed，终止 # Gate reason codes — HydraMind-inspired structured failure attribution # Maps to execution trace quad: node/attempt/gate/reason_code # Format: [node=,attempt=,gate=,action=] class VerifyReason: NO_OUTPUT = "NO_OUTPUT" # 无吸收产出记录 SKILL_MISSING = "SKILL_MISSING" # 声称创建的 skill 不存在 MEMORY_MISSING = "MEMORY_MISSING" # memory.json 文件缺失 NO_VERIFIABLE = "NO_VERIFIABLE" # 无可验证项 FINGERPRINT_FAIL = "FINGERPRINT_FAIL" # 指纹不匹配（可能空内容） CONTENT_THIN = "CONTENT_THIN" # 吸收产出内容过短（可能假成功） NO_BEHAVIOR_CHANGE = "NO_BEHAVIOR_CHANGE" # 吸收报告缺少「行为变化」章节 APPLY_CALLBACK_ERR = "APPLY_CALLBACK_ERR" # Step6 应用审计回调异常 PASS = "PASS" # 通过 PASS_LAX = "PASS_LAX" # 柔性通过（memory unconfirmed） # ── Spawn/Harness 层 reason codes（原 HARNESS_REASON，统一到此）── SPAWN_TIMEOUT = "RUNTIME_TIMEOUT" # 子进程超时 SPAWN_NONZERO = "RUNTIME_NONZERO_EXIT" # 子进程非零退出 SPAWN_FAILED = "RUNTIME_SPAWN_FAILED" # 子进程启动失败 SPAWN_ON_DONE_ERR = "CONTEXT_CALLBACK_FAILED" # 完成回调异常 VERIFY_REASON_CODE = { VerifyReason.NO_OUTPUT: "吸收子进程未产出任何结果记录", VerifyReason.SKILL_MISSING: "声称创建的 skill 文件不存在于文件系统", VerifyReason.MEMORY_MISSING: "memory.json 文件缺失", VerifyReason.NO_VERIFIABLE: "吸收产出中无可验证的实体项", VerifyReason.FINGERPRINT_FAIL: "content 指纹不在 memory.json 中", VerifyReason.CONTENT_THIN: "吸收产出内容过短（<50字符），疑似假成功", VerifyReason.NO_BEHAVIOR_CHANGE: "吸收报告缺少「行为变化」章节，无法确认系统影响", VerifyReason.APPLY_CALLBACK_ERR: "Step6 应用审计回调异常", # Spawn 层 VerifyReason.SPAWN_TIMEOUT: "Hermes 子进程超时被杀", VerifyReason.SPAWN_NONZERO: "Hermes 子进程非零退出码", VerifyReason.SPAWN_FAILED: "Hermes 子进程启动失败", VerifyReason.SPAWN_ON_DONE_ERR: "Hermes 子进程完成后回调异常", } def _verify_and_apply(item_id: int, ltype: str, target: str): """吸收完成后的 Step 5（确定性验证）+ Step 6（知识应用审计）。 Step 5 Gate: 确定性检查点，依据可验证事实产生控制动作。 动作: CONTINUE (放行 Step 6) / REVISE (自动重试) / REPLAN (人工判断) / BLOCK (标记 failed) 输出: reason_code + GateAction 用于离线评估归因。 Step 6: Hermes 子进程加载新知识 → 扫描相关系统 → 审计 → 自动 patch 或写 Feishu。 """ item = local_db.learn_get(item_id) if not item: return result = local_db.learn_result_get_by_learn_id(item_id) if not result: local_db.learn_update_status(item_id, "failed", f"[node=step5_verify,attempt=1,gate={VerifyReason.NO_OUTPUT}," f"action={GateAction.BLOCK}] {VERIFY_REASON_CODE[VerifyReason.NO_OUTPUT]}") return # ── Step 5 Gate: 确定性验证 ── # Gate consumes boundary facts (file existence, content fingerprint), # NOT LLM scores — per HydraMind principle verified = [] reason_code = None # skill 验证 cskill = result.get("created_skill", "") or "" otype = result.get("result_type", "") or "" if cskill: skill_dir = Path("/data/data/com.termux/files/home/.hermes/skills") found = list(skill_dir.rglob(f"*/{cskill}/SKILL.md")) or \ list(skill_dir.rglob(f"{cskill}/SKILL.md")) if found: verified.append(f"skill:{cskill} found") else: # HydraMind REVISE: one-shot retry — skill 可能还在磁盘写入中 import time as _time _time.sleep(0.5) found2 = list(skill_dir.rglob(f"*/{cskill}/SKILL.md")) or \ list(skill_dir.rglob(f"{cskill}/SKILL.md")) if found2: verified.append(f"skill:{cskill} found (retry)") else: # REVISE path: reset to pending_learn for one more cron-managed retry # (before BLOCK), consistent with MEMORY_MISSING/FINGERPRINT_FAIL behavior import re as _re_sk cur = local_db.learn_get(item_id) or {} prev_msgs = (cur.get("error_msg") or "") attempts = [int(m) for m in _re_sk.findall(r"attempt=(\d+)", prev_msgs)] next_att = max(attempts) + 1 if attempts else 2 reason_code = VerifyReason.SKILL_MISSING if next_att <= 3: local_db.learn_update_status(item_id, "pending_learn", f"[node=step5_verify,attempt={next_att},gate={reason_code}," f"action={GateAction.REVISE}] " f"{VERIFY_REASON_CODE[reason_code]}: {cskill}") else: local_db.learn_update_status(item_id, "failed", f"[node=step5_verify,attempt={next_att},gate={reason_code}," f"action={GateAction.BLOCK}] " f"{VERIFY_REASON_CODE[reason_code]}: {cskill}，已重试 {next_att-1} 次") return # memory 验证：检查 learn_content 里是否有实际产出标记 if otype in ("memory", "reference"): content = item.get("learn_content", "") or "" memory_path = Path("/data/data/com.termux/files/home/.hermes/memory.json") if memory_path.exists(): mem_text = memory_path.read_text() fingerprint = content.strip()[:30] if fingerprint and fingerprint in mem_text: verified.append("memory:content confirmed") elif len(content.strip()) < 100: # 内容过短（测试/微型吸收）→ PASS_LAX verified.append("memory:unconfirmed (pass)") else: # ── HydraMind Gate: primary check failed → try MEMORY.md fallback ── mem_md_path = Path("/data/data/com.termux/files/home/.hermes/memories/MEMORY.md") if mem_md_path.exists(): mem_md_text = mem_md_path.read_text() if fingerprint and fingerprint in mem_md_text: verified.append("memory:MEMORY.md confirmed (fallback)") else: reason_code = VerifyReason.FINGERPRINT_FAIL # Only enter REVISE path if both memory.json and MEMORY.md failed if reason_code is not None: import re as _re2 cur = local_db.learn_get(item_id) or {} prev_msgs = (cur.get("error_msg") or "") attempts = [int(m) for m in _re2.findall(r"attempt=(\d+)", prev_msgs)] next_att = max(attempts) + 1 if attempts else 2 if next_att <= 3: # Revise: 重置为 pending_learn，让 cron 执行器重试 local_db.learn_update_status(item_id, "pending_learn", f"[node=step5_verify,attempt={next_att},gate={reason_code}," f"action={GateAction.REVISE}] " f"{VERIFY_REASON_CODE[reason_code]}: " f"content 指纹 '{fingerprint}' 不在 memory.json 或 MEMORY.md 中（len={len(content.strip())}）") else: local_db.learn_update_status(item_id, "failed", f"[node=step5_verify,attempt={next_att},gate={reason_code}," f"action={GateAction.BLOCK}] " f"{VERIFY_REASON_CODE[reason_code]}: " f"content 指纹 '{fingerprint}' 不在 memory.json 或 MEMORY.md 中，已重试 {next_att-1} 次") return else: reason_code = VerifyReason.MEMORY_MISSING # REVISE: memory.json might be momentarily locked/absent, retry once import re as _re3 cur = local_db.learn_get(item_id) or {} prev_msgs = (cur.get("error_msg") or "") attempts = [int(m) for m in _re3.findall(r"attempt=(\d+)", prev_msgs)] next_att = max(attempts) + 1 if attempts else 2 if next_att <= 2: local_db.learn_update_status(item_id, "pending_learn", f"[node=step5_verify,attempt={next_att},gate={reason_code}," f"action={GateAction.REVISE}] " f"{VERIFY_REASON_CODE[reason_code]}") else: local_db.learn_update_status(item_id, "failed", f"[node=step5_verify,attempt={next_att},gate={reason_code}," f"action={GateAction.BLOCK}] " f"{VERIFY_REASON_CODE[reason_code]}，已重试") return if not verified: reason_code = VerifyReason.NO_VERIFIABLE local_db.learn_update_status(item_id, "failed", f"[node=step5_verify,attempt=1,gate={reason_code}," f"action={GateAction.REPLAN}] " f"{VERIFY_REASON_CODE[reason_code]}") return # Gate decision: CONTINUE — all boundary facts verified rc = VerifyReason.PASS if "unconfirmed" not in verified[0] else VerifyReason.PASS_LAX ga = GateAction.CONTINUE verify_log = (f"[node=step5_verify,attempt=1,gate={rc},action={ga}] " f"{'; '.join(verified)}") # 追加验证日志到 error_msg（保留原状态 + reason_code） item_cur = local_db.learn_get(item_id) or {} prev = (item_cur.get("error_msg") or "")[:200] local_db.learn_update_status(item_id, item_cur.get("status", "applied"), f"{prev} | {verify_log}"[:400]) # ── Step 6: 知识应用审计 ── _run_apply(item_id, ltype, target, verify_log) def _run_apply(item_id: int, ltype: str, target: str, verify_log: str): """Step 6: spawn Hermes 子进程 → 加载新知识 → 审计相关系统 → 产出改进。 用 learn-and-build skill 让 Hermes 扫描刚学的知识关联的系统， 用学到的框架/方法论审计，自动修复可自动的 gap，写 Feishu task 给人工判断的。 """ item = local_db.learn_get(item_id) if not item: return import uuid tid = uuid.uuid4().hex[:12] content_file = f"/data/data/com.termux/files/home/tmp/learn_apply_{tid}.md" # 准备应用内容：吸收摘要 + 产物信息 apply_content = item.get("learn_content", "") or "" if len(apply_content) < 50: Path(content_file).write_text( f"# 吸收产物\n- 类型: {ltype}\n- 目标: {target}\n- 验证: {verify_log}", encoding="utf-8") else: Path(content_file).write_text(apply_content, encoding="utf-8") prompt = f"""你是 Hermes 的知识应用引擎。Step 5 验证已通过，现在执行 Step 6：把刚学的知识应用到 Hermes 系统中。 ## 刚吸收的知识 文件: {content_file} 类型: {ltype} 目标: {target} 验证结果: {verify_log} {'**⚠️ WARNING: memory 验证为 PASS_LAX（指纹未匹配），吸收内容可能未真正写入 memory.json。审计时请 double-check 目标系统。**' if 'PASS_LAX' in verify_log else ''} ## 审计框架：多层栅格（HydraMind Gate Audit） 用刚学的知识当「验收标准」，逐层审计相关系统： ### 第一层：Gate 密度审计 - 相关系统在关键步骤间有几个运行时门控（Gate）？ - 每个 Gate 是否依据可验证事实（文件存在/测试通过/diff合理）决策，而非 LLM 评分？ - Gate 间距是否过大导致失败定位退化到 Agent 级（14% 步级准确率）而非步级？ ### 第二层：Gate 动作审计 - 每个 Gate 支持哪些控制动作？（continue/revise/replan/block） - 是否只有 binary pass/fail？缺 revise/replan 路径意味着什么？ ### 第三层：执行轨迹审计 - 失败时能否追溯到四元组：node/attempt/gate/reason_code？ - 如果只能看到"失败了"而不知道哪个节点、第几次尝试、哪个 gate 拦截、什么原因码 → O（可观测性）gap ### 第四层：过程评测闭环 - 是否有：运行时 Gate → 离线评估识别失败模式 → 修订 Gate 规则 → 回归测试？ - 缺失 → L（生命周期）gap，系统不会自我改进 ## 任务 1. 读 {content_file} 的完整内容，理解刚学到的知识 2. 用 search_files / read_file 扫描 Hermes 中与这些知识相关的系统： - 相关的 skill（读取 SKILL.md） - 相关的代码（search_files 找 .py 文件中的 gate/verify/validate/check 模式） - 相关的配置 3. 用上述四层栅格审计这些系统： - 当前实现 vs 新知识的差距在哪？ - 哪些 gap 可以立即自动修复（加 reason_code、加中间 gate、改 prompt）？ - 哪些需要人工判断（改 DB schema、加离线评估管线）？ 4. **执行修复**： - 可自动的 gap → 直接 patch 代码或 skill_manage - 需人工的 → 记入报告，标注「建议 Feishu task」 5. 写报告到 {content_file}.apply.md： ``` # Step 6 知识应用报告 - 学习内容: [一句话] - 审计了哪些系统: [skill名 / 文件路径] - Gate 密度评分: [X/10] — [说明] - Gate 动作评分: [X/10] — [说明] - 执行轨迹评分: [X/10] — [说明] - 过程评测闭环评分: [X/10] — [说明] - 自动修复: [改了什么，具体到行] - 发现但未修复: [gap 描述 + 为什么不能自动修] - 建议: [下一步行动] - Human gate: [哪些需要人看一眼再执行] ``` 注意：实际执行工具调用，不要只用文字描述。超时 5 分钟。""" def _on_apply_done(tid2, ok, spawn_reason=None): try: result_file = f"{content_file}.apply.md" if ok and Path(result_file).exists(): apply_text = Path(result_file).read_text()[:1500] local_db.learn_result_add( learn_queue_id=item_id, article_url=item.get("article_url", ""), article_title=item.get("title", ""), outline=f"Step6 知识应用审计\n{verify_log}", result_type="apply_audit", result_summary=apply_text, created_skill="", auto_applied=1 ) # Trace quad: gate action emitted for offline audit has_human = "Human gate" in apply_text or "建议 Feishu" in apply_text item_cur = local_db.learn_get(item_id) if item_cur: ga = GateAction.REPLAN if has_human else GateAction.CONTINUE local_db.learn_update_status(item_id, item_cur.get("status", "applied"), f"{verify_log} | " f"[node=step6_apply,attempt=1,gate={VerifyReason.PASS},action={ga}] " f"{'人工事项待处理' if has_human else '应用审计完成，无待办'}") # 如果有 Human gate 项 → 写 Feishu if has_human: try: import requests requests.post("http://127.0.0.1:5001/api/feishu/todo", json={"type": "learn_apply", "item_id": item_id, "title": f"学习应用: {target[:30]}", "content": apply_text[:800]}, timeout=10) except Exception: pass else: # apply 失败不改变 learn status——吸收本身是成功的 # ── HydraMind Gate: propagate spawn reason_code ── rc = spawn_reason if spawn_reason else VerifyReason.NO_OUTPUT ga = GateAction.REVISE if rc in (VerifyReason.SPAWN_TIMEOUT,) else GateAction.REPLAN item_cur = local_db.learn_get(item_id) if item_cur: local_db.learn_update_status(item_id, item_cur.get("status", "applied"), f"{verify_log} | " f"[node=step6_apply,attempt=1,gate={rc}," f"action={ga}] " f"Step6: 应用审计执行失败(吸收仍有效，可 /relearn 重试)") for f in [content_file, result_file]: try: Path(f).unlink() except: pass except Exception as e: item_cur = local_db.learn_get(item_id) if item_cur: local_db.learn_update_status(item_id, item_cur.get("status", "applied"), f"[node=step6_apply,attempt=1,gate=APPLY_CALLBACK_ERR," f"action={GateAction.BLOCK}] " f"Step6: 异常 {str(e)[:100]}") _spawn_hermes_job("learn_apply", f"apply: {target[:40]}", ["hermes", "chat", "-q", prompt, "--yolo", "-t", "terminal,file,web", "-s", "learn-and-build"], timeout=300, on_done=_on_apply_done, task_id=tid) @app.route("/api/learn//approve", methods=["POST"]) def api_learn_approve(item_id): """人工点「吸收」→ 进入 pending_learn 队列，等 Cron 学习执行器处理。""" try: item = local_db.learn_get(item_id) if not item: return jsonify({"ok": False, "error": "not found"}), 404 local_db.learn_update_status(item_id, "pending_learn") return jsonify({"ok": True, "data": {"status": "pending_learn"}}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/learn//reject", methods=["POST"]) def api_learn_reject(item_id): try: item = local_db.learn_get(item_id) if not item: return jsonify({"ok": False, "error": "not found"}), 404 local_db.learn_update_status(item_id, "rejected") return jsonify({"ok": True}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 VALID_LEARN_STATUSES = {"pending_review", "pending_learn", "applied", "approved", "failed", "rejected"} @app.route("/api/learn//priority", methods=["PUT"]) def api_learn_update_priority(item_id): """手动调整学习项优先级（支持拖拽排序）""" try: item = local_db.learn_get(item_id) if not item: return jsonify({"ok": False, "error": "not found"}), 404 body = request.get_json(silent=True) if not body or "priority" not in body: return jsonify({"ok": False, "error": "missing priority field"}), 400 try: prio = int(body["priority"]) except (ValueError, TypeError): return jsonify({"ok": False, "error": "priority must be integer"}), 400 conn = local_db.get_conn() conn.execute("UPDATE learn_queue SET priority = ? WHERE id = ?", (prio, item_id)) conn.commit() conn.close() return jsonify({"ok": True, "data": {"id": item_id, "priority": prio}}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/learn//status", methods=["PUT"]) def api_learn_update_status(item_id): """手动修改学习项状态""" try: item = local_db.learn_get(item_id) if not item: return jsonify({"ok": False, "error": "not found"}), 404 body = request.get_json(silent=True) if not body or "status" not in body: return jsonify({"ok": False, "error": "missing status field"}), 400 new_status = body["status"] if new_status not in VALID_LEARN_STATUSES: return jsonify({"ok": False, "error": f"invalid status: {new_status}", "valid": sorted(VALID_LEARN_STATUSES)}), 400 local_db.learn_update_status(item_id, new_status) return jsonify({"ok": True, "data": {"id": item_id, "status": new_status}}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/learn//relearn", methods=["POST"]) def api_learn_relearn(item_id): """重新触发学习：保留历史 trace（带 attempt 递增），重置状态 → pending_learn""" try: item = local_db.learn_get(item_id) if not item: return jsonify({"ok": False, "error": "not found"}), 404 # ── HydraMind Gate: attempt 计数器 ── # 解析现有 error_msg 中的 attempt 计数，递增后再重置 # 格式: [node=xxx,attempt=N,...] → attempt 取最大值 + 1 import re as _re prev_msg = item.get("error_msg", "") or "" attempts = [int(m) for m in _re.findall(r"attempt=(\d+)", prev_msg)] next_attempt = max(attempts) + 1 if attempts else 2 # 保留历史 trace（带 attempt 标签）追加到 error_msg history_note = ( f"[retry,attempt={next_attempt}] " f"重新触发吸收（共 {len(attempts)} 次历史 gate 记录）| ") # 清除旧 learning_results conn = local_db.get_conn() conn.execute("DELETE FROM learning_results WHERE learn_queue_id = ?", (item_id,)) conn.commit() conn.close() # 重置状态：pending_learn + 保留历史 error_msg + 新 attempt local_db.learn_update_status(item_id, "pending_learn", history_note + prev_msg[:300]) _run_absorb(item_id) return jsonify({"ok": True, "data": {"status": "pending_learn", "attempt": next_attempt}}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/learn/execute", methods=["POST"]) def api_learn_execute(): """Cron 调用的学习执行器：扫描 pending_learn 队列，逐条执行 4 步吸收 Loop。 并发安全：使用文件锁确保同一时间只有一个吸收在执行，避免多个 hermes chat 进程同时运行撑爆内存（每个 ~150MB，3 个并发 = 450MB+ 加上其他 agent 直接 OOM）。 """ import fcntl as _fcntl_exec LOCK_FILE = "/data/data/com.termux/files/home/tmp/learn_executor.lock" lock_fd = open(LOCK_FILE, "w") try: _fcntl_exec.flock(lock_fd, _fcntl_exec.LOCK_EX | _fcntl_exec.LOCK_NB) except (IOError, OSError): lock_fd.close() return jsonify({"ok": True, "data": {"processed": 0, "message": "another absorption in progress"}}) try: conn = local_db.get_conn() items = conn.execute( "SELECT id, error_msg FROM learn_queue WHERE status='pending_learn' ORDER BY priority DESC, score DESC LIMIT 3" ).fetchall() conn.close() if not items: return jsonify({"ok": True, "data": {"processed": 0, "message": "无待学习项"}}) import re as _re_exec skipped = 0 MAX_RETRY = 5 # 并发安全：一次只处理 1 条（原来的 for 循环会 spawn 多个后台 hermes chat，撑爆内存） row = items[0] rid = row["id"] error_msg = row["error_msg"] or "" # ── HydraMind Gate: pre-execution attempt limit ── attempts = [int(m) for m in _re_exec.findall(r"attempt=(\d+)", error_msg)] max_att = max(attempts) if attempts else 0 if max_att >= MAX_RETRY: gate_trace = ( f"[node=executor,attempt={max_att+1},gate=MAX_RETRY_EXCEEDED," f"action={GateAction.REPLAN}] " f"已重试 {max_att} 次，超过上限 {MAX_RETRY}，需人工判断") local_db.learn_update_status(rid, "failed", gate_trace) return jsonify({"ok": True, "data": {"processed": 0, "skipped": 1, "message": f"task {rid} exceeded {MAX_RETRY} retries, moved to failed"}}) _run_absorb(rid) more = len(items) - 1 return jsonify({"ok": True, "data": {"processed": 1, "more_pending": more}}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 finally: try: _fcntl_exec.flock(lock_fd, _fcntl_exec.LOCK_UN) lock_fd.close() except: pass @app.route("/api/learn/articles") def api_learn_articles(): """返回分类器处理过的所有文章及评分""" try: conn = local_db.get_conn() rows = conn.execute(""" SELECT id, title, learn_type, learn_target, score, confidence, status, created_at, reviewed_at, error_msg, (SELECT result_summary FROM learning_results lr WHERE lr.learn_queue_id = learn_queue.id LIMIT 1) as result_summary, (SELECT created_skill FROM learning_results lr WHERE lr.learn_queue_id = learn_queue.id LIMIT 1) as created_skill FROM learn_queue ORDER BY score DESC, created_at DESC """).fetchall() conn.close() return jsonify({"ok": True, "data": [dict(r) for r in rows]}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/learn/stats") def api_learn_stats(): try: conn = local_db.get_conn() pending = conn.execute("SELECT COUNT(*) FROM learn_queue WHERE status='pending_review'").fetchone()[0] pending_learn = conn.execute("SELECT COUNT(*) FROM learn_queue WHERE status='pending_learn'").fetchone()[0] approved = conn.execute("SELECT COUNT(*) FROM learn_queue WHERE status='approved'").fetchone()[0] applied = conn.execute("SELECT COUNT(*) FROM learn_queue WHERE status='applied'").fetchone()[0] rejected = conn.execute("SELECT COUNT(*) FROM learn_queue WHERE status='rejected'").fetchone()[0] failed = conn.execute("SELECT COUNT(*) FROM learn_queue WHERE status='failed'").fetchone()[0] unclassified = len(local_db.learn_get_unclassified_summaries(999)) high = conn.execute("SELECT COUNT(*) FROM learn_queue WHERE score >= 7 AND status IN ('pending_review','pending_learn')").fetchone()[0] results_count = conn.execute("SELECT COUNT(*) FROM learning_results").fetchone()[0] conn.close() return jsonify({"ok": True, "data": { "pending": pending, "pending_learn": pending_learn, "approved": approved, "applied": applied, "rejected": rejected, "failed": failed, "unclassified": unclassified, "high_score": high, "results_count": results_count }}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 # ── HydraMind Layer 4: Gate Audit (过程评测闭环) ── @app.route("/api/learn/gate-audit") def api_learn_gate_audit(): """扫描 learn_queue 的 error_msg 字段，提取所有 gate trace 四元组，产出聚合统计。 用于离线评估：识别失败模式 → 修订 Gate 规则 → 回归测试。 """ try: import re conn = local_db.get_conn() rows = conn.execute( "SELECT id, status, error_msg FROM learn_queue WHERE error_msg IS NOT NULL AND error_msg != ''" ).fetchall() conn.close() # Parse gate traces: [node=X,attempt=N,gate=RC,action=A] trace_pat = re.compile( r'\[node=([^,]+),attempt=(\d+),gate=([^,]+),action=([^\]]+)\]' ) gate_stats = {} # {reason_code: {count, actions: {action: count}, nodes: {node: count}}} item_traces = {} # {item_id: [traces]} total_gates = 0 for row in rows: rid, status, msg = row["id"], row["status"], row["error_msg"] or "" matches = trace_pat.findall(msg) if not matches: continue item_traces[rid] = [] for node, att, rc, action in matches: total_gates += 1 item_traces[rid].append({ "node": node.strip(), "attempt": int(att), "gate": rc.strip(), "action": action.strip() }) key = rc.strip() if key not in gate_stats: gate_stats[key] = {"count": 0, "actions": {}, "nodes": {}} gs = gate_stats[key] gs["count"] += 1 gs["actions"][action.strip()] = gs["actions"].get(action.strip(), 0) + 1 gs["nodes"][node.strip()] = gs["nodes"].get(node.strip(), 0) + 1 # Layer 4 health indicators has_offline_eval = False # no automated eval pipeline yet has_regression = False # no regression tests for gate rules has_feedback_loop = False # gate rules are hardcoded, don't evolve return jsonify({"ok": True, "data": { "total_items_with_traces": len(item_traces), "total_gate_events": total_gates, "gate_stats_by_reason_code": gate_stats, "layer4_health": { "offline_evaluation": has_offline_eval, "regression_testing": has_regression, "feedback_loop": has_feedback_loop, "score": "3/10 — traces exist but no automated evaluation or feedback loop" }, "recommendation": "建议：加 cron job 定期调此 API → 分析 TOP N 失败 reason_code → " "人工审核 gate 规则是否需要修订 → 回归验证" }}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 # ── Page: 学习结果 ───────────────────────────────── @app.route("/learn-results") def learn_results_page(): return render_template("learn_results.html") @app.route("/api/learn-results") def api_learn_results(): try: results = local_db.learn_result_list(limit=200) return jsonify({"ok": True, "data": results}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/learn-result//feedback", methods=["POST"]) def api_learn_result_feedback(item_id): try: data = request.get_json() feedback = (data or {}).get("feedback", "") if not feedback: return jsonify({"ok": False, "error": "feedback is required"}), 400 local_db.learn_result_add_feedback(item_id, feedback) return jsonify({"ok": True}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 # ── Page: 自动回复白名单 ───────────────────────────── @app.route("/whitelist") def whitelist_page(): return render_template("whitelist.html") @app.route("/api/whitelist") def api_whitelist_list(): try: return jsonify({"ok": True, "data": local_db.whitelist_list()}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/whitelist", methods=["POST"]) def api_whitelist_add(): """支持单条 {wxid,name,kind,role} 或批量 {items:[...]}（弹窗多选）。""" try: d = request.json or {} items = d.get("items") if isinstance(d.get("items"), list) else [d] n = 0 for it in items: wxid = (it.get("wxid") or "").strip() if not wxid: continue kind = "group" if wxid.endswith("@chatroom") else (it.get("kind") or "contact") local_db.whitelist_add(kind, wxid, (it.get("name") or "").strip(), role=(it.get("role") or "member")) n += 1 if n == 0: return jsonify({"ok": False, "error": "没有有效条目"}), 400 return jsonify({"ok": True, "data": {"added": n}}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 POLICY_CFG = "/data/data/com.termux/files/home/.hermes/wechat_reply/policy_config.json" def _read_auto_send(): """读策略总开关。文件 root 属主但 644，termux 用户可直接读。""" try: with open(POLICY_CFG) as f: return bool(json.load(f).get("auto_send_enabled", False)) except Exception: return False def _write_auto_send(enabled): """只翻转 auto_send_enabled，保留其余字段。文件 root 属主，viewer 经 su 写回。""" try: with open(POLICY_CFG) as f: cfg = json.load(f) except Exception: cfg = {} cfg["auto_send_enabled"] = bool(enabled) blob = json.dumps(cfg, ensure_ascii=False, indent=2) tmp = "/data/data/com.termux/files/home/.hermes/wechat_reply/.policy_config.tmp" with open(tmp, "w") as f: f.write(blob) # viewer 是 termux 用户无权直接覆盖 root 属主文件 → 写 tmp 再 su cp（沿用本 app su 模式） subprocess.run(["su", "-c", f"cp {tmp} {POLICY_CFG}"], capture_output=True, timeout=8) try: os.remove(tmp) except OSError: pass @app.route("/api/auto_send") def api_auto_send_get(): return jsonify({"ok": True, "enabled": _read_auto_send()}) @app.route("/api/auto_send", methods=["POST"]) def api_auto_send_set(): """总开关：True=白名单可自动发（仍受敏感词等硬规则拦截），False=一切进人工审批。""" try: d = request.json or {} _write_auto_send(bool(d.get("enabled"))) return jsonify({"ok": True, "enabled": _read_auto_send()}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 def _talker_row(wxid, name, ttype): return {"wxid": wxid, "name": name or wxid, "kind": "group" if ttype == "group" else "contact"} @app.route("/api/contacts") def api_contacts(): """弹窗默认列表：会话中出现过的联系人 + 群（排除公众号），去重带昵称，按最近排序。""" try: sql = """ SELECT wxid, name, talker_type FROM ( SELECT m.talker AS wxid, CASE WHEN m.talker LIKE '%@chatroom' THEN COALESCE(NULLIF(r.nickname,''), c.displayname, m.talker) ELSE COALESCE(NULLIF(r.conRemark,''), NULLIF(r.nickname,''), m.talker) END AS name, CASE WHEN m.talker LIKE '%@chatroom' THEN 'group' WHEN m.talker LIKE 'gh_%' THEN 'official' ELSE 'personal' END AS talker_type, MAX(m.createTime) AS t FROM message m LEFT JOIN rcontact r ON m.talker = r.username LEFT JOIN chatroom c ON m.talker = c.chatroomname GROUP BY m.talker ) WHERE talker_type != 'official' ORDER BY t DESC LIMIT 500 """ rows = wechat_db.query(sql) return jsonify({"ok": True, "data": [ _talker_row(r.get("wxid"), r.get("name"), r.get("talker_type")) for r in rows]}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/contacts/search") def api_contacts_search(): """按名字/wxid 搜全量 rcontact + 群（弹窗搜索框）。""" try: q = (request.args.get("q") or "").strip().replace("'", "").replace('"', "") if not q: return jsonify({"ok": True, "data": []}) rows = wechat_db.search_talker(q, limit=50) out = [_talker_row(r.get("talker"), r.get("display_name"), r.get("talker_type")) for r in rows if r.get("talker_type") != "official"] return jsonify({"ok": True, "data": out}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/whitelist//delete", methods=["POST"]) def api_whitelist_delete(item_id): try: local_db.whitelist_remove(item_id) return jsonify({"ok": True}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/whitelist//role", methods=["POST"]) def api_whitelist_set_role(item_id): """设置白名单角色：{"role": "admin"|"member"}""" try: role = (request.json or {}).get("role", "member") if role not in ("admin", "member"): return jsonify({"ok": False, "error": "role 必须是 admin 或 member"}), 400 local_db.whitelist_set_role(item_id, role) return jsonify({"ok": True}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 # ── Page: 微信自动回复审批 ─────────────────────────── @app.route("/approvals") def approvals_page(): return render_template("approvals.html") @app.route("/api/replies") def api_replies_list(): """待审批 + 最近已处理（发送/自动/取消）。""" try: status = request.args.get("status", "") return jsonify({"ok": True, "data": local_db.reply_list(status, limit=200)}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 def _send_wechat_text(to_wxid: str, text: str) -> dict: import sys sys.path.insert(0, "/data/data/com.termux/files/home/.hermes/event_bus") from wechat_hook_send import send_text_via_hook return send_text_via_hook(to_wxid, text) @app.route("/api/reply//send", methods=["POST"]) def api_reply_send(item_id): """人工点发送：用（可编辑后的）草稿发到该会话，标记 sent。""" try: item = local_db.reply_get(item_id) if not item: return jsonify({"ok": False, "error": "not found"}), 404 text = ((request.json or {}).get("text") or item.get("draft") or "").strip() if not text: return jsonify({"ok": False, "error": "回复内容为空"}), 400 to = item.get("talker") or "" res = _send_wechat_text(to, text) ok = bool(res.get("ok") or res.get("success") or res.get("status") == "sent") local_db.reply_update(item_id, "sent" if ok else "failed", draft=text) return jsonify({"ok": ok, "data": res}) except Exception as e: try: local_db.reply_update(item_id, "failed") except Exception: pass return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/reply//cancel", methods=["POST"]) def api_reply_cancel(item_id): try: local_db.reply_update(item_id, "cancelled") return jsonify({"ok": True}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 # ── AI 聊天 API ────────────────────────────────────── def call_hermes(user_input: str, timeout: int = 120) -> str: """调用 Hermes CLI 生成 AI 回复。 Args: user_input: 用户输入文本（可包含历史上下文） timeout: 超时秒数（默认 120） Returns: Hermes 的 stdout 输出，或错误提示字符串 """ try: r = subprocess.run( ["hermes", "chat", "-q", user_input, "-Q"], capture_output=True, text=True, timeout=timeout, cwd="/data/data/com.termux/files/home" ) ai_reply = (r.stdout or "").strip() if not ai_reply and r.stderr: ai_reply = f"[Hermes 错误] {r.stderr[:200]}" elif not ai_reply: ai_reply = "[Hermes 返回空回复]" return ai_reply except subprocess.TimeoutExpired: return "[Hermes:gate=TIMEOUT] 调用超时" except FileNotFoundError: return "[Hermes:gate=CMD_NOT_FOUND] 命令未找到" except Exception as hermes_err: return f"[Hermes:gate=RUNTIME_ERR] 调用失败: {str(hermes_err)[:200]}" def call_hermes_stream(user_input: str, timeout: int = 120): """调用 Hermes CLI 并以生成器方式流式输出。 使用 subprocess.Popen 逐行读取 stdout，yield 每个文本块。 Args: user_input: 用户输入文本（可包含历史上下文） timeout: 超时秒数（默认 120） Yields: str: 文本块（逐行或逐词） """ from threading import Timer proc = None try: proc = subprocess.Popen( ["hermes", "chat", "-q", user_input, "-Q"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd="/data/data/com.termux/files/home" ) # Read line by line with timeout start_time = time.time() for line in proc.stdout: elapsed = time.time() - start_time if elapsed > timeout: proc.kill() yield "\n[Hermes:gate=TIMEOUT] 调用超时" return if line: yield line # Check for stderr if no stdout if proc.poll() is not None and proc.returncode != 0: err = (proc.stderr.read() or "")[:200] if err: yield f"\n[Hermes 错误: {err}]" except FileNotFoundError: yield "[Hermes:gate=CMD_NOT_FOUND] 命令未找到" except Exception as e: yield f"\n[Hermes:gate=RUNTIME_ERR] 调用失败: {str(e)[:200]}]" finally: if proc is not None: try: proc.kill() except Exception: pass @app.route("/api/conversations//stream") def api_conversations_stream(conv_id): """SSE 流式对话：发送用户消息并流式返回 AI 回复。""" from flask import Response conv = local_db.conv_get(conv_id) if not conv: return jsonify({"ok": False, "error": "会话不存在"}), 404 message = request.args.get("message", "").strip() if not message: return jsonify({"ok": False, "error": "需要 message 参数"}), 400 # 保存用户消息 local_db.msg_add(conv_id, "user", message) # 如果是首条消息，自动更新标题 if local_db.msg_count(conv_id) == 1: title = message[:30] + ("..." if len(message) > 30 else "") local_db.conv_update_title(conv_id, title) # 构建上下文 history = local_db.msg_list(conv_id) context_lines = [] for msg in history[:-1]: # exclude just-added user message role_label = "用户" if msg["role"] == "user" else "助手" context_lines.append(f"{role_label}: {msg['content']}") hermes_input = message if context_lines: hermes_input = "\n".join(context_lines[-10:]) + f"\n用户: {message}" def generate(): """SSE generator：流式输出 AI 回复并累积全文。""" full_reply = [] try: for chunk in call_hermes_stream(hermes_input): full_reply.append(chunk) yield f"data: {chunk}\n\n" yield "data: [DONE]\n\n" except GeneratorExit: pass finally: # 流结束后保存完整回复到 DB reply_text = "".join(full_reply).strip() if reply_text: try: local_db.msg_add(conv_id, "assistant", reply_text) except Exception: pass return Response( generate(), mimetype="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", } ) @app.route("/api/conversations") def api_conversations(): """列出所有会话，按 updated_at 倒序""" try: convs = local_db.conv_list() return jsonify({"ok": True, "data": convs}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/conversations", methods=["POST"]) def api_conversations_create(): """新建会话""" try: data = request.get_json(silent=True) or {} title = data.get("title", "新对话") cid = local_db.conv_create(title) conv = local_db.conv_get(cid) return jsonify({"ok": True, "data": conv}), 201 except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/conversations/", methods=["DELETE"]) def api_conversations_delete(conv_id): """删除会话（级联删除消息）""" try: conv = local_db.conv_get(conv_id) if not conv: return jsonify({"ok": False, "error": "会话不存在"}), 404 local_db.conv_delete(conv_id) return jsonify({"ok": True}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/conversations/", methods=["PUT"]) def api_conversations_rename(conv_id): """重命名会话""" try: conv = local_db.conv_get(conv_id) if not conv: return jsonify({"ok": False, "error": "会话不存在"}), 404 data = request.get_json(silent=True) or {} title = (data.get("title") or "").strip() if not title: return jsonify({"ok": False, "error": "标题不能为空"}), 400 local_db.conv_update_title(conv_id, title) return jsonify({"ok": True}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/conversations//messages") def api_messages_list(conv_id): """获取某会话的所有消息（按 created_at 正序）""" try: conv = local_db.conv_get(conv_id) if not conv: return jsonify({"ok": False, "error": "会话不存在"}), 404 msgs = local_db.msg_list(conv_id) return jsonify({"ok": True, "data": msgs}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/api/conversations//messages", methods=["POST"]) def api_messages_send(conv_id): """发送用户消息并获取 AI 回复""" try: conv = local_db.conv_get(conv_id) if not conv: return jsonify({"ok": False, "error": "会话不存在"}), 404 data = request.get_json(silent=True) or {} content = data.get("content", "").strip() if not content: return jsonify({"ok": False, "error": "消息内容不能为空"}), 400 # 保存用户消息 user_msg_id = local_db.msg_add(conv_id, "user", content) # 如果是首条消息，自动更新标题 if local_db.msg_count(conv_id) == 1: title = content[:30] + ("..." if len(content) > 30 else "") local_db.conv_update_title(conv_id, title) # 构建 Hermes 上下文（历史消息） history = local_db.msg_list(conv_id) context_lines = [] for msg in history[:-1]: # exclude the just-added user message role_label = "用户" if msg["role"] == "user" else "助手" context_lines.append(f"{role_label}: {msg['content']}") # 构建 Hermes 上下文 hermes_input = content if context_lines: hermes_input = "\n".join(context_lines[-10:]) + f"\n用户: {content}" # 调用 Hermes（使用封装的 call_hermes 函数） ai_reply = call_hermes(hermes_input) # 保存 AI 回复 ai_msg_id = local_db.msg_add(conv_id, "assistant", ai_reply) return jsonify({ "ok": True, "data": { "user_message": {"id": user_msg_id, "role": "user", "content": content}, "assistant_message": {"id": ai_msg_id, "role": "assistant", "content": ai_reply} } }), 201 except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 # ── S6: 消息编辑 ── @app.route("/api/conversations//messages/", methods=["PUT"]) def api_messages_edit(conv_id, msg_id): """编辑消息内容（仅限 user 角色）""" try: conv = local_db.conv_get(conv_id) if not conv: return jsonify({"ok": False, "error": "会话不存在"}), 404 msg = local_db.msg_get(msg_id) if not msg or msg["conversation_id"] != conv_id: return jsonify({"ok": False, "error": "消息不存在"}), 404 if msg["role"] != "user": return jsonify({"ok": False, "error": "只能编辑用户消息"}), 400 data = request.get_json(silent=True) or {} content = (data.get("content") or "").strip() if not content: return jsonify({"ok": False, "error": "内容不能为空"}), 400 local_db.msg_update(msg_id, content) local_db.conv_update_time(conv_id) return jsonify({"ok": True, "data": {"id": msg_id, "content": content}}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 # ── S6: 重新生成 ── @app.route("/api/conversations//regenerate", methods=["POST"]) def api_regenerate(conv_id): """重新生成 AI 回复：删除最后一条 assistant 消息 → 重新调 LLM → SSE 流式返回""" try: conv = local_db.conv_get(conv_id) if not conv: return jsonify({"ok": False, "error": "会话不存在"}), 404 # 找到最后一条消息 msgs = local_db.msg_list(conv_id) if not msgs: return jsonify({"ok": False, "error": "没有消息可以重新生成"}), 400 last_msg = msgs[-1] if last_msg["role"] != "assistant": return jsonify({"ok": False, "error": "最后一条不是 AI 回复"}), 400 # 找到对应的用户消息（最后一条 assistant 之前的第一条 user） last_user_content = "" for m in reversed(msgs[:-1]): if m["role"] == "user": last_user_content = m["content"] break if not last_user_content: return jsonify({"ok": False, "error": "没有找到对应的用户消息"}), 400 # 删除最后一条 assistant 消息 local_db.msg_delete_since(conv_id, last_msg["id"] - 1) # 重新构建上下文并调 Hermes history = local_db.msg_list(conv_id) context_lines = [] for msg in history: role_label = "用户" if msg["role"] == "user" else "助手" context_lines.append(f"{role_label}: {msg['content']}") hermes_input = "\n".join(context_lines[-10:]) + f"\n用户: {last_user_content}" def generate(): proc = subprocess.Popen( ["hermes", "ask", hermes_input], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env={**os.environ, "NO_COLOR": "1"} ) full_reply = "" if proc.stdout: for line in proc.stdout: line = line.rstrip("\n") if line: full_reply += line yield f"data: {line}\n\n" proc.wait() # Save the new AI reply if full_reply.strip(): local_db.msg_add(conv_id, "assistant", full_reply.strip()) local_db.conv_update_time(conv_id) yield "data: [DONE]\n\n" return Response( generate(), mimetype="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive", } ) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 # ── Dashboard Page ────────────────────────────────── @app.route("/dashboard") def dashboard_page(): return render_template("dashboard.html") # ── Dashboard API ──────────────────────────────────── @app.route("/api/dashboard") def api_dashboard(): """Aggregate metrics from all modules for the data dashboard.""" try: conn = local_db.get_conn() data = { "chats": {}, "articles": {}, "learn": {}, "reports": {}, "jobs": {}, "auto_reply": {}, "circle": {}, } # ── Chats (WeChat DB + local) ── try: chats_rows = wechat_db.query( "SELECT COUNT(DISTINCT talker) as groups_count FROM message WHERE talker LIKE '%@chatroom'") data["chats"]["groups"] = chats_rows[0]["groups_count"] if chats_rows else 0 msg_rows = wechat_db.query("SELECT COUNT(*) as cnt FROM message") data["chats"]["messages"] = msg_rows[0]["cnt"] if msg_rows else 0 speaker_rows = wechat_db.query("SELECT COUNT(DISTINCT talker) as cnt FROM message") data["chats"]["speakers"] = speaker_rows[0]["cnt"] if speaker_rows else 0 except Exception: data["chats"]["groups"] = None data["chats"]["messages"] = None data["chats"]["speakers"] = None data["chats"]["shares"] = conn.execute( "SELECT COUNT(*) FROM article_shares").fetchone()[0] # ── Articles ── total_articles = conn.execute("SELECT COUNT(*) FROM articles").fetchone()[0] summarized = conn.execute( "SELECT COUNT(DISTINCT article_url) FROM daily_reports WHERE report_date='summary' AND article_url IS NOT NULL" ).fetchone()[0] learned = conn.execute( "SELECT COUNT(DISTINCT article_url) FROM daily_reports WHERE report_date='learn' AND article_url IS NOT NULL" ).fetchone()[0] data["articles"] = { "total": total_articles, "summarized": summarized, "unsummarized": max(0, total_articles - summarized), "learned": learned, "unlearned": max(0, total_articles - learned), } # ── Learn ── data["learn"] = { "pending_review": conn.execute( "SELECT COUNT(*) FROM learn_queue WHERE status='pending_review'").fetchone()[0], "approved": conn.execute( "SELECT COUNT(*) FROM learn_queue WHERE status='approved'").fetchone()[0], "applied": conn.execute( "SELECT COUNT(*) FROM learn_queue WHERE status='applied'").fetchone()[0], "rejected": conn.execute( "SELECT COUNT(*) FROM learn_queue WHERE status='rejected'").fetchone()[0], "unclassified": conn.execute( """SELECT COUNT(*) FROM daily_reports dr WHERE dr.report_date='summary' AND dr.dedup_key IS NOT NULL AND dr.dedup_key NOT IN ( SELECT DISTINCT lq.dedup_key FROM learn_queue lq WHERE lq.dedup_key IS NOT NULL )""").fetchone()[0], "by_type": { "skill": conn.execute( "SELECT COUNT(*) FROM learn_queue WHERE status='applied' AND learn_type='skill'").fetchone()[0], "memory": conn.execute( "SELECT COUNT(*) FROM learn_queue WHERE status='applied' AND learn_type='memory'").fetchone()[0], "todo": conn.execute( "SELECT COUNT(*) FROM learn_queue WHERE status='applied' AND learn_type='todo'").fetchone()[0], "reference": conn.execute( "SELECT COUNT(*) FROM learn_queue WHERE status='applied' AND learn_type='reference'").fetchone()[0], } } # ── Reports ── last_report = conn.execute( "SELECT created_at FROM daily_reports WHERE report_date NOT IN ('summary','learn') ORDER BY id DESC LIMIT 1" ).fetchone() data["reports"] = { "daily": conn.execute( "SELECT COUNT(*) FROM daily_reports WHERE report_date NOT IN ('summary','learn')").fetchone()[0], "summaries": conn.execute( "SELECT COUNT(*) FROM daily_reports WHERE report_date='summary'").fetchone()[0], "learn_records": conn.execute( "SELECT COUNT(*) FROM daily_reports WHERE report_date='learn'").fetchone()[0], "images": conn.execute("SELECT COUNT(*) FROM report_images").fetchone()[0], "last_generated": last_report["created_at"] if last_report else None, } # ── Jobs ── jobs_total = conn.execute("SELECT COUNT(*) FROM hermes_jobs").fetchone()[0] jobs_completed = conn.execute( "SELECT COUNT(*) FROM hermes_jobs WHERE status='completed'").fetchone()[0] jobs_failed = conn.execute( "SELECT COUNT(*) FROM hermes_jobs WHERE status='failed'").fetchone()[0] jobs_running = conn.execute( "SELECT COUNT(*) FROM hermes_jobs WHERE status='running'").fetchone()[0] last_job = conn.execute( "SELECT completed_at FROM hermes_jobs WHERE completed_at IS NOT NULL ORDER BY id DESC LIMIT 1" ).fetchone() done_total = jobs_completed + jobs_failed data["jobs"] = { "total": jobs_total, "running": jobs_running, "completed": jobs_completed, "failed": jobs_failed, "success_rate": round(jobs_completed / done_total * 100, 1) if done_total > 0 else 0, "last_completed": last_job["completed_at"] if last_job else None, } # ── Auto Reply ── data["auto_reply"] = { "whitelist": conn.execute("SELECT COUNT(*) FROM reply_whitelist").fetchone()[0], "pending": conn.execute( "SELECT COUNT(*) FROM reply_queue WHERE status='pending'").fetchone()[0], "sent": conn.execute( "SELECT COUNT(*) FROM reply_queue WHERE status IN ('sent','auto_sent')").fetchone()[0], } # ── Circle (AI Chat) ── data["circle"] = { "conversations": conn.execute("SELECT COUNT(*) FROM conversations").fetchone()[0], "messages": conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0], } conn.close() return jsonify({"ok": True, "data": data}) except Exception as e: import traceback traceback.print_exc() return jsonify({"ok": False, "error": str(e)}), 500 # ── AI Chat Page ──────────────────────────────────── @app.route("/chat") def ai_chat_page(): """AI 聊天页面：左侧会话列表 + 右侧消息区域""" return render_template("aichat.html") # ── Cron Management APIs ───────────────────────────── CRON_JOBS_FILE = Path("/data/data/com.termux/files/home/.hermes/cron/jobs.json") CRON_OUTPUT_DIR = Path("/data/data/com.termux/files/home/.hermes/cron/output") _cron_cache = {"ts": 0, "data": []} def _load_cron_jobs(force=False): """Load cron jobs from jobs.json with 30s cache.""" global _cron_cache now = time.time() if not force and (now - _cron_cache["ts"]) < 30 and _cron_cache["data"]: return _cron_cache["data"] if not CRON_JOBS_FILE.exists(): return [] try: with open(CRON_JOBS_FILE) as f: raw = json.load(f) except Exception: return _cron_cache["data"] if _cron_cache["data"] else [] jobs = [] for job in raw.get("jobs", []): jobs.append({ "id": job.get("id", ""), "name": job.get("name", ""), "schedule": job.get("schedule_display", job.get("schedule", {}).get("display", "")), "enabled": job.get("enabled", False), "state": job.get("state", "unknown"), "last_run": job.get("last_run_at"), "last_status": job.get("last_status"), "next_run": job.get("next_run_at"), "skills": job.get("skills", []), "deliver": job.get("deliver", "local"), "no_agent": job.get("no_agent", False), }) _cron_cache = {"ts": now, "data": jobs} return jobs @app.route("/cron") def cron_page(): """Cron 管理页面""" return render_template("cron.html") @app.route("/api/cron/list") def api_cron_list(): """返回所有 cron job 列表""" try: jobs = _load_cron_jobs() return jsonify({"ok": True, "data": jobs}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.route("/cron/") def cron_job_page(job_id): """独立全屏时间线日志详情页""" jobs = _load_cron_jobs() job = next((j for j in jobs if j["id"] == job_id), None) if not job: return "Job not found", 404 return render_template("cron_job.html", job=job) @app.route("/api/cron/logs/") def api_cron_logs(job_id): """返回指定 job 的执行历史""" try: out_dir = CRON_OUTPUT_DIR / job_id if not out_dir.is_dir(): return jsonify({"ok": False, "error": f"Job {job_id} not found or no logs"}), 404 logs = [] for fpath in sorted(out_dir.iterdir(), reverse=True): if not fpath.suffix == ".md": continue stat = fpath.stat() try: full_content = fpath.read_text(encoding="utf-8", errors="replace") summary = full_content[:500] if len(full_content) > 500 else full_content if len(full_content) > 500: summary += f"\n\n... (共 {len(full_content)} 字符，截取前 500)" except Exception: full_content = "" summary = "[读取失败]" logs.append({ "timestamp": fpath.stem, "filename": fpath.name, "content": summary, "full_content": full_content, "size": stat.st_size, "mtime": stat.st_mtime, }) return jsonify({"ok": True, "data": logs}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 if __name__ == "__main__": app.config['TEMPLATES_AUTO_RELOAD'] = True app.run(host="0.0.0.0", port=5001, debug=False)