# photofresh - adaptive photo touch-up for phone photos

> Source: <https://gist.github.com/ludostaatopstal/7ea70d1408e20648ab5a7662fcbcf823>
> Published: 2026-10-06 19:30:25+00:00

|  | # photofresh.py - adaptive, per-photo touch-up for phone photos (no AI, nothing invented) | 
|  | # | 
|  | # Watches an inbox folder, gently improves each new JPEG and writes the result to an | 
|  | # output folder (e.g. an Immich external library), sorted into year/month subfolders. | 
|  | # | 
|  | # Folders (inside the container): | 
|  | #   /data/inbox      <- new photos arrive here (e.g. via Syncthing) | 
|  | #   /data/processed  <- originals after processing (auto-deleted after KEEP_DAYS) | 
|  | #   /data/failed     <- photos that could not be processed (never deleted) | 
|  | #   /data/redo       <- put an original here to process it again and overwrite the old result | 
|  | #   /output          <- processed photos, in year/month subfolders | 
|  | # | 
|  | # Per photo it MEASURES first and only then adjusts: | 
|  | #   1. white balance: corrects half of the measured colour cast (max 12%) | 
|  | #   2. contrast range: stretched only if the histogram is compressed (haze, flat light) | 
|  | #   3. midtone lift: only for photos that are too dark | 
|  | #   4. local contrast (CLAHE), mixed in at 40% | 
|  | #   5. noise reduction: only if measured noise is high, and only on flat areas (walls, sky) | 
|  | #   "Night mode": high ISO + dark photo -> keep the darkness, only clean up noise. | 
|  | #   All EXIF data (date, GPS, orientation) is copied unchanged. | 
|  | import os, time, shutil, subprocess, traceback | 
|  | import cv2, numpy as np | 
|  |  | 
|  | DATA = "/data" | 
|  | INBOX, PROCESSED, FAILED, REDO = (f"{DATA}/{d}" for d in ("inbox", "processed", "failed", "redo")) | 
|  | OUTPUT = "/output" | 
|  | LEDGER, LOG = f"{DATA}/processed.list", f"{DATA}/photofresh.log" | 
|  | KEEP_DAYS = float(os.environ.get("KEEP_DAYS", "30"))     # 0 = delete originals right after processing | 
|  | SHARPEN = float(os.environ.get("SHARPEN", "0"))          # 0 = off (default), 1 = light sharpening | 
|  | NIGHT_ISO = float(os.environ.get("NIGHT_ISO", "1000"))   # ISO from which a dark photo counts as a night shot (0 = off) | 
|  | PUID, PGID = int(os.environ.get("PUID", "99")), int(os.environ.get("PGID", "100"))  # 99/100 = Unraid default | 
|  | JPEG_QUALITY = int(os.environ.get("JPEG_QUALITY", "95")) | 
|  | INTERVAL = 120        # seconds between scans | 
|  | MIN_AGE = 30          # a file must be unchanged for this long (still syncing otherwise) | 
|  | EXT = (".jpg", ".jpeg") | 
|  |  | 
|  | def log(msg): | 
|  | line = time.strftime("%Y-%m-%d %H:%M:%S ") + msg | 
|  | print(line, flush=True) | 
|  | with open(LOG, "a") as f: f.write(line + "\n") | 
|  |  | 
|  | def set_owner(path, is_dir=False): | 
|  | try: | 
|  | os.chown(path, PUID, PGID) | 
|  | os.chmod(path, 0o777 if is_dir else 0o666) | 
|  | except Exception: pass | 
|  |  | 
|  | def make_dir(path): | 
|  | if not os.path.isdir(path): | 
|  | os.makedirs(path, exist_ok=True) | 
|  | set_owner(path, True) | 
|  |  | 
|  | # ---------- the algorithm ---------- | 
|  | def noise_sigma(gray): | 
|  | """Estimate noise only in the 15% flattest blocks, so texture (grass, hair) doesn't count as noise.""" | 
|  | g = gray.astype(np.float32) | 
|  | hp = g - cv2.GaussianBlur(g, (0, 0), 1.5) | 
|  | grad = cv2.GaussianBlur(np.abs(cv2.Sobel(cv2.GaussianBlur(g, (0, 0), 3), cv2.CV_32F, 1, 1)), (0, 0), 3) | 
|  | B = 32; vals = []; flat = [] | 
|  | for y in range(0, g.shape[0] - B, B): | 
|  | for x in range(0, g.shape[1] - B, B): | 
|  | m = g[y:y+B, x:x+B].mean() | 
|  | if 25 < m < 235: | 
|  | flat.append(grad[y:y+B, x:x+B].mean()) | 
|  | vals.append(np.median(np.abs(hp[y:y+B, x:x+B])) * 1.4826) | 
|  | if not vals: return 0.0 | 
|  | flat = np.array(flat); vals = np.array(vals) | 
|  | return float(np.median(vals[flat <= np.percentile(flat, 15)])) | 
|  |  | 
|  | def enhance(bgr, iso=None): | 
|  | info = {} | 
|  | img = bgr.astype(np.float32) / 255.0 | 
|  | # night shot? (high ISO and dark image): keep the darkness, it is the mood | 
|  | mean = float(np.mean(cv2.cvtColor(bgr, cv2.COLOR_BGR2GRAY))) / 255 | 
|  | night = bool(NIGHT_ISO > 0 and iso is not None and iso >= NIGHT_ISO and mean < 0.40) | 
|  | info["night"] = night | 
|  | # 1. white balance: measure the cast on near-grey pixels, correct 50% (25% at night), max 12% | 
|  | hsv = cv2.cvtColor(bgr, cv2.COLOR_BGR2HSV) | 
|  | v, s = hsv[..., 2], hsv[..., 1] | 
|  | mask = (s < 60) & (v > 40) & (v < 240) | 
|  | if mask.sum() < 0.02 * mask.size: mask = (v > 20) & (v < 245) | 
|  | means = np.array([img[..., c][mask].mean() for c in range(3)]) | 
|  | strength = 0.25 if night else 0.5 | 
|  | gains = np.clip(1 + strength * (means.mean() / means - 1), 0.88, 1.12) | 
|  | img = np.clip(img * gains, 0, 1) | 
|  | info["wb"] = [round(float(x), 3) for x in gains] | 
|  | # 2. stretch the contrast range (may clip a tiny bit) | 
|  | lab = cv2.cvtColor((img * 255).astype(np.uint8), cv2.COLOR_BGR2LAB) | 
|  | L = lab[..., 0].astype(np.float32) | 
|  | lo, hi = np.percentile(L, 0.5), np.percentile(L, 99.5) | 
|  | lo_new = lo - 0.7 * max(lo - 6, 0) | 
|  | hi_new = hi + 0.8 * max(252 - hi, 0) | 
|  | if hi - lo > 10 and not night: | 
|  | L = (L - lo) / (hi - lo) * (hi_new - lo_new) + lo_new | 
|  | # 3. photo too dark: gentle midtone lift | 
|  | meanL = float(np.mean(np.clip(L, 0, 255))) / 255 | 
|  | if meanL < 0.42 and not night: | 
|  | gamma = max(0.85, np.log(0.45) / np.log(max(meanL, 1e-3))) | 
|  | L = 255 * np.power(np.clip(L / 255, 0, 1), gamma) | 
|  | # 4. adaptive local contrast (CLAHE), mixed in at 40%; skipped at night (it lifts shadows and grain) | 
|  | L8 = np.clip(L, 0, 255).astype(np.uint8) | 
|  | if not night: | 
|  | L8 = cv2.addWeighted(L8, 0.6, cv2.createCLAHE(clipLimit=1.6, tileGridSize=(8, 8)).apply(L8), 0.4, 0) | 
|  | lab[..., 0] = L8 | 
|  | out = cv2.cvtColor(lab, cv2.COLOR_LAB2BGR) | 
|  | # 5. noise reduction only when needed, and only on flat areas | 
|  | sig = noise_sigma(cv2.cvtColor(bgr, cv2.COLOR_BGR2GRAY)) | 
|  | info["noise"] = round(sig, 2) | 
|  | s_ref = max(sig, 0.8) | 
|  | g = cv2.cvtColor(out, cv2.COLOR_BGR2GRAY).astype(np.float32) | 
|  | tex = cv2.GaussianBlur(np.abs(g - cv2.GaussianBlur(g, (0, 0), 2)), (0, 0), 4) | 
|  | texture = np.clip((tex - s_ref) / (3.0 * s_ref), 0, 1)   # 0 = flat, 1 = texture/edge | 
|  | if sig > 1.2: | 
|  | # luminance noise (grain): gentle; colour noise (speckles): firm | 
|  | h_lum = float(np.clip((sig - 1.2) * 0.8 + 2.5, 2.5, 5.0)) | 
|  | h_col = float(np.clip((sig - 1.2) * 2.0 + 6.0, 6.0, 14.0)) | 
|  | den = cv2.fastNlMeansDenoisingColored(out, None, h_lum, h_col, 5, 15) | 
|  | w = (1 - texture)[..., None] | 
|  | out = (den * w + out * (1 - w)).astype(np.uint8) | 
|  | # 6. optional light sharpening (OFF by default): luminance only, textured areas only, | 
|  | #    ignores differences below the noise level, clamped to avoid halos | 
|  | if SHARPEN > 0: | 
|  | lab = cv2.cvtColor(out, cv2.COLOR_BGR2LAB) | 
|  | Lf = lab[..., 0].astype(np.float32) | 
|  | detail = Lf - cv2.GaussianBlur(Lf, (0, 0), 1.0) | 
|  | thr = max(1.5, 1.2 * sig) | 
|  | detail = np.sign(detail) * np.maximum(np.abs(detail) - thr, 0) | 
|  | detail = np.clip(detail, -12, 12) | 
|  | amount = (0.7 if sig < 3 else 0.5) * SHARPEN | 
|  | lab[..., 0] = np.clip(Lf + amount * detail * texture, 0, 255).astype(np.uint8) | 
|  | out = cv2.cvtColor(lab, cv2.COLOR_LAB2BGR) | 
|  | return out, info | 
|  |  | 
|  | # ---------- file handling ---------- | 
|  | def load_ledger(): | 
|  | if not os.path.exists(LEDGER): return set() | 
|  | with open(LEDGER) as f: return set(l.strip() for l in f if l.strip()) | 
|  |  | 
|  | def read_iso(src): | 
|  | try: | 
|  | r = subprocess.run(["exiftool", "-s3", "-n", "-ISO", src], capture_output=True, text=True, timeout=30) | 
|  | return float(r.stdout.strip().split()[0]) | 
|  | except Exception: | 
|  | return None | 
|  |  | 
|  | def target_subdir(src): | 
|  | try: | 
|  | r = subprocess.run(["exiftool", "-s3", "-d", "%Y/%m", "-DateTimeOriginal", src], | 
|  | capture_output=True, text=True, timeout=30) | 
|  | sub = r.stdout.strip() | 
|  | if len(sub) == 7 and sub[4] == "/": return sub | 
|  | except Exception: pass | 
|  | return time.strftime("%Y/%m", time.localtime(os.path.getmtime(src))) | 
|  |  | 
|  | def unique(path): | 
|  | if not os.path.exists(path): return path | 
|  | base, ext = os.path.splitext(path); i = 1 | 
|  | while os.path.exists(f"{base}_{i}{ext}"): i += 1 | 
|  | return f"{base}_{i}{ext}" | 
|  |  | 
|  | def process(name, ledger, folder=INBOX, redo=False): | 
|  | src = os.path.join(folder, name) | 
|  | key = f"{name}\|{os.path.getsize(src)}" | 
|  | if key in ledger and not redo: | 
|  | # already done (e.g. re-synced): don't duplicate it in the library, but don't delete it either | 
|  | shutil.move(src, unique(os.path.join(PROCESSED, name))) | 
|  | log(f"skipped (already processed), moved to 'processed': {name}  -> put it in 'redo' to process it again"); return | 
|  | # IGNORE_ORIENTATION: don't rotate pixels, so the original Orientation tag stays correct | 
|  | bgr = cv2.imread(src, cv2.IMREAD_COLOR \| cv2.IMREAD_IGNORE_ORIENTATION) | 
|  | if bgr is None: raise RuntimeError("could not read photo") | 
|  | iso = read_iso(src) | 
|  | out, info = enhance(bgr, iso) | 
|  | sub = target_subdir(src) | 
|  | make_dir(os.path.join(OUTPUT, sub[:4])); make_dir(os.path.join(OUTPUT, sub)) | 
|  | dst = os.path.join(OUTPUT, sub, name) | 
|  | if not redo: dst = unique(dst)          # in 'redo' the old result is overwritten | 
|  | tmp = dst + ".tmp.jpg" | 
|  | if not cv2.imwrite(tmp, out, [cv2.IMWRITE_JPEG_QUALITY, JPEG_QUALITY]): raise RuntimeError("write failed") | 
|  | r = subprocess.run(["exiftool", "-q", "-overwrite_original", "-TagsFromFile", src, "-all:all", tmp], | 
|  | capture_output=True, text=True, timeout=60) | 
|  | if r.returncode != 0: raise RuntimeError("copying metadata failed: " + r.stderr.strip()) | 
|  | if cv2.imread(tmp) is None: raise RuntimeError("result check failed") | 
|  | if not redo:   # in 'redo' the file gets the current time so the library notices the change | 
|  | st = os.stat(src); os.utime(tmp, (st.st_atime, st.st_mtime)) | 
|  | os.rename(tmp, dst); set_owner(dst) | 
|  | if key not in ledger: | 
|  | with open(LEDGER, "a") as f: f.write(key + "\n") | 
|  | ledger.add(key) | 
|  | if KEEP_DAYS > 0: | 
|  | keep = unique(os.path.join(PROCESSED, name)) | 
|  | shutil.move(src, keep) | 
|  | os.utime(keep, None)   # retention clock starts now, not at the capture date | 
|  | else: | 
|  | os.remove(src) | 
|  | log(f"{'redo ' if redo else ''}done: {name} -> {sub}/  iso={iso} night={'yes' if info['night'] else 'no'} wb={info['wb']} noise={info['noise']}") | 
|  |  | 
|  | def cleanup(): | 
|  | limit = time.time() - KEEP_DAYS * 86400 | 
|  | for n in os.listdir(PROCESSED): | 
|  | p = os.path.join(PROCESSED, n) | 
|  | if os.path.isfile(p) and os.path.getmtime(p) < limit: | 
|  | os.remove(p); log(f"original deleted after {KEEP_DAYS:g} days: {n}") | 
|  |  | 
|  | REPORTED = set()   # so each skipped file is logged only once | 
|  |  | 
|  | def report_once(name, reason): | 
|  | if (name, reason) not in REPORTED: | 
|  | REPORTED.add((name, reason)); log(f"skipped: {name} ({reason})") | 
|  |  | 
|  | def scan(ledger): | 
|  | for name in sorted(os.listdir(INBOX)): | 
|  | p = os.path.join(INBOX, name) | 
|  | if name.startswith(".") or not os.path.isfile(p): continue | 
|  | if not name.lower().endswith(EXT): | 
|  | report_once(name, "not a .jpg/.jpeg file"); continue | 
|  | age = time.time() - os.path.getmtime(p) | 
|  | if age < -60: | 
|  | # file date in the future: would otherwise never be processed | 
|  | report_once(name, "file date is in the future, processing anyway") | 
|  | elif age < MIN_AGE: | 
|  | continue | 
|  | try: | 
|  | process(name, ledger) | 
|  | except Exception as e: | 
|  | log(f"ERROR with {name}: {e}") | 
|  | try: shutil.move(p, unique(os.path.join(FAILED, name))) | 
|  | except Exception: log(traceback.format_exc()) | 
|  | for name in sorted(os.listdir(REDO)): | 
|  | p = os.path.join(REDO, name) | 
|  | if name.startswith(".") or not os.path.isfile(p): continue | 
|  | if not name.lower().endswith(EXT): | 
|  | report_once(name, "not a .jpg/.jpeg file (redo folder)"); continue | 
|  | if 0 <= time.time() - os.path.getmtime(p) < 10: continue   # still being copied | 
|  | try: | 
|  | process(name, ledger, REDO, redo=True) | 
|  | except Exception as e: | 
|  | log(f"ERROR with {name} (redo): {e}") | 
|  | try: shutil.move(p, unique(os.path.join(FAILED, name))) | 
|  | except Exception: log(traceback.format_exc()) | 
|  |  | 
|  | if __name__ == "__main__": | 
|  | for d in (INBOX, PROCESSED, FAILED, REDO): make_dir(d) | 
|  | log(f"photofresh started (keeping originals: {KEEP_DAYS:g} days, scanning every {INTERVAL} s)") | 
|  | log(f"inbox currently contains: {sorted(os.listdir(INBOX)) or 'nothing'}") | 
|  | ledger = load_ledger() | 
|  | while True: | 
|  | try: | 
|  | scan(ledger); cleanup() | 
|  | except Exception: | 
|  | log(traceback.format_exc()) | 
|  | time.sleep(INTERVAL) |