gengxin
This commit is contained in:
Ankle committed
1 parent
71c4711282
commit
3c997033cb
4 files changed
+139
-32
No files matched your search
+111
-12
@@ -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
|
||||
|
||||
Reference in new issue
Block a user