"""Upwork Proposal Strategist — Streamlit entry point. End users never run this directly — they double-click the launcher (see README), which opens the app in their web browser. Developers launch it via ``python desktop_app.py`` (same thing — starts the server and opens the browser) or, after ``pip install -e .``, ``python -m streamlit run app/main.py`` — no PYTHONPATH needed in either case, because the project is an installable package and the launcher puts the project root on ``sys.path``. See DEVELOPER.md. """ from __future__ import annotations import streamlit as st from app.config import ( APP_TITLE, APP_TAGLINE, get_settings, reload_settings, get_saved_dossier_path, save_dossier_path, apply_session_overrides, ) from app.services import browser_tag, key_store from app.services.api_gate import ApiGateError, run_capability_test from app.services.folder_validator import validate from app.services.dossier_reader import read_dossier from app.services.evidence_index import build_evidence_index from app.ui import ( setup_screen, dossier_screen, screenshot_screen, confirmation_screen, analysis_screen, proposal_screen, ) from app.ui import api_usage_panel, theme # (step key, sidebar label) in flow order. STEPS: list[tuple[str, str]] = [ ("setup", "Setup"), ("dossier", "Dossier"), ("screenshot", "Job Screenshot"), ("confirmation", "Confirm Details"), ("analysis", "Analysis"), ("proposal", "Proposal"), ] # "Confirm Details" is a developer/admin-only screen. Normal users get a # clean Setup → Dossier → Job Screenshot → Analysis → Proposal flow; the # extracted job fields are confirmed automatically in the backend right # after extraction, so these steps are hidden unless SHOW_DEBUG_PANEL=true. DEBUG_ONLY_STEPS: frozenset[str] = frozenset({"confirmation"}) def _visible_steps(show_debug: bool) -> list[tuple[str, str]]: """Return the steps shown in the sidebar for the current mode.""" if show_debug: return list(STEPS) return [(key, label) for key, label in STEPS if key not in DEBUG_ONLY_STEPS] RENDERERS = { "setup": setup_screen.render, "dossier": dossier_screen.render, "screenshot": screenshot_screen.render, "confirmation": confirmation_screen.render, "analysis": analysis_screen.render, "proposal": proposal_screen.render, } SESSION_DEFAULTS: dict[str, object] = { # Setup gate "api_status": None, "api_ok": False, # Theme (light default) "theme_dark": False, # Navigation "current_step": "setup", # Dossier flow "dossier_folder_path": "", "dossier_validation": None, "dossier_chunks": None, "dossier_read": False, "dossier_read_error": False, "evidence_index": None, "canonical_profile": None, "evidence_index_meta": None, # Screenshots "screenshots_uploaded": False, "uploaded_screenshots": [], "pasted_screenshots": [], "screenshot_upload_sig": (), "extracted_job_fields": None, # Confirmation "confirmed_job_fields": None, "fields_confirmed": False, "current_job_fingerprint": None, # Analysis output "scoring_result": None, "recommendation_result": None, "match_data": None, "generated_proposal": None, "verified_proposal": None, "proposal_context": None, "selected_evidence_for_proposal": None, # API usage tracking "api_usage_log": [], } def _init_session_state() -> None: for key, value in SESSION_DEFAULTS.items(): if key not in st.session_state: st.session_state[key] = value def _is_unlocked(step_key: str) -> bool: if step_key == "setup": return True if step_key == "dossier": return bool(st.session_state.get("api_ok")) if step_key == "screenshot": return bool(st.session_state.get("evidence_index")) if step_key == "confirmation": return bool(st.session_state.get("extracted_job_fields")) if step_key == "analysis": return bool(st.session_state.get("fields_confirmed")) if step_key == "proposal": return bool( st.session_state.get("fields_confirmed") and st.session_state.get("recommendation_result") ) return False def _is_completed(step_key: str) -> bool: if step_key == "setup": return bool(st.session_state.get("api_ok")) if step_key == "dossier": return bool(st.session_state.get("evidence_index")) if step_key == "screenshot": return bool(st.session_state.get("extracted_job_fields")) if step_key == "confirmation": return bool(st.session_state.get("fields_confirmed")) if step_key == "analysis": return bool(st.session_state.get("recommendation_result")) if step_key == "proposal": return bool(st.session_state.get("verified_proposal")) return False def _lock_reason(step_key: str) -> str: if step_key == "dossier": return "Locked — run the API check first." if step_key == "screenshot": return "Locked — read your dossier and build proof points first." if step_key == "confirmation": return "Locked — extract job details from a screenshot first." if step_key == "analysis": return "Locked — confirm the job details first." if step_key == "proposal": return "Locked — run the analysis first." return "Locked." def _step_status(step_key: str, current: str) -> str: if step_key == current: return "current" if not _is_unlocked(step_key): return "locked" if _is_completed(step_key): return "completed" return "available" # Quiet glyphs that read as a clean status, not technical noise. _STATUS_GLYPH = { "completed": "✓", "current": "●", "available": "○", "locked": "🔒", } def _header_chip() -> tuple[str, str]: if st.session_state.get("verified_proposal"): return "Proposal Ready", "ready" if st.session_state.get("recommendation_result"): return "Analysis Ready", "info" if st.session_state.get("api_ok"): return "API Ready", "ready" return "API Missing", "missing" def _render_sidebar(show_debug: bool) -> None: theme.sidebar_title("Steps") current = st.session_state.current_step for index, (key, label) in enumerate(_visible_steps(show_debug), start=1): status = _step_status(key, current) unlocked = status != "locked" glyph = _STATUS_GLYPH.get(status, "○") btn_label = f"{glyph} {index}. {label}" clicked = st.button( btn_label, key=f"nav_btn_{key}", type="primary" if status == "current" else "secondary", disabled=not unlocked, use_container_width=True, help=None if unlocked else _lock_reason(key), ) if clicked and unlocked and status != "current": st.session_state.current_step = key st.rerun() # API status warning — shown any time the key is no longer working so the # user knows why things are failing without hunting through the screens. if st.session_state.get("api_ok") is False and st.session_state.get("api_status") not in (None,): st.warning("⚠️ AI service issue — go to Setup to fix it.") # ── Always-visible settings link ─────────────────────────────────── # The Setup screen is auto-skipped once an API key is saved, so users # need an explicit way back to change it. st.divider() if st.button( "⚙️ Change API Settings", key="goto_setup_btn", use_container_width=True, type="secondary", help="Go back to Setup to change your API key or AI provider.", ): st.session_state.current_step = "setup" st.rerun() # Dark / light theme toggle. dark_on = st.toggle( "🌙 Dark mode", value=bool(st.session_state.get("theme_dark", False)), key="theme_toggle", ) if dark_on != st.session_state.get("theme_dark"): st.session_state.theme_dark = dark_on st.rerun() # Developer-only API usage panel. Hidden unless SHOW_DEBUG_PANEL=true. api_usage_panel.render(key_suffix="sidebar") def _try_restore_saved_config() -> None: """Load this browser's saved (encrypted) Setup config from Supabase. Runs once per session. Only acts when key storage is configured and the session has no API key yet. On success it applies the saved provider / model / key / vision settings into the session so the rest of the app — including smart-landing's API check — proceeds as if the user just set up. Entirely fail-safe (see :mod:`app.services.key_store`): any failure leaves the session untouched and the user simply lands on Setup as before. """ ss = st.session_state if ss.get("_restore_done"): return if not key_store.is_configured(): ss["_restore_done"] = True return if get_settings().has_api_key: ss["_restore_done"] = True return tag = browser_tag.get_or_create_tag() if not tag: return # tag not available yet; try again on the next render stored = key_store.load_config(tag) ss["_restore_done"] = True if stored: apply_session_overrides(stored) ss["_config_restored"] = True def _try_smart_landing(settings) -> None: """On a fresh session, silently restore Setup + Dossier if already done. Setup: re-run the API check. If it passes, mark api_ok = True. Dossier: if a folder path was previously entered AND is still valid, silently re-load the evidence index. If the path is gone/broken, show a sidebar warning instead of forcing the user back to the Dossier screen. On success both steps are skipped and the user lands on Job Screenshot. On any failure the user lands on the broken step as normal. """ ss = st.session_state # --- Step 1: API check ------------------------------------------- # Only attempt if a key is already saved; don't block if it's missing. if not ss.get("api_ok") and settings.has_api_key: try: result = run_capability_test(settings=settings, live=True) if result.ok: ss["api_ok"] = True ss["api_status"] = ApiGateError.API_OK else: ss["api_ok"] = False ss["api_status"] = result.status # Can't go further without a working API key. ss["current_step"] = "setup" return except Exception: ss["api_ok"] = False ss["current_step"] = "setup" return if not ss.get("api_ok"): # No key saved at all → stay on Setup. ss["current_step"] = "setup" return # --- Step 2: Dossier auto-load ----------------------------------- if ss.get("evidence_index"): # Evidence index already loaded in this session → skip to screenshot. ss["current_step"] = "screenshot" return # Prefer the session-state path; fall back to the persisted .env path. path = (ss.get("dossier_folder_path") or "").strip() if not path: path = get_saved_dossier_path() if not path: # No path saved anywhere → go to Dossier so user can enter one. ss["current_step"] = "dossier" return # Always keep session_state in sync so the dossier screen shows it. ss["dossier_folder_path"] = path try: result = validate(path) if not result.can_continue: # Path exists but unusable → land on Dossier with a clear error. ss["dossier_validation"] = result ss["current_step"] = "dossier" return chunks = read_dossier(path) if not chunks: ss["current_step"] = "dossier" return proofs, profile, meta = build_evidence_index(chunks) ss["dossier_chunks"] = chunks ss["dossier_read"] = True ss["dossier_read_error"] = False ss["dossier_validation"] = result ss["evidence_index"] = proofs ss["canonical_profile"] = profile ss["evidence_index_meta"] = meta ss["current_step"] = "screenshot" except Exception: # Folder moved or deleted → land on Dossier, warn there. ss["dossier_read_error"] = True ss["current_step"] = "dossier" def main() -> None: st.set_page_config( page_title=APP_TITLE, page_icon="🧭", layout="wide", initial_sidebar_state="expanded", ) theme.inject_css() _init_session_state() settings = get_settings() show_debug = bool(getattr(settings, "show_debug_panel", False)) # Confirm Details is debug-only — keep normal users out of it even if a # stale navigation target points there. if not show_debug and st.session_state.current_step == "confirmation": st.session_state.current_step = ( "analysis" if st.session_state.get("fields_confirmed") else "screenshot" ) # ---- Smart landing: skip screens the user has already completed -------- # On a fresh Streamlit session (first page load or page refresh) we check # whether Setup and Dossier are already done and land the user directly on # Job Screenshot so they never have to re-enter their key or folder path. # We only do this once per session (guarded by the _smart_land_done flag), # and only when the session default "setup" is still the current step. # Restore a previously saved (encrypted) config for this browser BEFORE # smart-landing, so a returning visitor's key is in place and they get # auto-advanced past Setup without re-entering anything. _try_restore_saved_config() settings = get_settings() if not st.session_state.get("_smart_land_done"): st.session_state["_smart_land_done"] = True _try_smart_landing(settings) if not _is_unlocked(st.session_state.current_step): st.session_state.current_step = "setup" # Re-tint the whole screen to the current step's signature color. dark = bool(st.session_state.get("theme_dark", False)) current_step = st.session_state.current_step theme.apply_step_accent(current_step, dark=dark) # "Step X of N" indicator over the visible steps. visible = _visible_steps(show_debug) step_label = None for idx, (key, _label) in enumerate(visible, start=1): if key == current_step: step_label = f"Step {idx} of {len(visible)}" break chip_label, chip_kind = _header_chip() theme.render_app_header( APP_TITLE, APP_TAGLINE, chip_label=chip_label, chip_kind=chip_kind, step_label=step_label, ) with st.sidebar: _render_sidebar(show_debug) RENDERERS[st.session_state.current_step]() if __name__ == "__main__": main()