fix: address 3 quality review findings — batch reject tracker, skip flush when clean, hoist frigate url check
This commit is contained in:
@@ -28,6 +28,7 @@ from .quality import blur_score_from_image
|
|||||||
from .reconcile import enrich_asset_with_face_data, reconcile_frigate_mappings
|
from .reconcile import enrich_asset_with_face_data, reconcile_frigate_mappings
|
||||||
from .upload_tracker import (
|
from .upload_tracker import (
|
||||||
UPLOAD_TRACKER_FILE,
|
UPLOAD_TRACKER_FILE,
|
||||||
|
REJECT_TRACKER_FILE,
|
||||||
get_lowest_quality_mapped_file,
|
get_lowest_quality_mapped_file,
|
||||||
get_most_redundant_mapped_file,
|
get_most_redundant_mapped_file,
|
||||||
get_tracked_frigate_file_count,
|
get_tracked_frigate_file_count,
|
||||||
@@ -368,6 +369,7 @@ def upload_to_frigate(jobs: list[dict]) -> None:
|
|||||||
person_has_fscores: bool = has_frigate_scores(name)
|
person_has_fscores: bool = has_frigate_scores(name)
|
||||||
|
|
||||||
begin_batch(UPLOAD_TRACKER_FILE)
|
begin_batch(UPLOAD_TRACKER_FILE)
|
||||||
|
begin_batch(REJECT_TRACKER_FILE)
|
||||||
try:
|
try:
|
||||||
for fname in person_files:
|
for fname in person_files:
|
||||||
fpath = os.path.join(person_dir, fname)
|
fpath = os.path.join(person_dir, fname)
|
||||||
@@ -604,6 +606,10 @@ def upload_to_frigate(jobs: list[dict]) -> None:
|
|||||||
flush_batch(UPLOAD_TRACKER_FILE)
|
flush_batch(UPLOAD_TRACKER_FILE)
|
||||||
except Exception as _flush_exc:
|
except Exception as _flush_exc:
|
||||||
logger.warning("flush_batch failed during cleanup — batch will be recovered on next begin_batch: %s", _flush_exc)
|
logger.warning("flush_batch failed during cleanup — batch will be recovered on next begin_batch: %s", _flush_exc)
|
||||||
|
try:
|
||||||
|
flush_batch(REJECT_TRACKER_FILE)
|
||||||
|
except Exception as _flush_exc:
|
||||||
|
logger.warning("flush_batch failed during cleanup — batch will be recovered on next begin_batch: %s", _flush_exc)
|
||||||
|
|
||||||
# Batch-map Frigate filenames to asset IDs now that all uploads are done.
|
# Batch-map Frigate filenames to asset IDs now that all uploads are done.
|
||||||
if actually_uploaded and not _skip_reconcile:
|
if actually_uploaded and not _skip_reconcile:
|
||||||
|
|||||||
@@ -44,6 +44,7 @@ REJECT_TRACKER_FILE = "frigate_rejected_ids.json"
|
|||||||
# Keyed by full path so tests with isolated tmp dirs never share entries.
|
# Keyed by full path so tests with isolated tmp dirs never share entries.
|
||||||
_cache: dict[str, dict] = {}
|
_cache: dict[str, dict] = {}
|
||||||
_deferred: set[str] = set() # paths whose disk writes are batched until flush_batch()
|
_deferred: set[str] = set() # paths whose disk writes are batched until flush_batch()
|
||||||
|
_dirty: set[str] = set() # deferred paths that received at least one _save during the batch
|
||||||
|
|
||||||
|
|
||||||
def _tracker_path(filename: str) -> Path:
|
def _tracker_path(filename: str) -> Path:
|
||||||
@@ -87,6 +88,7 @@ def _save(filename: str, data: dict) -> None:
|
|||||||
key = str(path)
|
key = str(path)
|
||||||
if key in _deferred:
|
if key in _deferred:
|
||||||
_cache[key] = data # accumulate in cache; disk write deferred until flush_batch()
|
_cache[key] = data # accumulate in cache; disk write deferred until flush_batch()
|
||||||
|
_dirty.add(key)
|
||||||
return
|
return
|
||||||
_write_to_disk(path, data)
|
_write_to_disk(path, data)
|
||||||
_cache[key] = data # update cache only after successful write
|
_cache[key] = data # update cache only after successful write
|
||||||
@@ -109,6 +111,7 @@ def begin_batch(filename: str) -> None:
|
|||||||
except Exception:
|
except Exception:
|
||||||
logger.warning("begin_batch: could not flush leftover deferred state for %s — partial progress may be lost", path)
|
logger.warning("begin_batch: could not flush leftover deferred state for %s — partial progress may be lost", path)
|
||||||
_deferred.discard(key)
|
_deferred.discard(key)
|
||||||
|
_dirty.discard(key)
|
||||||
_deferred.add(key)
|
_deferred.add(key)
|
||||||
|
|
||||||
|
|
||||||
@@ -116,9 +119,10 @@ def flush_batch(filename: str) -> None:
|
|||||||
"""Write the accumulated cache state for filename to disk."""
|
"""Write the accumulated cache state for filename to disk."""
|
||||||
path = _tracker_path(filename)
|
path = _tracker_path(filename)
|
||||||
key = str(path)
|
key = str(path)
|
||||||
if key in _cache:
|
if key in _dirty and key in _cache:
|
||||||
_write_to_disk(path, _cache[key])
|
_write_to_disk(path, _cache[key])
|
||||||
_deferred.discard(key)
|
_deferred.discard(key)
|
||||||
|
_dirty.discard(key)
|
||||||
|
|
||||||
|
|
||||||
def _flat_key(filename: str) -> str:
|
def _flat_key(filename: str) -> str:
|
||||||
@@ -379,17 +383,19 @@ def reset_all_people() -> None:
|
|||||||
approach is O(P²) because each call rebuilds the flat list from all remaining entries.
|
approach is O(P²) because each call rebuilds the flat list from all remaining entries.
|
||||||
"""
|
"""
|
||||||
upload_data = _load(UPLOAD_TRACKER_FILE)
|
upload_data = _load(UPLOAD_TRACKER_FILE)
|
||||||
|
frigate_url = _get_frigate_url()
|
||||||
|
if not frigate_url:
|
||||||
|
logger.info("FRIGATE_URL not set — skipping Frigate file deletion")
|
||||||
for person_name, raw_entry in upload_data.get("by_person", {}).items():
|
for person_name, raw_entry in upload_data.get("by_person", {}).items():
|
||||||
entry = _migrate_entry(raw_entry)
|
entry = _migrate_entry(raw_entry)
|
||||||
frigate_filenames = list(entry.get("frigate_files", {}).keys())
|
frigate_filenames = list(entry.get("frigate_files", {}).keys())
|
||||||
if not frigate_filenames:
|
if not frigate_filenames:
|
||||||
continue
|
continue
|
||||||
if not _get_frigate_url():
|
if frigate_url:
|
||||||
logger.info(f"FRIGATE_URL not set — skipping Frigate file deletion for {person_name}")
|
if delete_frigate_person_files(person_name, frigate_filenames):
|
||||||
elif delete_frigate_person_files(person_name, frigate_filenames):
|
logger.info(f"Deleted {len(frigate_filenames)} Frigate file(s) for {person_name}")
|
||||||
logger.info(f"Deleted {len(frigate_filenames)} Frigate file(s) for {person_name}")
|
else:
|
||||||
else:
|
logger.warning(f"Could not delete Frigate files for {person_name} — tracker reset proceeding anyway")
|
||||||
logger.warning(f"Could not delete Frigate files for {person_name} — tracker reset proceeding anyway")
|
|
||||||
_save(UPLOAD_TRACKER_FILE, {})
|
_save(UPLOAD_TRACKER_FILE, {})
|
||||||
_save(REJECT_TRACKER_FILE, {})
|
_save(REJECT_TRACKER_FILE, {})
|
||||||
logger.info("Reset all tracking data")
|
logger.info("Reset all tracking data")
|
||||||
|
|||||||
Reference in New Issue
Block a user