Vito
534c95248a
* fix(review): anchor files on their saver in the scan and Needs review (#390) The scan and Apply now resolve with the file's name, as the workspace inspector does since #383, so the one saver whose filename_prefix names the file anchors its branch. A file whose branch that anchor resolves is never queued (from the inspection profile it offers no choice; a default-ambiguous file asks saver_anchored) and its pending, undecided items leave on the next scan. REVIEW_RULE_VERSION is folded into inspection item versions so old items are re-evaluated. No match, several matches or an unfinished walk anchor nothing. Apply refuses a choice the anchor contradicts. * fix(review): refuse a default choice the saver anchor contradicts (#390) Apply re-resolves a default item with the file's name: when its saver anchors the branch, no candidate is offered and the choice is refused. The scan's anchor check skips bytes carrying a decision and remembers unanchored file states per process, so a rescan does not inspect an unchanged file again. * fix(review): queue an anchored file with its saver's pair as the choice (#390) Maintainer ruling: a file never falls through both the link and the queue. When only the saver anchor resolves a file the default profile and the parser floor leave unlinked, the scan queues it with the anchor's pair as its one candidate (evidence anchor: saver), never linking it. Files already linked or resolved by the default path are unchanged. Apply accepts the anchored pair when the bytes now resolve to it, and an anchored item is a choice the untraced settle keeps. * fix(review): upgrade old default items of anchored files, tag anchored results (#390) A pending default item recorded before the saver anchor no longer shortcuts the rescan when only the anchor resolves the file: the anchored item supersedes it. Apply accepts a resolved re-resolution only when it is tagged as the saver anchor's, the anchored pair must carry a positive text to choose, and limited anchored evidence still counts as a choice. TESTING.md describes the ruling.
401 lines
17 KiB
Python
401 lines
17 KiB
Python
"""Needs review recording for the output scan and the live monitor (#249).
|
|
|
|
The rule (maintainer test of #284/#288): a file is queued only when its
|
|
final outcome is ambiguous and unlinked. The strict resolver must find it
|
|
``ambiguous``, the scan's legacy fallback (the browser parser,
|
|
``_graph_prompt_text``) must not link it, and no saved row may link it
|
|
already. A file the fallback links, an exact pair the ingester stored and an
|
|
already linked file are never queued. Neither is an ambiguous file with
|
|
nothing to choose: fewer than two of its candidates carry a prompt
|
|
(``utils.review_capture.reviewable``); the scan reports it unresolved ("could
|
|
not be traced") unless the inspection profile finds a choice in it.
|
|
|
|
Only scans add items (maintainer ruling, 2026-10-23). The monitor's pass
|
|
records nothing: it only removes the pending items of an output it linked.
|
|
|
|
The queue heals itself: a pending item whose file is now linked (by the
|
|
fallback, a later resolution or the monitor) is removed by the next scan
|
|
that lists the file or by the monitor's next pass over it. The next scan
|
|
also removes a pending item without a choice whose file is still ambiguous
|
|
without one (``settle_untraced_review_items``), and an older version's
|
|
item of bytes now untraced (``settle_stale_review_items``). Only an undecided item goes
|
|
(``pending``, revision 0, no decision row); a decision is never deleted or
|
|
changed (``settle_linked_review_items``).
|
|
|
|
The scan records through ``_record_scanned_review`` (best effort: a failure
|
|
is logged and counted), which goes through
|
|
``utils.review_capture.record_ambiguous_file`` on the API's capture store;
|
|
the monitor calls ``record_review_file`` from its worker thread after its
|
|
own link. Nothing here links a file.
|
|
|
|
Recording changes records, so it follows the registry's ``records``
|
|
admission (#289). The scan records under its own ``records`` job. The
|
|
monitor's pass lasts one file and is not a job: it is refused while a
|
|
``records`` or ``guarded`` job runs (``deferred``), and the next scan reads
|
|
the file again. Thumbnail work (the job and the gallery's generate and clear
|
|
holds, kind ``thumbnails``) is ``guarded`` but only writes and removes preview
|
|
files under ``thumbnails/`` and never touches the database, so it does not
|
|
defer the recording (#249 S5 M7): a long rebuild no longer holds Needs review
|
|
back for its whole run.
|
|
"""
|
|
|
|
import hashlib
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from .admin import SCAN_LOG_PREFIXES, _plural, _scan_location, _scan_roots
|
|
from .jobs import THUMBNAILS_JOB_KIND
|
|
|
|
# The kind the monitor's recording checks admission under (never registered).
|
|
REVIEW_RECORDING_KIND = "review_recording"
|
|
# Guarded kinds that change no records, so the recording may run beside them.
|
|
PREVIEW_ONLY_KINDS = (THUMBNAILS_JOB_KIND,)
|
|
# record_ambiguous_file outcomes that leave the file in the pending queue.
|
|
QUEUED_OUTCOMES = ("recorded", "limited", "duplicate")
|
|
NEW_OUTCOMES = ("recorded", "limited")
|
|
|
|
try:
|
|
from ...utils import review_capture
|
|
from ...utils.review_capture import record_ambiguous_file
|
|
except ImportError:
|
|
from utils import review_capture
|
|
from utils.review_capture import record_ambiguous_file
|
|
|
|
# Untraced files remembered per process so a rescan of an unchanged file does
|
|
# not run the inspection profile again (oldest forgotten first). A plain dict
|
|
# of 16-byte digests (the locator, and its size, mtime_ns and item version):
|
|
# measured at 8.6 MiB with 65,536 entries (tracemalloc, CPython 3.14).
|
|
MAX_UNTRACED_MEMORY = 65_536
|
|
|
|
|
|
def _digest(text: str) -> bytes:
|
|
data = text.encode("utf-8", "surrogatepass")
|
|
return hashlib.blake2b(data, digest_size=16).digest()
|
|
|
|
|
|
def _memory_key(locator: str) -> bytes:
|
|
"""A 16-byte digest of ``locator`` (the untraced memory's key)."""
|
|
return _digest(locator)
|
|
|
|
|
|
# The scan log's reason for items that leave the queue with nothing to choose.
|
|
NOTHING_TO_CHOOSE = "nothing to choose"
|
|
|
|
|
|
class ReviewRecordingMixin:
|
|
"""Record the scan's and the monitor's ambiguous files for Needs review."""
|
|
|
|
def _record_scanned_review(self, media_file, roots, review=None, scanned=None):
|
|
"""Record one ambiguous, unlinked scanned file (logged and counted).
|
|
|
|
``scanned`` is the scan's ``(metadata, inspection_resolution)``; the
|
|
resolution is None when the scan did not run the inspection profile.
|
|
"""
|
|
try:
|
|
outcome = self._record_review_sync(
|
|
media_file, roots, review, inspection=True, scanned=scanned
|
|
)
|
|
except Exception:
|
|
self.logger.warning(
|
|
f"{media_file.name} could not be added to Needs review",
|
|
exc_info=True,
|
|
)
|
|
if review is not None:
|
|
review["errors"] += 1
|
|
return
|
|
if review is None:
|
|
return
|
|
if outcome in NEW_OUTCOMES:
|
|
review["added"] += 1
|
|
if outcome in QUEUED_OUTCOMES:
|
|
review["queued"] += 1
|
|
|
|
def _record_review_sync(
|
|
self, media_file, roots, review=None, inspection=False, scanned=None
|
|
):
|
|
"""``record_ambiguous_file`` for a file under one of ``roots`` (blocking).
|
|
|
|
``inspection`` (the scan only) lets a file without a choice under the
|
|
default profile be recorded from the inspection profile's choices.
|
|
An ``untraced`` file (its captured bytes offer nothing to choose) also
|
|
loses its pending items that offer nothing either.
|
|
"""
|
|
located = self._review_location(media_file, roots)
|
|
if located is None:
|
|
return "skipped"
|
|
root, index, relative, _spelled = located
|
|
outcome = record_ambiguous_file(
|
|
self.db,
|
|
self._capture_store(),
|
|
Path(root),
|
|
index,
|
|
relative,
|
|
inspection=inspection,
|
|
scanned=scanned,
|
|
)
|
|
if outcome == "untraced":
|
|
self._settle_untraced_reviews([media_file], roots, review)
|
|
return outcome
|
|
|
|
def _untraced_scan_outcome(self, media_file, roots, canonical, review=None):
|
|
"""The scan status of an unlinked file the default profile left
|
|
ambiguous without a choice (blocking).
|
|
|
|
``ambiguous`` when the inspection profile offers a choice: the file is
|
|
recorded for review from that profile (never linked), or ``unchanged``
|
|
when a saved row links it already. Otherwise ``unresolved``, and its
|
|
pending items without a choice leave the queue. The profile runs at
|
|
most once per file state: not for bytes already queued (or decided)
|
|
at the file's current size and ``mtime_ns`` under the version that
|
|
its records depend on, and not again for a state it already found
|
|
untraced in this process.
|
|
"""
|
|
try:
|
|
choice = self._inspection_choice(media_file, roots, canonical)
|
|
except Exception:
|
|
# One bad file never stops the scan: it stays "could not be traced".
|
|
self.logger.warning(
|
|
f"{media_file.name} could not be inspected for Needs review",
|
|
exc_info=True,
|
|
)
|
|
if review is not None:
|
|
review["errors"] += 1
|
|
return "unresolved"
|
|
if choice is None:
|
|
self._settle_untraced_reviews([media_file], roots, review)
|
|
self._settle_stale_reviews(media_file, roots, canonical, review)
|
|
return "unresolved"
|
|
if self._review_file_linked(media_file, roots):
|
|
return "unchanged"
|
|
self._record_scanned_review(media_file, roots, review, scanned=choice)
|
|
return "ambiguous"
|
|
|
|
def _inspection_choice(self, media_file, roots, canonical):
|
|
"""``(canonical, resolution_or_None)`` when the file offers a choice.
|
|
|
|
The resolution is None when the file's current state is already in
|
|
the queue (so the profile did not run). None when it offers none.
|
|
"""
|
|
if canonical is None:
|
|
return None
|
|
version = review_capture.inspection_review_version(canonical)
|
|
located = self._review_location(media_file, roots)
|
|
if version is None or located is None:
|
|
return None
|
|
locator = os.path.normcase(located[3])
|
|
try:
|
|
facts = os.lstat(located[3])
|
|
except OSError:
|
|
facts = None
|
|
state = None
|
|
if facts is not None:
|
|
state = _digest(f"{facts.st_size}:{facts.st_mtime_ns}:{version}")
|
|
if self.db.review_decided_at(
|
|
locator, facts.st_size, facts.st_mtime_ns
|
|
) or self.db.review_fingerprint_queued(
|
|
locator, facts.st_size, facts.st_mtime_ns, version
|
|
):
|
|
return canonical, None
|
|
memory = self._untraced_memory()
|
|
key = _memory_key(locator)
|
|
if state is not None and memory.get(key) == state:
|
|
return None
|
|
resolution = review_capture.inspection_choice(canonical, media_file.name)
|
|
if resolution is not None:
|
|
memory.pop(key, None)
|
|
return canonical, resolution
|
|
if state is not None:
|
|
memory.pop(key, None) # re-inserted last: the newest state
|
|
memory[key] = state
|
|
while len(memory) > MAX_UNTRACED_MEMORY:
|
|
del memory[next(iter(memory))]
|
|
return None
|
|
|
|
def _settle_stale_reviews(self, media_file, roots, canonical, review=None):
|
|
"""Remove the pending, undecided items of an untraced file's current
|
|
bytes recorded under another version (best effort, counted in
|
|
``review["untraced"]``).
|
|
|
|
The current resolver and table offer nothing to choose in these
|
|
bytes, so an item of an older version would only refuse Apply.
|
|
"""
|
|
located = self._review_location(media_file, roots)
|
|
if located is None or canonical is None:
|
|
return 0
|
|
# An unusable semantics table says nothing about the current
|
|
# versions; never let it retire items a working table recorded.
|
|
if review_capture.inspection_review_version(canonical) is None:
|
|
return 0
|
|
try:
|
|
facts = os.lstat(located[3])
|
|
removed = self.db.settle_stale_review_items(
|
|
os.path.normcase(located[3]),
|
|
facts.st_size,
|
|
facts.st_mtime_ns,
|
|
review_capture.item_versions(canonical),
|
|
)
|
|
except Exception:
|
|
self.logger.warning(
|
|
f"Stale Needs review items of {media_file.name} could not be "
|
|
"removed",
|
|
exc_info=True,
|
|
)
|
|
if review is not None:
|
|
review["errors"] += 1
|
|
return 0
|
|
if review is not None:
|
|
review["untraced"] += removed
|
|
return removed
|
|
|
|
def _untraced_memory(self):
|
|
memory = getattr(self, "_untraced_inspections", None)
|
|
if memory is None:
|
|
memory = self._untraced_inspections = {}
|
|
return memory
|
|
|
|
@staticmethod
|
|
def _review_location(media_file, roots, real_roots=None):
|
|
"""``(root, root_index, relative, spelled)`` of a listed file, or None.
|
|
|
|
``spelled`` is the capture's spelling of the file (the real root joined
|
|
with the relative path); ``os.path.normcase`` of it is the locator
|
|
review items are stored under. ``real_roots`` caches each root's
|
|
``realpath`` across a batch.
|
|
"""
|
|
relative, _label, index = _scan_location(media_file, roots)
|
|
root = next((r for r, _l, i in roots if i == index), None)
|
|
if index is None or root is None:
|
|
return None
|
|
if real_roots is None:
|
|
real_roots = {}
|
|
real = real_roots.get(index)
|
|
if real is None:
|
|
real = real_roots[index] = os.path.realpath(os.fspath(root))
|
|
spelled = os.path.join(real, *relative.split("/"))
|
|
return root, index, relative, spelled
|
|
|
|
def _review_file_linked(self, media_file, roots):
|
|
"""True when a saved row links the scanned file (any spelling)."""
|
|
located = self._review_location(media_file, roots)
|
|
spellings = (str(media_file),) + ((located[3],) if located else ())
|
|
return self.db.review_file_linked(spellings)
|
|
|
|
def _settle_scanned_reviews(self, media_files, roots, review=None):
|
|
"""Remove the pending items of ``media_files`` that are now linked.
|
|
|
|
Blocking; best effort like recording (a failure is logged and
|
|
counted in ``review["errors"]``). Returns how many items left.
|
|
"""
|
|
real_roots = {}
|
|
files = []
|
|
for media_file in media_files:
|
|
located = self._review_location(media_file, roots, real_roots)
|
|
if located is not None:
|
|
spelled = located[3]
|
|
files.append((os.path.normcase(spelled), (str(media_file), spelled)))
|
|
try:
|
|
settled = self.db.settle_linked_review_items(files) if files else 0
|
|
except Exception:
|
|
self.logger.warning(
|
|
"Needs review items of linked files could not be removed",
|
|
exc_info=True,
|
|
)
|
|
if review is not None:
|
|
review["errors"] += 1
|
|
return 0
|
|
if review is not None:
|
|
review["settled"] += settled
|
|
return settled
|
|
|
|
def _settle_untraced_reviews(self, media_files, roots, review=None):
|
|
"""Remove the pending items without a choice of untraced ``media_files``.
|
|
|
|
``media_files`` are files found ambiguous with fewer than two
|
|
candidates that carry a prompt. Blocking; best effort like the linked
|
|
settle (a failure is logged and counted in ``review["errors"]``).
|
|
Returns how many items left.
|
|
"""
|
|
locators = []
|
|
for media_file in media_files:
|
|
located = self._review_location(media_file, roots)
|
|
if located is not None:
|
|
locators.append(os.path.normcase(located[3]))
|
|
try:
|
|
removed = self.db.settle_untraced_review_items(locators) if locators else 0
|
|
except Exception:
|
|
self.logger.warning(
|
|
"Needs review items without a choice could not be removed",
|
|
exc_info=True,
|
|
)
|
|
if review is not None:
|
|
review["errors"] += 1
|
|
return 0
|
|
if review is not None:
|
|
review["untraced"] += removed
|
|
return removed
|
|
|
|
def record_review_file(self, image_path):
|
|
"""Apply the Needs review rule to one monitor output (its worker thread).
|
|
|
|
Only scans add items (maintainer ruling, 2026-10-23): the monitor
|
|
only clears the items of outputs it links. Maps ``image_path`` to the
|
|
gallery root that lists it; a file under no current root is
|
|
``skipped``. A linked file's pending items are removed and the answer
|
|
is ``linked``. Any other file is ``skipped``: nothing is captured or
|
|
recorded, and the next scan reads it. While a job runs that a
|
|
``records`` job may not run beside, other than thumbnail work
|
|
(``PREVIEW_ONLY_KINDS``), nothing is written and the answer is
|
|
``deferred``. Errors propagate: the monitor logs them and keeps its
|
|
link.
|
|
"""
|
|
busy = self._jobs.blocker(
|
|
REVIEW_RECORDING_KIND, admission="records", compatible=PREVIEW_ONLY_KINDS
|
|
)
|
|
if busy is not None:
|
|
self.logger.info(
|
|
f"Needs review was not updated for "
|
|
f"{os.path.basename(str(image_path))} while {busy.title} runs; "
|
|
"the next scan reads it again"
|
|
)
|
|
return "deferred"
|
|
full = os.path.abspath(str(image_path))
|
|
located = Path(os.path.realpath(os.path.dirname(full))) / os.path.basename(full)
|
|
roots = _scan_roots(self._get_all_output_dirs())
|
|
if self._review_location(located, roots) is None:
|
|
return "skipped"
|
|
if self._review_file_linked(located, roots):
|
|
self._settle_scanned_reviews([located], roots)
|
|
return "linked"
|
|
return "skipped"
|
|
|
|
@staticmethod
|
|
def _log_review_items(job, review):
|
|
"""The scan log's Needs review summary lines (none when nothing moved)."""
|
|
prefix = SCAN_LOG_PREFIXES["review"]
|
|
if review["added"]:
|
|
job.log_line(
|
|
f"{prefix} {_plural(review['added'], 'image')} added to Needs review"
|
|
)
|
|
already = review["queued"] - review["added"]
|
|
if already:
|
|
job.log_line(
|
|
f"{prefix} {_plural(already, 'image')} already in Needs review"
|
|
)
|
|
if review["settled"]:
|
|
job.log_line(
|
|
f"{SCAN_LOG_PREFIXES['known']} "
|
|
f"{_plural(review['settled'], 'image')} left Needs review · now linked"
|
|
)
|
|
if review["untraced"]:
|
|
job.log_line(
|
|
f"{SCAN_LOG_PREFIXES['known']} "
|
|
f"{_plural(review['untraced'], 'image')} left Needs review · "
|
|
f"{NOTHING_TO_CHOOSE}"
|
|
)
|
|
if review["errors"]:
|
|
job.log_line(
|
|
f"{SCAN_LOG_PREFIXES['errors']} "
|
|
f"{_plural(review['errors'], 'image')} could not be added to "
|
|
"Needs review"
|
|
)
|