Spaces:
Sleeping
Sleeping
| """Screen 3: Job screenshot upload and detail extraction. | |
| The user uploads one or more screenshots of an Upwork job post; the tool | |
| reads them and pulls out the structured job details. In the normal flow | |
| those details are confirmed automatically in the backend and the analysis | |
| unlocks immediately — there is no separate review step. Missing details | |
| are kept as "Not visible"; the tool never guesses hidden information. | |
| The manual "Confirm Details" review page is developer/admin-only and | |
| appears only when ``SHOW_DEBUG_PANEL=true``. Provider/model labels and | |
| fallback details likewise only appear in the Developer Debug Panel. | |
| """ | |
| from __future__ import annotations | |
| import io | |
| from typing import Any | |
| import streamlit as st | |
| from app.config import ( | |
| GROQ_VISION_UNSUPPORTED_MESSAGE, | |
| clear_opportunity_state, | |
| get_settings, | |
| ) | |
| from app.services.screenshot_parser import ( | |
| NOT_VISIBLE, | |
| SCREENSHOT_FIELDS, | |
| confirm_fields, | |
| extract_fields, | |
| get_meta as get_screenshot_meta, | |
| ) | |
| from app.ui import theme | |
| # Optional clipboard-paste component. It lets a user copy a screenshot with | |
| # the snipping tool and paste it straight into the app. It is an *optional* | |
| # dependency: if it isn't installed the screen falls back to drag-drop/browse | |
| # only, and headless tests (which don't ship the component) keep working. | |
| # pip install streamlit-paste-button | |
| try: # pragma: no cover - import guard, exercised only when the pkg is present | |
| from streamlit_paste_button import paste_image_button as _paste_image_button | |
| except Exception: # noqa: BLE001 - any import failure → paste simply unavailable | |
| _paste_image_button = None | |
| # The uploader widget key. Stable across reruns (so Streamlit keeps the | |
| # selected files) and unique to this screen. A counter suffix lets us RESET | |
| # the widget on demand (Streamlit can't clear a file_uploader any other way): | |
| # bumping the counter mounts a fresh, empty uploader for the next job. | |
| UPLOADER_KEY = "screenshot_uploader" | |
| def _uploader_key() -> str: | |
| """Current uploader widget key, including the reset counter.""" | |
| return f"{UPLOADER_KEY}_{st.session_state.get('uploader_reset_count', 0)}" | |
| def _clear_screenshots_for_new_job() -> None: | |
| """Wipe the current screenshots + analysis so the user can analyze a new job. | |
| Resets the uploader (via the counter), clears pasted images, the extracted | |
| fields, and every per-opportunity result (match, score, recommendation, | |
| proposal), then sends the user back to the screenshot step. | |
| """ | |
| st.session_state["uploader_reset_count"] = ( | |
| st.session_state.get("uploader_reset_count", 0) + 1 | |
| ) | |
| st.session_state[PASTED_KEY] = [] | |
| st.session_state[UPLOADS_KEY] = [] | |
| st.session_state[UPLOADS_SHOWN_KEY] = False | |
| st.session_state.uploaded_screenshots = [] | |
| st.session_state.screenshots_uploaded = False | |
| st.session_state.screenshot_upload_sig = () | |
| st.session_state.extracted_job_fields = None | |
| st.session_state.extraction_failed = False | |
| st.session_state.extraction_failed_meta = None | |
| clear_opportunity_state(st.session_state) | |
| st.session_state.current_step = "screenshot" | |
| # Session key holding the most recent clipboard-pasted screenshot as a | |
| # ``{name, mime, bytes}`` record (same shape as an uploaded file), so it can | |
| # survive reruns and feed the same downstream handler as the file uploader. | |
| PASTED_KEY = "pasted_screenshots" | |
| # Session keys holding the uploader's screenshots as ``{name, mime, bytes}`` | |
| # records. Streamlit empties a file_uploader whenever its step is left, so the | |
| # records are kept here: coming back to this step only to look must not count | |
| # as removing them (that used to wipe the finished analysis). The second key | |
| # is True while the records are the files the uploader is showing, and False | |
| # once they are carried over from an earlier visit to the step. | |
| UPLOADS_KEY = "screenshot_upload_records" | |
| UPLOADS_SHOWN_KEY = "screenshot_upload_records_shown" | |
| # User-facing message shown if a file was picked but its bytes never | |
| # reached the server (e.g. a residual upload failure). Never exposes the | |
| # underlying Axios/HTTP status. | |
| UPLOAD_FAILED_MESSAGE = ( | |
| "Screenshot upload failed. Please refresh and try again with a " | |
| "PNG or JPG file." | |
| ) | |
| # Shown when the AI could not read ANY field from the screenshot(s) — almost | |
| # always a provider/key problem, not a bad screenshot. We stop here instead of | |
| # running an analysis on empty data (which would wrongly read "Do Not Proceed"). | |
| EXTRACTION_FAILED_MESSAGE = ( | |
| "The AI could not read any details from your screenshot. This is almost " | |
| "always an AI-service issue, not your screenshot. Please check:\n\n" | |
| "1. **Your AI provider can read images.** Use **OpenAI**, **Anthropic**, " | |
| "or **Gemini** on the Setup page — **Groq cannot read images**.\n" | |
| "2. **Your API key is active and has credit.** An expired key or empty " | |
| "balance makes the read fail silently.\n" | |
| "3. **The screenshot is clear** — a full, legible image of the job post " | |
| "(PNG or JPG), not blurry or tiny.\n\n" | |
| "Fix the above (the **Change API Settings** button is in the sidebar), " | |
| "then click **Analyze Opportunity** again." | |
| ) | |
| # Map a file extension to a MIME type when the browser didn't supply one. | |
| _MIME_BY_SUFFIX = { | |
| "png": "image/png", | |
| "jpg": "image/jpeg", | |
| "jpeg": "image/jpeg", | |
| "webp": "image/webp", | |
| } | |
| def _guess_mime(name: str) -> str: | |
| suffix = name.lower().rsplit(".", 1)[-1] if "." in name else "" | |
| return _MIME_BY_SUFFIX.get(suffix, "image/png") | |
| def _collect_uploaded_screenshots(files: Any) -> list[dict]: | |
| """Read each Streamlit ``UploadedFile`` into an in-memory record. | |
| For every successfully-read upload we keep only ``{name, mime, | |
| bytes}`` in session state — the image is held in memory for this | |
| session and never written to disk. Files whose bytes can't be read | |
| (a failed/partial upload) are skipped so they don't count as present. | |
| """ | |
| collected: list[dict] = [] | |
| for uploaded_file in files or []: | |
| if uploaded_file is None: | |
| continue | |
| getvalue = getattr(uploaded_file, "getvalue", None) | |
| if not callable(getvalue): | |
| # Not a real UploadedFile (or an unreadable selection) — skip. | |
| continue | |
| try: | |
| data = getvalue() | |
| except Exception: # noqa: BLE001 - treat any read error as a failed upload | |
| continue | |
| if not data: | |
| continue | |
| name = getattr(uploaded_file, "name", None) or "screenshot" | |
| mime = getattr(uploaded_file, "type", None) or _guess_mime(name) | |
| collected.append({"name": name, "mime": mime, "bytes": bytes(data)}) | |
| return collected | |
| def _capture_pasted_screenshot() -> tuple[list[dict], bool]: | |
| """Render the clipboard-paste button and return the pasted image record(s). | |
| Returns the persisted pasted screenshot as a ``[{name, mime, bytes}]`` | |
| list (same shape as an uploaded file) so it can be merged with the file | |
| uploader and feed the identical downstream handler, plus whether that | |
| image was pasted during this visit to the step (the component still holds | |
| it; it comes back empty after the user navigates away and back). When a | |
| fresh image is pasted it replaces the previously-pasted one. If the | |
| component isn't installed, nothing is rendered and any previously-pasted | |
| image is kept. | |
| """ | |
| if _paste_image_button is None: | |
| # Component not installed — drag-drop/browse remains fully functional. | |
| return list(st.session_state.get(PASTED_KEY) or []), False | |
| # Colour the paste button to match the active theme (the component renders | |
| # in its own iframe, so it can't inherit our CSS — we pass colours in). | |
| _dark = bool(st.session_state.get("theme_dark", True)) | |
| if _dark: | |
| # Aurora secondary surface, solid because the component takes a | |
| # single colour rather than the glass gradient. | |
| _btn_bg, _btn_hover, _btn_fg = "#151D30", "#1D2740", "#F8FAFC" | |
| else: | |
| _btn_bg, _btn_hover, _btn_fg = "#E7EAF3", "#DCE1EE", "#0F172A" | |
| try: | |
| paste_result = _paste_image_button( | |
| "Paste screenshot from clipboard", | |
| # Key carries the same reset counter as the uploader, so "Clear | |
| # screenshots" mounts a FRESH paste widget with no retained image — | |
| # otherwise the component keeps re-adding the old pasted screenshot. | |
| key=f"screenshot_paste_button_{st.session_state.get('uploader_reset_count', 0)}", | |
| background_color=_btn_bg, | |
| hover_background_color=_btn_hover, | |
| text_color=_btn_fg, | |
| errors="ignore", | |
| ) | |
| except Exception: # noqa: BLE001 - component unavailable (e.g. headless) → upload-only | |
| # Rendering the clipboard widget must never break the screen; fall back | |
| # to upload-only and keep any image pasted on a previous run. | |
| return list(st.session_state.get(PASTED_KEY) or []), False | |
| image = getattr(paste_result, "image_data", None) | |
| pasted_this_visit = False | |
| if image is not None: | |
| # PIL image → PNG bytes, the same {name, mime, bytes} record an upload | |
| # produces. Held in memory only; never written to disk. | |
| try: | |
| buffer = io.BytesIO() | |
| image.save(buffer, format="PNG") | |
| st.session_state[PASTED_KEY] = [ | |
| { | |
| "name": "pasted-screenshot.png", | |
| "mime": "image/png", | |
| "bytes": buffer.getvalue(), | |
| } | |
| ] | |
| pasted_this_visit = True | |
| except Exception: # noqa: BLE001 - a bad paste must never crash the screen | |
| pass | |
| return list(st.session_state.get(PASTED_KEY) or []), pasted_this_visit | |
| def _sync_uploaded_screenshots( | |
| uploader_screenshots: list[dict], uploader_mounted: bool | |
| ) -> list[dict]: | |
| """Keep the uploader's screenshots in session state across visits. | |
| Streamlit empties a file_uploader whenever its step is left, so coming | |
| back mounts it empty although the user removed nothing: | |
| * files in the uploader replace the kept records (the user added or | |
| removed some); | |
| * an uploader the user emptied (its x buttons) clears them; | |
| * an uploader that is empty because it was just mounted keeps them, so | |
| opening this step only to look changes nothing. | |
| Returns the kept records. | |
| """ | |
| if not uploader_mounted: | |
| # Freshly mounted: anything kept was uploaded on an earlier visit. | |
| st.session_state[UPLOADS_SHOWN_KEY] = False | |
| if uploader_screenshots or st.session_state.get(UPLOADS_SHOWN_KEY): | |
| st.session_state[UPLOADS_KEY] = uploader_screenshots | |
| st.session_state[UPLOADS_SHOWN_KEY] = bool(uploader_screenshots) | |
| return list(st.session_state.get(UPLOADS_KEY) or []) | |
| def _signature(screenshots: list[dict]) -> tuple[tuple[str, int], ...]: | |
| """A light fingerprint of the uploaded set used to detect new uploads. | |
| Built from (name, byte-length) per file — enough to notice a swapped | |
| or added screenshot without hashing or logging the image itself. | |
| """ | |
| return tuple((s["name"], len(s["bytes"])) for s in screenshots) | |
| def _clear_stale_analysis() -> None: | |
| """Drop everything derived from a previous screenshot. | |
| Clears the extracted job fields plus every per-opportunity result | |
| (confirmed fields, match, score, recommendation, fingerprint, | |
| proposals) so a prior opportunity's analysis can never leak into a | |
| newly-uploaded one. | |
| """ | |
| st.session_state.extracted_job_fields = None | |
| clear_opportunity_state(st.session_state) | |
| def _reset_after_extract() -> None: | |
| """Clear every per-opportunity result before the new screenshot's | |
| job details are confirmed, so a previous opportunity's confirmed | |
| fields, match, score, recommendation, fingerprint, or proposal can | |
| never leak into this one.""" | |
| clear_opportunity_state(st.session_state) | |
| def _visible_field_count(extracted: dict) -> int: | |
| """How many job fields the vision model actually read (not 'Not visible').""" | |
| return sum( | |
| 1 | |
| for key in SCREENSHOT_FIELDS | |
| if (extracted.get(key) or {}).get("value") | |
| and (extracted.get(key) or {}).get("value") != NOT_VISIBLE | |
| ) | |
| def _extraction_failed(extracted: dict) -> bool: | |
| """True when the screenshot read produced no usable data. | |
| Treated as a failure when the vision API call did not succeed, OR it | |
| succeeded but read zero fields — both mean we must NOT run an analysis on | |
| empty data (which would wrongly conclude "Do Not Proceed"). | |
| """ | |
| meta = get_screenshot_meta(extracted) | |
| used_api = bool(meta.get("used_api")) | |
| status = str(meta.get("status") or "").lower() | |
| ok_status = status in ("", "ok", "success", "succeeded") | |
| if not used_api or not ok_status: | |
| return True | |
| return _visible_field_count(extracted) == 0 | |
| def _auto_confirm(extracted: dict) -> None: | |
| """Confirm the extracted job details in the backend (normal flow). | |
| No user review: extracted values become the confirmed values, with any | |
| missing field kept as "Not visible". This unlocks the analysis without | |
| a separate confirmation page. | |
| """ | |
| st.session_state.confirmed_job_fields = confirm_fields(extracted) | |
| st.session_state.fields_confirmed = True | |
| def _render_debug_panel(meta: dict) -> None: | |
| settings = get_settings() | |
| with st.expander("Developer Debug Panel", expanded=False): | |
| st.markdown( | |
| f"**Provider:** `{settings.llm_provider}` • " | |
| f"**Model:** `{settings.active_model}` • " | |
| f"**Task:** `screenshot_extraction`" | |
| ) | |
| st.markdown( | |
| "**Vision extraction:** " | |
| + ("LLM API used" if meta.get("used_api") else "local fallback") | |
| ) | |
| if meta.get("error_message"): | |
| st.caption(f"Reason: {meta['error_message']}") | |
| def _render_debug_handoff(extracted: dict) -> None: | |
| """Developer/admin handoff: review the fields on the Confirm Details page.""" | |
| with st.container(border=True, key="ups_panel_screenshot_1"): | |
| theme.section_label("Detected details") | |
| visible = sum( | |
| 1 | |
| for key in SCREENSHOT_FIELDS | |
| if (extracted.get(key) or {}).get("value") | |
| and (extracted.get(key) or {}).get("value") != NOT_VISIBLE | |
| ) | |
| st.success(f"{visible} of {len(SCREENSHOT_FIELDS)} details captured.") | |
| st.caption("Review and correct them on the next step.") | |
| if st.button( | |
| "Continue to Confirm Details", | |
| type="primary", | |
| key="continue_to_confirmation_btn", | |
| ): | |
| st.session_state.current_step = "confirmation" | |
| st.rerun() | |
| _render_debug_panel(get_screenshot_meta(extracted)) | |
| def render() -> None: | |
| if not st.session_state.get("evidence_index"): | |
| st.error( | |
| "This step is locked. Read your dossier and create proof points first." | |
| ) | |
| if st.button("Back to Dossier", key="back_to_dossier_from_screenshot"): | |
| st.session_state.current_step = "dossier" | |
| st.rerun() | |
| return | |
| settings = get_settings() | |
| debug = bool(getattr(settings, "show_debug_panel", False)) | |
| theme.screen_head( | |
| "screenshot", | |
| "Add the job post", | |
| "Drop a screenshot of the Upwork job — or paste one you copied. " | |
| "The tool reads the details for you. Anything it can't see is marked <b>Not visible</b> — it never guesses.", | |
| ) | |
| with st.container(border=True, key="ups_panel_screenshot_2"): | |
| theme.section_label("Add screenshot") | |
| # Streamlit drops a widget's state on any run that doesn't draw it, so | |
| # an uploader key missing here means the uploader is being mounted | |
| # afresh (first visit, or back from another step) and comes up empty. | |
| uploader_mounted = _uploader_key() in st.session_state | |
| # Native Streamlit uploader — no custom axios/fetch, no external | |
| # endpoint. The bytes stay on this server in memory only. | |
| uploaded = st.file_uploader( | |
| "Drop one or more screenshots of the job post", | |
| type=["png", "jpg", "jpeg", "webp"], | |
| accept_multiple_files=True, | |
| key=_uploader_key(), | |
| help="Files are held in this session only and never saved to disk.", | |
| ) | |
| # Clipboard paste — copy a screenshot (e.g. with the snipping tool) | |
| # and paste it here. It produces the same {name, mime, bytes} record | |
| # as an upload, so both input methods feed one downstream handler. | |
| if _paste_image_button is not None: | |
| st.caption("…or paste a screenshot you copied to the clipboard:") | |
| pasted, pasted_this_visit = _capture_pasted_screenshot() | |
| uploader_screenshots = _collect_uploaded_screenshots(uploaded) | |
| kept_uploads = _sync_uploaded_screenshots(uploader_screenshots, uploader_mounted) | |
| # A screenshot added after the ones on hand were already read starts a | |
| # new job. Screenshots kept from an earlier visit to this step (a | |
| # paste, or uploads the uploader no longer shows) belong to the job | |
| # already read: drop them so they are never merged into the new one. | |
| previous = set(st.session_state.get("screenshot_upload_sig") or ()) | |
| added = any(item not in previous for item in _signature(kept_uploads + pasted)) | |
| if added and st.session_state.get("extracted_job_fields"): | |
| if not pasted_this_visit: | |
| pasted = [] | |
| st.session_state[PASTED_KEY] = [] | |
| if not st.session_state.get(UPLOADS_SHOWN_KEY): | |
| kept_uploads = [] | |
| st.session_state[UPLOADS_KEY] = [] | |
| # Both inputs merge into one set the rest of the screen treats | |
| # uniformly (extraction already merges multiple screenshots). | |
| screenshots = kept_uploads + pasted | |
| st.session_state.uploaded_screenshots = screenshots | |
| st.session_state.screenshots_uploaded = bool(screenshots) | |
| # A new (or swapped) screenshot — uploaded OR pasted — invalidates any | |
| # analysis built from a previous one; clear it before re-analyzing. | |
| signature = _signature(screenshots) | |
| if screenshots and signature != st.session_state.get("screenshot_upload_sig"): | |
| _clear_stale_analysis() | |
| # A new screenshot is a fresh attempt — clear any prior failure note. | |
| st.session_state.extraction_failed = False | |
| st.session_state.extraction_failed_meta = None | |
| st.session_state.screenshot_upload_sig = signature | |
| # The widget shows the file(s) but none of their bytes reached the | |
| # server — surface a clean message instead of a raw upload error. | |
| if uploaded and not uploader_screenshots: | |
| st.error(UPLOAD_FAILED_MESSAGE) | |
| # A previous Analyze attempt couldn't read the screenshot — explain why | |
| # (provider/key issue) and let the user fix Setup and retry. | |
| if st.session_state.get("extraction_failed"): | |
| st.error(EXTRACTION_FAILED_MESSAGE) | |
| if screenshots: | |
| col_ready, col_clear = st.columns([2, 1]) | |
| with col_ready: | |
| st.caption(f"{len(screenshots)} screenshot(s) ready.") | |
| with col_clear: | |
| if st.button( | |
| "🗑️ Clear screenshots", | |
| key="clear_screenshots_btn", | |
| use_container_width=True, | |
| help="Remove these screenshots so you can analyze a different job.", | |
| ): | |
| _clear_screenshots_for_new_job() | |
| st.rerun() | |
| st.caption( | |
| "Include the **About the client** box (hire rate, jobs posted, " | |
| "payment verified) — add a second screenshot if it doesn't fit in " | |
| "one. Missing details are marked “Not visible”; the tool won't " | |
| "guess hidden information." | |
| ) | |
| # Groq is text-only: a Groq vision provider can't read screenshots. | |
| # Show a clean message and block extraction until the operator points | |
| # vision at a supported provider on the Setup page. | |
| vision_unsupported = settings.active_vision_provider == "groq" | |
| if vision_unsupported: | |
| st.warning(GROQ_VISION_UNSUPPORTED_MESSAGE) | |
| # Reading a screenshot needs a passing API check. Saving Setup again | |
| # clears it while this step stays open, so say so here — otherwise | |
| # Analyze is greyed out with no visible reason. | |
| api_ok = bool(st.session_state.get("api_ok")) | |
| if not api_ok: | |
| st.warning( | |
| "Your AI service needs a passing API check before a screenshot " | |
| "can be read. Run the API check on the Setup screen first." | |
| ) | |
| if st.button("Go to Setup", key="goto_setup_from_screenshot"): | |
| st.session_state.current_step = "setup" | |
| st.rerun() | |
| can_extract = bool( | |
| api_ok | |
| and st.session_state.get("uploaded_screenshots") | |
| and not vision_unsupported | |
| ) | |
| if can_extract: | |
| extract_help = None | |
| elif vision_unsupported: | |
| extract_help = ( | |
| "Choose a vision provider that supports images " | |
| "(OpenAI, Anthropic, or Gemini) on the Setup page." | |
| ) | |
| elif not api_ok: | |
| extract_help = "Run the API check on the Setup screen first." | |
| else: | |
| extract_help = "Add at least one screenshot first." | |
| # Normal flow merges extraction + analysis into ONE button: it reads | |
| # the screenshot, confirms the details in the backend, and advances | |
| # straight to Analysis. Debug mode keeps the granular "Extract Job | |
| # Details" → manual Confirm Details handoff for developers/admins. | |
| analyze_label = "Extract Job Details" if debug else "Analyze Opportunity" | |
| analyze_key = "extract_job_details_btn" if debug else "analyze_opportunity_btn" | |
| if st.button( | |
| analyze_label, | |
| type="primary", | |
| key=analyze_key, | |
| disabled=not can_extract, | |
| help=extract_help, | |
| ): | |
| # Pass raw image bytes (not file paths) to the vision parser. | |
| image_inputs = [ | |
| (s["bytes"], s["mime"]) | |
| for s in st.session_state.get("uploaded_screenshots") or [] | |
| ] | |
| spinner_msg = ( | |
| "Reading your screenshot…" | |
| if debug | |
| else "Reading your screenshot and analyzing the opportunity…" | |
| ) | |
| with st.spinner(spinner_msg): | |
| fields = extract_fields(image_inputs) | |
| _reset_after_extract() | |
| # Guard: if the vision read produced nothing usable, STOP here and | |
| # explain why — never run an analysis on empty data (it would | |
| # wrongly read "Do Not Proceed" with every signal "Not visible"). | |
| # The failed read is not kept as the job's details either (that | |
| # would tick this step done and unlock Confirm Details); only its | |
| # meta is kept, so the debug panel can still show the reason. | |
| if _extraction_failed(fields): | |
| st.session_state.extracted_job_fields = None | |
| st.session_state.extraction_failed_meta = get_screenshot_meta(fields) | |
| st.session_state.extraction_failed = True | |
| st.rerun() | |
| st.session_state.extracted_job_fields = fields | |
| st.session_state.extraction_failed = False | |
| st.session_state.extraction_failed_meta = None | |
| if not debug: | |
| # One go: auto-confirm in the backend and jump to Analysis. | |
| _auto_confirm(fields) | |
| st.session_state.current_step = "analysis" | |
| # Debug mode reruns too, so an earlier failure banner is not left | |
| # showing next to the details just read. | |
| st.rerun() | |
| # Debug/admin path keeps the manual review handoff to Confirm Details. A | |
| # failed read has nothing to hand off, but the panel still says why. | |
| extracted = st.session_state.get("extracted_job_fields") | |
| if debug and st.session_state.get("extraction_failed"): | |
| _render_debug_panel(st.session_state.get("extraction_failed_meta") or {}) | |
| elif extracted and debug: | |
| _render_debug_handoff(extracted) | |