Merge branch 'cleanup-final' into main

Author: crispasr integration

PLAN.md

@@ -120,7 +120,6 @@
 key_wrapper = KeyWrapper()
 
 try:
-
     FORMAT_TRANSPLANT_OK = True
     FORMAT_TRANSPLANT_ERROR = None
 except Exception as _ft_err:
@@ -136,7 +135,6 @@ def _ft_get_provider_choices(user_state):
     return ft_get_provider_choices(user_state, FORMAT_TRANSPLANT_OK)
 
 
-
 try:
     from tool_executor import (
         DEFAULT_TOOLS_ADMIN,
@@ -491,7 +489,6 @@ def clear_chat():
             # versions at lines ~885, ~1183, ~1385 (which pass session_state).
             # These were missing session_state and ran without auth.
 
-
     # ── Session initialization on page load ──────────────────────────
     # With auth=, the user is already authenticated when this fires.
     # We use request.username to look up the user and set the UI state.
@@ -508,6 +505,7 @@ def initialize_session(request: gr.Request):
 
         # Look up user from DB + retrieve cached UMK (derived during auth=)
         from db_ops import get_cached_umk, get_user_by_username, get_user_settings
+
         user = get_user_by_username(username)
         if not user:
             print(f"[SESSION] User {username} not found in DB")
@@ -615,6 +613,7 @@ def _on_logout(user_state):
         username = (user_state or {}).get("username")
         if username:
             from db_ops import clear_cached_umk
+
             clear_cached_umk(username)
 
     logout_btn.click(

app.py

@@ -183,7 +183,9 @@ def logout_user() -> tuple[str, dict, dict, dict]:
 # ---------------------------------------------------------------------------
 
 
-def update_transcript_chat_ui(prov: str, force_all: bool = False, user_state: dict | None = None) -> tuple[dict, str]:
+def update_transcript_chat_ui(
+    prov: str, force_all: bool = False, user_state: dict | None = None
+) -> tuple[dict, str]:
     """
     Same logic as update_c_ui but for the transcript "Send to Chat" section.
     Returns (Dropdown update, badge HTML).
@@ -223,7 +225,9 @@ def update_transcript_chat_ui(prov: str, force_all: bool = False, user_state: di
     )
 
 
-def update_c_ui(prov: str, force_all: bool = False, user_state: dict | None = None) -> tuple[dict, str]:
+def update_c_ui(
+    prov: str, force_all: bool = False, user_state: dict | None = None
+) -> tuple[dict, str]:
     """Chat-tab provider dropdown updater."""
     print(f"\n{'#' * 30} [UI-DEBUG] START {'#' * 30}")
     print(f"[UI-DEBUG] Provider Input: '{prov}' | Force All: {force_all}")
@@ -297,7 +301,9 @@ def update_c_ui(prov: str, force_all: bool = False, user_state: dict | None = No
     )
 
 
-def update_v_ui(prov: str, force_all: bool = False, user_state: dict | None = None) -> tuple[str, dict, int]:
+def update_v_ui(
+    prov: str, force_all: bool = False, user_state: dict | None = None
+) -> tuple[str, dict, int]:
     """Vision-tab provider dropdown updater."""
     if not prov:
         return "", gr.update(choices=[], value=None), 0
@@ -341,7 +347,9 @@ def update_v_ui(prov: str, force_all: bool = False, user_state: dict | None = No
     )
 
 
-def update_g_ui(prov: str, force_all: bool = False, user_state: dict | None = None) -> tuple[str, dict]:
+def update_g_ui(
+    prov: str, force_all: bool = False, user_state: dict | None = None
+) -> tuple[str, dict]:
     """Image-generation-tab provider dropdown updater."""
     if not prov:
         return "", gr.update(choices=[], value=None)
@@ -378,7 +386,9 @@ def update_g_ui(prov: str, force_all: bool = False, user_state: dict | None = No
     return styled_badge, gr.update(choices=final_choices, value=default_val, label="Modell")
 
 
-def update_t_ui(prov: str, force_all: bool = False, user_state: dict | None = None) -> tuple[str, dict, dict, dict]:
+def update_t_ui(
+    prov: str, force_all: bool = False, user_state: dict | None = None
+) -> tuple[str, dict, dict, dict]:
     """Transcription-tab provider dropdown updater.
 
     Returns: (badge, t_model_update, whisper_options_visibility).
@@ -405,6 +415,7 @@ def update_t_ui(prov: str, force_all: bool = False, user_state: dict | None = No
         if is_crispasr:
             # Filter CrispASR backends by user's max model size preference
             from crispasr_tool import get_filtered_backends
+
             max_b = 2.0
             if user_state and user_state.get("max_model_size_b"):
                 try:

app_helpers.py

@@ -182,6 +182,7 @@ def _last_image_file(user_state: dict) -> str:
     """Per-user path where the most recently generated/edited image is cached,
     so edit_image can target it via image_url='last'."""
     import os
+
     uid = (user_state or {}).get("id", "anon")
     d = os.path.join(STORAGE_MOUNT_POINT, f"user_{uid}", "images")
     os.makedirs(d, exist_ok=True)
@@ -190,6 +191,7 @@ def _last_image_file(user_state: dict) -> str:
 
 def _save_last_image(user_state: dict, src_path: str) -> None:
     import shutil
+
     try:
         shutil.copyfile(src_path, _last_image_file(user_state))
     except Exception:  # nosec B110
@@ -319,22 +321,27 @@ def _handler(args: dict):
 
         image_url = (args.get("image_url") or "").strip()
         prompt = (args.get("prompt") or "").strip()
-        model = ("flux-kontext-max" if args.get("high_fidelity")
-                 else "flux-kontext-pro")
+        model = "flux-kontext-max" if args.get("high_fidelity") else "flux-kontext-pro"
 
         if not prompt:
-            return ToolResult(text="Error: 'prompt' (the edit) is required.",
-                              error=True)
+            return ToolResult(text="Error: 'prompt' (the edit) is required.", error=True)
         bfl_key = api_keys.get("BFL", "")
         if not bfl_key:
             return ToolResult(
-                text="Image editing needs a BFL API key, which is not configured.",
-                error=True)
+                text="Image editing needs a BFL API key, which is not configured.", error=True
+            )
 
         # Resolve the input image: a public URL, a local path, or 'last'
         # (the most recently generated/edited image in this session).
-        _LAST_KW = {"", "last", "previous", "last_generated", "last image",
-                    "the last image", "previous image"}
+        _LAST_KW = {
+            "",
+            "last",
+            "previous",
+            "last_generated",
+            "last image",
+            "the last image",
+            "previous image",
+        }
         if image_url.startswith(("http://", "https://")):
             input_image, src_label = image_url, image_url[:80]
         else:
@@ -343,22 +350,22 @@ def _handler(args: dict):
                 if not os.path.exists(local):
                     return ToolResult(
                         text="No previous image to edit. Generate one first or "
-                             "pass a public image_url.", error=True)
+                        "pass a public image_url.",
+                        error=True,
+                    )
                 src_label = "last generated image"
             elif os.path.exists(image_url):
                 local, src_label = image_url, os.path.basename(image_url)
             else:
                 return ToolResult(
-                    text="Provide a public image_url, a local path, or 'last'.",
-                    error=True)
+                    text="Provide a public image_url, a local path, or 'last'.", error=True
+                )
             with open(local, "rb") as f:
                 input_image = base64.b64encode(f.read()).decode()  # BFL accepts base64
 
