""" title: Personal Status author: Blah version: 0.1.0 tested_on_open_webui_version: v0.6.15 description: Adds user status context to messages for personalised AI responses. """ import os import json import logging import sys import functools import inspect import time from typing import Optional, Callable, Any, Dict from pydantic import BaseModel, Field # Setup logging LOGGER: logging.Logger = logging.getLogger("FUNC:STATUS_INJECTOR") def set_logs(logger: logging.Logger, level: int, force: bool = False): """Configure logger with proper formatting and handlers.""" logger.setLevel(level) for handler in logger.handlers: if not force and isinstance(handler, logging.StreamHandler): handler.setLevel(level) return logger handler = logging.StreamHandler(sys.stdout) handler.setLevel(level) formatter = logging.Formatter( "%(levelname)s[%(name)s]%(lineno)s:%(asctime)s: %(message)s" ) handler.setFormatter(formatter) logger.addHandler(handler) return logger set_logs(LOGGER, logging.INFO) def log_exceptions(func: Callable[..., Any]): """Decorator to log exceptions in functions.""" @functools.wraps(func) async def _wrapper(*args, **kwargs): try: return await func(*args, **kwargs) except Exception as exc: LOGGER.error("Error in %s: %s", func, exc, exc_info=True) # Reraise to ensure the filter doesn't silently fail raise return _wrapper class Filter: class Valves(BaseModel): priority: int = Field( default=-10, description="Filter execution order (lower = earlier)" ) default_status: str = Field( default="Available", description="Default status when none is set" ) status_format: str = Field( default="prefix", description="How to inject status: 'prefix', 'system', or 'context'" ) debug_mode: bool = Field( default=False, description="Enable detailed logging" ) class UserValves(BaseModel): enabled: bool = Field( default=True, description="🔴 Enable or disable status injection for this user" ) custom_status: str = Field( default="", description="🔄 Override status from sidebar (leave empty to use stored)" ) def __init__(self): self.valves = self.Valves() self.uservalves = self.UserValves() self.store_path = os.path.join(os.getenv("DATA_DIR", "/app/backend/data"), "status_store.json") def _read_status(self, user_id: str) -> Dict: """Read status data from JSON store""" try: if os.path.exists(self.store_path): with open(self.store_path, "r") as f: data = json.load(f) return data.get(user_id, {}) return {} except Exception as e: LOGGER.error(f"Error reading status store: {e}", exc_info=True) return {} def _write_status(self, user_id: str, status: str, expires: Optional[float] = None) -> None: """Write status to the store file""" try: data = self._read_status(user_id) data[user_id] = { "status": status, "updated": time.time(), "expires": expires } os.makedirs(os.path.dirname(self.store_path), exist_ok=True) with open(self.store_path, "w") as f: json.dump(data, f) LOGGER.info(f"Status for user '{user_id}' updated to '{status}'.") except Exception as e: LOGGER.error(f"Failed to write status: {e}", exc_info=True) @log_exceptions async def inlet( self, body: dict, __user__: Optional[dict] = None, __event_emitter__: Optional[Callable] = None, **kwargs, ) -> dict: """Inject user status into messages before they reach the LLM""" if self.valves.debug_mode: set_logs(LOGGER, logging.DEBUG, force=True) LOGGER.debug("=== Status Filter Inlet Processing Starting ===") LOGGER.debug(f"Inlet body keys: {body.keys()}") if __user__ and hasattr(__user__, "valves") and not getattr(__user__.valves, "enabled", True): LOGGER.debug("Status injection disabled by user valve.") return body user_id = __user__.get("id") if __user__ else "anonymous" LOGGER.debug(f"Processing for user: {user_id}") if body.get("messages"): last_message = body["messages"][-1] if last_message.get("role") == "user": content = last_message.get("content", "").strip() if content.lower().startswith("!status "): new_status = content[8:].strip() if new_status: self._write_status(user_id, new_status) LOGGER.debug(f"Status updated to '{new_status}' via !status command.") if __event_emitter__: await __event_emitter__({ "type": "status", "data": { "description": f"Status updated to '{new_status}'", "done": True } }) # Halt processing by clearing messages, but return body to avoid downstream errors body["messages"] = [] return body status = self.valves.default_status status_source = "default" if __user__ and hasattr(__user__, "valves"): custom_status = getattr(__user__.valves, "custom_status", "").strip() if custom_status: status = custom_status status_source = "user_valve" LOGGER.debug(f"Using custom status from user valve: '{status}'") if status_source == "default": stored = self._read_status(user_id) if stored.get("status"): if stored.get("expires") and stored["expires"] < time.time(): LOGGER.debug("Stored status has expired, using default.") else: status = stored["status"] status_source = "stored" LOGGER.debug(f"Using stored status: '{status}'") if not body.get("messages"): LOGGER.debug("No messages found, skipping injection.") return body LOGGER.debug(f"Final status: '{status}' (from {status_source}). Injecting with format: '{self.valves.status_format}'") if self.valves.status_format == "system": system_msg = {"role": "system", "content": f"The user's current status is: {status}"} body["messages"].insert(0, system_msg) LOGGER.debug("Injected status as system message.") elif self.valves.status_format == "context": user_idx = next((i for i, msg in enumerate(body["messages"]) if msg.get("role") == "user"), None) if user_idx is not None: content = body["messages"][user_idx].get("content", "") context_xml = ( f'<details type="filters_context">\n' f'<summary>Filters context</summary>\n' f'<context id="status_injector">\n' f'<status>{status}</status>\n' f'</context>\n' f'</details>\n\n{content}' ) body["messages"][user_idx]["content"] = context_xml LOGGER.debug("Injected status as XML context.") else: # prefix user_idx = next((i for i, msg in enumerate(body["messages"]) if msg.get("role") == "user"), None) if user_idx is not None: content = body["messages"][user_idx].get("content", "") body["messages"][user_idx]["content"] = f"[UserStatus: {status}]\n\n{content}" LOGGER.debug("Injected status as prefix.") LOGGER.debug("=== Status Filter Inlet Processing Finished ===") return body