"""Local web UI for Image Processor Pro. A thin Flask front-end over the EXISTING pipeline. It does not change any processing logic — it builds a Config and calls ImagePipeline.process_many() exactly like the CLI (main.py) does, then shows the results in the browser. Run: .venv/bin/python webapp.py # http://127.0.0.1:5001 .venv/bin/python webapp.py --port 8080 """ from __future__ import annotations import argparse import threading from pathlib import Path from flask import Flask, jsonify, render_template_string, request, send_from_directory from config import Config from pipeline import ImagePipeline from utils import human_size, setup_logging app = Flask(__name__) OUTPUT_ROOT = Path("output").resolve() # One pipeline per (remove_watermark, engine) combo so the LaMa model is loaded # once and reused across requests. Batches are serialised — this is a local, # single-user tool and it keeps us clear of any shared-state questions. _pipelines: dict[tuple[bool, str], ImagePipeline] = {} _pipelines_lock = threading.Lock() _process_lock = threading.Lock() def get_pipeline(remove_watermark: bool, engine: str) -> ImagePipeline: key = (remove_watermark, engine) with _pipelines_lock: pipeline = _pipelines.get(key) if pipeline is None: config = Config() config.remove_watermark = remove_watermark config.watermark_engine = engine pipeline = ImagePipeline(config) _pipelines[key] = pipeline return pipeline def _output_url(output_path: Path) -> str | None: try: rel = output_path.resolve().relative_to(OUTPUT_ROOT) except (ValueError, OSError): return None return "/output/" + str(rel).replace("\\", "/") @app.route("/") def index() -> str: return render_template_string(PAGE) @app.route("/output/") def serve_output(subpath: str): # send_from_directory rejects path traversal outside OUTPUT_ROOT. return send_from_directory(OUTPUT_ROOT, subpath) @app.route("/process", methods=["POST"]) def process(): data = request.get_json(silent=True) or {} raw = data.get("urls", "") urls = [line.strip() for line in raw.splitlines() if line.strip()] if not urls: return jsonify(error="Please paste at least one image URL."), 400 remove_watermark = bool(data.get("remove_watermark", True)) engine = "classical" if data.get("engine") == "classical" else "lama" pipeline = get_pipeline(remove_watermark, engine) with _process_lock: # one batch at a time results = pipeline.process_many(urls) by_url: dict[str, object] = {} for r in results: by_url.setdefault(r.url, r) payload = [] for url in urls: r = by_url.get(url) if r is None: payload.append(dict(url=url, success=False, error="No result returned.")) continue dims = f"{r.final_dimensions[0]}×{r.final_dimensions[1]}" if r.final_dimensions else None payload.append(dict( url=url, success=r.success, output_url=_output_url(r.output_path) if r.output_path else None, dimensions=dims, size=human_size(r.final_size_bytes) if r.success else None, wm_seconds=round(r.watermark_seconds, 2), error=r.error, )) succeeded = sum(1 for p in payload if p["success"]) return jsonify(results=payload, succeeded=succeeded, failed=len(payload) - succeeded, engine=engine, remove_watermark=remove_watermark) PAGE = r""" Image Processor Pro

Image Processor Pro

Paste image URLs (one per line) — they’re downloaded, watermark-removed, and saved as optimized JPEGs.

Tip: paste 10+ URLs at once — they’re processed together.

Remove watermark

""" def main() -> None: parser = argparse.ArgumentParser(description="Local web UI for Image Processor Pro.") parser.add_argument("--host", default="127.0.0.1", help="Bind host (default 127.0.0.1, local only).") parser.add_argument("--port", type=int, default=5001, help="Port (default 5001).") args = parser.parse_args() setup_logging(Path("logs"), verbose=False) OUTPUT_ROOT.mkdir(parents=True, exist_ok=True) print(f"\n Image Processor Pro UI -> http://{args.host}:{args.port}\n") app.run(host=args.host, port=args.port, threaded=True) if __name__ == "__main__": main()