-        print(f"[TOOLS] 🖌️ edit_image | model={model} src={src_label!r} "
-              f"prompt={prompt[:80]!r}…")
+        print(f"[TOOLS] 🖌️ edit_image | model={model} src={src_label!r} prompt={prompt[:80]!r}…")
         try:
-            file_path, status = run_image_edit(
-                prompt, input_image, model, bfl_key, user_state)
+            file_path, status = run_image_edit(prompt, input_image, model, bfl_key, user_state)
         except Exception as e:
             print(f"[TOOLS] ⚠️ edit_image failed: {e}")
             return ToolResult(text=f"Image edit failed: {e}", error=True)
@@ -371,12 +378,14 @@ def _handler(args: dict):
                 b64 = base64.b64encode(f.read()).decode()
             ext = os.path.splitext(file_path)[1].lstrip(".").lower() or "jpeg"
             mime = f"image/{ext}" if ext != "jpg" else "image/jpeg"
-            img_html = (f'<img src="data:{mime};base64,{b64}" '
-                        f'style="max-width:100%;border-radius:8px;margin-top:8px;" />')
+            img_html = (
+                f'<img src="data:{mime};base64,{b64}" '
+                f'style="max-width:100%;border-radius:8px;margin-top:8px;" />'
+            )
         except Exception as e:
             img_html = f"*(edited image saved to {file_path} — could not embed: {e})*"
 
