# app.py — Diag Studio (Customer UI) # AI Destekli Araç Motor Ses Analizi — sade ticari arayüz # İstekler: Ses oynatıcı, ilerleme çubuğu, kullanıcı dostu akış. # Teknik sağlayıcı/kitaplık adları ve loglar UI’da gösterilmez. # Yönetici ayarları yalnızca DEV_MODE=1 iken görünür. import os import io import json import time import requests import streamlit as st from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry # ---------------- Safe defaults ---------------- os.environ.setdefault("HOME", "/tmp") os.environ.setdefault("STREAMLIT_BROWSER_GATHER_USAGE_STATS", "false") os.environ.setdefault("STREAMLIT_SERVER_HEADLESS", "true") # ---------------- Theme ---------------- try: st._config.set_option("theme.base", "dark") st._config.set_option("theme.primaryColor", "#ff7a1a") st._config.set_option("theme.backgroundColor", "#0e0f12") st._config.set_option("theme.secondaryBackgroundColor", "#171a1f") st._config.set_option("theme.textColor", "#e6e8ea") except Exception: pass # ---------------- Env helpers ---------------- def _env(k: str) -> str: return (os.getenv(k) or "").strip().strip("'\"") DEV_MODE = (_env("DEV_MODE") in {"1", "true", "yes", "on"}) API_BASE_DEFAULT = (_env("CARDIAG_AI_API") or _env("APP_API") or "http://127.0.0.1:7860").rstrip("/") TOKEN = _env("CARDIAG_AI_TOKEN") or _env("APP_TOKEN") HEADERS = {"Authorization": f"Bearer {TOKEN}"} if TOKEN else {} # ---------------- i18n ---------------- LANGS = { "tr": { "title": "AI Destekli Araç Motor Ses Analizi", "about": "Motor ses kaydını yükleyin veya bağlantı girin. Sistem sesi analiz eder, olası arızaları ve önerileri özetler.", "upload": "Ses dosyası yükle", "or": "veya", "paste_url": "Ses bağlantısı (opsiyonel)", "analyze": "Analiz Et", "reset": "Sıfırla", "preview": "Önizleme", "result": "Sonuç", "commentary": "Özet ve Öneriler", "json": "Detaylı Bulgular", "artifacts": "Çıktılar", "download_summary": "Özet indir", "download_parsed": "Detay indir (JSON)", "download_raw": "Ham metin indir", "upload_hint": "mp3 / wav / ogg önerilir. İsterseniz herkese açık bir bağlantı da girebilirsiniz.", "warming": "Hazırlanıyor…", "analyzing": "Analiz ediliyor…", "health_ok": "Sistem hazır.", "health_fail": "Sisteme erişilemedi.", "need_backend": "Sistem adresi yapılandırılmadı.", "need_input": "Lütfen bir dosya yükleyin veya bağlantı girin.", "error_generic": "İşlem başarısız. Lütfen tekrar deneyin.", "try_again": "Yoğunluk olabilir. Biraz sonra tekrar deneyin.", "timeline_title": "İşlem Durumu", "step_prepare": "Hazırlık", "step_upload": "Yükleme", "step_process": "Analiz", "step_summarize": "Özet oluşturma", "admin": "Yönetici", "admin_desc": "Bu bölüm yalnızca yetkili kullanım içindir.", "backend_url": "Sistem adresi", "use_data_uri": "Gizli yükleme", "warmup": "Ön ısıtma", "auto_retry": "Otomatik yeniden dene", "max_wait": "Azami bekleme (sn)", "show_debug": "Hata ayıklama bilgileri", }, "en": { "title": "AI-Powered Engine Sound Analysis", "about": "Upload an engine sound or provide a public URL. The system analyzes the audio and summarizes likely issues and next steps.", "upload": "Upload audio file", "or": "or", "paste_url": "Audio URL (optional)", "analyze": "Analyze", "reset": "Reset", "preview": "Preview", "result": "Result", "commentary": "Summary & Recommendations", "json": "Detailed Findings", "artifacts": "Downloads", "download_summary": "Download summary", "download_parsed": "Download details (JSON)", "download_raw": "Download raw text", "upload_hint": "mp3 / wav / ogg recommended. You may paste a public URL instead of uploading.", "warming": "Preparing…", "analyzing": "Analyzing…", "health_ok": "Service is ready.", "health_fail": "Cannot reach the service.", "need_backend": "Service URL is not configured.", "need_input": "Please upload a file or enter a URL.", "error_generic": "Request failed. Please try again.", "try_again": "Service may be busy. Try again shortly.", "timeline_title": "Progress", "step_prepare": "Preparing", "step_upload": "Uploading", "step_process": "Processing", "step_summarize": "Summarizing", "admin": "Admin", "admin_desc": "This section is for authorized use only.", "backend_url": "Service URL", "use_data_uri": "Private upload", "warmup": "Warm up", "auto_retry": "Auto-retry", "max_wait": "Max wait (sec)", "show_debug": "Show debug info", }, } # ---------------- Session ---------------- if "lang" not in st.session_state: st.session_state.lang = "tr" if "api_base" not in st.session_state: st.session_state.api_base = API_BASE_DEFAULT if "use_data_uri" not in st.session_state: st.session_state.use_data_uri = True if "warmup" not in st.session_state: st.session_state.warmup = True if "auto_retry" not in st.session_state: st.session_state.auto_retry = True if "max_wait" not in st.session_state: st.session_state.max_wait = 60 if "show_debug" not in st.session_state: st.session_state.show_debug = False if "prompt_override" not in st.session_state: st.session_state.prompt_override = "" if "res" not in st.session_state: st.session_state.res = None T = LANGS[st.session_state.lang] # ---------------- Header ---------------- left, right = st.columns([0.7, 0.3]) with left: st.title(T["title"]) with right: st.session_state.lang = st.selectbox( "Language / Dil", ["tr", "en"], index=(0 if st.session_state.lang == "tr" else 1), ) T = LANGS[st.session_state.lang] st.caption(T["about"]) # ---------------- Admin (DEV only) ---------------- if DEV_MODE: with st.expander(f"🛠 {T['admin']}", expanded=False): st.write(T["admin_desc"]) st.session_state.api_base = st.text_input(T["backend_url"], value=st.session_state.api_base) st.session_state.use_data_uri = st.checkbox(T["use_data_uri"], value=st.session_state.use_data_uri) st.session_state.warmup = st.checkbox(T["warmup"], value=st.session_state.warmup) st.session_state.auto_retry = st.checkbox(T["auto_retry"], value=st.session_state.auto_retry) st.session_state.max_wait = st.slider(T["max_wait"], 10, 240, st.session_state.max_wait, step=10) st.session_state.show_debug = st.checkbox(T["show_debug"], value=st.session_state.show_debug) st.session_state.prompt_override = st.text_area("Advanced prompt (optional)", value=st.session_state.prompt_override, height=120) st.markdown("---") # ---------------- Upload / URL ---------------- st.write(T["upload_hint"]) file = st.file_uploader(T["upload"], type=["mp3", "wav", "ogg"]) st.write(f"**{T['or']}**") audio_url = st.text_input(T["paste_url"], value="", placeholder="https://...") # Audio preview block (player) if file is not None: st.markdown(f"**{T['preview']}**") st.audio(file, format=file.type or "audio/mpeg") elif audio_url.strip(): st.markdown(f"**{T['preview']}**") st.audio(audio_url) # ---------------- Controls ---------------- c1, c2, c3 = st.columns([0.22, 0.22, 0.56]) go = c1.button(f"▶ {T['analyze']}", type="primary") reset = c2.button(f"↩ {T['reset']}") health = c3.button("🔄 Health") # ---------------- HTTP helpers ---------------- def build_session(auto_retry: bool) -> requests.Session: s = requests.Session() s.headers.update(HEADERS) if auto_retry: retry = Retry( total=6, connect=6, read=6, backoff_factor=0.8, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=frozenset(["GET", "POST"]), raise_on_status=False, ) adapter = HTTPAdapter(max_retries=retry) s.mount("http://", adapter) s.mount("https://", adapter) return s def warm_up(sess: requests.Session, base: str, progress_cb=None, total_budget: int = 25): start = time.time() paths = ["/", "/health"] for i, path in enumerate(paths, 1): try: _ = sess.get(f"{base.rstrip('/')}{path}", timeout=8) except Exception: pass if progress_cb: progress_cb(min(int(i/len(paths)*18), 18)) # 0-18% if time.time() - start > total_budget: break time.sleep(0.4) def fetch_bytes(sess: requests.Session, base: str, url: str) -> bytes: full = url if url.startswith("http") else f"{base.rstrip('/')}{url}" r = sess.get(full, timeout=90) r.raise_for_status() return r.content def post_analyze(sess: requests.Session, base: str, _file, _url: str, progress_cb=None) -> dict: data = { "use_data_uri": "true" if st.session_state.use_data_uri else "false", "lang": st.session_state.lang, } if DEV_MODE and st.session_state.prompt_override.strip(): data["prompt"] = st.session_state.prompt_override.strip() files = None if _file is not None: # read into BytesIO for repeatable upload content = _file.read() buf = io.BytesIO(content) mime = getattr(_file, "type", None) or "application/octet-stream" files = {"file": (_file.name, buf, mime)} if progress_cb: progress_cb(30) # after upload prep elif _url.strip(): data["audio_url"] = _url.strip() if not files and "audio_url" not in data: return {"ok": False, "error": T["need_input"]} url = f"{base.rstrip('/')}/analyze" deadline = time.time() + st.session_state.max_wait last_err = None while True: try: if progress_cb: progress_cb(35) # request start resp = sess.post(url, data=data, files=files, timeout=240) if progress_cb: progress_cb(78) # server responded resp.raise_for_status() if progress_cb: progress_cb(92) # parse return resp.json() except Exception as e: last_err = e if time.time() > deadline or not st.session_state.auto_retry: return {"ok": False, "error": f"{T['error_generic']} ({last_err})"} time.sleep(1.2) # ---------------- Actions ---------------- sess = build_session(st.session_state.auto_retry) if health: try: r = sess.get(f"{st.session_state.api_base.rstrip('/')}/health", timeout=10) st.success(T["health_ok"] if r.ok else T["health_fail"]) if DEV_MODE and r.ok: st.code(json.dumps(r.json(), ensure_ascii=False, indent=2), language="json") except Exception: st.error(T["health_fail"]) if reset: st.session_state.res = None st.rerun() # Progress widgets (player stays above) progress_box = st.container() timeline = progress_box.expander(T["timeline_title"], expanded=True) p_text = progress_box.empty() p_bar = progress_box.progress(0) def progress_update(pct: int, label: str = ""): p_bar.progress(min(max(pct, 0), 100)) if label: p_text.write(label) if go: if not st.session_state.api_base: st.error(T["need_backend"]) elif not file and not audio_url.strip(): st.error(T["need_input"]) else: # timeline with timeline: st.write(f"1) {T['step_prepare']}") st.write(f"2) {T['step_upload']}") st.write(f"3) {T['step_process']}") st.write(f"4) {T['step_summarize']}") # warmup (optional) if st.session_state.warmup: progress_update(5, T["warming"]) warm_up(sess, st.session_state.api_base, progress_cb=lambda v: progress_update(v, T["warming"])) # analyze progress_update(22, T["analyzing"]) res = post_analyze( sess, st.session_state.api_base, file, audio_url, progress_cb=lambda v: progress_update(v, T["analyzing"]) ) st.session_state.res = res progress_update(100, "") # ---------------- Results ---------------- res = st.session_state.res if res: if not res.get("ok", True) and res.get("error"): st.error(T["error_generic"]) st.write(res.get("error")) st.info(T["try_again"]) if DEV_MODE and st.session_state.show_debug and res.get("fal_debug"): st.expander("Debug").code(json.dumps(res["fal_debug"], ensure_ascii=False, indent=2), language="json") else: st.subheader(T["result"]) # Summary for drivers comm = (res.get("commentary") or "").strip() if comm: st.markdown(f"**{T['commentary']}**") st.write(comm) # Detailed findings (JSON). This is the structured diagnosis — no provider names. parsed = res.get("fal_parsed") if parsed: st.markdown(f"**{T['json']}**") st.json(parsed) # Downloads dls = res.get("downloads") or {} if any(dls.get(k) for k in ("summary_txt", "fal_parsed_json", "fal_raw_txt")): st.markdown(f"**{T['artifacts']}**") col_a, col_b, col_c = st.columns(3) if dls.get("summary_txt"): try: data = fetch_bytes(sess, st.session_state.api_base, dls["summary_txt"]) col_a.download_button(label=T["download_summary"], data=data, file_name="ozet.txt" if st.session_state.lang=="tr" else "summary.txt") except Exception: col_a.error("✖") if dls.get("fal_parsed_json"): try: data = fetch_bytes(sess, st.session_state.api_base, dls["fal_parsed_json"]) col_b.download_button(label=T["download_parsed"], data=data, file_name="detay.json" if st.session_state.lang=="tr" else "details.json", mime="application/json") except Exception: col_b.error("✖") if dls.get("fal_raw_txt"): try: data = fetch_bytes(sess, st.session_state.api_base, dls["fal_raw_txt"]) col_c.download_button(label=T["download_raw"], data=data, file_name="ham.txt" if st.session_state.lang=="tr" else "raw.txt") except Exception: col_c.error("✖") # ---------------- Footer ---------------- st.caption("© 2025 Diag Studio — Araç Motor Ses Analizi")