import html
import time
import csv
from collections import deque
import json
import os
import tempfile
import traceback
from pathlib import Path
import gradio as gr
from pipeline.orchestrator import analyse
# Keep a short in-memory history of recent analyses (most recent first)
HISTORY: deque = deque(maxlen=5)
HISTORY_PATH = Path(__file__).resolve().parent / ".lth_history.json"
def _load_history():
try:
if HISTORY_PATH.exists():
with open(HISTORY_PATH, "r", encoding="utf-8") as fh:
arr = json.load(fh)
# maintain order most recent first
HISTORY.clear()
for item in arr[:HISTORY.maxlen]:
HISTORY.append(item)
except Exception:
pass
# Load existing history on import
_load_history()
def _save_history():
try:
with open(HISTORY_PATH, "w", encoding="utf-8") as fh:
json.dump(list(HISTORY), fh, ensure_ascii=False, indent=2)
except Exception:
pass
def export_json_to_pdf(json_path: str) -> str:
# Minimal PDF export using reportlab: write summary and key fields
from reportlab.lib.pagesizes import A4
from reportlab.lib.units import mm
from reportlab.pdfgen import canvas
if not json_path:
raise ValueError("No JSON path provided")
with open(json_path, "r", encoding="utf-8") as fh:
data = json.load(fh)
pdf_path = Path(tempfile.gettempdir()) / f"luxury_truth_lens_report_{int(time.time())}.pdf"
c = canvas.Canvas(str(pdf_path), pagesize=A4)
w, h = A4
margin = 20 * mm
x = margin
y = h - margin
# Title
c.setFont("Helvetica-Bold", 18)
c.drawString(x, y, "Luxury Truth Lens — Report")
y -= 12 * mm
# Layers
l2 = data.get("layer2", {})
l3 = data.get("layer3", {})
l4 = data.get("layer4", {})
l5 = data.get("layer5", {})
c.setFont("Helvetica-Bold", 12)
c.drawString(x, y, "Brand:")
c.setFont("Helvetica", 12)
c.drawString(x + 40 * mm, y, str(l2.get("brand", "-")))
y -= 8 * mm
c.setFont("Helvetica-Bold", 12)
c.drawString(x, y, "Category:")
c.setFont("Helvetica", 12)
c.drawString(x + 40 * mm, y, str(l2.get("category", "-")))
y -= 8 * mm
c.setFont("Helvetica-Bold", 12)
c.drawString(x, y, "Confidence:")
c.setFont("Helvetica", 12)
c.drawString(x + 40 * mm, y, f"{l3.get('confidence_score', 0)}/100")
y -= 12 * mm
c.setFont("Helvetica-Bold", 12)
c.drawString(x, y, "Provenance:")
c.setFont("Helvetica", 12)
c.drawString(x + 40 * mm, y, str(l4.get("provenance_status", "-")))
y -= 12 * mm
# Actions
c.setFont("Helvetica-Bold", 12)
c.drawString(x, y, "Actions:")
y -= 8 * mm
c.setFont("Helvetica", 11)
actions = l5.get("actions", [])
if isinstance(actions, list):
for act in actions:
text = act["text"] if isinstance(act, dict) and act.get("text") else str(act)
# wrap
for chunk in [text[i:i+80] for i in range(0, len(text), 80)]:
if y < margin + 20 * mm:
c.showPage()
y = h - margin
c.drawString(x + 6 * mm, y, chunk)
y -= 6 * mm
c.showPage()
c.save()
return str(pdf_path)
def _confidence_breakdown_html(l3: dict, l2: dict, l4: dict) -> str:
# Accepts layer3 dict and builds a 3-component breakdown: visual, caption, provenance
vs = int(l3.get("visual_similarity", l3.get("confidence_score", 0) * 0.6))
ct = int(l2.get("confidence", 0) * 100 * 0.3) if l2.get("confidence") is not None else int((l3.get("confidence_score", 0)) * 0.2)
pv = int(l4.get("match_score", 0)) if l4.get("match_score") is not None else 0
# Normalize to max 100
vs = min(100, vs)
ct = min(100, ct)
pv = min(100, pv)
return (
'
'
f'
Visual similarity
{vs}%
'
f'
Caption match
{ct}%
'
f'
Provenance match
{pv}%
'
'
'
)
def _risk_matrix_html(source_type: str, score: int) -> str:
# Map source type to an X coordinate (0 left safe, 100 right risky)
src = (source_type or "").lower()
if "ai" in src or "generated" in src:
x = 85
elif "screenshot" in src:
x = 60
elif "render" in src:
x = 70
else:
x = 20
# Y coordinate from confidence (low confidence => high risk on Y)
y = 100 - score
# Constrain
x = max(5, min(95, x))
y = max(5, min(95, y))
# Simple SVG 120x120 with grid and dot
svg = (
f''
)
return f'
{svg}
'
def export_json_to_csv(json_path: str) -> str:
if not json_path:
raise ValueError("No JSON path provided")
with open(json_path, "r", encoding="utf-8") as fh:
data = json.load(fh)
rows = []
l1 = data.get("layer1", {})
l2 = data.get("layer2", {})
l3 = data.get("layer3", {})
l4 = data.get("layer4", {})
l5 = data.get("layer5", {})
rows.append(
{
"timestamp": time.time(),
"brand": l2.get("brand"),
"category": l2.get("category"),
"source_type": l1.get("source_type"),
"confidence_score": l3.get("confidence_score"),
"signal_label": l3.get("signal_label"),
"provenance_status": l4.get("provenance_status"),
"actions": " | ".join(l5.get("actions", [])),
}
)
csv_path = Path(tempfile.gettempdir()) / f"luxury_truth_lens_report_{int(time.time())}.csv"
with open(csv_path, "w", newline="", encoding="utf-8") as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=rows[0].keys())
writer.writeheader()
for r in rows:
writer.writerow(r)
return str(csv_path)
TOP_DISCLAIMER = (
"Research tool, not a substitute for professional authentication. "
"Do not rely on it alone for high-value purchase decisions."
)
def _example_paths():
base = Path(__file__).resolve().parent / "examples"
if not base.is_dir():
return []
return [
[str(path)]
for path in sorted(base.iterdir())
if path.suffix.lower() in {".png", ".jpg", ".jpeg", ".webp"}
]
def _hf_token_status():
token = (
os.getenv("HF_TOKEN")
or os.getenv("HUGGING_FACE_HUB_TOKEN")
or os.getenv("HUGGINGFACEHUB_API_TOKEN")
)
if token:
return "Detected"
return "Missing"
def _severity_label(severity: str) -> str:
sev = (severity or "info").lower()
labels = {
"info": "Measured confidence",
"caution": "Guarded assessment",
"warning": "Elevated risk",
"critical": "Immediate concern",
}
return labels.get(sev, "Guarded assessment")
def _severity_class(severity: str) -> str:
sev = (severity or "info").lower()
if sev not in {"info", "caution", "warning", "critical"}:
sev = "caution"
return sev
def _confidence_band(score: int) -> str:
if score >= 75:
return "high"
if score >= 45:
return "medium"
return "low"
def _confidence_tone(score: int) -> str:
if score >= 75:
return "high"
if score >= 45:
return "medium"
return "low"
def _status_badge(label: str, kind: str) -> str:
safe_kind = kind if kind in {"info", "caution", "warning", "critical", "success"} else "info"
return f'{html.escape(label)}'
def _summary_metric(label: str, value: str, tone: str = "info") -> str:
return (
f'
'
f'{html.escape(label)}'
f'{html.escape(value)}'
f'
'
)
def _confidence_visual_html(score: int, signal_label: str) -> str:
tone = _confidence_tone(score)
return (
f'
"
)
provenance_rows = [
("Status", l4["provenance_status"]),
("Reference matches", str(l4.get("db_entry_count", 0))),
]
if l4.get("match_source"):
provenance_rows.append(("Reference source", l4["match_source"]))
if l4.get("match_date"):
provenance_rows.append(("Recorded date", l4["match_date"]))
if l4.get("note"):
provenance_rows.append(("Note", l4["note"]))
md4 = "".join(
f"
{html.escape(label)}{html.escape(value)}
"
for label, value in provenance_rows
)
action_items_list = []
for item in actions:
if isinstance(item, dict):
text = item.get("text") or item.get("label") or "(action)"
evidence = item.get("evidence")
if evidence and isinstance(evidence, dict):
ev_layer = evidence.get("layer") or evidence.get("source") or ""
ev_note = evidence.get("note") or evidence.get("id") or ""
ev_html = f" (via {html.escape(ev_layer)} {html.escape(str(ev_note))})"
else:
ev_html = ""
action_items_list.append(f"
{html.escape(str(text))}{ev_html}
")
else:
action_items_list.append(f"
{html.escape(str(item))}
")
action_items = "".join(action_items_list) or "
(none)
"
md5 = f"
{action_items}
"
meta = (
f"**Token status:** `{_hf_token_status()}`\n\n"
f"**Disclaimer:** {result.get('global_disclaimer', TOP_DISCLAIMER)}"
)
json_path = Path(tempfile.gettempdir()) / "luxury_truth_lens_report.json"
with open(json_path, "w", encoding="utf-8") as handle:
json.dump(result, handle, ensure_ascii=False, indent=2)
# Record in-memory history (keep recent 5)
try:
HISTORY.appendleft(
{
"time": int(time.time()),
"path": str(json_path),
"brand": l2.get("brand"),
"score": l3.get("confidence_score"),
"severity": severity,
}
)
# persist
_save_history()
except Exception:
# non-fatal
pass
# Confidence breakdown and risk matrix (phase 3)
breakdown_html = _confidence_breakdown_html(l3, l2, l4)
matrix_html = _risk_matrix_html(l1.get("source_type", ""), l3.get("confidence_score", 0))
# Attach breakdown into md3 display and include risk matrix near summary
md3 = (
f"{_confidence_visual_html(l3['confidence_score'], l3['signal_label'])}"
f"{breakdown_html}"
f"
A restrained review surface for luxury image triage.
Submit a single frame and read five structured lenses:
origin, identity, confidence, provenance, and recommended next action.
Accepted ImageJPG, PNG, or WebP, up to 10 MB
HF TokenOptional—speeds model downloads significantly
Use CaseScreen risk quickly, then escalate to specialist review
"""
)
with gr.Row(elem_classes=["masthead-row"]):
with gr.Column(scale=18):
gr.HTML(
"""
Luxury image review
Luxury Truth Lens
Review one image at a time with a denser two-panel workspace built for fast visual triage,
provenance checks, and cleaner decision support.
Mode
Five-lens review
Canvas
Editorial workspace
"""
)
with gr.Column(scale=5):
status = gr.Markdown("Status: standing by.", elem_classes=["soft-status"])
with gr.Row(equal_height=False, elem_classes=["workspace-row"]):
with gr.Column(scale=9, min_width=440):
with gr.Group(elem_classes=["soft-card", "submission-card"]):
gr.Markdown("## Submission", elem_classes=["section-title"])
gr.Markdown(
"Upload a frame or choose a sample, then run a structured review.",
elem_classes=["section-copy"],
)
with gr.Group(elem_classes=["well"]):
img_in = gr.Image(
label="Luxury item image",
type="numpy",
height=560,
sources=["upload"],
)
gr.HTML(
"""
Accepted imageJPG, PNG, WebP up to 10 MB
UseScreen quickly, then escalate to specialist review
"""
)
examples = _example_paths()
if examples:
gr.Examples(
examples=examples,
inputs=[img_in],
label="Examples",
)
btn = gr.Button("Run Review", elem_id="analyze-btn", variant="primary")
with gr.Column(scale=14, min_width=620):
with gr.Group(elem_classes=["soft-card"]):
gr.Markdown("## Review", elem_classes=["section-title"])
gr.Markdown(
"Read the top-line judgment first, then move through the five supporting lenses.",
elem_classes=["section-copy"],
)
with gr.Group(elem_classes=["summary-box"]):
summary = gr.HTML("
Submit an image to generate a review.
")
with gr.Group(elem_classes=["lens-stack"]):
with gr.Accordion("Lens I Origin", open=True):
with gr.Group(elem_classes=["layer-card"]):
out1 = gr.Markdown()
gr.Markdown("*How the image appears to have been produced, and how certain that read is.*", elem_classes=["layer-note"])
with gr.Accordion("Lens II Identity", open=True):
with gr.Group(elem_classes=["layer-card"]):
out2 = gr.Markdown()
gr.Markdown("*Brand, category, caption, and alternate interpretations from the model.*", elem_classes=["layer-note"])
with gr.Accordion("Lens III Confidence", open=True):
with gr.Group(elem_classes=["layer-card"]):
out3 = gr.Markdown()
gr.Markdown("*Visual confidence score and supporting signal. Not a professional authentication result.*", elem_classes=["layer-note"])
with gr.Accordion("Lens IV Provenance", open=True):
with gr.Group(elem_classes=["layer-card"]):
out4 = gr.Markdown()
gr.Markdown("*Reference lookups against known flagged entries and stored provenance notes.*", elem_classes=["layer-note"])
with gr.Accordion("Lens V Actions", open=True):
with gr.Group(elem_classes=["layer-card"]):
out5 = gr.Markdown()
gr.Markdown("*Recommended follow-up actions shaped by the full review.*", elem_classes=["layer-note"])
meta = gr.Markdown(
f"**Token status:** `{_hf_token_status()}`\n\n**Disclaimer:** {TOP_DISCLAIMER}",
elem_classes=["soft-footer"],
)
with gr.Group(elem_classes=["download-box"]):
json_file = gr.File(label="Download JSON review")
pdf_button = gr.Button("Export PDF Review")
pdf_file = gr.File(label="Download PDF review")
recent = gr.Dropdown(choices=[], label="Recent reviews", interactive=True)
recent_summary = gr.HTML("", visible=True)
btn.click(
fn=run_analysis,
inputs=[img_in],
outputs=[summary, out1, out2, out3, out4, out5, meta, status, json_file, recent, recent_summary],
api_name="analyze",
show_progress="full",
)
def _export_pdf(path: str):
return export_json_to_pdf(path)
pdf_button.click(fn=_export_pdf, inputs=[json_file], outputs=[pdf_file])
def _load_recent(selected_label: str):
if not selected_label:
return ""
# Find matching history entry by label
hist = list(HISTORY)
target = None
for item in hist:
label = f"{item.get('brand') or 'unknown'} - {item.get('score')}/100"
if label == selected_label or selected_label.startswith(label):
target = item
break
if target is None:
return ""
try:
with open(target["path"], "r", encoding="utf-8") as fh:
data = json.load(fh)
except Exception as exc:
return f"