diff --git a/device/pixelphone/event_bus/device_tools.py b/device/pixelphone/event_bus/device_tools.py index aee610c..ef640ee 100644 --- a/device/pixelphone/event_bus/device_tools.py +++ b/device/pixelphone/event_bus/device_tools.py @@ -47,6 +47,8 @@ import json import os import re import subprocess + +import ui_guard from pathlib import Path # ── shell routing (on-device su / off-device adb) ────────────────────────── @@ -74,6 +76,8 @@ def _run(cmd: str, timeout: int = _TIMEOUT) -> subprocess.CompletedProcess: `su -c`. Off-device we drive the phone through `adb shell`, whose uid already holds those permissions — no `su` needed. """ + if ui_guard.check(cmd): # ui.paused：拒绝动屏幕的命令 + return subprocess.CompletedProcess(cmd, 75, "", "UI paused (ui.paused)") if _ON_DEVICE: argv = ["su", "-c", cmd] else: diff --git a/device/pixelphone/event_bus/ui_control_tools.py b/device/pixelphone/event_bus/ui_control_tools.py index 7b977c2..08df194 100644 --- a/device/pixelphone/event_bus/ui_control_tools.py +++ b/device/pixelphone/event_bus/ui_control_tools.py @@ -48,6 +48,8 @@ import json import os import re import subprocess + +import ui_guard import time from pathlib import Path from typing import Any, Optional @@ -78,6 +80,8 @@ def _run(cmd: str, timeout: int = _TIMEOUT) -> subprocess.CompletedProcess: uid already holds those permissions — no `su` needed (and it avoids the nested-quoting trap of `adb shell su -c '...'`). """ + if ui_guard.check(cmd): # ui.paused：拒绝动屏幕的命令 + return subprocess.CompletedProcess(cmd, 75, "", "UI paused (ui.paused)") if _ON_DEVICE: argv = ["su", "-c", cmd] else: diff --git a/device/pixelphone/event_bus/ui_guard.py b/device/pixelphone/event_bus/ui_guard.py new file mode 100644 index 0000000..7a31408 --- /dev/null +++ b/device/pixelphone/event_bus/ui_guard.py @@ -0,0 +1,43 @@ +"""UI 操作总闸：ui.paused 存在 → 所有会动屏幕的命令（input / am start / monkey）被拒（rc=75）。 +所有 UI 驱动方（ui_control_tools/device_tools 的 _run、orchestrator 的前后台切换、reply-sender、orig-save.sh） +共用这一个文件。wechat-viewer 的 /ops 页面负责创建/删除它。动作审计写 ui-actions.log 供 /ops 展示。""" +import os +import re +import sys +import time + +PAUSE_FILE = "/data/data/com.termux/files/home/pixelphone/ui.paused" +AUDIT_FILE = "/data/data/com.termux/files/home/pixelphone/ui-actions.log" +_UI_RE = re.compile(r"\binput\s+\w+|\bam\s+start\b|\bmonkey\b") # 只管"动手"的；dump/screencap 只读不拦 +_MAX = 512 * 1024 + + +def paused() -> bool: + return os.path.exists(PAUSE_FILE) + + +def is_ui(cmd: str) -> bool: + return bool(_UI_RE.search(cmd)) + + +def audit(what: str, blocked: bool = False) -> None: + try: + if os.path.exists(AUDIT_FILE) and os.path.getsize(AUDIT_FILE) > _MAX: + os.replace(AUDIT_FILE, AUDIT_FILE + ".1") + new = not os.path.exists(AUDIT_FILE) + who = os.path.basename(sys.argv[0] or "?") + with open(AUDIT_FILE, "a") as f: + f.write(f"{time.strftime('%F %T')}\t{'BLOCKED' if blocked else 'ok'}\t{who}[{os.getpid()}]\t{what[:200]}\n") + if new: + os.chmod(AUDIT_FILE, 0o666) + except OSError: + pass + + +def check(cmd: str) -> bool: + """是 UI 命令则记审计；返回 True = 该拦截。""" + if not is_ui(cmd): + return False + b = paused() + audit(cmd, b) + return b diff --git a/device/pixelphone/scripts/orig-save.sh b/device/pixelphone/scripts/orig-save.sh index 8cc17a1..1dbbfe8 100755 --- a/device/pixelphone/scripts/orig-save.sh +++ b/device/pixelphone/scripts/orig-save.sh @@ -13,6 +13,8 @@ LOG=$HOME/tmp/orig_save_$(date +%Y%m%d_%H%M).log mkdir -p "$HOME/tmp" say() { echo "[$(date +%H:%M:%S)] $*" | tee -a "$LOG"; } +[ -e "$HOME/pixelphone/ui.paused" ] && { say "UI 已暂停，跳过原图保存"; exit 0; } + LOCK=$TMPDIR/orig-save.lock # send 阶段看到它在就等下一轮（两边都要用微信界面） if ! mkdir "$LOCK" 2>/dev/null; then [ -n "$(find "$LOCK" -maxdepth 0 -mmin +90 2>/dev/null)" ] && { rmdir "$LOCK"; mkdir "$LOCK"; } || { say "上一轮还在跑，跳过"; exit 0; } diff --git a/device/pixelphone/scripts/reply-sender.py b/device/pixelphone/scripts/reply-sender.py index b1810f2..9835c4b 100644 --- a/device/pixelphone/scripts/reply-sender.py +++ b/device/pixelphone/scripts/reply-sender.py @@ -139,6 +139,9 @@ def _rm(path): def main(): + if os.path.exists("/data/data/com.termux/files/home/pixelphone/ui.paused"): + print(f"[{time.strftime('%H:%M:%S')}] SKIP: UI paused (队列保留，恢复后再发)") + return # ── WeChat Login Gate ── gate = check_wechat_ready() if not gate["ready"]: diff --git a/device/pixelphone/wechat_reply/wechat_reply_daemon.sh b/device/pixelphone/wechat_reply/wechat_reply_daemon.sh index c734575..ce860ec 100755 --- a/device/pixelphone/wechat_reply/wechat_reply_daemon.sh +++ b/device/pixelphone/wechat_reply/wechat_reply_daemon.sh @@ -18,7 +18,8 @@ mkdir -p "$W" # ── 单实例：已有活着的监督器就退出 ── if [ -f "$LOCK" ]; then OLD=$(cat "$LOCK" 2>/dev/null) - if [ -n "$OLD" ] && kill -0 "$OLD" 2>/dev/null; then + if [ -n "$OLD" ] && kill -0 "$OLD" 2>/dev/null \ + && grep -q wechat_reply_daemon "/proc/$OLD/cmdline" 2>/dev/null; then # 重启后 pid 会被复用，必须校验 cmdline echo "$(date) 监督器已在跑 pid=$OLD，退出" >> "$LOG" exit 0 fi diff --git a/device/pixelphone/wechat_reply/wechat_reply_orchestrator.py b/device/pixelphone/wechat_reply/wechat_reply_orchestrator.py index d1ee443..bcc9a1e 100644 --- a/device/pixelphone/wechat_reply/wechat_reply_orchestrator.py +++ b/device/pixelphone/wechat_reply/wechat_reply_orchestrator.py @@ -347,10 +347,21 @@ VERIFY_POLL = 3 # 前台停留期每次查 status 的间隔 DWELL_POLLS = 4 # 一次 onResume 后停留前台查几次(给网络发送时间,不立刻再背景化打断) +def _ui_paused(what): + """UI 总闸（见 event_bus/ui_guard.py）；非暂停时记一笔审计。""" + sys.path.insert(0, EVENT_BUS) + import ui_guard + b = ui_guard.paused() + ui_guard.audit(what, b) + return b + + def _onresume_cycle(): """一次真实的 onPause→onResume:HOME 背景化 → 停顿 → 拉回微信前台。 触发 WeChat NetSceneQueue flush 卡在 status=1 的发送。切换要给足时间——太快 (背景/前台 <2s)微信来不及走完 onPause/onResume+网络 tick,flush 不了。""" + if _ui_paused("HOME+am start LauncherUI (onResume cycle)"): + return try: subprocess.run(["input", "keyevent", "KEYCODE_HOME"], capture_output=True, timeout=6) time.sleep(2.5) # 让 onPause 走完、微信真的进后台 @@ -465,6 +476,8 @@ def _send_text_verified(to_wxid, text, key): 超时仍卡在发送中/失败 → 返回 ok=False,由调用方标 failed 进审核(不假装成功)。 ponytail: 校验靠反复拷库(~96MB/次)读 status；发送稀疏可接受,首轮多半即达。 上限=拷库 I/O;升级=让 hook 直接回报送达 / 常驻前台。""" + if _ui_paused("send_text_verified " + to_wxid): + return {"ok": False, "reason": "ui_paused"} _, db, _ = w.account_paths() st = w.copy_db(db, "_stage_verify") try: diff --git a/device/wechat-viewer/app.py b/device/wechat-viewer/app.py index 62aadcd..6ae1bb7 100644 --- a/device/wechat-viewer/app.py +++ b/device/wechat-viewer/app.py @@ -6810,6 +6810,177 @@ def _sync_scores_to_db(): db.close() +# ── /ops：后台任务监管 + UI 总闸 ─────────────────────────────────── +# 路径刻意不放 /api/ 下：/api/ 对本机 127.0.0.1 免登录，而这里能杀进程，必须登录（且仅 admin，见 check_auth）。 +# UI 总闸 = 文件 ui.paused，由 event_bus/ui_guard.py 等各 UI 驱动方共同遵守。 +import re as _re +_PX = "/data/data/com.termux/files/home/pixelphone" +_UI_PAUSE = f"{_PX}/ui.paused" +_UI_AUDIT = f"{_PX}/ui-actions.log" +_TERMUX_ENV = ("HOME=/data/data/com.termux/files/home PREFIX=/data/data/com.termux/files/usr " + "PATH=/data/data/com.termux/files/usr/bin:/system/bin TMPDIR=/data/data/com.termux/files/usr/tmp") +# (key, 显示名, ARGS 正则, 类型, 是否会动 UI) +_OPS_PROCS = [ + ("reply-daemon", "微信回复·监督器", r"wechat_reply_daemon\.sh", "service", False), + ("reply-watcher", "微信回复·watcher", r"wechat_reply_watcher\.py", "service", False), + ("orchestrator", "微信回复·orchestrator(生成+发送)", r"wechat_reply_orchestrator\.py", "task", True), + ("reply-sender", "回复队列发送器", r"reply-sender\.py", "task", True), + ("orig-save", "原图保存(接管屏幕)", r"orig-save\.sh|save_originals_batch|upload_originals", "task", True), + ("dsh-headless", "dsh headless 任务(手机MCP)", r"dsh --profile headless|run-headless\.sh|run-reply-member\.sh", "task", True), + ("digest", "日报链/预摘要", r"daily-digest\.sh|gen_daily|send_digest|tgo_summarize_batch", "task", False), + ("boss-resume", "BOSS 简历管线", r"boss-resume\.sh", "task", False), + ("dsh-web", "dsh web", r"\.bin/dsh web", "service", False), + ("glm-proxy", "GLM failover proxy", r"glm_failover_proxy", "service", False), + ("crond", "crond", r"^crond", "service", False), + ("viewer", "wechat-viewer(本页)", r"python3 app\.py", "service", False), +] +_OPS_SVC = { # 可启停的常驻服务：key -> (start 命令, stop 用的 pkill -f 模式；None=pkill -x key) + "reply-daemon": (f"{_TERMUX_ENV} nohup sh {_PX}/wechat_reply/wechat_reply_daemon.sh >/dev/null 2>&1 &", + ["wechat_reply_daemon.sh", "wechat_reply_watcher.py"]), + "dsh-web": (f"su -G 3003 10329 -c '{_TERMUX_ENV} nohup /data/data/com.termux/files/home/dsh/run-web.sh " + f"> /data/data/com.termux/files/home/dsh/web.log 2>&1 < /dev/null &'", + [r"\.bin/dsh web"]), + "crond": (f"su -G 3003 10329 -c '{_TERMUX_ENV} /data/data/com.termux/files/usr/bin/crond'", None), +} + + +def _ops_su(cmd, timeout=15): + r = subprocess.run(["su", "-c", cmd], capture_output=True, text=True, timeout=timeout) + return (r.stdout or "").strip() + + +def _ops_ps(): + out = _ops_su("ps -A -o PID,ETIME,ARGS") + rows = [] + for line in out.splitlines()[1:]: + parts = line.split(None, 2) + if len(parts) < 3 or not parts[0].isdigit(): + continue + pid, etime, args = int(parts[0]), parts[1], parts[2] + if args.startswith(("su ", "ps ", "sh -c", "grep ")) or "ps -A -o" in args: + continue + for key, name, rx, kind, ui in _OPS_PROCS: + if _re.search(rx, args): + rows.append({"pid": pid, "etime": etime, "key": key, "name": name, "kind": kind, + "ui": ui, "args": args[:300]}) + break + return rows + + +def _ops_ui_actions(n=60): + try: + with open(_UI_AUDIT, errors="replace") as f: + lines = f.readlines()[-n:] + except OSError: + return [] + out = [] + for l in reversed(lines): + p = l.rstrip("\n").split("\t", 3) + if len(p) == 4: + out.append({"ts": p[0], "result": p[1], "who": p[2], "what": p[3]}) + return out + + +def _ops_guard(): + if request.headers.get("X-Requested-With") != "ops": + return jsonify(error="forbidden"), 403 + + +@app.route("/ops") +def ops_page(): + return render_template("ops.html") + + +@app.route("/ops/state") +def ops_state(): + procs = _ops_ps() + screen = _ops_su("dumpsys power | grep -m1 mWakefulness=; dumpsys window | grep -m1 mCurrentFocus") + try: + cron = subprocess.run(["crontab", "-l"], capture_output=True, text=True, timeout=5).stdout + except Exception: # noqa: BLE001 + cron = "" + paused_at = "" + if os.path.exists(_UI_PAUSE): + paused_at = time.strftime("%F %T", time.localtime(os.path.getmtime(_UI_PAUSE))) + present = {p["key"] for p in procs} + services = [{"key": k, "name": n, "up": k in present, "controllable": k in _OPS_SVC, + "pids": [p["pid"] for p in procs if p["key"] == k]} + for k, n, _, kind, _ in _OPS_PROCS if kind == "service"] + return jsonify(paused=bool(paused_at), paused_at=paused_at, screen=screen, services=services, + tasks=[p for p in procs if p["kind"] == "task"], ui_actions=_ops_ui_actions(), + cron=[l for l in cron.splitlines() if l.strip() and not l.startswith(("PATH", "HOME", "TMPDIR", "D=", "L="))]) + + +@app.route("/ops/pause", methods=["POST"]) +def ops_pause(): + bad = _ops_guard() + if bad: + return bad + on = bool((request.get_json(silent=True) or {}).get("on")) + killed = [] + if on: + with open(_UI_PAUSE, "w") as f: + f.write(time.strftime("%F %T") + "\n") + os.chmod(_UI_PAUSE, 0o666) + # 长时间接管屏幕的脚本直接掐掉（其余 UI 驱动方每条命令都会被总闸拒绝） + for p in _ops_ps(): + if p["key"] == "orig-save": + _ops_su(f"kill {p['pid']}") + killed.append(p["pid"]) + else: + try: + os.remove(_UI_PAUSE) + except FileNotFoundError: + pass + return jsonify(ok=True, paused=on, killed=killed) + + +@app.route("/ops/kill", methods=["POST"]) +def ops_kill(): + bad = _ops_guard() + if bad: + return bad + d = request.get_json(silent=True) or {} + pid = int(d.get("pid", 0)) + tgt = next((p for p in _ops_ps() if p["pid"] == pid and p["key"] != "viewer"), None) # 只能杀受管列表里的，且不能杀自己 + if not tgt: + return jsonify(ok=False, error="pid 不在受管列表"), 404 + _ops_su(f"kill {'-9 ' if d.get('force') else ''}{pid}") + return jsonify(ok=True, killed=pid) + + +@app.route("/ops/service", methods=["POST"]) +def ops_service(): + bad = _ops_guard() + if bad: + return bad + d = request.get_json(silent=True) or {} + key, act = d.get("key"), d.get("action") + if key not in _OPS_SVC or act not in ("start", "stop", "restart"): + return jsonify(ok=False, error="不支持"), 400 + start_cmd, pats = _OPS_SVC[key] + if act in ("stop", "restart"): + for pat in (pats or []): + _ops_su(f"pkill -f '{pat}'") + if not pats: + _ops_su(f"pkill -x {key}") + time.sleep(1) + if act in ("start", "restart"): + _ops_su(start_cmd) + return jsonify(ok=True) + + +@app.route("/ops/log") +def ops_log(): + paths = {"reply-daemon": f"{_PX}/wechat_reply/daemon.log", "dsh-web": "/data/data/com.termux/files/home/dsh/web.log", + "reply-sender": f"{_PX}/cron-logs/reply-sender.log", "orig-save": f"{_PX}/cron-logs/orig-save.log", + "digest": f"{_PX}/cron-logs/digest-send.log", "ui": _UI_AUDIT} + p = paths.get(request.args.get("name", "")) + if not p: + return jsonify(text="无此日志"), 404 + return jsonify(text=_ops_su(f"tail -n 80 '{p}' 2>&1")[-8000:]) + + if __name__ == "__main__": app.config['TEMPLATES_AUTO_RELOAD'] = True app.run(host="0.0.0.0", port=5001, debug=False) diff --git a/device/wechat-viewer/templates/base.html b/device/wechat-viewer/templates/base.html index adfc3c1..ba614ab 100644 --- a/device/wechat-viewer/templates/base.html +++ b/device/wechat-viewer/templates/base.html @@ -198,6 +198,9 @@ body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif; b ⏱ Cron 管理 + + 🛠 后台监管 + {% endif %} 🏢 BOSS 岗位 diff --git a/device/wechat-viewer/templates/ops.html b/device/wechat-viewer/templates/ops.html new file mode 100644 index 0000000..d9b0141 --- /dev/null +++ b/device/wechat-viewer/templates/ops.html @@ -0,0 +1,72 @@ +{% extends "base.html" %} +{% block title %}后台监管 - WeChat Viewer{% endblock %} +{% block body %} + +
+

🛠 后台监管

+ +
+ +

🖐 UI 操作记录 谁在动屏幕（最近 60 条，红色 = 被总闸拦下）

+
+ +

▶ 运行中的任务

+

⚙ 常驻服务

+

⏱ Crontab


+  
📜 日志

+  

+  
点上面的按钮看日志

+

+
+{% endblock %}