kushidou/padsleep_adb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 1: 守护程序、TUI配置工具、安装脚本

Browse Source 
核心变更:
- padsleep.py: 完整重写，实现守护进程主循环
  - ConfigManager: JSON配置加载/保存/变更检测
  - TimeChecker: 亮屏时间段判断（支持跨天）
  - AdbManager: ADB操作封装，自动检测设备，命令后验证
  - AuditLogger: 审计日志，超50MB自动裁剪保留后10%
  - IpcServer: Unix Socket JSON协议IPC服务端
  - 自动恢复: 亮屏时段内检测到屏幕熄灭立即点亮
  - 外部状态检测: 记录用户手动开关屏事件
- padsleep_tui.py: curses TUI配置工具
  - 亮屏时间段增删改查
  - ADB设备列表查看/选择
  - 守护进程状态监控
  - 审计日志查看
- padsleep.service: systemd服务单元（路径自适应）
- install.sh: 安装到 /opt/apps/padsleep，自动设置权限

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
kushidou
2026-07-09 23:42:09 +08:00
co-authored by Claude Fable 5
parent 4d0db4c390
commit 458b67213f
4 changed files with 427 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
padsleep.py
+217 -37
View File
@@ -6,7 +6,9 @@
阶段一功能：
- 根据配置的亮屏时间段自动亮屏/息屏
- 屏幕亮起超过设定时长且不在亮屏时间段内 → 自动息屏
- 在亮屏时间段内检测到屏幕熄灭 → 自动恢复亮屏
- 通过 Unix Socket 提供 IPC 接口，供 TUI 工具管理
- 审计日志记录所有事件，自动维护日志大小不超过 50MB
"""
import json
@@ -24,8 +26,10 @@ from socketserver import ThreadingUnixStreamServer, StreamRequestHandler
# ── 常量 ────────────────────────────────────────────────────────
SCRIPT_DIR = Path(__file__).parent.resolve()
# 注意: .resolve().parent 确保即使通过 symlink 调用也能正确找到真实目录
SCRIPT_DIR = Path(__file__).resolve().parent
CONFIG_PATH = SCRIPT_DIR / "config.json"
LOG_PATH = SCRIPT_DIR / "padsleep.log"
SOCKET_PATH = "/tmp/padsleep.sock"
DEFAULT_CONFIG = {
@@ -48,6 +52,81 @@ logging.basicConfig(
log = logging.getLogger("padsleep")
# ═══════════════════════════════════════════════════════════════════
# AuditLogger — 审计日志
# ═══════════════════════════════════════════════════════════════════
class AuditLogger:
"""审计日志，记录到 padsleep.log。
日志格式: [YYYY-MM-DD HH:MM:SS] 事件 — 详情
自动维护: 超过 50MB 时保留最后 10% 的内容。
"""
MAX_BYTES = 50 * 1024 * 1024 # 50 MB
_CHECK_INTERVAL = 100 # 每写入 100 次检查一次文件大小
def __init__(self, path: Path = LOG_PATH):
self.path = path
self._write_count = 0
# 启动时检查并裁剪
self._maybe_trim()
def log(self, event: str, detail: str = ""):
"""写入一条审计日志。"""
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
entry = f"[{now}] {event}"
if detail:
entry += f" — {detail}"
try:
with open(self.path, "a", encoding="utf-8") as f:
f.write(entry + "\n")
except OSError as e:
log.error("审计日志写入失败: %s", e)
self._write_count += 1
if self._write_count >= self._CHECK_INTERVAL:
self._write_count = 0
self._maybe_trim()
def _maybe_trim(self):
"""检查日志文件大小，超过上限则裁剪（保留最后 10% 内容）。"""
try:
if self.path.exists() and self.path.stat().st_size > self.MAX_BYTES:
self._trim()
except OSError:
pass
def _trim(self):
"""删除日志前 90% 行，只保留最后 10%。"""
try:
with open(self.path, "r", encoding="utf-8") as f:
lines = f.readlines()
if not lines:
return
keep = lines[-max(1, len(lines) // 10):]
old_size = self.path.stat().st_size
with open(self.path, "w", encoding="utf-8") as f:
f.writelines(keep)
new_size = self.path.stat().st_size
saved_mb = (old_size - new_size) / 1048576
log.info("审计日志已裁剪: 释放 %.1f MB (保留 %.1f MB)",
saved_mb, new_size / 1048576)
except OSError as e:
log.error("审计日志裁剪失败: %s", e)
def read_tail(self, n: int = 50) -> str:
"""读取日志末尾 N 行。"""
try:
if not self.path.exists():
return "(日志文件不存在)"
with open(self.path, "r", encoding="utf-8") as f:
lines = f.readlines()
return "".join(lines[-n:])
except OSError as e:
return f"(读取日志失败: {e})"
# ═══════════════════════════════════════════════════════════════════
# ConfigManager — 配置加载/保存/监控
# ═══════════════════════════════════════════════════════════════════
@@ -165,6 +244,16 @@ class TimeChecker:
now_m = TimeChecker.now_minutes()
return any(TimeChecker.is_within_period(p, now_m) for p in periods)
@staticmethod
def nearest_period_description(periods: list) -> str:
"""返回时间段的文字描述，用于审计日志。"""
if not periods:
return "无配置"
descs = []
for p in periods:
descs.append(f"{p['start']}-{p['end']}")
return "; ".join(descs)
# ═══════════════════════════════════════════════════════════════════
# AdbManager — ADB 操作封装
@@ -173,13 +262,15 @@ class TimeChecker:
class AdbManager:
"""封装所有 ADB CLI 操作，所有异常都被捕获，不会导致进程退出。"""
def __init__(self, config_getter):
def __init__(self, config_getter, audit_logger: AuditLogger | None = None):
self._config_getter = config_getter
self._audit = audit_logger
self._ready = False
self._last_error = ""
self._screen_on_since: float | None = None
self._last_known_screen_on = False
self._last_adb_check = 0 # 限频用
self._current_serial: str | None = None # 自动检测到的设备串号
# ── 底层命令执行 ──
@@ -187,10 +278,11 @@ class AdbManager:
"""执行 ADB 命令，返回 (returncode, stdout, stderr)。永不抛出。"""
config = self._config_getter()
adb_path = config.get("adb_path", "adb")
serial = config.get("adb_device_serial", "")
serial = config.get("adb_device_serial", "") or self._current_serial or ""
cmd = [adb_path]
if serial:
# "devices" / "--version" 命令不加 -s
if serial and args and args[0] not in ("devices", "--version"):
cmd.extend(["-s", serial])
cmd.extend(args)
@@ -237,7 +329,6 @@ class AdbManager:
def is_ready(self) -> bool:
"""检查是否有已授权的 ADB 设备可用。"""
now = time.time()
# 限频：每秒最多检测一次
if now - self._last_adb_check < 1 and self._ready:
return True
self._last_adb_check = now
@@ -247,10 +338,12 @@ class AdbManager:
devices = self.list_devices()
for d in devices:
if d["status"] == "device" and (not target or d["serial"] == target):
self._ready = True
self._last_error = ""
return True
if d["status"] == "device":
if not target or d["serial"] == target:
self._ready = True
self._current_serial = d["serial"]
self._last_error = f"已就绪: {d['serial']}"
return True
# 诊断
unauthorized = [d for d in devices if d["status"] == "unauthorized"]
@@ -263,7 +356,7 @@ class AdbManager:
self._ready = False
return False
# ── 屏幕控制 ──
# ── 屏幕状态 ──
def get_screen_state(self) -> tuple[bool, str]:
"""返回 (is_on, wakefulness)，失败时返回上次已知状态。"""
@@ -296,26 +389,74 @@ class AdbManager:
return time.time() - self._screen_on_since
return 0.0
# ── 屏幕控制（含验证）──
def screen_off(self) -> bool:
"""发送息屏命令。"""
"""发送息屏命令，等待 2 秒后验证，失败时重试一次。"""
log.info("执行息屏操作")
if self._audit:
self._audit.log("息屏", "发出息屏指令")
rc, _, err = self._exec(["shell", "input", "keyevent", "26"])
if rc == 0:
self._screen_on_since = None
return True
log.error("息屏失败: %s", err)
return False
if rc != 0:
log.error("息屏命令失败: %s", err)
if self._audit:
self._audit.log("息屏失败", err)
return False
# 等待并验证
time.sleep(2)
is_on, _ = self.get_screen_state()
if is_on:
log.warning("息屏命令已发出但屏幕未熄灭，重试一次")
rc, _, err = self._exec(["shell", "input", "keyevent", "26"])
time.sleep(2)
is_on, _ = self.get_screen_state()
if is_on:
log.error("息屏重试后仍未熄灭")
if self._audit:
self._audit.log("息屏异常", "屏幕状态未变化")
# 不更新 _screen_on_since
return False
self._screen_on_since = None
log.info("息屏成功")
if self._audit:
self._audit.log("息屏成功", "屏幕已熄灭")
return True
def screen_on(self) -> bool:
"""发送亮屏并解锁命令。"""
"""发送亮屏命令，等待 2 秒后验证，失败时重试一次。"""
log.info("执行亮屏操作")
if self._audit:
self._audit.log("亮屏", "发出亮屏指令")
rc, _, err = self._exec(["shell", "input", "keyevent", "26"])
if rc == 0:
self._screen_on_since = time.time()
self._exec(["shell", "input", "keyevent", "82"]) # 解锁（MENU）
return True
log.error("亮屏失败: %s", err)
return False
if rc != 0:
log.error("亮屏命令失败: %s", err)
if self._audit:
self._audit.log("亮屏失败", err)
return False
# 等待并验证
time.sleep(2)
is_on, _ = self.get_screen_state()
if not is_on:
log.warning("亮屏命令已发出但屏幕未点亮，重试一次")
rc, _, err = self._exec(["shell", "input", "keyevent", "26"])
time.sleep(2)
is_on, _ = self.get_screen_state()
if not is_on:
log.error("亮屏重试后仍未点亮")
if self._audit:
self._audit.log("亮屏异常", "屏幕状态未变化")
return False
self._screen_on_since = time.time()
# 解锁屏幕
self._exec(["shell", "input", "keyevent", "82"])
log.info("亮屏成功")
if self._audit:
self._audit.log("亮屏成功", "屏幕已点亮")
return True
# ═══════════════════════════════════════════════════════════════════
@@ -390,8 +531,9 @@ class PadSleepApp:
def __init__(self):
self.running = True
self.audit = AuditLogger()
self.config_mgr = ConfigManager()
self.adb = AdbManager(self._get_config)
self.adb = AdbManager(self._get_config, self.audit)
self.ipc = IpcServer(self)
self._lock = threading.Lock()
self._start_time = time.time()
@@ -401,6 +543,7 @@ class PadSleepApp:
"within_period": False,
"adb_ready": False,
"device": None,
"device_serial": "",
"uptime": 0,
"start_time": self._start_time,
}
@@ -431,6 +574,8 @@ class PadSleepApp:
cfg = self.config_mgr.get()
status["periods_count"] = len(cfg.get("screen_on_periods", []))
status["check_interval"] = cfg.get("check_interval_seconds", 30)
status["config_path"] = str(self.config_mgr.path)
status["log_path"] = str(LOG_PATH)
return {"ok": True, "data": status}
if cmd == "get_config":
@@ -443,9 +588,6 @@ class PadSleepApp:
if "start" not in p or "end" not in p:
return {"ok": False, "error": "时间段格式错误"}
self.config_mgr.update(new)
cfg = self.config_mgr.get()
# 重设串号以使 is_ready 立即用新串号检查
self.config_mgr.config["adb_device_serial"] = cfg.get("adb_device_serial", "")
return {"ok": True, "data": {"reloaded": True}}
if cmd == "reload":
@@ -458,10 +600,12 @@ class PadSleepApp:
if cmd == "select_device":
serial = request.get("serial", "")
self.config_mgr.update({"adb_device_serial": serial})
self.audit.log("设备选择", f"串号: {serial or '自动'}")
return {"ok": True, "data": {"selected": serial}}
if cmd == "shutdown":
log.info("收到 IPC 关闭命令")
self.audit.log("守护进程", "收到关闭命令")
self.running = False
return {"ok": True}
@@ -473,6 +617,10 @@ class PadSleepApp:
ok = self.adb.screen_on()
return {"ok": ok}
if cmd == "log_tail":
n = request.get("lines", 50)
return {"ok": True, "data": {"log": self.audit.read_tail(n)}}
return {"ok": False, "error": f"未知命令: {cmd}"}
# ── 主循环 ──
@@ -480,15 +628,18 @@ class PadSleepApp:
def run(self):
log.info("=" * 50)
log.info("padsleep_adb 守护程序启动")
log.info("日志路径: %s", LOG_PATH)
log.info("=" * 50)
self.audit.log("守护进程启动", f"配置文件: {self.config_mgr.path}")
# 加载配置
self.config_mgr.load()
# 检测 ADB
if not self.adb.check_available():
log.warning("ADB 未就绪，请安装: sudo apt install adb")
log.warning("安装后将自动检测到 ADB，无需重启本程序")
log.warning("安装后自动检测到 ADB，无需重启本程序")
else:
log.info("ADB 可执行文件已就绪")
@@ -498,15 +649,19 @@ class PadSleepApp:
# 注册信号
def sighandler(sig, _frame):
log.info("收到信号 %s，正在退出...", sig)
self.audit.log("守护进程", "收到退出信号")
self.running = False
signal.signal(signal.SIGTERM, sighandler)
signal.signal(signal.SIGINT, sighandler)
# 记录上一次屏幕状态用于外部变化检测
prev_screen_on: bool | None = None
# 主循环
while self.running:
try:
self._tick()
prev_screen_on = self._tick(prev_screen_on)
except Exception as e:
log.error("主循环异常: %s", e, exc_info=True)
@@ -522,10 +677,11 @@ class PadSleepApp:
# 清理
self.ipc.cleanup()
self.audit.log("守护进程", "已退出")
log.info("padsleep_adb 守护程序已退出")
def _tick(self):
"""单次检查逻辑。"""
def _tick(self, prev_screen_on: bool | None) -> bool:
"""单次检查逻辑。返回当前屏幕亮灭状态。"""
config = self.config_mgr.get()
# ── 检查 ADB 就绪 ──
@@ -535,7 +691,7 @@ class PadSleepApp:
if not hasattr(self, "_last_adb_warn") or now - self._last_adb_warn > 60:
log.warning("ADB 未就绪: %s", self.adb._last_error)
self._last_adb_warn = now
return
return prev_screen_on if prev_screen_on is not None else False
# ── 获取屏幕状态 ──
screen_on, wakefulness = self.adb.get_screen_state()
@@ -545,30 +701,54 @@ class PadSleepApp:
device = self.adb.get_ready_device()
self._update_status(
adb_ready=True,
device=device,
device=device or "无",
device_serial=config.get("adb_device_serial", ""),
screen_on=screen_on,
wakefulness=wakefulness,
within_period=within_period,
)
# ── 检测外部屏幕状态变化 ──
if prev_screen_on is not None and prev_screen_on != screen_on:
if screen_on:
log.info("检测到屏幕被点亮 (外部操作)")
self.audit.log("检测到屏幕被点亮", "外部操作或用户手动")
else:
log.info("检测到屏幕被熄灭 (外部操作)")
self.audit.log("检测到屏幕被熄灭", "外部操作或用户手动")
# ── 决策逻辑 ──
delay_sec = config.get("auto_sleep_delay_minutes", 5) * 60
period_desc = TimeChecker.nearest_period_description(periods)
if within_period and not screen_on:
# 在亮屏时段但屏幕灭着 → 立即亮屏（自动恢复）
log.info("在亮屏时间段内但屏幕已熄灭 → 立即亮屏")
self.audit.log("亮屏", f"在亮屏时间段 ({period_desc}) 内检测到屏幕熄灭，自动恢复")
self.adb.screen_on()
return True
if not within_period and screen_on:
# 非亮屏时段但屏幕亮着 → 超时后息屏
on_duration = self.adb.get_screen_on_duration()
if on_duration >= delay_sec:
log.info("不在亮屏时段（已亮 %.0fs）→ 自动息屏", on_duration)
self.audit.log("息屏", f"不在亮屏时间段 ({period_desc})，超时 {config.get('auto_sleep_delay_minutes')} 分钟")
self.adb.screen_off()
return False
else:
remaining = delay_sec - on_duration
if remaining <= 10:
log.info("不在亮屏时段，%.0f 秒后将息屏", remaining)
elif within_period and not screen_on:
# 在亮屏时段但屏幕灭着 → 亮屏
log.info("在亮屏时段内 → 自动亮屏")
self.adb.screen_on()
elif within_period and screen_on:
# 在亮屏时段且屏幕已亮 — 一切正常，但记录一个调试信息
if prev_screen_on is False:
# 这是上一步我们刚刚点亮了屏幕
pass
return screen_on
# ═══════════════════════════════════════════════════════════════════