feat(mongo-file): add file download cards and preview functionality

flexus_client_kit/ckit_ask_model.py

@@ -188,7 +188,7 @@ async def bot_subchat_create_multiple(
             variable_values={
                 "who_is_asking": who_is_asking,
                 "persona_id": persona_id,
-                "first_question": first_question,
+                "first_question": [json.dumps(q) for q in first_question],
                 "first_calls": first_calls,
                 "title": title,
                 "fcall_id": fcall_id,

flexus_client_kit/integrations/fi_mongo_store.py

@@ -2,6 +2,7 @@
 import os
 import logging
 import re
+import urllib.parse
 from typing import Dict, Any, Optional
 from flexus_client_kit.integrations.fi_localfile import _validate_file_security
 from pymongo.collection import Collection
@@ -21,8 +22,8 @@
         "properties": {
             "op": {
                 "type": "string",
-                "enum": ["help", "list", "ls", "cat", "grep", "delete", "upload", "save"],
-                "description": "Operation: list/ls (list files), cat (read file), grep (search), delete, upload (from disk), save (content directly)",
+                "enum": ["help", "list", "ls", "cat", "grep", "delete", "upload", "save", "render"],
+                "description": "Operation: list/ls (list files), cat (read file), grep (search), delete, upload (from disk), save (content directly), render (show download card to user)",
             },
             "args": {
                 "type": "object",
@@ -67,22 +68,43 @@
           Sometimes you need to grep .json files on disk, remember that all the strings inside are escaped in that case, making
           it a bit harder to match.
 
+render  - Show a download card to the user for an already-stored file.
+          The user sees a styled card with file icon, name, and download/preview button.
+          args: path (required)
+
 Examples:
   mongo_store(op="list", args={"path": "folder1/"})
   mongo_store(op="cat", args={"path": "folder1/something_20250803.json", "lines_range": "0:40", "safety_valve": "10k"})
   mongo_store(op="save", args={"path": "investigations/abc123.json", "content": "{...json...}"})
   mongo_store(op="delete", args={"path": "folder1/something_20250803.json"})
   mongo_store(op="grep", args={"path": "tasks.txt", "pattern": "TODO", "context": 2})
+  mongo_store(op="render", args={"path": "reports/monthly.pdf"})
 """
 
 # There's also a secret op="undelete" command that can bring deleted files
 
 
+_MIME_TYPES = {
+    ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
+    ".webp": "image/webp", ".gif": "image/gif", ".svg": "image/svg+xml",
+    ".pdf": "application/pdf", ".txt": "text/plain", ".csv": "text/csv",
+    ".json": "application/json", ".html": "text/html", ".htm": "text/html",
+    ".xml": "application/xml", ".md": "text/markdown",
+    ".yaml": "application/yaml", ".yml": "application/yaml",
+}
+
+
+def _guess_mime_type(path: str) -> str:
+    ext = os.path.splitext(path)[1].lower()
+    return _MIME_TYPES.get(ext, "application/octet-stream")
+
+
 async def handle_mongo_store(
     workdir: str,
     mongo_collection: Collection,
     toolcall: ckit_cloudtool.FCloudtoolCall,
     model_produced_args: Optional[Dict[str, Any]],
+    persona_id: Optional[str] = None,
 ) -> str:
     if not model_produced_args:
         return HELP
@@ -211,6 +233,23 @@ async def handle_mongo_store(
         else:
             return f"Error: File {path} not found in MongoDB"
 
+    elif op == "render":
+        if not path:
+            return f"Error: path parameter required for render operation\n\n{HELP}"
+        if not persona_id:
+            return "Error: render operation requires persona_id (pass it to handle_mongo_store)"
+        path_error = validate_path(path)
+        if path_error:
+            return f"Error: {path_error}"
+        document = await ckit_mongo.mongo_retrieve_file(mongo_collection, path)
+        if not document:
+            return f"Error: File {path} not found in MongoDB"
+        display_name = os.path.basename(path)
+        mime = _guess_mime_type(path)
+        enc_path = urllib.parse.quote(path, safe="/")
+        enc_name = urllib.parse.quote(display_name, safe="")
+        return f"📎DOWNLOAD:{persona_id}:{enc_path}:{enc_name}:{mime}"
+
     else:
         return HELP
 

flexus_simple_bots/avatar_from_idea_imagine.py

@@ -1,8 +1,190 @@
-import os, sys, asyncio, base64, io
+import os, sys, asyncio, base64, io, json
+from pathlib import Path
 from PIL import Image
-import xai_sdk
+try:
+    import xai_sdk
+except ImportError:
+    xai_sdk = None
 
-client = xai_sdk.Client()
+_default_client = None
+
+
+DEFAULT_MODEL = "grok-imagine-image"
+DEFAULT_RESOLUTION = "1k"
+_STYLE_BANK_MANIFEST = Path(__file__).parent / "bot_pictures" / "style_bank" / "manifest.json"
+
+
+def create_xai_client(api_key: str | None = None):
+    if xai_sdk is None:
+        raise RuntimeError("xai-sdk package is required")
+    if api_key:
+        return xai_sdk.Client(api_key=api_key)
+    return xai_sdk.Client()
+
+
+def _get_default_client():
+    global _default_client
+    if _default_client is None:
+        _default_client = create_xai_client()
+    return _default_client
+
+
+def _image_to_data_url(image_bytes: bytes, mime: str = "image/png") -> str:
+    return f"data:{mime};base64,{base64.b64encode(image_bytes).decode('utf-8')}"
+
+
+def style_bank_manifest() -> list[dict]:
+    if not _STYLE_BANK_MANIFEST.exists():
+        return []
+    with open(_STYLE_BANK_MANIFEST, "r", encoding="utf-8") as f:
+        rows = json.load(f)
+    if not isinstance(rows, list):
+        raise ValueError(f"Bad style-bank manifest: {_STYLE_BANK_MANIFEST}")
+    return rows
+
+
+def default_style_bank_files() -> dict[str, bytes]:
+    root = Path(__file__).parent
+    files = {}
+    for row in style_bank_manifest():
+        rel = str(row.get("source_path", "")).strip()
+        target_name = str(row.get("target_name", "")).strip()
+        if not rel or not target_name:
+            continue
+        path = root / rel
+        if not path.exists():
+            continue
+        files[target_name] = path.read_bytes()
+    return files
+
+
+async def _sample_image(
+    xclient,
+    *,
+    prompt: str,
+    image_urls: list[str],
+    resolution: str = DEFAULT_RESOLUTION,
+) -> bytes:
+    kwargs = {
+        "prompt": prompt,
+        "model": DEFAULT_MODEL,
+        "aspect_ratio": None,
+        "resolution": resolution,
+        "image_format": "base64",
+    }
+    image_urls = image_urls[:5]
+    if len(image_urls) == 1:
+        kwargs["image_url"] = image_urls[0]
+    else:
+        kwargs["image_urls"] = image_urls
+
+    def _api_call():
+        return xclient.image.sample(**kwargs)
+
+    rsp = await asyncio.to_thread(_api_call)
+    return rsp.image
+
+
+_FULLSIZE_TARGET = (1024, 1536)
+_MAX_IMAGE_BYTES = 250_000
+
+
+def _encode_webp_within_limit(
+    im: Image.Image,
+    quality: int = 100,
+    max_bytes: int = _MAX_IMAGE_BYTES,
+    min_quality: int = 40,
+) -> bytes:
+    for q in range(quality, min_quality - 1, -5):
+        out = io.BytesIO()
+        im.save(out, "WEBP", quality=q, method=6)
+        data = out.getvalue()
+        if len(data) <= max_bytes:
+            return data
+    out = io.BytesIO()
+    im.save(out, "WEBP", quality=min_quality, method=6)
+    return out.getvalue()
+
+
+def _save_fullsize_webp_bytes(
+    png_bytes: bytes,
+    quality: int = 100,
+    target_size: tuple[int, int] = _FULLSIZE_TARGET,
+) -> tuple[bytes, tuple[int, int]]:
+    tw, th = target_size
+    with Image.open(io.BytesIO(png_bytes)) as im:
+        im = make_transparent(im)
+        iw, ih = im.size
+        if (iw, ih) != (tw, th):
+            scale = min(tw / iw, th / ih)
+            new_w, new_h = int(iw * scale), int(ih * scale)
+            im = im.resize((new_w, new_h), Image.LANCZOS)
+            canvas = Image.new("RGBA", (tw, th), (0, 0, 0, 0))
+            canvas.paste(im, ((tw - new_w) // 2, (th - new_h) // 2))
+            im = canvas
+        data = _encode_webp_within_limit(im, quality)
+        size = im.size
+    return data, size
+
+
+def _save_avatar_256_webp_bytes(avatar_png_bytes: bytes, quality: int = 100) -> tuple[bytes, tuple[int, int]]:
+    with Image.open(io.BytesIO(avatar_png_bytes)) as im:
+        im = make_transparent(im)
+        s = min(im.size)
+        cx, cy = im.size[0] // 2, im.size[1] // 2
+        im = im.crop((cx - s // 2, cy - s // 2, cx + s // 2, cy + s // 2)).resize((256, 256), Image.LANCZOS)
+        data = _encode_webp_within_limit(im, quality)
+        size = im.size
+    return data, size
+
+
+async def generate_avatar_assets_from_idea(
+    *,
+    input_image_bytes: bytes,
+    description: str,
+    style_reference_images: list[bytes],
+    api_key: str,
+    count: int = 5,
+) -> list[dict]:
+    if not description or not description.strip():
+        raise ValueError("description is required")
+    if count < 1 or count > 10:
+        raise ValueError("count must be in range [1, 10]")
+
+    xclient = create_xai_client(api_key)
+    refs = [_image_to_data_url(input_image_bytes)]
+    refs += [_image_to_data_url(x) for x in style_reference_images]
+    refs = refs[:5]
+
+    fullsize_prompt = (
+        f"{description.strip()}. "
+        "Create a full-size variation of the character on pure solid bright green background (#00FF00)."
+    )
+    avatar_prompt = (
+        f"{description.strip()}. "
+        "Make avatar suitable for small pictures, face much bigger exactly in the center, "
+        "use a pure solid bright green background (#00FF00)."
+    )
+
+    async def _one(i: int):
+        fullsize_png = await _sample_image(xclient, prompt=fullsize_prompt, image_urls=refs)
+        fullsize_webp, fullsize_size = _save_fullsize_webp_bytes(fullsize_png)
+
+        avatar_png = await _sample_image(
+            xclient,
+            prompt=avatar_prompt,
+            image_urls=[_image_to_data_url(fullsize_png)],
+        )
+        avatar_webp_256, avatar_size_256 = _save_avatar_256_webp_bytes(avatar_png)
+        return {
+            "index": i,
+            "fullsize_webp": fullsize_webp,
+            "fullsize_size": fullsize_size,
+            "avatar_webp_256": avatar_webp_256,
+            "avatar_size_256": avatar_size_256,
+        }
+
+    return await asyncio.gather(*[_one(i) for i in range(count)])
 
 
 def make_transparent(im, fringe_fight: float = 0.30, defringe_radius: int = 4):
@@ -64,19 +246,28 @@ async def make_fullsize_variations(input_path: str, base_name: str, out_dir: str
 
     async def generate_one(i):
         def api_call():
-            return client.image.sample(
+            return _get_default_client().image.sample(
                 prompt="Make variations of the charactor on solid bright green background (#00FF00).",
-                model="grok-imagine-image",
+                model=DEFAULT_MODEL,
                 image_url=image_url,
                 aspect_ratio=None,  # does not work for image edit
-                resolution="1k",
+                resolution=DEFAULT_RESOLUTION,
                 image_format="base64"
             )
         rsp = await asyncio.to_thread(api_call)
         png_bytes = rsp.image
 
+        tw, th = _FULLSIZE_TARGET
         with Image.open(io.BytesIO(png_bytes)) as im:
             im = make_transparent(im)
+            iw, ih = im.size
+            if (iw, ih) != (tw, th):
+                scale = min(tw / iw, th / ih)
+                new_w, new_h = int(iw * scale), int(ih * scale)
+                im = im.resize((new_w, new_h), Image.LANCZOS)
+                canvas = Image.new("RGBA", (tw, th), (0, 0, 0, 0))
+                canvas.paste(im, ((tw - new_w) // 2, (th - new_h) // 2))
+                im = canvas
             fn = os.path.join(out_dir, f"i{i:02d}-{base_name}-{im.size[0]}x{im.size[1]}.webp")
             im.save(fn, 'WEBP', quality=85, method=6)
             print(f"Saved {fn}")
@@ -90,12 +281,12 @@ async def make_avatar(i: int, png_bytes: bytes, base_name: str, out_dir: str):
     image_url = f"data:image/png;base64,{base64.b64encode(png_bytes).decode('utf-8')}"
 
     def api_call():
-        return client.image.sample(
+        return _get_default_client().image.sample(
             prompt="Make avatar suitable for small pictures, face much bigger exactly in the center, use a pure solid bright green background (#00FF00).",
-            model="grok-imagine-image",
+            model=DEFAULT_MODEL,
             image_url=image_url,
             aspect_ratio=None,  # does not work for image edit
-            resolution="1k",
+            resolution=DEFAULT_RESOLUTION,
             image_format="base64"
         )
     rsp = await asyncio.to_thread(api_call)

flexus_simple_bots/bot_pictures/style_bank/manifest.json

@@ -0,0 +1,27 @@
+[
+  {
+    "target_name": "frog.webp",
+    "source_path": "frog/frog-256x256.webp",
+    "label": "Cute mascot style"
+  },
+  {
+    "target_name": "strategist.webp",
+    "source_path": "strategist/strategist-256x256.webp",
+    "label": "Professional portrait style"
+  },
+  {
+    "target_name": "ad_monster.webp",
+    "source_path": "admonster/ad_monster-256x256.webp",
+    "label": "Playful monster style"
+  },
+  {
+    "target_name": "karen.webp",
+    "source_path": "karen/karen-256x256.webp",
+    "label": "Clean assistant style"
+  },
+  {
+    "target_name": "boss.webp",
+    "source_path": "boss/boss-256x256.webp",
+    "label": "Founder portrait style"
+  }
+]

setup.py

@@ -58,6 +58,7 @@ def run(self):
         "pandas",
         "playwright",
         "openai",
+        "xai-sdk",
         "mcp",
         "python-telegram-bot>=20.0",
     ],