rack:固件下载目录自动清理(每小时,FwStore.prune)

- 固件库里登记过的(界面上传)不自动删;没登记的(ota_push.py 直传、台架镜像)放满 7 天删;
  上传中断的 .tmp 1 小时删;非 <16 hex>.bin / .tmp 的文件名一律不碰
- 板子下载一次几秒就完、回滚用的是板上另一个槽,删旧文件不影响任何板子
- main.py:有 --fw-dir 时起清理循环,出错只记日志;test_firmware.py 加 2 例,共 41 例(本机 + 服务器)
- 已部署;本机模拟板实测放一个 8 天前的孤儿镜像,启动即被清掉

Co-Authored-By: Claude Opus 5.5 <[email protected]>
这个提交包含在:
zhou yang
2026-09-27 14:48:12 +08:00
共同撰写人 Claude Opus 5.5
父节点 284571de20
当前提交 1cd4fd4082
共修改 5 个文件,包含 105 行新增和 4 行删除
+42
查看文件
@@ -17,6 +17,7 @@ import hashlib
import os
import re
import struct
import time
from typing import List, Optional
from .db import Db, now_ms
@@ -26,6 +27,13 @@ URL_MAX = 64 # OTA_URL_MAX(命令 JSON 字符串上限)
PROJECT = "phonerail"
SHA_RE = re.compile(r"^[0-9a-f]{64}$")
# 自动清理(prune)。下载目录里只有两种文件归本模块管:`<16 hex>.bin` 与上传中途的 `.<16 hex>.bin.tmp`;
# 别的名字一律不碰(目录是给 nginx 的,谁往里放了别的东西,不是这里该删的)。
FILE_RE = re.compile(r"^[0-9a-f]{16}\.bin$")
TMP_RE = re.compile(r"^\.[0-9a-f]{16}\.bin\.tmp$")
ORPHAN_KEEP_S = 7 * 86400 # 库里没登记的镜像(ota_push.py 直接传的、台架镜像)留 7 天
TMP_KEEP_S = 3600 # 上传中断留下的半截文件
# esp_app_desc_t 紧跟在镜像头(24 B)+ 第一个段头(8 B)之后
_DESC = 32
_DESC_MAGIC = 0xABCD5432
@@ -116,6 +124,40 @@ class FwStore:
return [self.get(r["sha256"]) for r in
self.db.q("SELECT sha256 FROM firmware ORDER BY uploaded DESC")]
def prune(self, now_s: Optional[float] = None) -> List[str]:
"""清掉**库里没登记**、而且放了 ORPHAN_KEEP_S 以上的镜像,以及 TMP_KEEP_S 以上的半截上传。
返回删掉的文件名。
⚠️ 固件库里登记过的(界面上传的)**不自动删** —— 那是人挑出来的版本,界面上有删除按钮。
没登记的删了也不影响板子:板子下载一次几秒就完,之后不再需要服务器上的文件
(回滚用的是板子自己另一个槽里的旧版,不是这里的)。7 天是给「推了一半、第二天接着推」留的余地。
"""
now = time.time() if now_s is None else now_s
known = {self.name_for(r["sha256"]) for r in self.db.q("SELECT sha256 FROM firmware")}
gone = []
try:
names = os.listdir(self.dir)
except FileNotFoundError:
return gone
for name in names:
if FILE_RE.match(name):
if name in known:
continue
keep = ORPHAN_KEEP_S
elif TMP_RE.match(name):
keep = TMP_KEEP_S
else:
continue
path = os.path.join(self.dir, name)
try:
if now - os.path.getmtime(path) < keep:
continue
os.remove(path)
gone.append(name)
except OSError:
pass # 刚被别人删了 / 权限:下一轮再说
return gone
def delete(self, sha: str) -> bool:
if self.get(sha) is None:
return False