nodes/ComfyAssets-ComfyUI_PromptManager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/feat/249-needs-review-backend' into feat/249-needs-review-ui

Browse Source 
# Conflicts:
#	web/js/workspace/images.mjs
This commit is contained in:
Vito Sansevero
2026-10-23 09:53:22 -07:00
parent 5467cdbcc8 95f88d3966
commit 828b5f9db3
38 changed files with 2492 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/api/admin.py
+42 -14
View File
@@ -211,6 +211,28 @@ def _time_left(done, total, batches, elapsed):
return f" · about {minutes} min left"
def new_review_counts():
"""The scan's Needs review tally: ``added`` new items, ``queued`` files
in the queue after this scan (new or already there), ``settled`` items
removed because their file is linked now, and ``errors``."""
return {"added": 0, "queued": 0, "settled": 0, "errors": 0}
def _review_scope(review):
"""The scan job's ``scope`` keys for its Needs review tally.
``review_items`` is the number of this scan's files that are in the
pending queue after it: the review group's ambiguous files, less those a
capture could not read or a decision already settled.
"""
return {
"review_items": review["queued"],
"review_added": review["added"],
"review_settled": review["settled"],
"review_errors": review["errors"],
}
def _scan_status_line(done, total, batches, elapsed):
""" "Processing file N of M", with an estimate once enough batches ran."""
line = f"Processing file {done:,} of {total:,}"
@@ -1445,8 +1467,8 @@ class AdminRoutesMixin:
known = set()
if media_files and job.scope.get("files") != "all":
known = await self._known_scan_files(job, media_files)
review = {"recorded": 0, "errors": 0}
job.scope.update(review_items=0, review_errors=0)
review = new_review_counts()
job.scope.update(_review_scope(review))
async for event in self._scan_batches(
media_files, counts, job, roots, known, review
):
@@ -1524,11 +1546,12 @@ class AdminRoutesMixin:
)
for key in counts:
counts[key] += batch_counts[key]
if review is not None:
job.scope.update(
review_items=review["recorded"],
review_errors=review["errors"],
)
if review is not None:
# After the batch's links: a file linked now leaves the queue.
await self._run_in_executor(
self._settle_scanned_reviews, batch, roots, review
)
job.scope.update(_review_scope(review))
for group, row, log_line in outcomes:
job.record(group, row, log_line)
@@ -1585,10 +1608,11 @@ class AdminRoutesMixin:
Returns the legacy counts. When ``outcomes`` is a list, every processed
file appends its ``(group, row, log_line)`` for the scan job, with the
path relative to its root in ``roots`` (see ``_scan_roots``). A file the
resolver finds ambiguous is also recorded for Needs review (best
effort, counted in ``review`` when given); statuses and counts do not
depend on it.
path relative to its root in ``roots`` (see ``_scan_roots``). A file
whose final outcome is ambiguous and unlinked (the resolver found it
ambiguous and the fallback did not link it) is recorded for Needs
review (best effort, counted in ``review`` when given); one a saved
row already links is reported ``unchanged`` and never queued.
"""
counts = dict.fromkeys(SCAN_COUNT_KEYS, 0)
@@ -1610,8 +1634,11 @@ class AdminRoutesMixin:
counts["failed"] += 1
note(media_file, "error")
continue
if ambiguous:
self._record_scanned_review(media_file, roots, review)
if ambiguous and outcome == "ambiguous":
if self._review_file_linked(media_file, roots):
outcome = "unchanged" # already linked: not a review case
else:
self._record_scanned_review(media_file, roots, review)
if outcome is None:
note(media_file, None)
continue
@@ -1655,7 +1682,8 @@ class AdminRoutesMixin:
``status``, the prompt's display ``title`` and, for a resolved pair,
the pair counts. When the resolver finds the file ambiguous (whether
or not the browser parser then links it) and ``ambiguous`` is a list,
``media_file`` is appended to it.
``media_file`` is appended to it; only a file that then returns
``"ambiguous"`` (nothing linked it) is a Needs review case.
"""
if not metadata:
return None
py/api/review_recording.py
+126 -28
View File
@@ -1,16 +1,29 @@
"""Needs review recording for the output scan and the live monitor (#249).
The scan hands every file the ingestion resolver finds ambiguous to
``_record_scanned_review`` (best effort: a failure is logged and counted);
the monitor calls ``record_review_file`` from its worker thread after its
fallback link. Both go through ``utils.review_capture.record_ambiguous_file``
on the API's capture store. Nothing here links a file.
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.
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. 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); the monitor calls ``record_review_file`` from its
worker thread after its own link. Both go through
``utils.review_capture.record_ambiguous_file`` on the API's capture store.
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 recording lasts one file and is not a job: it is refused while a
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 ambiguous file again.
the file again.
"""
import os
@@ -20,6 +33,9 @@ from .admin import SCAN_LOG_PREFIXES, _plural, _scan_location, _scan_roots
# The kind the monitor's recording checks admission under (never registered).
REVIEW_RECORDING_KIND = "review_recording"
# record_ambiguous_file outcomes that leave the file in the pending queue.
QUEUED_OUTCOMES = ("recorded", "limited", "duplicate")
NEW_OUTCOMES = ("recorded", "limited")
try:
from ...utils.review_capture import record_ambiguous_file
@@ -31,7 +47,7 @@ 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):
"""Record one ambiguous scanned file; a failure is logged and counted."""
"""Record one ambiguous, unlinked scanned file (logged and counted)."""
try:
outcome = self._record_review_sync(media_file, roots)
except Exception:
@@ -42,48 +58,130 @@ class ReviewRecordingMixin:
if review is not None:
review["errors"] += 1
return
if review is not None and outcome in ("recorded", "limited"):
review["recorded"] += 1
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):
"""``record_ambiguous_file`` for a file under one of ``roots`` (blocking)."""
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:
located = self._review_location(media_file, roots)
if located is None:
return "skipped"
root, index, relative, _spelled = located
return record_ambiguous_file(
self.db, self._capture_store(), Path(root), index, relative
)
def record_review_file(self, image_path):
"""Needs review recording for the live monitor (its worker thread).
@staticmethod
def _review_location(media_file, roots, real_roots=None):
"""``(root, root_index, relative, spelled)`` of a listed file, or None.
Maps ``image_path`` to the gallery root that lists it (the listing's
``root_index``); a file under no current root is ``skipped``. While a
job runs that a ``records`` job may not run beside, nothing is recorded
and the answer is ``deferred``. Errors propagate: the monitor logs them
and keeps its link.
``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 _fallback_links(self, metadata):
"""True when the scan's legacy fallback would link these records."""
return bool(metadata.get("prompt")) and bool(self._graph_prompt_text(metadata))
def record_review_file(self, image_path):
"""Apply the Needs review rule to one monitor output (its worker thread).
Maps ``image_path`` to the gallery root that lists it; a file under no
current root is ``skipped``. A linked file is never queued: its
pending items are removed and the answer is ``linked``. An unlinked
file the scan's fallback would link is ``skipped`` (the next scan
links it); any other file is recorded when it is ambiguous. While a
job runs that a ``records`` job may not run beside, 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")
if busy is not None:
self.logger.info(
f"{os.path.basename(str(image_path))} was not added to Needs "
f"review while {busy.title} runs; the next scan reads it again"
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)
return self._record_review_sync(
located, _scan_roots(self._get_all_output_dirs())
)
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"
if self._fallback_links(self._extract_comfyui_metadata(str(located))):
return "skipped"
return self._record_review_sync(located, roots)
@staticmethod
def _log_review_items(job, review):
"""The scan log's Needs review summary lines (none when nothing moved)."""
if review["recorded"]:
prefix = SCAN_LOG_PREFIXES["review"]
if review["added"]:
job.log_line(
f"{SCAN_LOG_PREFIXES['review']} "
f"{_plural(review['recorded'], 'image')} added to Needs review"
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["errors"]:
job.log_line(
py/api/system.py
+6 -7
View File
@@ -41,12 +41,11 @@ _HEALTH_STATUS_LINES = {
_StoredReport = namedtuple("_StoredReport", "generation report")
async def _settled(*reads):
"""Every read's result once all of them ended; the first error raises.
async def _gather_settled(*reads):
"""Await every read, then raise the first failure.
``asyncio.gather`` answers at the first error while the other executor
reads still run, so a route could answer (and its caller replace or
delete the database) with a sibling read still inside SQLite.
Unlike a plain gather, a failure never answers while the other executor
reads still run against the database.
"""
results = await asyncio.gather(*reads, return_exceptions=True)
for result in results:
@@ -169,7 +168,7 @@ class SystemRoutesMixin:
from ..config import GalleryConfig
try:
(output_dirs, auto_detected), counts = await _settled(
(output_dirs, auto_detected), counts = await _gather_settled(
self._run_in_executor(self._output_dirs_with_origin),
self._run_in_executor(self.db.library_counts),
)
@@ -203,7 +202,7 @@ class SystemRoutesMixin:
async def get_system_statistics(self, request):
try:
by_category, totals, counts = await _settled(
by_category, totals, counts = await _gather_settled(
self._run_in_executor(self.db.usage_by_category),
self._run_in_executor(self.db.get_statistics),
self._run_in_executor(self.db.library_counts),