Quality scan v4: calibrated PIL verdicts (dark/blurry, 300px downscale, threshold 200); persist status+flag_reason; lazy thumbnails (11.5s→0.08s page load)
This commit is contained in:
211
quality_scan.py
211
quality_scan.py
@@ -1,153 +1,118 @@
|
||||
"""photo-pipeline: quality scan flow (v3) using CleanVision.
|
||||
"""photo-pipeline: quality scan flow (v4 — calibrated thresholds).
|
||||
|
||||
Audits a folder for quality issues (blurry, dark, light, grayscale,
|
||||
low-information, odd aspect/size) and near/exact duplicates.
|
||||
Fast PIL-based verdicts written to DB:
|
||||
dark → mean luminance < 40 (downscaled 300px)
|
||||
blurry → edge variance < 200 (downscaled 300px; calibrated on real corpus:
|
||||
median 649, p25 335, so 200 flags the clearly-blurry tail)
|
||||
unreadable → corrupt image
|
||||
|
||||
Writes a per-image verdict: keep / review / delete-candidate, and moves
|
||||
files into the /mnt/data/{01_keep,02_review,03_delete} staging dirs.
|
||||
Nothing is deleted — 03_delete is a holding area for human confirmation.
|
||||
Verdicts: delete_candidate (dark/blurry/unreadable) or keep. Written to
|
||||
image_hashes.status + flag_reason. No file moves here.
|
||||
"""
|
||||
|
||||
import shutil
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
|
||||
from prefect import flow, task
|
||||
from PIL import Image, ImageFilter, ImageStat
|
||||
|
||||
STAGING = Path("/mnt/data")
|
||||
KEEP = STAGING / "01_keep"
|
||||
REVIEW = STAGING / "02_review"
|
||||
DELETE = STAGING / "03_delete"
|
||||
import photo_db as db
|
||||
|
||||
# Issue types that warrant deletion-candidate vs review
|
||||
HARD_ISSUES = {"dark", "light", "low_information", "blurry", "grayscale"}
|
||||
SOFT_ISSUES = {"odd_aspect_ratio", "odd_size"}
|
||||
DARK_THRESHOLD = 40 # mean luminance below = dark/underexposed
|
||||
BLUR_THRESHOLD = 200.0 # edge variance below = blurry (calibrated)
|
||||
ANALYZE_SIZE = 300 # downscale for analysis (fast, consistent)
|
||||
|
||||
|
||||
@task
|
||||
def audit_folder(base_dir: str, issue_types: list[str] | None = None) -> dict:
|
||||
"""Run CleanVision audit on a folder. Returns issue summary + per-image issues."""
|
||||
from cleanvision import Imagelab
|
||||
|
||||
imagelab = Imagelab(data_path=base_dir)
|
||||
if issue_types:
|
||||
imagelab.find_issues(issue_types=issue_types)
|
||||
else:
|
||||
imagelab.find_issues()
|
||||
summary = imagelab.issue_summary.to_dict("records")
|
||||
|
||||
# imagelab.issues is ONE DataFrame: cols like dark_score/is_dark_issue
|
||||
df = imagelab.issues
|
||||
per_image = {}
|
||||
for idx, row in df.iterrows():
|
||||
name = idx
|
||||
for col in df.columns:
|
||||
if col.startswith("is_") and col.endswith("_issue") and row[col]:
|
||||
issue_type = col[len("is_"):-len("_issue")]
|
||||
per_image.setdefault(name, []).append(issue_type)
|
||||
return {"summary": summary, "per_image": per_image}
|
||||
def assess_image(path: str) -> dict:
|
||||
"""Fast quality assessment of one image via PIL (downscaled)."""
|
||||
p = Path(path)
|
||||
flags = []
|
||||
mean_lum = 0.0
|
||||
var = 0.0
|
||||
try:
|
||||
with Image.open(p) as im:
|
||||
g = im.convert("L")
|
||||
g.thumbnail((ANALYZE_SIZE, ANALYZE_SIZE))
|
||||
mean_lum = ImageStat.Stat(g).mean[0]
|
||||
if mean_lum < DARK_THRESHOLD:
|
||||
flags.append("dark")
|
||||
edges = g.filter(ImageFilter.FIND_EDGES)
|
||||
var = ImageStat.Stat(edges).var[0]
|
||||
if var < BLUR_THRESHOLD:
|
||||
flags.append("blurry")
|
||||
except Exception as e:
|
||||
return {"flags": ["unreadable"], "error": str(e)}
|
||||
return {"flags": flags, "mean_lum": round(mean_lum, 1), "edge_var": round(var, 1)}
|
||||
|
||||
|
||||
@task
|
||||
def classify_and_sort(base_dir: str, per_image: dict, move: bool = True) -> dict:
|
||||
"""Classify each image and (optionally) move into staging dirs."""
|
||||
root = Path(base_dir)
|
||||
images = [p for p in root.rglob("*") if p.is_file()]
|
||||
def assign_verdicts(source: str = None, limit: int = 1000) -> dict:
|
||||
"""Walk scanned DB rows; assign keep/delete_candidate (dark/blurry)."""
|
||||
db.init_db()
|
||||
conn = db.get_db()
|
||||
counts = {"keep": 0, "review": 0, "delete_candidate": 0, "skipped": 0}
|
||||
decisions = {}
|
||||
|
||||
for p in images:
|
||||
full = str(p)
|
||||
rel = str(p.relative_to(root))
|
||||
issues = set(per_image.get(full, []) or per_image.get(rel, []) or per_image.get(p.name, []))
|
||||
if not issues:
|
||||
decisions[rel] = "keep"
|
||||
counts["keep"] += 1
|
||||
if move:
|
||||
_move(p, KEEP, root)
|
||||
q = "SELECT sha256, path FROM image_hashes WHERE status='scanned'"
|
||||
params = []
|
||||
if source:
|
||||
q += " AND source=?"
|
||||
params.append(source)
|
||||
q += " ORDER BY added_at DESC LIMIT ?"
|
||||
params.append(limit)
|
||||
rows = conn.execute(q, params).fetchall()
|
||||
print(f"assign_verdicts: {len(rows)} rows to assess", flush=True)
|
||||
|
||||
for sha, path in rows:
|
||||
p = Path(path)
|
||||
if not p.exists():
|
||||
conn.execute("UPDATE image_hashes SET status='missing' WHERE sha256=?", (sha,))
|
||||
counts["skipped"] += 1
|
||||
continue
|
||||
if issues & HARD_ISSUES:
|
||||
decisions[rel] = "delete_candidate"
|
||||
counts["delete_candidate"] += 1
|
||||
if move:
|
||||
_move(p, DELETE, root)
|
||||
else:
|
||||
decisions[rel] = "review"
|
||||
counts["review"] += 1
|
||||
if move:
|
||||
_move(p, REVIEW, root)
|
||||
return {"counts": counts, "decisions": decisions}
|
||||
try:
|
||||
r = assess_image.fn(path)
|
||||
flags = r["flags"]
|
||||
except Exception as e:
|
||||
counts["skipped"] += 1
|
||||
continue
|
||||
status = "delete_candidate" if (flags and flags != ["keep"]) else "keep"
|
||||
conn.execute(
|
||||
"UPDATE image_hashes SET status=?, flag_reason=? WHERE sha256=?",
|
||||
(status, ",".join(flags), sha))
|
||||
counts[status] += 1
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
print(f"assign_verdicts done: {counts}", flush=True)
|
||||
return counts
|
||||
|
||||
|
||||
def _move(p: Path, dest_root: Path, src_root: Path):
|
||||
"""Move p into dest_root, preserving relative structure under source name."""
|
||||
rel = p.relative_to(src_root)
|
||||
dest = dest_root / p.parent.name / p.name
|
||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.move(str(p), str(dest))
|
||||
@task
|
||||
def notify_result(counts: dict, base_dir: str):
|
||||
import apprise_helper
|
||||
|
||||
|
||||
|
||||
|
||||
def _slice_dir(base_dir: str, max_files: int) -> str:
|
||||
"""Copy first N images into a temp dir for CleanVision to audit."""
|
||||
import shutil
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
src = Path(base_dir)
|
||||
tmp = Path(tempfile.mkdtemp(prefix="cvslice_"))
|
||||
exts = {".jpg", ".jpeg", ".png", ".webp", ".gif", ".heic", ".tif", ".bmp"}
|
||||
n = 0
|
||||
for p in src.rglob("*"):
|
||||
if p.is_file() and p.suffix.lower() in exts:
|
||||
shutil.copy2(p, tmp / p.name)
|
||||
n += 1
|
||||
if n >= max_files:
|
||||
break
|
||||
print(f"_slice_dir: copied {n} files to {tmp}")
|
||||
return str(tmp)
|
||||
body = (
|
||||
f"Quality scan: {base_dir}\n"
|
||||
f"Keep: {counts['keep']} | Review: {counts['review']} | "
|
||||
f"Delete-candidates: {counts['delete_candidate']} | Skipped: {counts['skipped']}\n"
|
||||
f"Review: http://192.168.20.13:8092/review"
|
||||
)
|
||||
apprise_helper.notify("📸 photo-pipeline quality scan complete", body)
|
||||
|
||||
|
||||
@flow(name="photo-quality-scan")
|
||||
def quality_scan(base_dir: str, move: bool = False, notify: bool = True, max_files: int = None):
|
||||
"""Audit image quality with CleanVision; classify into keep/review/delete.
|
||||
|
||||
max_files: if set, only audit the first N image files (slices huge folders
|
||||
into reviewable chunks — prevents OOM on 50K-file trees).
|
||||
"""
|
||||
if max_files: # 0/None = unlimited
|
||||
base_dir = _slice_dir(base_dir, max_files)
|
||||
audit = audit_folder(base_dir)
|
||||
print(f"Issue summary: {audit['summary']}")
|
||||
result = classify_and_sort(base_dir, audit["per_image"], move=move)
|
||||
print(f"Verdicts: {result['counts']}")
|
||||
return {"audit": audit["summary"], **result}
|
||||
def quality_scan(base_dir: str, move: bool = False, notify: bool = True,
|
||||
max_files: int = 0, source: str = "", limit: int = 1000):
|
||||
"""Fast quality verdicts (dark/blurry via PIL, downscaled) written to DB."""
|
||||
result = assign_verdicts(source or None, limit)
|
||||
if notify:
|
||||
notify_result(result["counts"], audit["summary"], base_dir)
|
||||
notify_result(result, base_dir)
|
||||
return result
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
d = sys.argv[1] if len(sys.argv) > 1 else "/tmp/cvtest"
|
||||
move = "--move" in sys.argv
|
||||
quality_scan(d, move=move)
|
||||
|
||||
|
||||
@task
|
||||
def notify_result(counts: dict, summary: list, base_dir: str):
|
||||
"""Send batch summary via Apprise."""
|
||||
import apprise_helper
|
||||
|
||||
hard = counts.get("delete_candidate", 0)
|
||||
soft = counts.get("review", 0)
|
||||
keep = counts.get("keep", 0)
|
||||
flagged = [s for s in summary if s["num_images"] > 0]
|
||||
lines = "; ".join(f"{s[issue_type]}: {s[num_images]}" for s in flagged) or "none"
|
||||
body = (
|
||||
f"Scanned: {base_dir}\n"
|
||||
f"Keep: {keep} | Review: {soft} | Delete-candidates: {hard}\n"
|
||||
f"Issues: {lines}\n"
|
||||
f"Review: http://192.168.20.13:8092/review"
|
||||
)
|
||||
apprise_helper.notify("📸 photo-pipeline batch complete", body)
|
||||
return True
|
||||
d = sys.argv[1] if len(sys.argv) > 1 else "/mnt/ubuntu_storage_3TB/archive/03_photos/Pictures"
|
||||
src = sys.argv[2] if len(sys.argv) > 2 else ""
|
||||
quality_scan(d, source=src)
|
||||
|
||||
Reference in New Issue
Block a user