feat(tools): reviewed repair of the foreign dataset links (#247)

tools/repair_foreign_links.py judges every link row of a file that only
foreign runs reported with the shared ladder (scan policy, recorded saver).
The dry run judges a copy, writes a JSON plan and a Markdown list, and
rehearses the apply on that copy to audit the prompts that would lose
every link (deleting them is off unless chosen). --apply refuses while
ComfyUI runs, takes a backup, re-judges every row inside one BEGIN
IMMEDIATE transaction, creates the targets with the scan's marks, relinks
and drops pair links, keeps UNIQUE(prompt_id, file_path), and rolls back
with exit 2 on any drift.
This commit is contained in:
Vito Sansevero
2026-10-28 07:01:30 -07:00
parent b107de70f8
commit 6b568de09b
2 changed files with 1181 additions and 0 deletions
+531
View File
@@ -0,0 +1,531 @@
"""tools/repair_foreign_links.py: the reviewed repair of the #247 dataset links.
The fixture is a library whose foreign (no PromptManager node) dataset run
wrote shared-prefix files that the old scan linked to the wrong branch text.
Recorded history names each file's saver, so the ladder (``recorded-run``)
names the right branch.
"""
import contextlib
import hashlib
import io
import json
import os
import shutil
import sqlite3
import tempfile
import unittest
from pathlib import Path
from PIL import Image
from PIL.PngImagePlugin import PngInfo
try:
from tests.open_handles import assert_closed
except ImportError: # discovered with tests/ as the top-level directory
from open_handles import assert_closed
from database.models import PromptModel
from database.operations import PromptDatabase
from tests import saver_anchor_graphs as shapes
from tests.test_run_attribution import RecordedRuns
from tools import repair_foreign_links as tool
from utils.file_prompt import Decision
from utils.hashing import generate_prompt_hash
TEXTS = shapes.BRANCH_TEXTS
NAME = "IMG_GEN_{:05}_.png"
def png(path, graph):
info = PngInfo()
info.add_text("prompt", json.dumps(graph))
Image.new("RGB", (4, 4)).save(path, pnginfo=info)
def sha(text):
return hashlib.sha256(text.encode("utf-8")).hexdigest()
def file_sha(path):
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
def dump(path):
"""The database's logical content (independent of WAL checkpoints)."""
conn = sqlite3.connect(path)
try:
return list(conn.iterdump())
finally:
conn.close()
def not_running(_comfyui_dir, _port):
return []
class RepairFixture(RecordedRuns, unittest.TestCase):
"""Rows: A relink to a new prompt, B relink to a stored prompt (with a
pair link), C keep, D file gone, E two savers (skip), F the UNIQUE case
(a wrong row beside a correct row), H relink from a rated prompt, G a
PromptManager run (not a candidate)."""
def setUp(self):
self.tmp = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, self.tmp, True)
self.addCleanup(assert_closed, self, self.tmp)
self.addCleanup(PromptModel.reset_schema_cache)
self.output = os.path.join(self.tmp, "output")
os.makedirs(os.path.join(self.output, "dataset"))
self.db_path = os.path.join(self.tmp, "prompts.db")
self.plan_path = os.path.join(self.tmp, "plan.json")
self.backups = os.path.join(self.tmp, "backups")
os.makedirs(self.backups)
self.files = {}
for key, number in zip("ABCDEFHG", range(1, 9)):
path = os.path.join(self.output, "dataset", NAME.format(number))
png(path, shapes.shared_prefix())
self.files[key] = path
db = PromptDatabase(self.db_path)
try:
self.build(db)
finally:
db.close_all()
os.remove(self.files["D"])
def build(self, db):
save = lambda text, **kw: db.save_prompt( # noqa: E731
text=text, prompt_hash=generate_prompt_hash(text), **kw
)
self.p0 = save(TEXTS[0], category="scanned", tags=["auto-scanned"])
self.p2 = save(TEXTS[2], category="scanned", tags=["auto-scanned"])
self.orphan = save(
"an old guess",
category="scanned",
tags=["auto-scanned"],
notes="Auto-scanned from IMG_GEN_00001_.png",
)
self.rated = save("a rated guess", rating=5)
link = lambda prompt, key: db.link_image_to_prompt( # noqa: E731
prompt, self.files[key], metadata={}
)
self.rows = {
"A": link(self.orphan, "A"),
"B": link(self.p0, "B"),
"C": link(self.p0, "C"),
"D": link(self.p0, "D"),
"E": link(self.p0, "E"),
"F": link(self.p0, "F"),
"F_ok": link(self.p2, "F"),
"H": link(self.rated, "H"),
"G": link(self.p0, "G"),
}
conn = db.model.get_connection()
with conn:
pair = conn.execute(
"INSERT INTO prompt_pairs(positive_prompt_id) VALUES(?)", (self.p0,)
).lastrowid
conn.execute(
"INSERT INTO image_prompt_pairs(image_id, pair_id, provenance) "
"VALUES(?,?,'{}')",
(self.rows["B"], pair),
)
name = lambda key: ("dataset", os.path.basename(self.files[key])) # noqa
# Savers of the shared-prefix graph: "13", "23", "33", "43" (branches 1-4)
self.record_run(
db,
self.output,
[name(k) for k in "CDEAEBFH"],
[],
nodes=["13", "13", "13", "23", "23", "33", "33", "43"],
)
self.record_run(db, self.output, [name("G")], [TEXTS[0]], item=1, nodes=["23"])
def dry_run(self, *args):
out = io.StringIO()
code = tool.main(
[
self.db_path,
"--output-dir",
self.output,
"--plan",
self.plan_path,
*args,
],
out=out,
)
return code, out.getvalue()
def apply(self, *args, probe=not_running):
out = io.StringIO()
code = tool.main(
[
self.db_path,
"--apply",
self.plan_path,
"--comfyui-dir",
os.path.join(self.tmp, "comfy"),
"--backup-dir",
self.backups,
*args,
],
out=out,
probe=probe,
)
return code, out.getvalue()
def plan(self):
with open(self.plan_path, encoding="utf-8") as stream:
return json.load(stream)
def write_plan(self, plan):
with open(self.plan_path, "w", encoding="utf-8") as stream:
json.dump(plan, stream)
def query(self, sql, args=()):
conn = sqlite3.connect(self.db_path)
try:
return conn.execute(sql, args).fetchall()
finally:
conn.close()
def links(self):
return dict(self.query("SELECT id, prompt_id FROM generated_images"))
class DryRunTests(RepairFixture):
def test_plan_names_relink_keep_and_skip_rows(self):
code, out = self.dry_run()
self.assertEqual(code, 0, out)
items = {item["row_id"]: item for item in self.plan()["rows"]}
rows = self.rows
self.assertNotIn(rows["G"], items, "a PromptManager run is not foreign")
actions = {key: items[rows[key]]["action"] for key in rows if key != "G"}
self.assertEqual(
actions,
{
"A": "relink",
"B": "relink",
"C": "keep",
"D": "skip",
"E": "skip",
"F": "relink",
"F_ok": "keep",
"H": "relink",
},
)
a = items[rows["A"]]
self.assertEqual(a["current_prompt_id"], self.orphan)
self.assertEqual(a["current_text_sha256"], sha("an old guess"))
self.assertEqual(a["target_text"], TEXTS[1])
self.assertEqual(a["target_hash"], generate_prompt_hash(TEXTS[1]))
self.assertEqual(a["target_prompt_id"], "new")
self.assertEqual(a["file"], self.files["A"])
self.assertEqual(items[rows["B"]]["target_prompt_id"], self.p2)
self.assertEqual(items[rows["H"]]["target_prompt_id"], "new")
self.assertEqual(items[rows["D"]]["reason"], "file gone")
self.assertEqual(
items[rows["E"]]["reason"], "not decided by the recorded saver"
)
self.assertIn("relink: 4", out)
self.assertIn("keep: 2", out)
self.assertIn("skip: 2", out)
def test_orphan_audit_lists_the_prompts_losing_every_link(self):
self.dry_run()
plan = self.plan()
orphans = {o["id"]: o for o in plan["orphans"]}
self.assertEqual(set(orphans), {self.orphan, self.rated})
self.assertEqual(orphans[self.orphan]["blockers"], [])
self.assertTrue(orphans[self.orphan]["safe"])
self.assertEqual(orphans[self.rated]["blockers"], ["usage or rating"])
self.assertFalse(orphans[self.rated]["safe"])
self.assertFalse(plan["delete_orphans"], "off by default")
self.dry_run("--delete-orphans")
self.assertTrue(self.plan()["delete_orphans"])
def test_dry_run_writes_nothing_and_writes_the_markdown_list(self):
before = file_sha(self.db_path)
listing = sorted(os.listdir(self.tmp))
code, out = self.dry_run()
self.assertEqual(code, 0)
self.assertEqual(file_sha(self.db_path), before)
self.assertIn("Dry run: nothing was changed.", out)
# A read-only reader of a WAL library may leave its empty side files
sidecars = {"prompts.db-shm", "prompts.db-wal"}
self.assertEqual(
sorted(set(os.listdir(self.tmp)) - sidecars),
sorted(set(listing + ["plan.json", "plan.md"]) - sidecars),
)
wal = Path(self.tmp, "prompts.db-wal")
self.assertEqual(wal.stat().st_size if wal.exists() else 0, 0)
markdown = Path(self.tmp, "plan.md").read_text(encoding="utf-8")
self.assertIn(f"| {self.rows['A']} |", markdown)
self.assertIn("new", markdown)
self.assertIn("would lose every link", markdown)
class ApplyTests(RepairFixture):
def setUp(self):
super().setUp()
code, out = self.dry_run()
self.assertEqual(code, 0, out)
def test_apply_relinks_creates_targets_and_drops_pair_links(self):
code, out = self.apply()
self.assertEqual(code, 0, out)
links = self.links()
new = dict(self.query("SELECT hash, id FROM prompts"))
target_a = new[generate_prompt_hash(TEXTS[1])]
target_h = new[generate_prompt_hash(TEXTS[3])]
self.assertEqual(links[self.rows["A"]], target_a)
self.assertEqual(links[self.rows["B"]], self.p2)
self.assertEqual(links[self.rows["H"]], target_h)
self.assertEqual(links[self.rows["C"]], self.p0, "keep")
self.assertEqual(links[self.rows["E"]], self.p0, "skip")
self.assertEqual(links[self.rows["G"]], self.p0, "not foreign")
self.assertEqual(
self.query(
"SELECT COUNT(*) FROM image_prompt_pairs WHERE image_id=?",
(self.rows["B"],),
),
[(0,)],
)
category, notes = self.query(
"SELECT category, notes FROM prompts WHERE id=?", (target_a,)
)[0]
self.assertEqual(category, "scanned")
self.assertEqual(
notes, f"Auto-scanned from {os.path.basename(self.files['A'])}"
)
tags = self.query(
"SELECT t.name FROM prompt_tags pt JOIN tags t ON t.id=pt.tag_id "
"WHERE pt.prompt_id=?",
(target_a,),
)
self.assertEqual(tags, [("auto-scanned",)])
self.assertIn("integrity_check", out)
self.assertIn("foreign_key_check", out)
self.assertRegex(out, r"integrity_check\s+ok\s+ok")
# Orphans are kept unless the plan chose to delete them
self.assertTrue(self.query("SELECT 1 FROM prompts WHERE id=?", (self.orphan,)))
def test_unique_case_deletes_the_wrong_row(self):
code, out = self.apply()
self.assertEqual(code, 0, out)
links = self.links()
self.assertNotIn(self.rows["F"], links)
self.assertEqual(links[self.rows["F_ok"]], self.p2)
rows = self.query(
"SELECT prompt_id FROM generated_images WHERE image_path=?",
(self.files["F"],),
)
self.assertEqual(rows, [(self.p2,)])
def test_backup_is_taken_before_writing(self):
before = self.links()
code, out = self.apply()
self.assertEqual(code, 0, out)
names = os.listdir(self.backups)
self.assertEqual(len(names), 1)
self.assertRegex(names[0], r"^prompts\.db\.pre-247-repair_\d{8}_\d{6}$")
backup = os.path.join(self.backups, names[0])
conn = sqlite3.connect(backup)
try:
saved = dict(conn.execute("SELECT id, prompt_id FROM generated_images"))
finally:
conn.close()
self.assertEqual(saved, before)
self.assertIn(f"Backup: {backup}", out)
def test_backup_beside_the_database_by_default(self):
out = io.StringIO()
code = tool.main(
[
self.db_path,
"--apply",
self.plan_path,
"--comfyui-dir",
os.path.join(self.tmp, "comfy"),
],
out=out,
probe=not_running,
)
self.assertEqual(code, 0, out.getvalue())
self.assertTrue(
any(
n.startswith("prompts.db.pre-247-repair_") for n in os.listdir(self.tmp)
)
)
def test_refuses_while_comfyui_runs(self):
before = file_sha(self.db_path)
code, out = self.apply(probe=lambda _d, _p: ["process 42 (main.py)"])
self.assertEqual(code, 2)
self.assertIn("REFUSED", out)
self.assertIn("process 42", out)
self.assertEqual(os.listdir(self.backups), [], "no backup either")
self.assertEqual(file_sha(self.db_path), before)
def test_rehearsal_skips_the_running_check_on_a_copy_only(self):
code, out = self.apply("--rehearsal", probe=lambda _d, _p: ["port 8188"])
self.assertEqual(code, 0, out)
self.assertIn("rehearsal", out)
# The live library of that ComfyUI folder: --rehearsal is refused
live_dir = os.path.join(self.tmp, "comfy", "user", "__promptmanager")
os.makedirs(live_dir)
live = os.path.join(live_dir, "prompts.db")
shutil.copy(self.db_path, live)
out = io.StringIO()
code = tool.main(
[
live,
"--apply",
self.plan_path,
"--comfyui-dir",
os.path.join(self.tmp, "comfy"),
"--rehearsal",
],
out=out,
probe=lambda _d, _p: [],
)
self.assertEqual(code, 2)
self.assertIn("--rehearsal", out.getvalue())
def test_drift_refuses_and_changes_nothing(self):
cases = {
"row moved": lambda: self.query_write(
"UPDATE generated_images SET prompt_id=? WHERE id=?",
(self.p2, self.rows["H"]),
),
"text changed": lambda: self.query_write(
"UPDATE prompts SET text='an older guess' WHERE id=?", (self.orphan,)
),
"file changed": lambda: png(self.files["B"], shapes.branch_bench()),
"target stored since": lambda: self.query_write(
"INSERT INTO prompts(text, hash) VALUES(?,?)",
(TEXTS[1], generate_prompt_hash(TEXTS[1])),
),
}
pristine = os.path.join(self.tmp, "pristine.db")
shutil.copy(self.db_path, pristine)
image = Path(self.files["B"]).read_bytes()
for name, drift in cases.items():
with self.subTest(name):
shutil.copy(pristine, self.db_path)
Path(self.files["B"]).write_bytes(image)
drift()
before = dump(self.db_path)
code, out = self.apply()
self.assertEqual(code, 2, out)
self.assertIn("REFUSED (rolled back, nothing changed)", out)
self.assertEqual(dump(self.db_path), before)
def query_write(self, sql, args):
conn = sqlite3.connect(self.db_path)
try:
with conn:
conn.execute(sql, args)
finally:
conn.close()
def test_late_failure_rolls_back_every_write(self):
plan = self.plan()
plan["delete_orphans"] = True
self.write_plan(plan)
# The safe orphan gained a user tag after the dry run
self.query_write("INSERT INTO tags(name) VALUES('mine')", ())
self.query_write(
"INSERT INTO prompt_tags(prompt_id, tag_id) "
"SELECT ?, id FROM tags WHERE name='mine'",
(self.orphan,),
)
before = self.links()
prompts = self.query("SELECT COUNT(*) FROM prompts")
code, out = self.apply()
self.assertEqual(code, 2, out)
self.assertIn(f"prompt {self.orphan}", out)
self.assertIn("user tags", out)
self.assertEqual(self.links(), before, "the relinks were rolled back")
self.assertEqual(self.query("SELECT COUNT(*) FROM prompts"), prompts)
def test_orphans_are_deleted_only_when_chosen_and_unreferenced(self):
plan = self.plan()
plan["delete_orphans"] = True
self.write_plan(plan)
code, out = self.apply()
self.assertEqual(code, 0, out)
self.assertFalse(self.query("SELECT 1 FROM prompts WHERE id=?", (self.orphan,)))
self.assertTrue(
self.query("SELECT 1 FROM prompts WHERE id=?", (self.rated,)),
"a prompt with a rating is never deleted",
)
self.assertIn("orphans_deleted", out)
def test_an_applied_plan_refuses_a_second_time(self):
self.assertEqual(self.apply()[0], 0)
before = dump(self.db_path)
code, out = self.apply()
self.assertEqual(code, 2)
self.assertIn("REFUSED", out)
self.assertEqual(dump(self.db_path), before)
class ActionTests(unittest.TestCase):
"""Only the recorded saver relinks; a certain step naming the linked
prompt keeps it; the legacy floor and the rest are skipped."""
def test_action_by_decision(self):
linked = generate_prompt_hash("linked")
cases = [
(Decision("positive", "other", source="recorded-run"), "relink"),
(Decision("positive", "linked", source="recorded-run"), "keep"),
(Decision("positive", "linked", source="inspector"), "keep"),
(Decision("positive", "other", source="inspector"), "skip"),
(Decision("positive", "linked", source="legacy-parser"), "skip"),
(Decision("pair", "linked", source="resolver"), "skip"),
(Decision("never"), "skip"),
]
for decision, action in cases:
with self.subTest(decision.source, text=decision.text):
self.assertEqual(tool.action_of(decision, linked), action)
class CommandLineTests(unittest.TestCase):
def test_running_probe_reports_a_listening_port(self):
import socket
with socket.socket() as server:
server.bind(("127.0.0.1", 0))
server.listen(1)
port = server.getsockname()[1]
found = tool.comfyui_running("/nonexistent/comfy", port)
self.assertTrue(any(str(port) in item for item in found))
def test_running_probe_is_quiet_when_nothing_runs(self):
import socket
with socket.socket() as probe:
probe.bind(("127.0.0.1", 0))
port = probe.getsockname()[1]
self.assertEqual(tool.comfyui_running("/nonexistent/comfy", port), [])
def test_missing_database_is_an_argument_error(self):
with self.assertRaises(SystemExit), contextlib.redirect_stderr(io.StringIO()):
tool.main(
["/nonexistent/prompts.db", "--output-dir", "/tmp"], out=io.StringIO()
)
def test_apply_needs_the_comfyui_folder(self):
with tempfile.TemporaryDirectory() as tmp:
db = os.path.join(tmp, "prompts.db")
Path(db).write_bytes(b"")
with (
self.assertRaises(SystemExit),
contextlib.redirect_stderr(io.StringIO()),
):
tool.main([db, "--apply", "plan.json"], out=io.StringIO())
if __name__ == "__main__":
unittest.main()
+650
View File
@@ -0,0 +1,650 @@
"""Repair the links the old scan gave foreign dataset files (#247), on review.
Before #247 the output scan linked each file of a foreign run (a workflow
with no PromptManager node) through the legacy parser, so every file of a
shared-prefix dataset run got the first branch's text. Recorded history
knows which save node wrote each file; the shared ladder
(``utils.file_prompt.decide_file_prompt``, scan policy) uses it
(``recorded-run``) to name the right branch. This tool judges every
``generated_images`` row of a file whose recording runs all have no
PromptManager context:
``relink`` the ladder names a positive by the recorded saver, and its text
is not the linked prompt's
``keep`` the recorded saver or the inspector names the linked prompt
``skip`` anything else (file gone, no single recorded saver, ambiguous,
the legacy floor or another step decided)
Dry run (the default) judges a copy of the database (SQLite backup from a
read-only connection) and writes a JSON plan and a Markdown list beside it.
It also rehearses the apply on that copy, to list the prompts that would
lose every link with their remaining references. Deleting those is OFF
unless the dry run is given ``--delete-orphans``; the choice is recorded in
the plan. The library is never written.
``--apply <plan.json>`` (ComfyUI stopped):
1. refuses while a ``main.py`` runs from ``--comfyui-dir`` or anything listens
on ``--port`` (``--rehearsal`` skips this, never for the library of that
ComfyUI folder);
2. saves ``<db>.pre-247-repair_<stamp>`` (or into ``--backup-dir``);
3. in ONE ``BEGIN IMMEDIATE`` transaction: re-judges every planned row and
refuses on any drift (row moved, prompt text changed, file changed, target
stored since), creates the target prompts that do not exist (category
``scanned``, tag ``auto-scanned``, note ``Auto-scanned from <first file>``),
points each row at its target and drops its pair links (they named the old
positive). A file that already has a row for the target keeps that row and
loses the wrong one (UNIQUE(prompt_id, file_path)). Orphans are deleted only
when the plan chose it and each is still unreferenced.
Any failed check rolls everything back and exits 2. Before/after table
counts, ``integrity_check`` and ``foreign_key_check`` are printed.
python3 tools/repair_foreign_links.py <db> --output-dir <ComfyUI output> \\
--plan repair_plan.json
python3 tools/repair_foreign_links.py <db> --apply repair_plan.json \\
--comfyui-dir <ComfyUI folder>
Run from this repository with its venv. The process check reads /proc
(Linux); elsewhere only the port is checked, so stop ComfyUI first.
"""
import argparse
import datetime
import hashlib
import json
import logging
import os
import socket
import sqlite3
import sys
import tempfile
from pathlib import Path
from types import SimpleNamespace
_ROOT = Path(__file__).resolve().parents[1]
if str(_ROOT) not in sys.path:
sys.path.insert(0, str(_ROOT))
from database.maintenance import ORPHANED_PROMPTS_SQL, _delete_ids # noqa: E402
from database.operations import PromptDatabase # noqa: E402
from database.pair_ingestion import ( # noqa: E402
SCAN_MARKS,
file_link_rows,
mark_created,
)
from database.run_attribution import recorded_output # noqa: E402
from py.api import PromptManagerAPI # noqa: E402
from tools.review_image_links import backup_database # noqa: E402
from utils.execution_outputs import snapshot_output_roots # noqa: E402
from utils.file_prompt import POSITIVE, Policy, decide_file_prompt # noqa: E402
from utils.hashing import generate_prompt_hash # noqa: E402
PLAN_VERSION = 1
COMFY_PORT = 8188
LIVE_LIBRARY = ("user", "__promptmanager", "prompts.db")
EXCERPT = 80
COUNTED = (
"prompts",
"generated_images",
"prompt_tags",
"image_prompt_pairs",
"workflow_documents",
"prompt_pairs",
"prompt_library_state",
"execution_run_contexts",
"tags",
)
# The ladder steps whose answer is certain enough to keep a link.
KEEP_SOURCES = ("recorded-run", "inspector")
# What must be unchanged since the dry run for a row to be written.
DRIFT_KEYS = (
"action",
"file",
"current_prompt_id",
"current_text_sha256",
"file_size",
"file_mtime_ns",
"target_hash",
"target_prompt_id",
)
# Locations reported only by runs without any PromptManager context.
_FOREIGN_LOCATIONS = (
"SELECT o.subfolder, o.filename FROM execution_run_outputs o "
"WHERE o.folder_type = 'output' AND o.root_key = ? AND o.path_style = ? "
"GROUP BY o.subfolder, o.filename "
"HAVING SUM(EXISTS (SELECT 1 FROM execution_run_contexts c "
"WHERE c.run_id = o.run_id)) = 0 "
"ORDER BY o.subfolder, o.filename"
)
class Refused(Exception):
"""A check failed: nothing is written."""
def sha256(text):
return hashlib.sha256(text.encode("utf-8")).hexdigest()
def comfyui_running(comfyui_dir, port):
"""What shows ComfyUI is running: ``main.py`` processes of that folder
(Linux /proc) and a listener on ``port``."""
found = []
folder = os.path.realpath(comfyui_dir)
proc = Path("/proc")
for pid in filter(str.isdigit, os.listdir(proc) if proc.is_dir() else ()):
try:
parts = (proc / pid / "cmdline").read_bytes().split(b"\0")
cwd = os.readlink(proc / pid / "cwd")
except OSError:
continue
if any(part.endswith(b"main.py") for part in parts) and (
cwd == folder or any(folder.encode() in part for part in parts)
):
found.append(f"process {pid} (main.py in {cwd})")
with socket.socket() as probe:
probe.settimeout(1)
if probe.connect_ex(("127.0.0.1", port)) == 0:
found.append(f"something listens on port {port}")
return found
def metadata_reader():
"""The scan's metadata extraction (PNG text chunks, ffprobe for videos)."""
api = SimpleNamespace(logger=logging.getLogger("prompt_manager.repair_247"))
return lambda path: PromptManagerAPI._extract_comfyui_metadata(api, path) or {}
def foreign_rows(db, output_dir):
"""The link rows of every file only foreign runs reported, by id."""
style = "windows" if os.name == "nt" else "posix"
root_key = snapshot_output_roots(style, output_dir, output_dir).output_key
conn = db.model.get_connection()
rows = {}
for subfolder, filename in conn.execute(_FOREIGN_LOCATIONS, (root_key, style)):
path = os.path.join(output_dir, subfolder, filename)
rows.update((row["id"], row) for row in file_link_rows(conn, path))
return [rows[key] for key in sorted(rows)]
class Judge:
"""One file's decision per path, for all of its rows."""
def __init__(self, db, output_dir, read=None):
self.db = db
self.output_dir = output_dir
self.read = read or metadata_reader()
self.decisions = {}
def decide(self, path, size):
if path not in self.decisions:
self.decisions[path] = decide_file_prompt(
self.read(path),
file_name=os.path.basename(path),
file_size=size,
in_output=True,
lookup=lambda: recorded_output(self.db, path, self.output_dir),
policy=Policy.SCAN,
)
return self.decisions[path]
def row(self, row):
"""The plan entry of one ``generated_images`` row."""
conn = self.db.model.get_connection()
path = row["image_path"]
text, prompt_hash = conn.execute(
"SELECT text, hash FROM prompts WHERE id = ?", (row["prompt_id"],)
).fetchone()
item = {
"row_id": row["id"],
"file": path,
"current_prompt_id": row["prompt_id"],
"current_text_sha256": sha256(text),
}
try:
stat = os.stat(path)
except OSError:
return {**item, "action": "skip", "reason": "file gone"}
item.update(file_size=stat.st_size, file_mtime_ns=stat.st_mtime_ns)
decision = self.decide(path, stat.st_size)
action = action_of(decision, prompt_hash)
if action == "skip":
return {
**item,
"action": "skip",
"reason": "not decided by the recorded saver",
"decision": f"{decision.kind}/{decision.source}",
}
target_hash = generate_prompt_hash(decision.text)
stored = conn.execute(
"SELECT id FROM prompts WHERE hash = ?", (target_hash,)
).fetchone()
item.update(
target_text=decision.text,
target_hash=target_hash,
target_prompt_id=stored[0] if stored else "new",
action=action,
source=decision.source,
)
return item
def action_of(decision, linked_hash):
"""``relink``, ``keep`` or ``skip`` for a row linked to ``linked_hash``.
Only the recorded saver (``recorded-run``) names a new target. A certain
step (the recorded saver or the inspector) naming the linked prompt keeps
it; the legacy floor and every other answer are skipped.
"""
if decision.kind != POSITIVE or decision.source not in KEEP_SOURCES:
return "skip"
if generate_prompt_hash(decision.text) == linked_hash:
return "keep"
return "relink" if decision.source == "recorded-run" else "skip"
def build_plan(db, output_dir, read=None):
judge = Judge(db, output_dir, read)
return {
"version": PLAN_VERSION,
"created": datetime.datetime.now().isoformat(timespec="seconds"),
"output_dir": output_dir,
"rows": [judge.row(row) for row in foreign_rows(db, output_dir)],
"orphans": [],
"delete_orphans": False,
}
def blockers(conn, prompt_id, unused):
"""Why ``prompt_id`` must not be deleted (empty: nothing references it)."""
found = []
values = {"id": prompt_id, "tag": SCAN_MARKS.tags[0]}
checks = {
"linked images": "SELECT COUNT(*) FROM generated_images WHERE prompt_id=:id",
"saved pairs": "SELECT COUNT(*) FROM prompt_pairs "
"WHERE positive_prompt_id=:id OR negative_prompt_id=:id",
"recorded runs": "SELECT COUNT(*) FROM execution_run_contexts "
"WHERE positive_prompt_id=:id OR negative_prompt_id=:id",
"library state": "SELECT COUNT(*) FROM prompt_library_state "
"WHERE prompt_id=:id",
"user tags": "SELECT COUNT(*) FROM prompt_tags pt JOIN tags t "
"ON t.id=pt.tag_id WHERE pt.prompt_id=:id AND t.name != :tag",
}
for name, sql in checks.items():
count = conn.execute(sql, values).fetchone()[0]
if count:
found.append(f"{name}: {count}")
run_count, last_used, rating, category, notes = conn.execute(
"SELECT run_count, last_used_at, rating, category, notes FROM prompts "
"WHERE id=?",
(prompt_id,),
).fetchone()
if run_count or last_used or rating:
found.append("usage or rating")
if category not in (None, SCAN_MARKS.category):
found.append(f"category {category}")
if notes and not notes.startswith("Auto-scanned from "):
found.append("notes")
if prompt_id not in unused:
found.append("not unused by the app's rule")
return found
def _rejudge(conn, judge, plan):
for item in plan["rows"]:
if item["action"] == "skip":
continue
row = conn.execute(
"SELECT * FROM generated_images WHERE id = ?", (item["row_id"],)
).fetchone()
if row is None:
raise Refused(f"row {item['row_id']}: gone")
fresh = judge.row(row)
for key in DRIFT_KEYS:
if fresh.get(key) != item.get(key):
raise Refused(
f"row {item['row_id']}: {key} changed "
f"({item.get(key)!r} -> {fresh.get(key)!r})"
)
def _targets(conn, db, relinks):
"""``{target hash: prompt id}``, creating each "new" target once."""
targets = {}
for item in relinks:
digest = item["target_hash"]
if digest in targets:
continue
if item["target_prompt_id"] != "new":
targets[digest] = item["target_prompt_id"]
continue
prompt_id, created = db._pair_component_outcome(conn, item["target_text"])
if not created:
raise Refused(f"row {item['row_id']}: target prompt already stored")
mark_created(conn, db, prompt_id, SCAN_MARKS, os.path.basename(item["file"]))
targets[digest] = prompt_id
return targets
def _relink(conn, item, target):
"""Point the row at ``target``; True when the wrong row was deleted instead."""
row_id = item["row_id"]
correct = [
row
for row in file_link_rows(conn, item["file"])
if row["prompt_id"] == target and row["id"] != row_id
]
if correct: # UNIQUE(prompt_id, file_path): the correct row stays alone
if _delete_ids(conn, "generated_images", [row_id]) != 1:
raise Refused(f"row {row_id}: delete touched no row")
return True
# A pair link records the old positive: it goes with the old link
conn.execute("DELETE FROM image_prompt_pairs WHERE image_id = ?", (row_id,))
cursor = conn.execute(
"UPDATE generated_images SET prompt_id = ? WHERE id = ? AND prompt_id = ?",
(target, row_id, item["current_prompt_id"]),
)
if cursor.rowcount != 1:
raise Refused(f"row {row_id}: relink touched {cursor.rowcount} rows")
return False
def _audit(conn, relinks):
"""The prompts the relinks leave without any link, with their blockers."""
sources = {}
for item in relinks:
sources.setdefault(item["current_prompt_id"], []).append(item)
unused = {row[0] for row in conn.execute(ORPHANED_PROMPTS_SQL)}
audit = []
for prompt_id in sorted(sources):
if conn.execute(
"SELECT 1 FROM generated_images WHERE prompt_id = ?", (prompt_id,)
).fetchone():
continue
(text,) = conn.execute(
"SELECT text FROM prompts WHERE id = ?", (prompt_id,)
).fetchone()
found = blockers(conn, prompt_id, unused)
audit.append(
{
"id": prompt_id,
"text_excerpt": " ".join(text.split())[:EXCERPT],
"text_sha256": sha256(text),
"images_before": len(sources[prompt_id]),
"blockers": found,
"safe": not found,
}
)
return audit
def _delete_orphans(conn, plan):
chosen = [orphan for orphan in plan["orphans"] if orphan["safe"]]
unused = {row[0] for row in conn.execute(ORPHANED_PROMPTS_SQL)}
for orphan in chosen:
row = conn.execute(
"SELECT text FROM prompts WHERE id = ?", (orphan["id"],)
).fetchone()
if row is None or sha256(row[0]) != orphan["text_sha256"]:
raise Refused(f"prompt {orphan['id']}: gone or text changed")
found = blockers(conn, orphan["id"], unused)
if found:
raise Refused(f"prompt {orphan['id']}: {', '.join(found)}")
ids = [orphan["id"] for orphan in chosen]
if _delete_ids(conn, "prompts", ids) != len(ids):
raise Refused("an orphan prompt was not deleted")
return len(ids)
def apply_plan(db, plan, read=None):
"""Write ``plan`` in one transaction; returns ``(result, orphan audit)``.
Raises ``Refused`` (and rolls back) on any drift or failed check.
"""
judge = Judge(db, plan["output_dir"], read)
relinks = [item for item in plan["rows"] if item["action"] == "relink"]
with db._pair_transaction() as conn:
_rejudge(conn, judge, plan)
targets = _targets(conn, db, relinks)
dropped = sum(
_relink(conn, item, targets[item["target_hash"]]) for item in relinks
)
audit = _audit(conn, relinks)
deleted = _delete_orphans(conn, plan) if plan.get("delete_orphans") else 0
created = sum(1 for item in relinks if item["target_prompt_id"] == "new")
result = {
"relinked": len(relinks) - dropped,
"duplicates_dropped": dropped,
"prompts_created": len(
{i["target_hash"] for i in relinks if i["target_prompt_id"] == "new"}
),
"rows_to_new_prompts": created,
"kept": sum(1 for item in plan["rows"] if item["action"] == "keep"),
"orphans_deleted": deleted,
}
return result, audit
def counts(path):
"""Table counts plus the integrity and foreign-key checks (read-only)."""
uri = Path(os.path.abspath(path)).as_uri() + "?mode=ro"
conn = sqlite3.connect(uri, uri=True)
try:
found = {
table: conn.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0]
for table in COUNTED # nosec B608 - fixed table names
}
found["integrity_check"] = conn.execute("PRAGMA integrity_check").fetchone()[0]
found["foreign_key_check"] = len(
conn.execute("PRAGMA foreign_key_check").fetchall()
)
return found
finally:
conn.close()
def _print_counts(before, after, out):
out.write(f"{'':24}{'before':>12}{'after':>12}\n")
for key in before:
out.write(f"{key:24}{before[key]!s:>12}{after[key]!s:>12}\n")
def _summary(plan):
actions = [item["action"] for item in plan["rows"]]
return " ".join(
f"{name}: {actions.count(name)}" for name in ("relink", "keep", "skip")
)
def _cell(text):
return " ".join(str(text).split())[:EXCERPT].replace("|", "\\|")
def markdown(plan, prompt_texts):
"""The maintainer's list of the plan."""
root = plan["output_dir"]
rel = lambda path: os.path.relpath(path, root) # noqa: E731
rows = plan["rows"]
lines = [
"# #247 repair plan (dry run)",
"",
f"Created {plan['created']}; output folder `{root}`. {_summary(plan)}.",
"",
f"## Relink ({sum(i['action'] == 'relink' for i in rows)})",
"",
"| row | file | linked prompt | linked text | target | target text |",
"|---:|---|---:|---|---|---|",
]
for item in (i for i in rows if i["action"] == "relink"):
lines.append(
f"| {item['row_id']} | {_cell(rel(item['file']))} | "
f"{item['current_prompt_id']} | "
f"{_cell(prompt_texts.get(item['current_prompt_id'], ''))} | "
f"{item['target_prompt_id']} | {_cell(item['target_text'])} |"
)
lines += ["", f"## Keep ({sum(i['action'] == 'keep' for i in rows)})", ""]
lines += ["| row | file | prompt |", "|---:|---|---:|"]
for item in (i for i in rows if i["action"] == "keep"):
lines.append(
f"| {item['row_id']} | {_cell(rel(item['file']))} | "
f"{item['current_prompt_id']} |"
)
lines += ["", f"## Skip ({sum(i['action'] == 'skip' for i in rows)})", ""]
lines += ["| row | file | prompt | reason |", "|---:|---|---:|---|"]
for item in (i for i in rows if i["action"] == "skip"):
reason = item["reason"] + (
f" ({item['decision']})" if item.get("decision") else ""
)
lines.append(
f"| {item['row_id']} | {_cell(rel(item['file']))} | "
f"{item['current_prompt_id']} | {reason} |"
)
choice = "ON (--delete-orphans)" if plan["delete_orphans"] else "OFF"
lines += [
"",
f"## Prompts that would lose every link ({len(plan['orphans'])})",
"",
f"Deleting the safe ones: {choice}.",
"",
"| id | text | images before | blockers | safe to delete |",
"|---:|---|---:|---|---|",
]
for orphan in plan["orphans"]:
lines.append(
f"| {orphan['id']} | {_cell(orphan['text_excerpt'])} | "
f"{orphan['images_before']} | {', '.join(orphan['blockers']) or 'none'} | "
f"{'yes' if orphan['safe'] else 'no'} |"
)
simulated = plan.get("simulated")
if simulated:
lines += ["", "## Rehearsed on the dry run's copy", ""]
lines += ["| table | before | after |", "|---|---:|---:|"]
for key, value in simulated["before"].items():
lines.append(f"| {key} | {value} | {simulated['after'][key]} |")
return "\n".join(lines) + "\n"
def dry_run(args, out):
with tempfile.TemporaryDirectory() as scratch:
copy = os.path.join(scratch, "repair.db")
backup_database(args.database, copy)
db = PromptDatabase(copy)
try:
plan = build_plan(db, os.path.abspath(args.output_dir))
texts = dict(
db.model.get_connection().execute(
"SELECT id, text FROM prompts WHERE id IN (SELECT prompt_id "
"FROM generated_images)"
)
)
before = counts(copy)
result, audit = apply_plan(db, plan)
finally:
db.close_all()
after = counts(copy)
plan.update(
orphans=audit,
delete_orphans=bool(args.delete_orphans),
simulated={"before": before, "after": after, "result": result},
)
plan_path = Path(args.plan)
plan_path.write_text(
json.dumps(plan, indent=1, ensure_ascii=False) + "\n", encoding="utf-8"
)
list_path = plan_path.with_suffix(".md")
list_path.write_text(markdown(plan, texts), encoding="utf-8")
out.write(f"{_summary(plan)}\n")
out.write(
f"Prompts that would lose every link: {len(audit)} "
f"({sum(o['safe'] for o in audit)} safe to delete; deleting "
f"{'ON' if plan['delete_orphans'] else 'OFF'})\n"
)
out.write(f"Plan: {plan_path}\nList: {list_path}\n")
out.write("Dry run: nothing was changed.\n")
return 0
def _is_live(database, comfyui_dir):
live = os.path.join(comfyui_dir, *LIVE_LIBRARY)
return os.path.realpath(database) == os.path.realpath(live)
def apply(args, out, probe):
with open(args.apply, encoding="utf-8") as stream:
plan = json.load(stream)
if plan.get("version") != PLAN_VERSION:
out.write("REFUSED: not a plan of this tool\n")
return 2
if args.rehearsal and _is_live(args.database, args.comfyui_dir):
out.write(
"REFUSED: --rehearsal is for a copy; this is the library of "
f"{args.comfyui_dir}\n"
)
return 2
running = probe(args.comfyui_dir, args.port)
if running and not args.rehearsal:
out.write("REFUSED: ComfyUI is running: " + "; ".join(running) + "\n")
return 2
if args.rehearsal:
out.write(f"(rehearsal on a copy; running check: {running or 'nothing'})\n")
stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
folder = args.backup_dir or os.path.dirname(os.path.abspath(args.database))
backup = os.path.join(
folder, f"{os.path.basename(args.database)}.pre-247-repair_{stamp}"
)
backup_database(args.database, backup)
out.write(f"Backup: {backup}\n")
before = counts(args.database)
db = PromptDatabase(args.database)
try:
result, _audit = apply_plan(db, plan)
except (Refused, sqlite3.Error) as error:
out.write(f"REFUSED (rolled back, nothing changed): {error}\n")
return 2
finally:
db.close_all()
after = counts(args.database)
_print_counts(before, after, out)
out.write(f"Applied: {json.dumps(result, sort_keys=True)}\n")
if after["integrity_check"] != "ok" or after["foreign_key_check"]:
out.write(f"WARNING: integrity or foreign-key check failed; restore {backup}\n")
return 1
return 0
def main(argv=None, out=sys.stdout, probe=comfyui_running):
parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0])
parser.add_argument("database", help="the library database (prompts.db)")
parser.add_argument("--output-dir", help="ComfyUI's output folder (dry run)")
parser.add_argument(
"--plan", default="repair_foreign_links_plan.json", help="plan to write"
)
parser.add_argument(
"--delete-orphans",
action="store_true",
help="record in the plan that the safe orphan prompts are deleted",
)
parser.add_argument("--apply", metavar="PLAN", help="write this reviewed plan")
parser.add_argument("--comfyui-dir", help="the ComfyUI folder (apply)")
parser.add_argument("--backup-dir", help="where the backup goes (apply)")
parser.add_argument("--port", type=int, default=COMFY_PORT)
parser.add_argument(
"--rehearsal",
action="store_true",
help="apply to a copy while ComfyUI runs (never its own library)",
)
args = parser.parse_args(argv)
if not os.path.isfile(args.database):
parser.error(f"no database at {args.database}")
if args.apply:
if not args.comfyui_dir:
parser.error("--apply needs --comfyui-dir (the running check)")
return apply(args, out, probe)
if not args.output_dir:
parser.error("the dry run needs --output-dir")
return dry_run(args, out)
if __name__ == "__main__":
sys.exit(main())