"""Central log of failed or unprofitable compressions."""
import json
import logging
import os
from datetime import UTC, datetime
from pathlib import Path
from platformdirs import user_data_dir
logger = logging.getLogger("rcomp")
#: Recorded reasons. Deterministic failures only: transient issues (lost
#: connection, upload errors) are not recorded, so they are naturally retried.
REASONS = {"probe_failed", "ffmpeg_error", "not_profitable", "exception", "target_exists", "ori_exists"}
[docs]
def tracking_path():
"""
Returns
-------
:class:`~pathlib.Path`
Location of the tracking file.
"""
return Path(user_data_dir(appname="rcomp", appauthor=False)) / "tracking.json"
[docs]
def tracking_key(file):
"""
Parameters
----------
file: :class:`~pathlib.Path` or :class:`str`
File location.
Returns
-------
:class:`str`
Canonical key: resolved absolute path, case-normalized (Windows paths
are case-insensitive).
"""
return os.path.normcase(str(Path(file).resolve()))
[docs]
class Tracker:
"""
JSON-backed store of files not worth re-trying.
An entry is only honored while the file's size and mtime still match: a
replaced or edited file is retried. Writes are immediate and atomic
(temporary file + :func:`os.replace`), so a crash never loses the log.
Parameters
----------
path: :class:`~pathlib.Path` or :class:`str`, optional
Alternative file location (defaults to :func:`tracking_path`).
"""
def __init__(self, path=None):
self.path = Path(path) if path is not None else tracking_path()
self.entries = {}
self._removed = set()
self.load()
[docs]
def load(self):
"""Load the file; a corrupt or missing file yields an empty store (with a warning)."""
self._removed = set()
self.entries = {}
if not self.path.exists():
return
try:
data = json.loads(self.path.read_text(encoding="utf-8"))
self.entries = dict(data["entries"])
except (json.JSONDecodeError, KeyError, TypeError, UnicodeDecodeError):
logger.warning("Corrupt tracking file %s, starting afresh.", self.path)
self.entries = {}
def _read_disk(self):
"""Current on-disk entries, best effort (empty on any problem)."""
try:
data = json.loads(self.path.read_text(encoding="utf-8"))
return dict(data["entries"])
except (OSError, json.JSONDecodeError, KeyError, TypeError, UnicodeDecodeError):
return {}
[docs]
def save(self):
"""
Merge with the on-disk state, then write atomically.
Several rcomp runs may share this file (parallel batches on different
disks): starting from the disk state keeps their entries alive, our own
removals stay removed, and our entries win on conflicting keys.
"""
merged = self._read_disk()
for key in self._removed:
merged.pop(key, None)
merged.update(self.entries)
self.entries = merged
self.path.parent.mkdir(parents=True, exist_ok=True)
tmp = self.path.with_name(self.path.name + ".tmp")
tmp.write_text(
json.dumps({"version": 1, "entries": self.entries}, indent=1),
encoding="utf-8",
)
os.replace(tmp, self.path)
[docs]
def should_skip(self, file):
"""
Parameters
----------
file: :class:`~pathlib.Path` or :class:`str`
Candidate file.
Returns
-------
:class:`str` or None
The recorded reason if an entry exists *and* the file is unchanged
(same size and mtime), None otherwise.
"""
entry = self.entries.get(tracking_key(file))
if entry is None:
return None
try:
stat = Path(file).stat()
except OSError:
return None
if stat.st_size == entry.get("size") and stat.st_mtime == entry.get("mtime"):
return entry.get("reason")
return None
[docs]
def record(self, file, reason, detail=""):
"""
Add or replace an entry and save immediately.
Parameters
----------
file: :class:`~pathlib.Path` or :class:`str`
File to remember.
reason: :class:`str`
One of :data:`REASONS`.
detail: :class:`str`
Free-form context (e.g. truncated ffmpeg stderr).
"""
try:
stat = Path(file).stat()
except OSError:
return
self.entries[tracking_key(file)] = {
"size": stat.st_size,
"mtime": stat.st_mtime,
"reason": reason,
"detail": detail,
"date": datetime.now(UTC).isoformat(timespec="seconds"),
}
self.save()
[docs]
def clear(self, file):
"""Remove the entry for `file` (no-op if absent); called after a success."""
if self.discard_key(tracking_key(file)):
self.save()
[docs]
def discard_key(self, key):
"""
Remove one entry by canonical key, without saving.
Returns
-------
:class:`bool`
Whether an entry was removed. The removal is remembered so a later
:meth:`save` does not resurrect the key from the disk state.
"""
if self.entries.pop(key, None) is not None:
self._removed.add(key)
return True
return False
[docs]
def prune(self):
"""
Drop entries whose file has disappeared or changed.
Returns
-------
:class:`int`
Number of entries removed.
"""
stale = []
for key, entry in self.entries.items():
try:
stat = Path(key).stat()
except OSError:
stale.append(key)
continue
if stat.st_size != entry.get("size") or stat.st_mtime != entry.get("mtime"):
stale.append(key)
for key in stale:
self.discard_key(key)
if stale:
self.save()
return len(stale)