# streamlit_app.py — AI LightBox · Jewelry (dark-orange UI, fixed video refresh)
# Pre-run sabit 4 kart (Input/Packshot/Padded/Video loop) + Run sonrası native odaklı kartlar
import os, io, base64, zipfile, requests, streamlit as st
from PIL import Image
# ===== Theme (programmatic dark orange) =====
try:
st._config.set_option("theme.base", "dark")
st._config.set_option("theme.primaryColor", "#ff7a1a") # dark orange
st._config.set_option("theme.backgroundColor", "#0e0f12") # app bg
st._config.set_option("theme.secondaryBackgroundColor", "#161a1f")
st._config.set_option("theme.textColor", "#e7e9ed")
except Exception:
pass
# ================= Env & API =================
API_BASE = (os.getenv("AI_LIGHTBOX_API", "http://127.0.0.1:8000") or "http://127.0.0.1:8000").strip().strip("'\"").rstrip("/")
HF_TOKEN = (os.getenv("AI_LIGHTBOX_TOKEN", "") or "").strip().strip("'\"")
HEADERS = {"Authorization": f"Bearer {HF_TOKEN}"} if HF_TOKEN else {}
# UI vars
VIDEO_MAX_PX_DEFAULT = int(os.getenv("VIDEO_MAX_PX", "720"))
VIDEO_RES_OPTIONS = ["512P", "768P"]
VIDEO_DUR_OPTIONS = ["6", "10"]
ASPECT_OPTIONS = ["9:16", "3:4", "1:1"] # preview kadraj oranları
# ================= Session =================
if "results" not in st.session_state:
st.session_state["results"] = None
if "job_counter" not in st.session_state:
st.session_state["job_counter"] = 0
if "video_ui_max_px" not in st.session_state:
st.session_state["video_ui_max_px"] = VIDEO_MAX_PX_DEFAULT
# ================= HTTP/Backend helpers =================
def _needs_auth(url: str) -> bool:
return url.startswith(API_BASE) or url.startswith("outputs/") or url.startswith("/outputs/")
def _abs_backend_url(url_or_path: str) -> str:
if not url_or_path:
return ""
if url_or_path.startswith(("http://","https://")):
return url_or_path
if url_or_path.startswith("/outputs/"):
return f"{API_BASE}{url_or_path}"
if url_or_path.startswith("outputs/"):
return f"{API_BASE}/{url_or_path}"
return f"{API_BASE}/{url_or_path.lstrip('/')}"
def backend_ok() -> bool:
try:
r = requests.get(f"{API_BASE}/health", headers=HEADERS, timeout=10)
r.raise_for_status()
return True
except Exception:
return False
def post_chain(data: dict, files_payload):
r = requests.post(f"{API_BASE}/v1/tryon/chain", data=data, files=files_payload or None,
headers=HEADERS, timeout=600)
r.raise_for_status()
return r.json()
def post_video(image_url: str, duration="6", resolution="768P", prompt_optimizer=False, preview_aspect="9:16"):
payload = {
"image_url": image_url,
"duration": duration,
"resolution": resolution,
"prompt_optimizer": "true" if prompt_optimizer else "false",
"preview_aspect": preview_aspect,
}
r = requests.post(f"{API_BASE}/v1/video/from-image", data=payload, headers=HEADERS, timeout=600)
r.raise_for_status()
return r.json()
@st.cache_data(ttl=300, show_spinner=False)
def fetch_bytes(url_or_path: str):
if not url_or_path: return None
try:
url = _abs_backend_url(url_or_path)
headers = HEADERS if _needs_auth(url) else None
r = requests.get(url, headers=headers, timeout=180)
r.raise_for_status()
return r.content
except Exception:
return None
def _sniff_mime_from_bytes(b: bytes) -> str:
try:
fmt = (Image.open(io.BytesIO(b)).format or "JPEG").lower()
return {"jpeg":"image/jpeg","jpg":"image/jpeg","png":"image/png","webp":"image/webp"}.get(fmt,"image/jpeg")
except Exception:
return "image/jpeg"
def image_bytes_to_data_url(b: bytes, mime: str | None = None) -> str:
if not mime:
mime = _sniff_mime_from_bytes(b)
enc = base64.b64encode(b).decode("ascii")
return f"data:{mime};base64,{enc}"
@st.cache_data(ttl=300, show_spinner=False)
def image_size_from_bytes(b: bytes):
try:
im = Image.open(io.BytesIO(b)); return im.size
except Exception:
return None
def infer_ext(data: bytes) -> str:
try:
im = Image.open(io.BytesIO(data))
fmt = (im.format or "JPEG").lower()
return "." + {"jpeg":"jpg","jpg":"jpg","png":"png","webp":"webp"}.get(fmt,"jpg")
except Exception:
if data[:4] == b"\x00\x00\x00\x18" or data[4:8] == b"ftyp":
return ".mp4"
return ".bin"
def make_zip(named_bytes: list[tuple[str, bytes]]) -> bytes:
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", compression=zipfile.ZIP_DEFLATED) as z:
for fname, b in named_bytes:
if b: z.writestr(fname, b)
buf.seek(0); return buf.read()
# ================= Preview helpers (uniform cards) =================
def _parse_aspect_str(s: str):
try:
a, b = s.split(":"); return max(1, int(a)), max(1, int(b))
except Exception:
return (9, 16)
def inject_base_css():
st.markdown("""
""", unsafe_allow_html=True)
def inject_preview_css(aspect: str, fit_mode="contain"):
aw, ah = _parse_aspect_str(aspect)
st.markdown(f"""
""", unsafe_allow_html=True)
def show_tile_card_display_download(
col, title,
display_url: str | None = None,
display_bytes: bytes | None = None,
download_url: str | None = None,
download_bytes: bytes | None = None,
filename_stub: str = "image",
key_prefix: str = "",
):
"""Gösterimde HAFİF JPEG/PNG (display_bytes) tercih edilir; indirmede NATIVE."""
col.markdown(f'
{title}
', unsafe_allow_html=True)
# ---- Display ----
if not display_url and not display_bytes:
col.markdown('', unsafe_allow_html=True)
col.caption("—"); return None, None
if display_bytes is not None:
src = image_bytes_to_data_url(display_bytes) # mime otomatik
col.markdown(f'
""", unsafe_allow_html=True
)
col_vid.download_button("Download Video (MP4)", data=vid_demo, file_name="tryon.mp4", mime="video/mp4", key="pre_video_dl")
else:
col_vid.markdown('', unsafe_allow_html=True)
col_vid.caption("—")
# ================= ÇALIŞTIR =================
if run:
if not ok:
st.error("Backend erişilemiyor. API_BASE / TOKEN kontrol edin.")
elif not (file_list or image_url or demo_files.get("input.jpg")):
st.error("En az bir görsel girişi yapın (dosya yükle veya URL).")
else:
try:
img_urls = image_url or ""
files_payload = []
if file_list:
for f in file_list:
files_payload.append(("files", (f.name, f.getvalue(), f.type or "image/jpeg")))
data = {
"category": st.session_state.get("jewel_type","auto"),
"edit_prompt": "",
"num_images": "1",
"image_urls": img_urls,
"mannequin_image_url": "",
"identity_lock": "true" if identity_lock else "false",
# Preview controls (backend pad/preview için)
"preview_aspect": preview_aspect,
"preview_long_edge": str(int(preview_long_edge)),
"preview_ratio": str(float(preview_ratio)),
"padding_ratio": str(float(preview_ratio)), # backward-compat
# SR
"upscale": "true" if upscale else "false",
"upscale_factor": st.session_state.get("upscale_factor","2"),
"upscale_stage": st.session_state.get("upscale_stage","both"),
# Video (opsiyonel zincir)
"to_video": "true" if to_video else "false",
"video_prompt": "",
"duration": st.session_state.get("duration_sel", VIDEO_DUR_OPTIONS[0]),
"resolution": st.session_state.get("resolution_sel", VIDEO_RES_OPTIONS[1]),
"prompt_optimizer": "false",
}
with st.spinner("AI Chain Running…"):
out = post_chain(data, files_payload if files_payload else None)
st.session_state["job_counter"] += 1
# ---- ÇIKTILAR ----
# Packshot NATIVE → DISPLAY küçük JPEG (şeffaflık beyaza kompoze)
pack_native_url = None; pack_display_jpeg = None
try:
pack_native_url = (out["packshot_step"]["result"]["images_native"] or [{}])[0].get("url")
if pack_native_url:
pn = fetch_bytes(pack_native_url)
if pn: pack_display_jpeg = make_small_jpeg_from_native(pn, 1280, bg_rgb=(255,255,255))
except Exception: pass
# Padded NATIVE → DISPLAY küçük JPEG (pad etkisi görünür)
padded_native_url = None; padded_display_jpeg = None
try:
padded_native_url = (out["packshot_step"]["result"]["padded_native"] or [{}])[0].get("url")
if padded_native_url:
nb = fetch_bytes(padded_native_url)
if nb: padded_display_jpeg = make_small_jpeg_from_native(nb, 1280, bg_rgb=(255,255,255))
except Exception: pass
# Model NATIVE → DISPLAY küçük JPEG
placed_native_url = None; placed_display_jpeg = None
try:
placed_native_url = (out["image_step"]["result"]["images_native"] or [{}])[0].get("url")
if placed_native_url:
mn = fetch_bytes(placed_native_url)
if mn: placed_display_jpeg = make_small_jpeg_from_native(mn, 1280, bg_rgb=(255,255,255))
except Exception: pass
# Video (zincirde üretildiyse)
vurl = None
try:
vurl = (out.get("video_step") or {}).get("result", {}).get("video", {}).get("url")
except Exception:
vurl = None
# UI ipucu
ui_hints = (out.get("video_step") or {}).get("ui_hints") or {}
if isinstance(ui_hints, dict):
st.session_state["video_ui_max_px"] = int(ui_hints.get("suggested_video_max_px", VIDEO_MAX_PX_DEFAULT))
st.session_state["results"] = {
"job_id": out.get("job_id"),
"schema_version": out.get("schema_version"),
"packshot_native_url": pack_native_url,
"packshot_display_jpeg": pack_display_jpeg,
"padded_native_url": padded_native_url,
"padded_display_jpeg": padded_display_jpeg,
"placed_native_url": placed_native_url,
"placed_display_jpeg": placed_display_jpeg,
"video_url": vurl,
"api_base": API_BASE,
"has_auth_header": bool(HEADERS),
"preview_aspect": preview_aspect,
}
except requests.HTTPError as e:
st.error(f"HTTP {e.response.status_code}")
except Exception as e:
st.error(f"Hata: {e}")
# ================= SONUÇLAR =================
res = st.session_state.get("results")
vbytes = None
if res:
st.subheader("Product Results")
c_pack, c_pad, c_model, c_vid = st.columns(4)
# Packshot (backend) — DISPLAY = küçük JPEG, DOWNLOAD = NATIVE
show_tile_card_display_download(
c_pack, "Packshot (Removed BG)",
display_bytes=res.get("packshot_display_jpeg"),
download_url=res.get("packshot_native_url"),
filename_stub="packshot_native",
key_prefix=f"job{st.session_state['job_counter']}"
)
# Padded (backend) — DISPLAY = padded küçük JPEG, DOWNLOAD = padded NATIVE
show_tile_card_display_download(
c_pad, "Padded",
display_bytes=res.get("padded_display_jpeg"),
download_url=res.get("padded_native_url"),
filename_stub="padded_backend",
key_prefix=f"job{st.session_state['job_counter']}"
)
# Model üzerinde (backend) — DISPLAY = küçük JPEG (native’den), DOWNLOAD = NATIVE
show_tile_card_display_download(
c_model, "TryON Model",
display_bytes=res.get("placed_display_jpeg"),
download_url=res.get("placed_native_url"),
filename_stub="model_native",
key_prefix=f"job{st.session_state['job_counter']}"
)
# Video
vurl = res.get("video_url")
c_vid.markdown('
AI Generated Video
', unsafe_allow_html=True)
if vurl:
abs_v = _abs_backend_url(vurl)
c_vid.markdown(
f"""
""", unsafe_allow_html=True
)
vb = fetch_bytes(abs_v)
if vb:
vbytes = vb
c_vid.download_button("Download Video (MP4)", data=vb, file_name="ai_lightbox_video.mp4",
mime="video/mp4", key=f"job{st.session_state['job_counter']}_video_dl")
else:
c_vid.markdown('', unsafe_allow_html=True)
# ---- Generate video on demand ----
# unique key by job to avoid stale checkbox value on rerun
vid_checkbox_key = f"video_toggle_job{st.session_state['job_counter']}"
if st.checkbox("Generate Video From This Result", value=False, key=vid_checkbox_key):
src_for_video = res.get("placed_native_url") or res.get("packshot_native_url")
if not src_for_video:
st.warning("Video için kullanılacak native görsel bulunamadı.")
else:
with st.spinner("Generating Video …"):
try:
v = post_video(
src_for_video,
duration=st.session_state.get("duration_sel", VIDEO_DUR_OPTIONS[0]),
resolution=st.session_state.get("resolution_sel", VIDEO_RES_OPTIONS[1]),
preview_aspect=st.session_state.get("preview_aspect","9:16"),
prompt_optimizer=False,
)
except requests.HTTPError as e:
st.error(f"HTTP {e.response.status_code}")
v = None
if v:
vurl2 = (v.get("result") or {}).get("video", {}).get("url")
ui_hints = v.get("ui_hints") or {}
if vurl2:
# write to session and force rerun so the video appears immediately
st.session_state["results"]["video_url"] = vurl2
if isinstance(ui_hints, dict):
st.session_state["video_ui_max_px"] = int(ui_hints.get("suggested_video_max_px", VIDEO_MAX_PX_DEFAULT))
# reset checkbox before rerun to prevent re-trigger loop
st.session_state.pop(vid_checkbox_key, None)
st.success("Video is Ready! Showing preview…")
st.rerun()
else:
st.info("Video URL gelmedi.")
# Native ZIP
named = []
pn = fetch_bytes(res.get("packshot_native_url")) if res.get("packshot_native_url") else None
mn = fetch_bytes(res.get("placed_native_url")) if res.get("placed_native_url") else None
if pn: named.append((f"packshot_native{infer_ext(pn)}", pn))
if mn: named.append((f"model_native{infer_ext(mn)}", mn))
if vbytes: named.append(("video.mp4", vbytes))
if named:
zip_bytes = make_zip(named)
st.download_button("Download All (ZIP)", data=zip_bytes,
file_name="ai_lightbox_outputs.zip", mime="application/zip",
key=f"job{st.session_state['job_counter']}_zip_dl")
# ================= Sidebar =================
with st.sidebar:
st.caption(f"API_BASE: {API_BASE}")
if HF_TOKEN: st.caption("Auth: Bearer (aktif)")
st.slider("Video önizleme genişliği (px)", 360, 1080, st.session_state["video_ui_max_px"],
step=10, key="video_ui_max_px")
if st.button("Önbelleği temizle"): st.cache_data.clear(); st.success("Önbellek temizlendi.")
if st.button("Son çıktıları temizle"): st.session_state["results"] = None; st.success("Çıktılar temizlendi.")