-        text_summary = (f"✅ Image edited via BFL/{model}.\nEdit: {prompt[:120]}")
+        text_summary = f"✅ Image edited via BFL/{model}.\nEdit: {prompt[:120]}"
         return ToolResult(text=text_summary, ui=f"{text_summary}\n\n{img_html}")
 
     return _handler
@@ -456,6 +465,7 @@ def _handler(args: dict) -> str:
         # Load transcription history from DB
         try:
             from db_ops import get_user_transcriptions
+
             transcriptions = get_user_transcriptions(user_id, limit=100)
         except Exception as e:
             return f"Could not load transcription history: {e}"
@@ -471,18 +481,21 @@ def _handler(args: dict) -> str:
             if text and len(text) > 20:
                 # Truncate very long texts to keep embedding fast
                 docs.append(text[:2000])
-                meta.append({
-                    "id": t.get("id"),
-                    "provider": t.get("provider", "?"),
-                    "created": str(t.get("created_at", "?")),
-                    "full_len": len(text),
-                })
+                meta.append(
+                    {
+                        "id": t.get("id"),
+                        "provider": t.get("provider", "?"),
+                        "created": str(t.get("created_at", "?")),
+                        "full_len": len(text),
+                    }
+                )
 
         if not docs:
             return "No transcription texts found (all empty)."
 
         try:
             from crispembed_tool import semantic_search
+
             results = semantic_search(query, docs, top_k=top_k)
         except Exception as e:
             return f"Semantic search failed: {e}"
@@ -551,7 +564,9 @@ def no_creds_error(*args, **kwargs):  # type: ignore[no-untyped-def]
             "exchange_contact_edit": no_creds_error,
         }
 
-    logger.info("[EXCH] Credentials loaded for user_id=%s: email=%s", user_id, creds.get("email", "?"))
+    logger.info(
+        "[EXCH] Credentials loaded for user_id=%s: email=%s", user_id, creds.get("email", "?")
+    )
 
     def handle_exchange_list(args: dict) -> str:
         logger.debug("[EXCH] exchange_list called with args=%s", args)
@@ -1200,9 +1215,8 @@ def run_chat(
             # ppt-llm presentation tools (provider/model/key + per-user storage)
             try:
                 from pptx_tool import make_pptx_tool_handlers
-                handlers.update(make_pptx_tool_handlers(
-                    API_KEYS, user_state, provider, model, key
-                ))
+
+                handlers.update(make_pptx_tool_handlers(API_KEYS, user_state, provider, model, key))
             except Exception as _ppt_e:
                 logger.debug(f"[TOOLS] pptx handlers unavailable: {_ppt_e}")
 

chat_handlers.py

@@ -126,7 +126,9 @@ def summarize_turns(turns: list[dict[str, Any]], client: Any, model: str) -> str
         return ""
 
 
-def prune_messages(messages: list[dict[str, Any]], model_limit: int, max_output_tokens: int = 1000) -> list[dict[str, Any]]:
+def prune_messages(
+    messages: list[dict[str, Any]], model_limit: int, max_output_tokens: int = 1000
+) -> list[dict[str, Any]]:
     """
     Smartly trims conversation history with verbose CLI logging.
     Always keeps the system prompt and the latest user message.