Compare commits

...
5 Commits
Author SHA1 Message Date
kushidou da8ad4d17d 修复 install.sh create_wrapper 覆盖源文件的问题
旧版用 ln -sf 创建软链接指向 .py 文件,新版改用 cat 写入包装脚本时,
bash 重定向会跟随旧软链接,直接将包装脚本内容写入 .py 源文件。

修复: rm -f "$link" 先移除旧文件/链接再写入包装脚本
2026-07-10 20:44:37 +08:00
kushidou 5a75faf6fc 审计日志增加 ADB 连接/断开和设备 ID 记录
- is_ready(): 检测到 ADB 设备连接时记录串号和类型 (USB/无线)
- is_ready(): 设备断开时记录上次串号和断开原因
- connect_wireless(): 记录无线连接成功/失败
- disconnect_wireless(): 记录无线断开
2026-07-10 19:48:13 +08:00
kushidou e17d03cea9 集成 venv 方案: 安装脚本 + 服务文件适配虚拟环境
- install.sh: 创建 .venv、安装 adbutils、生成包装脚本使用 venv python
- padsleep.service: ExecStart 指向 .venv/bin/python3
- requirements.txt: 添加 venv 安装说明
- 解决 PEP 668 (externally-managed-environment) 问题
2026-07-10 19:45:50 +08:00
kushidou 36fe0bb4fd 合并远程变更: adbutils 改造 + 无线 ADB 支持 + TUI 优化
合并内容:
- 远程: adbutils 库替换 subprocess、无线 ADB 连接、TUI 闪烁修复
- 本地: 审计日志、命令执行后验证重试、安装脚本
- 合并后保留双方特性: adbutils + 审计日志 + 无线 ADB + 命令验证
2026-07-10 19:35:32 +08:00
kushidouandClaude Fable 5 458b67213f Phase 1: 守护程序、TUI配置工具、安装脚本
核心变更:
- 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>
2026-07-09 23:42:09 +08:00
5 changed files with 468 additions and 32 deletions
Executable
+160
View File
@@ -0,0 +1,160 @@
#!/bin/bash
# padsleep_adb 安装脚本
# 用法: sudo ./install.sh
#
# 将程序安装到 /opt/apps/padsleep,创建命令行软链接。
# 自动创建 Python 虚拟环境并安装依赖。
# 代码中 SCRIPT_DIR 使用 Path(__file__).resolve().parent
# 会自动适配安装路径,无需修改代码。
set -e
INSTALL_DIR="/opt/apps/padsleep"
BIN_DIR="/usr/local/bin"
SERVICE_DIR="/etc/systemd/system"
VENV_DIR="$INSTALL_DIR/.venv"
PYTHON_BIN="$VENV_DIR/bin/python3"
# 颜色
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m'
info() { echo -e "${GREEN}[INFO]${NC} $1"; }
warn() { echo -e "${YELLOW}[WARN]${NC} $1"; }
error() { echo -e "${RED}[ERROR]${NC} $1"; }
# ── 检查 root ──
if [ "$EUID" -ne 0 ]; then
error "请使用 sudo 运行此安装脚本"
echo " sudo $0"
exit 1
fi
# ── 确定源目录(脚本所在目录)──
SRC_DIR="$(cd "$(dirname "$0")" && pwd)"
if [ ! -f "$SRC_DIR/padsleep.py" ]; then
error "$SRC_DIR 中未找到 padsleep.py"
echo "请在 padsleep_adb 项目目录中运行此脚本"
exit 1
fi
# ── 确定运行用户(sudo 的原用户)──
RUN_USER="${SUDO_USER:-}"
if [ -z "$RUN_USER" ]; then
if [ -f "$SRC_DIR/padsleep.service" ]; then
RUN_USER=$(grep '^User=' "$SRC_DIR/padsleep.service" | head -1 | cut -d= -f2)
fi
fi
if [ -z "$RUN_USER" ]; then
RUN_USER="$(whoami)"
fi
RUN_USER_HOME=$(eval echo "~$RUN_USER" 2>/dev/null || echo "/home/$RUN_USER")
info "源目录: $SRC_DIR"
info "安装目标: $INSTALL_DIR"
info "运行用户: $RUN_USER"
# ── 创建目标目录 ──
mkdir -p "$INSTALL_DIR"
info "已创建目录 $INSTALL_DIR"
# ── 复制文件 ──
cp "$SRC_DIR/padsleep.py" "$INSTALL_DIR/"
cp "$SRC_DIR/padsleep_tui.py" "$INSTALL_DIR/"
cp "$SRC_DIR/requirements.txt" "$INSTALL_DIR/" 2>/dev/null || true
# 配置文件:目标不存在则复制,存在则保留
if [ ! -f "$INSTALL_DIR/config.json" ]; then
cp "$SRC_DIR/config.json" "$INSTALL_DIR/"
info "已创建默认配置文件 $INSTALL_DIR/config.json"
else
warn "配置文件已存在,保留现有配置: $INSTALL_DIR/config.json"
fi
# 日志文件:创建空文件
touch "$INSTALL_DIR/padsleep.log"
chmod 644 "$INSTALL_DIR/padsleep.log"
# ── 创建 Python 虚拟环境 ──
if [ ! -d "$VENV_DIR" ]; then
info "正在创建 Python 虚拟环境..."
python3 -m venv "$VENV_DIR"
info "虚拟环境已创建: $VENV_DIR"
else
info "虚拟环境已存在,跳过创建"
fi
# ── 安装 Python 依赖 ──
info "正在安装 Python 依赖..."
"$VENV_DIR/bin/pip" install --quiet adbutils 2>&1 || {
warn "pip 安装失败,尝试离线安装..."
if [ -f "$SRC_DIR/.venv" ]; then
cp -r "$SRC_DIR/.venv" "$VENV_DIR"
fi
}
info "Python 依赖安装完成"
# 设置执行权限
chmod 755 "$INSTALL_DIR/padsleep.py"
chmod 755 "$INSTALL_DIR/padsleep_tui.py"
# 修正所有权和权限:确保运行用户有权读写配置和日志
chown -R "$RUN_USER:$RUN_USER" "$INSTALL_DIR" 2>/dev/null || chown -R "$RUN_USER" "$INSTALL_DIR" 2>/dev/null
chmod 755 "$INSTALL_DIR"
chmod 644 "$INSTALL_DIR/padsleep.log" 2>/dev/null
info "文件所有权已设置为 $RUN_USER"
# ── 创建命令包装脚本(使用 venv Python)──
create_wrapper() {
local script="$1" # padsleep.py 或 padsleep_tui.py
local link="$2" # /usr/local/bin/padsleep
# 先移除旧文件/软链接,避免 cat 跟随链接覆盖源文件
rm -f "$link"
cat > "$link" <<WRAPPER
#!/bin/bash
exec "$PYTHON_BIN" "$INSTALL_DIR/$script" "\$@"
WRAPPER
chmod 755 "$link"
}
create_wrapper "padsleep.py" "$BIN_DIR/padsleep"
create_wrapper "padsleep_tui.py" "$BIN_DIR/padsleep-config"
info "已创建命令入口:"
info " $BIN_DIR/padsleep → 启动守护进程 ($PYTHON_BIN)"
info " $BIN_DIR/padsleep-config → 启动 TUI 配置工具 ($PYTHON_BIN)"
# ── 可选:复制 systemd 服务 ──
if [ -f "$SRC_DIR/padsleep.service" ]; then
cp "$SRC_DIR/padsleep.service" "$SERVICE_DIR/padsleep.service"
sed -i "s|WorkingDirectory=.*|WorkingDirectory=$INSTALL_DIR|" "$SERVICE_DIR/padsleep.service"
sed -i "s|ExecStart=.*|ExecStart=$PYTHON_BIN $INSTALL_DIR/padsleep.py|" "$SERVICE_DIR/padsleep.service"
sed -i "s|^User=.*|User=$RUN_USER|" "$SERVICE_DIR/padsleep.service"
systemctl daemon-reload
info "已复制 systemd 服务文件: $SERVICE_DIR/padsleep.service"
info "服务用户已设为: $RUN_USER"
info "可执行以下命令启用:"
info " sudo systemctl enable --now padsleep"
fi
# ── 检查 ADB ──
if command -v adb &>/dev/null; then
info "ADB 已安装: $(adb --version 2>&1 | head -1)"
else
warn "ADB 未安装,请执行: sudo apt install adb"
fi
# ── 完成 ──
echo ""
echo -e "${GREEN}═══════════════════════════════════════${NC}"
info "安装完成!"
echo ""
info "使用方法:"
echo " 1. 启动守护进程: padsleep"
echo " 2. 打开配置界面: padsleep-config"
echo " 3. 注册系统服务: sudo systemctl enable --now padsleep"
echo ""
info "配置文件: $INSTALL_DIR/config.json"
info "日志文件: $INSTALL_DIR/padsleep.log"
echo -e "${GREEN}═══════════════════════════════════════${NC}"
+222 -21
View File
@@ -6,7 +6,9 @@
阶段一功能: 阶段一功能:
- 根据配置的亮屏时间段自动亮屏/息屏 - 根据配置的亮屏时间段自动亮屏/息屏
- 屏幕亮起超过设定时长且不在亮屏时间段内 → 自动息屏 - 屏幕亮起超过设定时长且不在亮屏时间段内 → 自动息屏
- 在亮屏时间段内检测到屏幕熄灭 → 自动恢复亮屏
- 通过 Unix Socket 提供 IPC 接口,供 TUI 工具管理 - 通过 Unix Socket 提供 IPC 接口,供 TUI 工具管理
- 审计日志记录所有事件,自动维护日志大小不超过 50MB
""" """
import json import json
@@ -24,8 +26,10 @@ from adbutils import AdbClient, AdbDevice
# ── 常量 ──────────────────────────────────────────────────────── # ── 常量 ────────────────────────────────────────────────────────
SCRIPT_DIR = Path(__file__).parent.resolve() # 注意: .resolve().parent 确保即使通过 symlink 调用也能正确找到真实目录
SCRIPT_DIR = Path(__file__).resolve().parent
CONFIG_PATH = SCRIPT_DIR / "config.json" CONFIG_PATH = SCRIPT_DIR / "config.json"
LOG_PATH = SCRIPT_DIR / "padsleep.log"
SOCKET_PATH = "/tmp/padsleep.sock" SOCKET_PATH = "/tmp/padsleep.sock"
DEFAULT_CONFIG = { DEFAULT_CONFIG = {
@@ -50,6 +54,81 @@ logging.basicConfig(
log = logging.getLogger("padsleep") 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 — 配置加载/保存/监控 # ConfigManager — 配置加载/保存/监控
# ═══════════════════════════════════════════════════════════════════ # ═══════════════════════════════════════════════════════════════════
@@ -167,6 +246,16 @@ class TimeChecker:
now_m = TimeChecker.now_minutes() now_m = TimeChecker.now_minutes()
return any(TimeChecker.is_within_period(p, now_m) for p in periods) 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 操作封装(使用 adbutils 库) # AdbManager — ADB 操作封装(使用 adbutils 库)
@@ -175,15 +264,18 @@ class TimeChecker:
class AdbManager: class AdbManager:
"""封装所有 ADB 操作,使用 adbutils 库实现持久连接。""" """封装所有 ADB 操作,使用 adbutils 库实现持久连接。"""
def __init__(self, config_getter): def __init__(self, config_getter, audit_logger: AuditLogger | None = None):
self._config_getter = config_getter self._config_getter = config_getter
self._audit = audit_logger
self._client: AdbClient | None = None self._client: AdbClient | None = None
self._device: AdbDevice | None = None self._device: AdbDevice | None = None
self._ready = False self._ready = False
self._last_error = "" self._last_error = ""
self._screen_on_since: float | None = None self._screen_on_since: float | None = None
self._last_known_screen_on = False self._last_known_screen_on = False
self._last_adb_check = 0 self._last_adb_check = 0 # 限频用
self._prev_ready = False # 前次 ADB 就绪状态,用于检测变化
self._current_serial: str | None = None # 自动检测到的设备串号
self._wireless_addr: str = "" # 当前无线连接地址 (IP:PORT) self._wireless_addr: str = "" # 当前无线连接地址 (IP:PORT)
self._wireless_retries: int = 0 # 无线重连尝试次数 self._wireless_retries: int = 0 # 无线重连尝试次数
@@ -240,11 +332,15 @@ class AdbManager:
self._wireless_addr = addr self._wireless_addr = addr
self._wireless_retries = 0 self._wireless_retries = 0
log.info("无线 ADB 已连接: %s", addr) log.info("无线 ADB 已连接: %s", addr)
if self._audit:
self._audit.log("无线连接", f"已连接: {addr}")
return True, result return True, result
return False, result return False, result
except Exception as e: except Exception as e:
self._last_error = str(e) self._last_error = str(e)
log.error("无线连接失败: %s", e) log.error("无线连接失败: %s", e)
if self._audit:
self._audit.log("无线连接失败", f"{addr}: {e}")
return False, str(e) return False, str(e)
def disconnect_wireless(self, addr: str = "") -> bool: def disconnect_wireless(self, addr: str = "") -> bool:
@@ -256,6 +352,8 @@ class AdbManager:
client = self._ensure_client() client = self._ensure_client()
client.disconnect(target) client.disconnect(target)
log.info("无线 ADB 已断开: %s", target) log.info("无线 ADB 已断开: %s", target)
if self._audit:
self._audit.log("无线断开", f"已断开: {target}")
if target == self._wireless_addr: if target == self._wireless_addr:
self._wireless_addr = "" self._wireless_addr = ""
return True return True
@@ -371,6 +469,10 @@ class AdbManager:
if not target_serial or d.serial == target_serial: if not target_serial or d.serial == target_serial:
d.getprop("ro.build.version.sdk") d.getprop("ro.build.version.sdk")
self._device = d self._device = d
self._current_serial = d.serial
if not self._prev_ready and self._audit:
self._audit.log("ADB 设备已连接", f"串号: {d.serial}, 类型: USB")
self._prev_ready = True
self._ready = True self._ready = True
self._last_error = "" self._last_error = ""
return True return True
@@ -385,6 +487,9 @@ class AdbManager:
if d.serial == wireless_addr: if d.serial == wireless_addr:
d.getprop("ro.build.version.sdk") d.getprop("ro.build.version.sdk")
self._device = d self._device = d
if not self._prev_ready and self._audit:
self._audit.log("ADB 设备已连接", f"串号: {d.serial}, 类型: 无线")
self._prev_ready = True
self._ready = True self._ready = True
self._last_error = "" self._last_error = ""
self._wireless_addr = wireless_addr self._wireless_addr = wireless_addr
@@ -392,6 +497,9 @@ class AdbManager:
# 不在列表中,尝试连接 # 不在列表中,尝试连接
if self.ensure_wireless_connection(): if self.ensure_wireless_connection():
if not self._prev_ready and self._audit:
self._audit.log("ADB 设备已连接", f"串号: {wireless_addr}, 类型: 无线")
self._prev_ready = True
return True return True
if not devices and not wireless_host: if not devices and not wireless_host:
@@ -403,11 +511,16 @@ class AdbManager:
except Exception as e: except Exception as e:
self._last_error = str(e) self._last_error = str(e)
# 如果之前是就绪状态但检测失败 → 设备断开
if self._prev_ready and self._audit:
last_serial = self._current_serial or (self._device.serial if self._device else "未知")
self._audit.log("ADB 设备已断开", f"上次串号: {last_serial}, 原因: {self._last_error}")
self._prev_ready = False
self._ready = False self._ready = False
self._device = None self._device = None
return False return False
# ── 屏幕控制 ── # ── 屏幕状态 ──
def get_screen_state(self) -> tuple[bool, str]: def get_screen_state(self) -> tuple[bool, str]:
"""返回 (is_on, wakefulness),失败时返回上次已知状态。""" """返回 (is_on, wakefulness),失败时返回上次已知状态。"""
@@ -445,37 +558,86 @@ class AdbManager:
return time.time() - self._screen_on_since return time.time() - self._screen_on_since
return 0.0 return 0.0
# ── 屏幕控制(含验证)──
def screen_off(self) -> bool: def screen_off(self) -> bool:
"""发送息屏命令。""" """发送息屏命令,等待 2 秒后验证,失败时重试一次"""
log.info("执行息屏操作") log.info("执行息屏操作")
try: try:
device = self._ensure_device() device = self._ensure_device()
if device is None: if device is None:
log.error("息屏失败: 无可用设备") log.error("息屏失败: 无可用设备")
return False return False
if self._audit:
self._audit.log("息屏", "发出息屏指令")
device.shell("input keyevent 26") device.shell("input keyevent 26")
# 等待并验证
time.sleep(2)
is_on, _ = self.get_screen_state()
if is_on:
log.warning("息屏命令已发出但屏幕未熄灭,重试一次")
device.shell("input keyevent 26")
time.sleep(2)
is_on, _ = self.get_screen_state()
if is_on:
log.error("息屏重试后仍未熄灭")
self._last_error = "息屏验证失败"
if self._audit:
self._audit.log("息屏异常", "屏幕状态未变化")
return False
self._screen_on_since = None self._screen_on_since = None
log.info("息屏成功")
if self._audit:
self._audit.log("息屏成功", "屏幕已熄灭")
return True return True
except Exception as e: except Exception as e:
self._last_error = str(e) self._last_error = str(e)
log.error("息屏失败: %s", e) log.error("息屏失败: %s", e)
if self._audit:
self._audit.log("息屏失败", str(e))
return False return False
def screen_on(self) -> bool: def screen_on(self) -> bool:
"""发送亮屏并解锁命令。""" """发送亮屏命令,等待 2 秒后验证,失败时重试一次"""
log.info("执行亮屏操作") log.info("执行亮屏操作")
try: try:
device = self._ensure_device() device = self._ensure_device()
if device is None: if device is None:
log.error("亮屏失败: 无可用设备") log.error("亮屏失败: 无可用设备")
return False return False
if self._audit:
self._audit.log("亮屏", "发出亮屏指令")
device.shell("input keyevent 26") device.shell("input keyevent 26")
# 等待并验证
time.sleep(2)
is_on, _ = self.get_screen_state()
if not is_on:
log.warning("亮屏命令已发出但屏幕未点亮,重试一次")
device.shell("input keyevent 26")
time.sleep(2)
is_on, _ = self.get_screen_state()
if not is_on:
log.error("亮屏重试后仍未点亮")
self._last_error = "亮屏验证失败"
if self._audit:
self._audit.log("亮屏异常", "屏幕状态未变化")
return False
self._screen_on_since = time.time() self._screen_on_since = time.time()
device.shell("input keyevent 82") # 解锁(MENU # 解锁屏幕
device.shell("input keyevent 82")
log.info("亮屏成功")
if self._audit:
self._audit.log("亮屏成功", "屏幕已点亮")
return True return True
except Exception as e: except Exception as e:
self._last_error = str(e) self._last_error = str(e)
log.error("亮屏失败: %s", e) log.error("亮屏失败: %s", e)
if self._audit:
self._audit.log("亮屏失败", str(e))
return False return False
@@ -551,8 +713,9 @@ class PadSleepApp:
def __init__(self): def __init__(self):
self.running = True self.running = True
self.audit = AuditLogger()
self.config_mgr = ConfigManager() self.config_mgr = ConfigManager()
self.adb = AdbManager(self._get_config) self.adb = AdbManager(self._get_config, self.audit)
self.ipc = IpcServer(self) self.ipc = IpcServer(self)
self._lock = threading.Lock() self._lock = threading.Lock()
self._start_time = time.time() self._start_time = time.time()
@@ -562,6 +725,7 @@ class PadSleepApp:
"within_period": False, "within_period": False,
"adb_ready": False, "adb_ready": False,
"device": None, "device": None,
"device_serial": "",
"uptime": 0, "uptime": 0,
"start_time": self._start_time, "start_time": self._start_time,
} }
@@ -592,6 +756,8 @@ class PadSleepApp:
cfg = self.config_mgr.get() cfg = self.config_mgr.get()
status["periods_count"] = len(cfg.get("screen_on_periods", [])) status["periods_count"] = len(cfg.get("screen_on_periods", []))
status["check_interval"] = cfg.get("check_interval_seconds", 30) status["check_interval"] = cfg.get("check_interval_seconds", 30)
status["config_path"] = str(self.config_mgr.path)
status["log_path"] = str(LOG_PATH)
status["connection_type"] = self.adb.get_connection_type() status["connection_type"] = self.adb.get_connection_type()
status["wireless_host"] = cfg.get("adb_wireless_host", "") status["wireless_host"] = cfg.get("adb_wireless_host", "")
status["wireless_port"] = cfg.get("adb_wireless_port", 5555) status["wireless_port"] = cfg.get("adb_wireless_port", 5555)
@@ -607,9 +773,6 @@ class PadSleepApp:
if "start" not in p or "end" not in p: if "start" not in p or "end" not in p:
return {"ok": False, "error": "时间段格式错误"} return {"ok": False, "error": "时间段格式错误"}
self.config_mgr.update(new) 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}} return {"ok": True, "data": {"reloaded": True}}
if cmd == "reload": if cmd == "reload":
@@ -622,10 +785,12 @@ class PadSleepApp:
if cmd == "select_device": if cmd == "select_device":
serial = request.get("serial", "") serial = request.get("serial", "")
self.config_mgr.update({"adb_device_serial": serial}) self.config_mgr.update({"adb_device_serial": serial})
self.audit.log("设备选择", f"串号: {serial or '自动'}")
return {"ok": True, "data": {"selected": serial}} return {"ok": True, "data": {"selected": serial}}
if cmd == "shutdown": if cmd == "shutdown":
log.info("收到 IPC 关闭命令") log.info("收到 IPC 关闭命令")
self.audit.log("守护进程", "收到关闭命令")
self.running = False self.running = False
return {"ok": True} return {"ok": True}
@@ -637,6 +802,10 @@ class PadSleepApp:
ok = self.adb.screen_on() ok = self.adb.screen_on()
return {"ok": ok} return {"ok": ok}
if cmd == "log_tail":
n = request.get("lines", 50)
return {"ok": True, "data": {"log": self.audit.read_tail(n)}}
if cmd == "connect_wireless": if cmd == "connect_wireless":
host = request.get("host", "") host = request.get("host", "")
port = request.get("port", 5555) port = request.get("port", 5555)
@@ -663,15 +832,18 @@ class PadSleepApp:
def run(self): def run(self):
log.info("=" * 50) log.info("=" * 50)
log.info("padsleep_adb 守护程序启动") log.info("padsleep_adb 守护程序启动")
log.info("日志路径: %s", LOG_PATH)
log.info("=" * 50) log.info("=" * 50)
self.audit.log("守护进程启动", f"配置文件: {self.config_mgr.path}")
# 加载配置 # 加载配置
self.config_mgr.load() self.config_mgr.load()
# 检测 ADB # 检测 ADB
if not self.adb.check_available(): if not self.adb.check_available():
log.warning("ADB 未就绪,请安装: sudo apt install adb") log.warning("ADB 未就绪,请安装: sudo apt install adb")
log.warning("安装后自动检测到 ADB,无需重启本程序") log.warning("安装后自动检测到 ADB,无需重启本程序")
else: else:
log.info("ADB 可执行文件已就绪") log.info("ADB 可执行文件已就绪")
@@ -693,15 +865,19 @@ class PadSleepApp:
# 注册信号 # 注册信号
def sighandler(sig, _frame): def sighandler(sig, _frame):
log.info("收到信号 %s,正在退出...", sig) log.info("收到信号 %s,正在退出...", sig)
self.audit.log("守护进程", "收到退出信号")
self.running = False self.running = False
signal.signal(signal.SIGTERM, sighandler) signal.signal(signal.SIGTERM, sighandler)
signal.signal(signal.SIGINT, sighandler) signal.signal(signal.SIGINT, sighandler)
# 记录上一次屏幕状态用于外部变化检测
prev_screen_on: bool | None = None
# 主循环 # 主循环
while self.running: while self.running:
try: try:
self._tick() prev_screen_on = self._tick(prev_screen_on)
except Exception as e: except Exception as e:
log.error("主循环异常: %s", e, exc_info=True) log.error("主循环异常: %s", e, exc_info=True)
@@ -717,10 +893,11 @@ class PadSleepApp:
# 清理 # 清理
self.ipc.cleanup() self.ipc.cleanup()
self.audit.log("守护进程", "已退出")
log.info("padsleep_adb 守护程序已退出") log.info("padsleep_adb 守护程序已退出")
def _tick(self): def _tick(self, prev_screen_on: bool | None) -> bool:
"""单次检查逻辑。""" """单次检查逻辑。返回当前屏幕亮灭状态。"""
config = self.config_mgr.get() config = self.config_mgr.get()
# ── 检查 ADB 就绪 ── # ── 检查 ADB 就绪 ──
@@ -730,7 +907,7 @@ class PadSleepApp:
if not hasattr(self, "_last_adb_warn") or now - self._last_adb_warn > 60: if not hasattr(self, "_last_adb_warn") or now - self._last_adb_warn > 60:
log.warning("ADB 未就绪: %s", self.adb._last_error) log.warning("ADB 未就绪: %s", self.adb._last_error)
self._last_adb_warn = now 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() screen_on, wakefulness = self.adb.get_screen_state()
@@ -740,30 +917,54 @@ class PadSleepApp:
device = self.adb.get_ready_device() device = self.adb.get_ready_device()
self._update_status( self._update_status(
adb_ready=True, adb_ready=True,
device=device, device=device or "",
device_serial=config.get("adb_device_serial", ""),
screen_on=screen_on, screen_on=screen_on,
wakefulness=wakefulness, wakefulness=wakefulness,
within_period=within_period, 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 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: if not within_period and screen_on:
# 非亮屏时段但屏幕亮着 → 超时后息屏 # 非亮屏时段但屏幕亮着 → 超时后息屏
on_duration = self.adb.get_screen_on_duration() on_duration = self.adb.get_screen_on_duration()
if on_duration >= delay_sec: if on_duration >= delay_sec:
log.info("不在亮屏时段(已亮 %.0fs)→ 自动息屏", on_duration) log.info("不在亮屏时段(已亮 %.0fs)→ 自动息屏", on_duration)
self.audit.log("息屏", f"不在亮屏时间段 ({period_desc}),超时 {config.get('auto_sleep_delay_minutes')} 分钟")
self.adb.screen_off() self.adb.screen_off()
return False
else: else:
remaining = delay_sec - on_duration remaining = delay_sec - on_duration
if remaining <= 10: if remaining <= 10:
log.info("不在亮屏时段,%.0f 秒后将息屏", remaining) log.info("不在亮屏时段,%.0f 秒后将息屏", remaining)
elif within_period and not screen_on: elif within_period and screen_on:
# 在亮屏时段屏幕灭着 → 亮屏 # 在亮屏时段屏幕已亮 — 一切正常,但记录一个调试信息
log.info("在亮屏时段内 → 自动亮屏") if prev_screen_on is False:
self.adb.screen_on() # 这是上一步我们刚刚点亮了屏幕
pass
return screen_on
# ═══════════════════════════════════════════════════════════════════ # ═══════════════════════════════════════════════════════════════════
+4 -3
View File
@@ -1,12 +1,13 @@
[Unit] [Unit]
Description=padsleep_adb — 基于 ADB 的安卓平板亮屏/息屏控制 Description=padsleep_adb — 基于 ADB 的安卓平板亮屏/息屏控制
Documentation=https://github.com/kushidou/padsleep_adb
After=network.target After=network.target
[Service] [Service]
Type=simple Type=simple
User=kushidou User=_INSTALL_USER_
WorkingDirectory=/home/kushidou/apps/padsleep_adb WorkingDirectory=/opt/apps/padsleep
ExecStart=/usr/bin/python3 /home/kushidou/apps/padsleep_adb/padsleep.py ExecStart=/opt/apps/padsleep/.venv/bin/python3 /opt/apps/padsleep/padsleep.py
Restart=on-failure Restart=on-failure
RestartSec=10 RestartSec=10
StandardOutput=journal StandardOutput=journal
+77 -7
View File
@@ -21,7 +21,7 @@ from pathlib import Path
# ── 常量 ──────────────────────────────────────────────────────── # ── 常量 ────────────────────────────────────────────────────────
SOCKET_PATH = "/tmp/padsleep.sock" SOCKET_PATH = "/tmp/padsleep.sock"
SCRIPT_DIR = Path(__file__).parent.resolve() SCRIPT_DIR = Path(__file__).resolve().parent
CONFIG_PATH = SCRIPT_DIR / "config.json" CONFIG_PATH = SCRIPT_DIR / "config.json"
# ═══════════════════════════════════════════════════════════════════ # ═══════════════════════════════════════════════════════════════════
@@ -203,6 +203,10 @@ class TuiApp:
elif key == ord("h") or key == ord("H"): elif key == ord("h") or key == ord("H"):
self._show_help() self._show_help()
elif key == ord("v") or key == ord("V"):
if self.connected:
self._dlg_view_log()
# 定时自动刷新 # 定时自动刷新
if key == -1 and self.connected: if key == -1 and self.connected:
self.refresh() self.refresh()
@@ -311,16 +315,31 @@ class TuiApp:
y = self._draw_separator(y, "[2] ADB 配置") y = self._draw_separator(y, "[2] ADB 配置")
if self.status: if self.status:
dev = self.status.get("device") or "未选择" serial_cfg = self.status.get("device_serial", "")
detected = self.status.get("device", "")
conn_type = self.status.get("connection_type", "none") conn_type = self.status.get("connection_type", "none")
conn_icon = "USB" if conn_type == "usb" else "WiFi" if conn_type == "wireless" else "---" conn_icon = "USB" if conn_type == "usb" else "WiFi" if conn_type == "wireless" else "---"
if serial_cfg:
dev_str = f"{serial_cfg} (已指定)"
elif detected and detected != "":
dev_str = f"{detected} (自动检测)"
else:
dev_str = detected or "未连接"
has_device = conn_type != "none" or (detected and detected != "")
self._clear_line(y) self._clear_line(y)
try: try:
self.stdscr.addstr(y, 4, f"当前设备: {dev}") label = f"当前设备: {dev_str} [{conn_icon}]"
self.stdscr.addstr(f" [{conn_icon}]") if has_device:
self.stdscr.addstr(y, 4, label)
self.stdscr.addstr(y, 4 + len(label), "", curses.color_pair(3))
else:
self.stdscr.addstr(y, 4, "当前设备: 未连接", curses.color_pair(2))
except curses.error: except curses.error:
pass pass
y += 1 y += 1
y += 1
# 显示无线连接配置 # 显示无线连接配置
wireless_host = self.status.get("wireless_host", "") wireless_host = self.status.get("wireless_host", "")
@@ -352,7 +371,7 @@ class TuiApp:
if y < h: if y < h:
self._clear_line(y) self._clear_line(y)
try: try:
self.stdscr.addstr(y, 4, "(无设备 — 请连接平板)") self.stdscr.addstr(y, 4, "(无设备 — 请连接平板)", curses.color_pair(2))
except curses.error: except curses.error:
pass pass
y += 1 y += 1
@@ -410,6 +429,11 @@ class TuiApp:
except curses.error: except curses.error:
pass pass
y += 1 y += 1
try:
self.stdscr.addstr(y, 4, "配置: ")
self.stdscr.addstr(f"{s.get('config_path', '?')}", curses.A_DIM)
except curses.error:
pass
else: else:
if y < h: if y < h:
self._clear_line(y) self._clear_line(y)
@@ -421,7 +445,7 @@ class TuiApp:
if y < h: if y < h:
try: try:
self.stdscr.addstr(y, 4, "[L]重载 [F]息屏 [N]亮屏 [T]重连 [H]帮助 [Q]退出", self.stdscr.addstr(y, 4, "[L]重载 [F]息屏 [N]亮屏 [V]日志 [T]重连 [H]帮助 [Q]退出",
curses.A_DIM) curses.A_DIM)
except curses.error: except curses.error:
pass pass
@@ -445,7 +469,7 @@ class TuiApp:
pass # show message below separator pass # show message below separator
else: else:
if self.connected: if self.connected:
msg = "已连接 | ↑↓/jk 选择 A新增 E编辑 D删除 S选设备 W无线 H帮助" msg = "已连接 | ↑↓/jk 选择 A新增 E编辑 D删除 S选设备 W无线 V日志 H帮助"
else: else:
msg = "未检测到守护进程 | T 尝试连接" msg = "未检测到守护进程 | T 尝试连接"
try: try:
@@ -671,6 +695,51 @@ class TuiApp:
except (ValueError, IndexError): except (ValueError, IndexError):
return False return False
# ── 日志查看 ──
def _dlg_view_log(self):
"""查看审计日志"""
r = self.client.send("log_tail", lines=100)
if not r.get("ok"):
self.msg(f"获取日志失败: {r.get('error', '')}")
return
log_text = r["data"].get("log", "")
h, w = self.height, self.width
bh, bw = min(h - 2, 30), min(w - 4, 80)
by = max(1, (h - bh) // 2)
bx = max(1, (w - bw) // 2)
win = curses.newwin(bh, bw, by, bx)
lines = log_text.split("\n")
scroll = max(0, len(lines) - (bh - 3))
while True:
win.box()
win.addstr(0, 2, " 审计日志 (最近100行) ", curses.A_BOLD)
visible = lines[scroll:scroll + bh - 3]
for i, line in enumerate(visible):
if i >= bh - 3:
break
try:
win.addstr(i + 1, 2, line[:bw - 4])
except curses.error:
pass
footer = f"{scroll+1}-{scroll+len(visible)}/{len(lines)} ↑↓滚动 ESC返回 "
try:
win.addstr(bh - 1, 2, footer[:bw - 4], curses.A_DIM)
except curses.error:
pass
win.refresh()
k = win.getch()
if k == 27 or k == ord("q"): # ESC/q
break
elif k in (curses.KEY_UP, ord("k")) and scroll > 0:
scroll -= 1
elif k in (curses.KEY_DOWN, ord("j")) and scroll < len(lines) - (bh - 3):
scroll += 1
# ── 帮助 ── # ── 帮助 ──
def _show_help(self): def _show_help(self):
@@ -693,6 +762,7 @@ class TuiApp:
" W 无线连接设备", " W 无线连接设备",
" R 刷新状态", " R 刷新状态",
" L 重载守护进程配置", " L 重载守护进程配置",
" V 查看审计日志",
" F 强制息屏", " F 强制息屏",
" N 强制亮屏", " N 强制亮屏",
" T 重新连接守护进程", " T 重新连接守护进程",
+4
View File
@@ -1,4 +1,8 @@
# padsleep_adb 依赖 # padsleep_adb 依赖
# 安装方式(在项目目录下):
# python3 -m venv .venv
# .venv/bin/pip install -r requirements.txt
adbutils>=2.12.0 adbutils>=2.12.0
# 系统依赖:adb (Android Debug Bridge) # 系统依赖:adb (Android Debug Bridge)