"""
100% local server - makes no external network calls.
Runs on your machine only (127.0.0.1) to serve the UI and analyze files locally.
"""
import os
import sys
os.environ.setdefault("HF_HUB_DISABLE_SYMLINKS_WARNING", "1")
# Hidden "app" mode (launched via TruScan.vbs with pythonw.exe): there is no
# console, so send every print/log line to cache\logs\truescan.log instead.
if sys.stdout is None or os.environ.get("TRUESCAN_HIDDEN"):
_log_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "cache", "logs")
os.makedirs(_log_dir, exist_ok=True)
_log_path = os.path.join(_log_dir, "truescan.log")
try:
if os.path.getsize(_log_path) > 5 * 1024 * 1024: # keep it small
os.replace(_log_path, _log_path + ".old")
except OSError:
pass
sys.stdout = sys.stderr = open(_log_path, "a", encoding="utf-8", buffering=1)
# Keep the last lines of output in memory so the app window can show them
# (the "black screen" moved inside the app: click the status pill).
import collections as _collections
_RECENT_LINES = _collections.deque(maxlen=200)
class _Tee:
def __init__(self, stream):
self._stream = stream
self._buf = ""
def write(self, text):
try:
if self._stream is not None:
self._stream.write(text)
except Exception:
pass
self._buf += text
while "\n" in self._buf:
line, self._buf = self._buf.split("\n", 1)
line = line.rstrip("\r")
if line.strip():
_RECENT_LINES.append(line[-400:])
# tqdm progress bars use \r without \n - keep the latest one visible
if "\r" in self._buf:
parts = [p.strip() for p in self._buf.split("\r") if p.strip()]
last = parts[-1] if parts else ""
self._buf = ""
if last:
_RECENT_LINES.append(last[-400:])
return len(text)
def flush(self):
try:
if self._stream is not None:
self._stream.flush()
except Exception:
pass
def __getattr__(self, name):
return getattr(self._stream, name)
sys.stdout = _Tee(sys.stdout)
sys.stderr = _Tee(sys.stderr)
import json
import hashlib
import tempfile
import webbrowser
import traceback
import threading
import time
from concurrent.futures import ThreadPoolExecutor
# Project layout: this file lives at the project root; the actual app code
# lives in route/, and model weights/packages live in models/. Both need
# to be importable — route/ for `from analysis_engine import ...` and for
# analysis_engine's own sibling imports (ai_detector, face_detector, ...),
# and models/ because it holds Python PACKAGES (aidetector/, bioverify_engine/)
# that get imported directly (e.g. `import aidetector`), not just data files.
_PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
_ROUTE_DIR = os.path.join(_PROJECT_ROOT, "route")
_MODELS_DIR = os.path.join(_PROJECT_ROOT, "models")
for _p in (_ROUTE_DIR, _MODELS_DIR):
if _p not in sys.path:
sys.path.insert(0, _p)
from flask import Flask, request, jsonify, render_template
from werkzeug.utils import secure_filename
# ---------------------------------------------------------------------------
# Optional detector files. If one is missing from this installation (e.g. it
# was never committed to GitHub), the app still starts: that one signal
# reports "unavailable" (the same value the detector itself returns when its
# model can't load) and the status pill lists it as missing.
# ---------------------------------------------------------------------------
_OPTIONAL_MODULES = {
"nonescape_bridge": {"classify_image": None},
"steganographia_detector": {"classify_image": None},
"pixel_forensics_detector": {"classify_image": None},
"sdxl_face_detector": {"classify_image": None},
"video_metadata_detector": {"scan_video_metadata": (0, [], False)},
"voice_deepfake_detector": {"classify_audio": (0, None, False)},
"voice_descriptive_stats": {},
"speaker_verification": {},
}
MISSING_MODULES = []
def _install_missing_stubs():
import importlib.util
import types
for name, funcs in _OPTIONAL_MODULES.items():
if importlib.util.find_spec(name) is not None:
continue
mod = types.ModuleType(name)
mod.__truescan_stub__ = True
for fn, ret in funcs.items():
setattr(mod, fn, (lambda ret=ret: (lambda *a, **k: ret))())
def __getattr__(attr, _name=name):
if attr.startswith("__"):
raise AttributeError(attr)
def _missing(*a, **k):
raise RuntimeError(f"{_name}.py is missing from this installation")
return _missing
mod.__getattr__ = __getattr__
sys.modules[name] = mod
MISSING_MODULES.append(name)
if MISSING_MODULES:
print(f"[app] Missing detector files (those signals are skipped): {', '.join(MISSING_MODULES)}", flush=True)
_install_missing_stubs()
from analysis_engine import analyze_image_bytes, analyze_video_bytes, r
import forensic_scope_api
import document_forensics
import voice_deepfake_detector
import voice_descriptive_stats
import speaker_verification
import plate_reader
import vehicle_classifier
import image_enhancer
import video_enhancer
import audio_extractor
import ai_detector
import deepfake_face_detector
import face_detector
import efficientnetb3_detector
import sdxl_face_detector
import noise_residual_detector
import clone_detection_module
import pixel_forensics_detector
import community_forensics_detector
import steganographia_detector
import insightface_detector
app = Flask(__name__)
import logging
class _NoPolling(logging.Filter):
"""Hides the UI's every-few-seconds counter polling from the console so
real messages (warm-up, errors, timings) stay readable."""
_QUIET = ("/api/calibration-log/count", "/api/correct-plate/count", "/api/ready", "/api/activity", "/favicon.ico")
def filter(self, record):
msg = record.getMessage()
return not any(p in msg for p in self._QUIET)
logging.getLogger("werkzeug").addFilter(_NoPolling())
app.config["MAX_CONTENT_LENGTH"] = 300 * 1024 * 1024 # 300MB max
_SHARE_PW_FILE = os.path.join(_PROJECT_ROOT, "cache", "share_password.txt")
_share_pw_cache = {"mtime": None, "pw": ""}
def _share_password():
"""Password for remote visitors: APP_PASSWORD (server/Docker) or the one
set by "Share Online.bat" (Windows PC shared through Tailscale)."""
if os.environ.get("APP_PASSWORD"):
return os.environ["APP_PASSWORD"]
try:
m = os.path.getmtime(_SHARE_PW_FILE)
except OSError:
return ""
if m != _share_pw_cache["mtime"]:
with open(_SHARE_PW_FILE, "r", encoding="utf-8-sig") as f:
_share_pw_cache.update(mtime=m, pw=f.read().strip())
return _share_pw_cache["pw"]
_PROXY_HEADERS = ("X-Forwarded-For", "X-Forwarded-Host", "X-Forwarded-Proto",
"Tailscale-User-Login", "Tailscale-Funnel-Request")
def _is_local_user():
"""The person sitting at this PC (app window) — not a visitor arriving
through Tailscale serve/funnel, which also connects from 127.0.0.1 but
adds proxy headers."""
return (request.remote_addr in ("127.0.0.1", "::1")
and not any(h in request.headers for h in _PROXY_HEADERS)
and not os.environ.get("TRUESCAN_SERVER"))
_AUTH_COOKIE = "ts_auth"
_OPEN_PATHS = ("/login", "/logout", "/favicon.ico")
def _session_token(pw):
import hashlib
import hmac
return hmac.new(pw.encode(), b"truscan-session-v1", hashlib.sha256).hexdigest()
def _password_ok(given, pw):
import hmac
return given is not None and hmac.compare_digest(str(given).encode(), pw.encode())
_LOGIN_PAGE = """
TrueScan
"""
@app.route("/login", methods=["GET", "POST"])
def login_page():
from flask import Response, redirect
import html
pw = _share_password()
nxt = request.values.get("next") or "/"
if not nxt.startswith("/") or nxt.startswith("//"):
nxt = "/"
err = ""
if request.method == "POST" and pw:
if _password_ok(request.form.get("password"), pw):
resp = redirect(nxt)
# SameSite=None so the session also works inside the huggingface.co page frame.
resp.set_cookie(_AUTH_COOKIE, _session_token(pw), max_age=30 * 24 * 3600,
httponly=True, secure=request.is_secure or bool(os.environ.get("SPACE_ID")),
samesite="None" if (request.is_secure or os.environ.get("SPACE_ID")) else "Lax")
return resp
time.sleep(1.0) # slows down guessing
err = 'كلمة المرور غير صحيحة / Incorrect password
'
page = _LOGIN_PAGE.replace("__NEXT__", html.escape(nxt, quote=True)).replace("__ERR__", err)
return Response(page, 200, {"Content-Type": "text/html; charset=utf-8"})
@app.route("/logout")
def logout_page():
from flask import redirect
resp = redirect("/login")
resp.delete_cookie(_AUTH_COOKIE)
return resp
@app.before_request
def _require_password():
"""Remote visitors sign in once on the login page (a cookie keeps them
signed in for 30 days). The local app window never asks."""
if _is_local_user() or request.path in _OPEN_PATHS:
return None
from flask import Response, redirect, jsonify as _json
import hmac
pw = _share_password()
if not pw:
if os.environ.get("SPACE_ID"):
# Hosted on Hugging Face: never serve the app without a password,
# whether the Space is public or private.
return Response("TrueScan: set the APP_PASSWORD secret in GitHub and run the deploy again.", 403)
if os.environ.get("TRUESCAN_SERVER"):
return None # Docker without a password = private tailnet only
return Response("Sharing is on but no password is set - run Share Online.bat again.", 403)
tok = request.cookies.get(_AUTH_COOKIE)
if tok and hmac.compare_digest(tok, _session_token(pw)):
return None
if _password_ok(request.headers.get("X-App-Password"), pw):
return None # scripted checks (smoke test)
auth = request.authorization
if auth and auth.type == "basic" and _password_ok(auth.password, pw):
return None
if request.path.startswith("/api/"):
return _json({"error": "Login required"}), 401
from urllib.parse import quote
return redirect("/login?next=" + quote(request.full_path.rstrip("?")))
# ---------------------------------------------------------------------------
# FAST STARTUP (2026-09): the server + browser now start immediately and the
# models load in a background thread. Analysis endpoints wait for that load
# to finish (so no request ever sees a half-loaded model); the UI itself,
# settings and counters respond instantly.
# ---------------------------------------------------------------------------
_models_ready = threading.Event() # set when EVERY model finished
_model_events = {} # name -> Event, set when THAT model finished
_warmup_status = {"done": 0, "total": 0, "failed": [], "seconds": {}}
# Which models each endpoint actually needs. Plate reading waits ONLY for
# its own two models, not for all 13 detectors. None = wait for everything.
_ENDPOINT_NEEDS = {
"/api/read-plate": ["plate_reader"],
"/api/analyze": None,
"/api/analyze-document": None,
"/api/forensic-scope": None,
}
def _event(name):
return _model_events.setdefault(name, threading.Event())
@app.before_request
def _wait_for_models():
if request.path not in _ENDPOINT_NEEDS:
return
needs = _ENDPOINT_NEEDS[request.path]
if needs is None:
_models_ready.wait()
else:
for name in needs:
_event(name).wait()
# One heavy analysis at a time (TRUESCAN_PARALLEL to change). On a small CPU
# (e.g. a 4-core Intel N95) two analyses side by side are SLOWER than two in
# a row and can run out of memory; the second visitor just waits its turn.
_HEAVY_PATHS = {"/api/analyze", "/api/analyze-document", "/api/read-plate", "/api/forensic-scope",
"/api/analyze-voice", "/api/compare-voices"}
# (enhancement is plain OpenCV - light enough to run beside an analysis, so
# the preview doesn't wait behind the AI check that runs at the same time)
_heavy_gate = threading.BoundedSemaphore(max(1, int(os.environ.get("TRUESCAN_PARALLEL", "1"))))
@app.before_request
def _take_turn():
if request.path in _HEAVY_PATHS:
_heavy_gate.acquire()
request.environ["truescan.heavy"] = True
@app.teardown_request
def _give_turn(exc=None):
try:
if request.environ.pop("truescan.heavy", False):
_heavy_gate.release()
except Exception:
pass
@app.route("/favicon.ico")
def favicon():
"""Window + taskbar icon when running as an app window."""
from flask import send_file
# Prefer the project's original icon (TrueScan.ico); truscan.ico is the fallback.
for name in ("TrueScan.ico", "truscan.ico"):
ico = os.path.join(_PROJECT_ROOT, name)
if os.path.exists(ico):
return send_file(ico, mimetype="image/x-icon")
return ("", 404)
_active = {} # request id -> (label, start time)
_ACTIVITY_LABELS = {
"/api/analyze": "فحص الملف", "/api/analyze-document": "فحص المستند",
"/api/read-plate": "قراءة اللوحة",
"/api/analyze-voice": "فحص الصوت", "/api/compare-voices": "مقارنة الأصوات",
"/api/forensic-scope": "الفحص الجنائي",
"/api/enhance/render": "معالجة الصورة",
"/api/jpeg-forensics": "كشف التلاعب", "/api/prnu": "مطابقة بصمة الكاميرا", "/api/file-structure": "بنية الملف",
"/api/enhance/video/average": "دمج إطارات الفيديو",
}
@app.before_request
def _track_start():
label = _ACTIVITY_LABELS.get(request.path)
if label:
_active[id(request._get_current_object())] = (label, time.time())
@app.teardown_request
def _track_end(exc=None):
try:
_active.pop(id(request._get_current_object()), None)
except Exception:
pass
@app.route("/api/activity")
def activity():
now = time.time()
return jsonify({
"ready": _models_ready.is_set(),
"done": _warmup_status["done"], "total": _warmup_status["total"],
"busy": [{"label": l, "seconds": int(now - t0)} for l, t0 in list(_active.values())],
})
_STATUS_WIDGET = """
"""
_PLATE_PICKER = r"""
"""
_LETTER_SYNC = r"""
"""
def _with_status(html):
extra = '' + _STATUS_WIDGET + _PLATE_PICKER + _LETTER_SYNC
i = html.lower().rfind(" ")
return html[:i] + extra + html[i:] if i != -1 else html + extra
# ---------------------------------------------------------------------------
# Image enhancement tool (/enhance) — Amped-style filter chain for any photo.
# ---------------------------------------------------------------------------
_ENHANCE_CACHE = {} # token -> (PIL image, sha256); last few uploads only
@app.route("/enhance")
def enhance_page():
path = os.path.join(app.root_path, "templates", "enhance.html")
with open(path, "r", encoding="utf-8") as f:
return f.read(), 200, {"Content-Type": "text/html; charset=utf-8"}
_VIDEO_CACHE = {} # vtoken -> {"path", "sha256", "info"}; last 2 videos only
_ENHANCE_TMP = os.path.join(_PROJECT_ROOT, "cache", "enhance_tmp")
def _jpeg(rgb_or_pil, max_side=1280, quality=88):
import io
from PIL import Image
from flask import send_file
im = rgb_or_pil if isinstance(rgb_or_pil, Image.Image) else Image.fromarray(rgb_or_pil)
if max(im.size) > max_side:
s = max_side / max(im.size)
im = im.resize((int(im.width * s), int(im.height * s)), Image.LANCZOS)
buf = io.BytesIO()
im.convert("RGB").save(buf, "JPEG", quality=quality)
buf.seek(0)
r = send_file(buf, mimetype="image/jpeg")
r.headers["Cache-Control"] = "no-store"
return r
def _enhance_video_upload(file):
import uuid
os.makedirs(_ENHANCE_TMP, exist_ok=True)
ext = (file.filename or "video.mp4").rsplit(".", 1)[-1].lower()
vtoken = uuid.uuid4().hex
path = os.path.join(_ENHANCE_TMP, f"{vtoken}.{ext if ext.isalnum() else 'mp4'}")
file.save(path)
try:
meta = video_enhancer.info(path)
except Exception as e:
video_enhancer.cleanup(path)
return jsonify({"error": f"Could not read this video: {e}"}), 400
_VIDEO_CACHE[vtoken] = {"path": path, "sha256": video_enhancer.sha256_file(path), "info": meta}
while len(_VIDEO_CACHE) > 2:
old = _VIDEO_CACHE.pop(next(iter(_VIDEO_CACHE)))
video_enhancer.cleanup(old["path"])
return jsonify({"kind": "video", "vtoken": vtoken, "sha256": _VIDEO_CACHE[vtoken]["sha256"], **meta})
@app.route("/api/enhance/video/frame")
def enhance_video_frame():
v = _VIDEO_CACHE.get(request.args.get("vtoken", ""))
if not v:
return jsonify({"error": "Video expired - please load it again"}), 404
try:
return _jpeg(video_enhancer.frame(v["path"], int(request.args.get("i", 0))))
except Exception as e:
return jsonify({"error": str(e)}), 400
@app.route("/api/enhance/video/average", methods=["POST"])
def enhance_video_average():
"""Range Selector + Local Stabilization + Frame Averaging. The merged
frame becomes a normal image token, so every image filter applies."""
import uuid
from PIL import Image
body = request.get_json(silent=True) or {}
v = _VIDEO_CACHE.get(body.get("vtoken", ""))
if not v:
return jsonify({"error": "Video expired - please load it again"}), 404
roi = body.get("roi")
if roi is not None and (not isinstance(roi, list) or len(roi) != 4):
roi = None
t0 = time.perf_counter()
try:
merged, ref, stats = video_enhancer.average(
v["path"], body.get("start", 0), body.get("end", v["info"]["frames"] - 1),
roi=roi, stabilize=bool(body.get("stabilize", True)),
method="median" if body.get("method") == "median" else "mean")
except Exception as e:
return jsonify({"error": str(e)}), 400
print(f"[enhance] video average {stats} in {time.perf_counter() - t0:.1f}s", flush=True)
token = uuid.uuid4().hex
_ENHANCE_CACHE[token] = (Image.fromarray(merged), v["sha256"])
_ENHANCE_REF[token] = Image.fromarray(ref)
while len(_ENHANCE_CACHE) > 4:
k = next(iter(_ENHANCE_CACHE))
_ENHANCE_CACHE.pop(k)
_ENHANCE_REF.pop(k, None)
return jsonify({"token": token, "width": merged.shape[1], "height": merged.shape[0], "stats": stats})
_ENHANCE_REF = {} # token -> single reference frame with the same crop (the "before")
@app.route("/api/enhance/ref/