From 3c997033cbfacaf673c36245533e045f54a7e9b1 Mon Sep 17 00:00:00 2001 From: Ankle Date: Mon, 10 Aug 2026 18:38:25 +0800 Subject: [PATCH] gengxin --- README.md | 39 ++++--- bin/cleanup-smartcam-data | 123 +++++++++++++++++++-- scripts/install.sh | 6 +- systemd/cleanup-smartcam-data-resume.timer | 3 + 4 files changed, 139 insertions(+), 32 deletions(-) diff --git a/README.md b/README.md index 12eba98..4702e45 100644 --- a/README.md +++ b/README.md @@ -8,8 +8,7 @@ cleanup-smartcam-data/ ├── README.md ├── assets/ -│ ├── cleanup-smartcam-data-flow.svg（旧版源文件） -│ └── cleanup-smartcam-data-flow.png（当前流程图） +│ └── cleanup-smartcam-data-flow.png ├── bin/ │ └── cleanup-smartcam-data ├── systemd/ @@ -43,6 +42,10 @@ cleanup-smartcam-data/ - smart-cam 容器名：`smart-cam` - 数据目录：`/userdata/data` - 容器内成功记录：`/smart-cam/kafka_sender/processed_files.log` +- 强校验回执：`/userdata/data/upload_completed_types.json` +- scan 强校验回执：`/userdata/data/upload_completed_scans.json` +- 容器必须通过 `--hostname "$(hostname)"` 继承宿主机设备编号；`config.json` + 的 `device.id` 可留空自动使用该主机名，显式填写时必须与主机名一致 - `smart_camera` 支持读取 `/smart-cam/data/.recording-paused-low-disk` （与宿主机 `/userdata/data/.recording-paused-low-disk` 是同一挂载文件） - 宿主机使用 Docker 和 systemd @@ -75,7 +78,11 @@ sudo ./scripts/install.sh - 开机2分钟后首次检查； - 之后每60分钟检查一次； - 每次定时检查都会安全清理已确认上传成功的数据，不以空闲空间阈值作为清理条件； -- 只清理 `processed_files.log` 中确认成功、且超过60分钟的数据； +- 只清理 `processed_files.log` 与 `upload_completed_types.json` 同时确认、且超过60分钟的数据； +- 每条回执必须同时满足：当前启用的上传类型全部成功、回执 `device_id` 等于当前 + 有效设备号、回执 `remote_base` 等于当前配置计算出的后台目录；有效设备号优先取 + 非空 `config.json device.id`，否则取容器主机名；显式设备号与主机名不一致、旧式无 + 设备号回执、设备号不一致或目标目录不一致时一律不删除； - 同时删除同一任务的： ```text @@ -84,22 +91,17 @@ sudo ./scripts/install.sh /userdata/data/annotations/.xml ``` -- 超过60分钟的0字节 raw，如果未被进程占用，也会按上述三件套删除； -- 每次运行都会对`videos_raw/videos/annotations`按base进行完整性对账； - - 三个成员齐全才视为完整三件套； - - 缺少任意成员且所有现存成员均超过60分钟、未被进程打开时，删除该base的全部残留； - - 新录制、仍被进程打开或未满60分钟的文件不会被完整性对账误删； +- 不再自动删除未获得强校验上传回执的零字节、残缺三件套或 `.part` 残留；这些文件 + 即使看起来无效，也优先保留供人工审计，避免把上传故障误判为可清理数据； - 已确认上传的正常清理在删除前会再次要求raw、annotated和XML三者齐全，缺一项整组跳过； -- scan 以一次扫描目录为原子单位；只有全部小图和`meta.json`的SFTP - 上传均完成、Kafka消息确认成功，并写入 - `/userdata/data/upload_processed_scans.log`后才允许清理； -- 已确认的scan还必须超过60分钟、`meta.json`状态为`completed`且目录内 +- scan 以一次扫描目录为原子单位；只有全部小图、`meta.json` 和可选 + `patrol_plan.json` 均通过 SFTP 大小校验、Kafka 消息确认成功，并写入同时包含当前 + 设备号、远端路径、Kafka topic、图片数和文件数的 + `/userdata/data/upload_completed_scans.json` 强回执后才允许清理； +- `upload_processed_scans.log` 只是队列索引，旧版只有 scan_id 的记录不能单独证明 + 上传目标正确，不再触发删除； +- 已强确认的 scan 还必须超过60分钟、`meta.json`状态为`completed`且目录内 没有文件被进程打开，随后才会整目录删除； -- 对`_raw.mp4.partN.mp4`录制残留执行整组安全检查： - - 同一base的所有part都必须超过60分钟、全部为0字节且均未被进程打开； - - 如果正式`_raw.mp4`已经存在，只删除符合条件的残留part，保留正常三件套； - - 如果正式raw不存在，删除全部残留part及对应的annotated视频和XML； - - 任一part仍较新、非0字节或正在使用时，整组跳过； - 空闲低于3GiB时只创建“禁止录像”标记，立即结束正在生成的录像，并停止产生新的 `videos`、`videos_raw`、`annotations`；持续检测、推理、RTSP/公网直播、scan和 Producer上传均不暂停； @@ -125,13 +127,14 @@ systemctl status cleanup-smartcam-data.service journalctl -u cleanup-smartcam-data.service -n 100 --no-pager ``` -确认成功记录可读： +确认成功记录与强校验回执均可读： ```bash docker exec smart-cam \ test -r /smart-cam/kafka_sender/processed_files.log test -r /userdata/data/upload_processed_scans.log +test -r /userdata/data/upload_completed_types.json ``` 确认 Docker 未重启： diff --git a/bin/cleanup-smartcam-data b/bin/cleanup-smartcam-data index ff00d9f..be8d457 100755 --- a/bin/cleanup-smartcam-data +++ b/bin/cleanup-smartcam-data @@ -6,6 +6,7 @@ DATA_ROOT="${DATA_ROOT:-/userdata/data}" CONTAINER="smart-cam" PROCESSED_FILE="/smart-cam/kafka_sender/processed_files.log" SCAN_PROCESSED_FILE="${DATA_ROOT}/upload_processed_scans.log" +SCAN_RECEIPTS_FILE="${DATA_ROOT}/upload_completed_scans.json" STATE_DIR="${STATE_DIR:-/var/lib/cleanup-smartcam-data}" RECORDING_PAUSE_MARKER="${RECORDING_PAUSE_MARKER:-${DATA_ROOT}/.recording-paused-low-disk}" LEGACY_PAUSED_STATE="${STATE_DIR}/capture-paused" @@ -131,20 +132,29 @@ load_processed_files() { docker exec -i "$CONTAINER" /opt/venv/bin/python - 2>/dev/null <<'PY' | import json import re +import socket processed_path = "/smart-cam/kafka_sender/processed_files.log" config_path = "/smart-cam/config/config.json" receipts_path = "/smart-cam/data/upload_completed_types.json" -legacy_types = {"videos_raw", "annotations"} pattern = re.compile(r"^video_[0-9]{8}_[0-9]{6}_[0-9]{3}\.xml$") with open(config_path, encoding="utf-8") as stream: - upload = json.load(stream).get("upload", {}) + config = json.load(stream) +upload = config.get("upload", {}) +configured_device_id = str(config.get("device", {}).get("id", "")).strip() +host_device_id = socket.gethostname().strip() +device_id = configured_device_id or host_device_id +server = upload.get("server", {}) if isinstance(upload.get("server", {}), dict) else {} +root = str(server.get("root_dir", "/storage2/file/remote_sync")).rstrip("/") +namespace = str(upload.get("storage_namespace", "smart_cam")).strip("/") +remote_base = f"{root}/{namespace}/{device_id}" if device_id else "" required = { name for name, default in ( ("videos", False), ("videos_raw", True), ("annotations", True) ) if bool(upload.get(name, default)) } + try: with open(receipts_path, encoding="utf-8") as stream: receipts = json.load(stream) @@ -156,14 +166,103 @@ with open(processed_path, encoding="utf-8") as stream: name = line.strip() if not pattern.fullmatch(name): continue - confirmed = set(receipts.get(name, {}).get("types", legacy_types)) - if required.issubset(confirmed): + receipt = receipts.get(name, {}) + confirmed = set(receipt.get("types", [])) + if ( + device_id + and (not configured_device_id or configured_device_id == host_device_id) + and receipt.get("device_id") == device_id + and receipt.get("remote_base") == remote_base + and required.issubset(confirmed) + ): print(name) PY sort -u > "$destination" [[ -s "$destination" ]] } +load_processed_scans() { + local destination="$1" + if ! docker inspect -f '{{.State.Running}}' "$CONTAINER" 2>/dev/null | grep -qx true; then + return 1 + fi + docker exec -i "$CONTAINER" /opt/venv/bin/python - 2>/dev/null <<'PY' | +import json +import os +import re +import socket + +config_path = "/smart-cam/config/config.json" +processed_path = "/smart-cam/data/upload_processed_scans.log" +receipts_path = "/smart-cam/data/upload_completed_scans.json" +scan_root = "/smart-cam/data/scan" +pattern = re.compile(r"^[0-9]{4}-[0-9]{2}-[0-9]{2}_[0-9]{2}-[0-9]{2}-[0-9]{2}$") + +with open(config_path, encoding="utf-8") as stream: + config = json.load(stream) +upload = config.get("upload", {}) +configured_device_id = str(config.get("device", {}).get("id", "")).strip() +host_device_id = socket.gethostname().strip() +if configured_device_id and configured_device_id != host_device_id: + raise SystemExit(0) +device_id = configured_device_id or host_device_id +server = upload.get("server", {}) if isinstance(upload.get("server", {}), dict) else {} +root = str(server.get("root_dir", "/storage2/file/remote_sync")).rstrip("/") +namespace = str(upload.get("storage_namespace", "smart_cam")).strip("/") +remote_base = f"{root}/{namespace}/{device_id}" if device_id else "" +kafka_topic = str(upload.get("kafka_topic", "")).strip() + +try: + with open(receipts_path, encoding="utf-8") as stream: + receipts = json.load(stream) +except (FileNotFoundError, json.JSONDecodeError, OSError): + receipts = {} + +try: + with open(processed_path, encoding="utf-8") as stream: + processed = {line.strip() for line in stream if pattern.fullmatch(line.strip())} +except OSError: + processed = set() + +for scan_id in sorted(processed): + receipt = receipts.get(scan_id, {}) + image_count = int(receipt.get("image_count", 0)) + plan_included = receipt.get("patrol_plan_included") is True + expected_files = image_count + 1 + (1 if plan_included else 0) + valid = ( + device_id + and receipt.get("device_id") == device_id + and receipt.get("remote_base") == remote_base + and receipt.get("remote_scan_path") == f"{remote_base}/scan/{scan_id}" + and receipt.get("kafka_topic") == kafka_topic + and receipt.get("sftp_confirmed") is True + and receipt.get("kafka_confirmed") is True + and image_count > 0 + and int(receipt.get("file_count", 0)) == expected_files + ) + if not valid: + continue + try: + with open(os.path.join(scan_root, scan_id, "meta.json"), encoding="utf-8") as stream: + meta = json.load(stream) + state = meta.get("state", {}) + valid = ( + meta.get("scan_id") == scan_id + and state.get("state") == "completed" + and int(state.get("total", 0)) == image_count + and int(state.get("completed", -1)) == image_count + and len(meta.get("images", [])) == image_count + and bool(meta.get("patrol_plan")) == plan_included + ) + except (OSError, ValueError, TypeError, json.JSONDecodeError): + valid = False + if valid: + print(scan_id) +PY + sort -u > "$destination" + [[ -s "$destination" ]] +} + build_safe_bundles() { local processed="$1" output="$2" xml base raw annotated newest_ts path complete : > "$output" @@ -231,13 +330,15 @@ delete_safe_bundles() { } delete_confirmed_scans() { - local scan_id scan_dir resolved size + local scan_id scan_dir resolved size confirmed local deleted_scans=0 deleted_bytes=0 unsafe - [[ -r "$SCAN_PROCESSED_FILE" ]] || { - log INFO "no confirmed scan upload list yet; scan cleanup skipped" + confirmed="$(mktemp /tmp/confirmed-smartcam-scans.XXXXXX)" + if ! load_processed_scans "$confirmed"; then + rm -f -- "$confirmed" + log INFO "no current-device strong scan upload receipts; scan cleanup skipped" return 0 - } + fi while IFS= read -r scan_id; do [[ "$scan_id" =~ ^[0-9]{4}-[0-9]{2}-[0-9]{2}_[0-9]{2}-[0-9]{2}-[0-9]{2}$ ]] || @@ -306,7 +407,8 @@ PY else log ERROR "failed to delete confirmed uploaded scan: ${scan_dir}" fi - done < <(sort -u "$SCAN_PROCESSED_FILE") + done < "$confirmed" + rm -f -- "$confirmed" log INFO "scan cleanup result: deleted=${deleted_scans} scans bytes=${deleted_bytes}" } @@ -499,9 +601,6 @@ delete_stale_zero_part_bundles() { main() { local free_bytes processed bundles recover_legacy_capture_pause - delete_stale_zero_bundles - delete_stale_zero_part_bundles - delete_stale_incomplete_bundles free_bytes="$(available_bytes)" # Protect the disk before cleanup work. Only recording is disabled; continuous diff --git a/scripts/install.sh b/scripts/install.sh index 9d1849d..8609bc0 100755 --- a/scripts/install.sh +++ b/scripts/install.sh @@ -35,8 +35,10 @@ install -o root -g root -m 0644 \ systemctl daemon-reload systemctl enable --now cleanup-smartcam-data.timer -# The fast timer is started only while low disk space disables local recording. -systemctl disable cleanup-smartcam-data-resume.timer 2>/dev/null || true +# Enable the recovery timer at boot so a low-disk marker that survives a host +# restart is noticed promptly. With no marker, its first run stops itself; +# the 60-minute primary cleanup schedule is unchanged. +systemctl enable cleanup-smartcam-data-resume.timer if [[ -f /userdata/data/.recording-paused-low-disk || \ -f /var/lib/cleanup-smartcam-data/capture-paused || \ -f /var/lib/userdata-video-cleanup/capture-paused ]]; then diff --git a/systemd/cleanup-smartcam-data-resume.timer b/systemd/cleanup-smartcam-data-resume.timer index 18282ee..d1fb696 100644 --- a/systemd/cleanup-smartcam-data-resume.timer +++ b/systemd/cleanup-smartcam-data-resume.timer @@ -7,3 +7,6 @@ OnActiveSec=1min OnUnitActiveSec=2min AccuracySec=15s Unit=cleanup-smartcam-data.service + +[Install] +WantedBy=timers.target