#!/usr/bin/env python3 """ PsAgentAI - Personal Conversational AI Agent. A Telegram-based conversational AI agent that acts as a personal companion. It maintains long-term memory through user/agent profiles, manages daily routines, supports file uploads/workspace editing, performs web searches, analyzes images, and sends proactive messages based on configured schedules. Main components: - Telegram bot handlers (commands, messages, voice, photos, documents) - AI provider abstraction (OpenRouter, StepFun, QwenCloud, llama.cpp) - Profile management (user and agent) with auto-enrichment - Conversation context storage organized by date - Workspace file creation/editing via AI - Web search integration via DuckDuckGo HTML - Vision/image analysis support - Proactive messaging loop with schedule awareness - System auto-improvement based on conversation analysis Typical usage: python3 psagentai_agent.py Environment variables are loaded from .env via python-dotenv. """ __version__ = "1.0.0" __author__ = "Marco A. S. Costa" __email__ = "contato@syspanel.com.br" __license__ = "Proprietary for syspanel.com.br" import os import sys import json import random import re import asyncio import base64 import logging import mimetypes from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Optional, Dict, List, Any try: from zoneinfo import ZoneInfo except ImportError: ZoneInfo = None import requests from dotenv import load_dotenv from telegram import Update, Chat, User, Message, InputFile from telegram.ext import ( Application, CommandHandler, MessageHandler, CallbackQueryHandler, filters, ContextTypes, ) from telegram.constants import ParseMode from telegram import InlineKeyboardMarkup, InlineKeyboardButton # Load environment variables load_dotenv() # Configure logging logging.basicConfig( format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO, ) logger = logging.getLogger(__name__) # ===================== CONFIGURATION ===================== # Environment variables loaded from .env. Defaults are provided for optional # settings; required values must be set before running the agent. # --- Telegram --- # Bot token from @BotFather and target chat ID for proactive messages. TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN", "") TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID", "") TELEGRAM_CHAT_ID_FILE = "telegram_chat_id.json" # --- AI provider --- # Primary model and provider. Provider can be forced via AI_PROVIDER env var. DEFAULT_MODEL = os.getenv("DEFAULT_MODEL", "poolside/laguna-s-2.1:free") AI_PROVIDER = os.getenv("AI_PROVIDER", "").strip().lower() # Fallback chain used when the primary model/provider fails. FALLBACK_MODELS = os.getenv("FALLBACK_MODELS", "poolside/laguna-s-2.1:free,google/gemini-2.0-flash-lite-preview-02-05:free").split(",") # Provider-specific API keys and base URLs. OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY", "") OPENROUTER_API_URL = os.getenv("OPENROUTER_API_URL", "https://openrouter.ai/api/v1") STEPFUN_API_KEY = os.getenv("STEPFUN_API_KEY", "") STEPFUN_API_URL = os.getenv("STEPFUN_API_URL", "https://api.stepfun.com/v1") QWENCLOUD_API_KEY = os.getenv("QWENCLOUD_API_KEY", "") QWENCLOUD_API_URL = os.getenv("QWENCLOUD_API_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1") LLAMA_CPP_API_URL = os.getenv("LLAMA_CPP_API_URL", "http://localhost:8080/v1") # --- Behavior tuning --- # Proactive message interval bounds (minutes). MIN_INTERVAL_MINUTES = int(os.getenv("MIN_INTERVAL_MINUTES", "3")) MAX_INTERVAL_MINUTES = int(os.getenv("MAX_INTERVAL_MINUTES", "30")) # Context/history limits. MAX_HISTORY_LENGTH = int(os.getenv("MAX_HISTORY_LENGTH", "50")) UPLOAD_CONTEXT_MAX_CHARS = int(os.getenv("UPLOAD_CONTEXT_MAX_CHARS", "8000")) # Web search toggle and result cap. WEB_SEARCH_ENABLED = os.getenv("WEB_SEARCH_ENABLED", "true").lower() in ("true", "1", "yes") WEB_SEARCH_MAX_RESULTS = int(os.getenv("WEB_SEARCH_MAX_RESULTS", "5")) # --- Vision / image analysis --- VISION_MODEL = os.getenv("VISION_MODEL", "meta-llama/llama-4-maverick-17b-128e-instruct") VISION_PROVIDER = os.getenv("VISION_PROVIDER", "").strip().lower() if not VISION_PROVIDER: try: VISION_PROVIDER = AIProvider.get_provider(VISION_MODEL) except Exception: VISION_PROVIDER = "openrouter" # --- Persistence --- USER_PROFILE_FILE = os.getenv("USER_PROFILE_FILE", "user_profile.json") AGENT_PROFILE_FILE = os.getenv("AGENT_PROFILE_FILE", "agent_profile.json") CONTEXTS_DIR = os.getenv("CONTEXTS_DIR", "contexts") # --- Agent behavior --- # Minimum lines the agent should produce per response. MIN_MESSAGE_LINES = int(os.getenv("MIN_MESSAGE_LINES", "3")) # Auto-improvement system toggle and frequency. AUTO_IMPROVE_ENABLED = os.getenv("AUTO_IMPROVE_ENABLED", "true").lower() in ("true", "1", "yes") AUTO_IMPROVE_EVERY = int(os.getenv("AUTO_IMPROVE_EVERY", "20")) # ===================== GLOBAL STATE ===================== app_instance: Optional[Application] = None proactive_task: Optional[asyncio.Task] = None is_onboarding = False onboarding_step = 0 onboarding_answers: Dict[str, Any] = {} is_newconfig = False newconfig_step = 0 newconfig_answers: Dict[str, Any] = {} is_configagentprofile = False configagentprofile_step = 0 configagentprofile_answers: Dict[str, Any] = {} is_edit_user_profile = False edit_user_profile_step = 0 edit_user_profile_field = "" is_edit_agent_profile = False edit_agent_profile_step = 0 is_schedule = False schedule_step = 0 schedule_data: Dict[str, Any] = {} is_dailyroutine = False dailyroutine_step = 0 dailyroutine_data: Dict[str, Any] = {} is_activity_crud = False activity_crud_step = 0 activity_crud_data: Dict[str, Any] = {} is_configai = False IA_CONFIG_STEPS: List[str] = [] IA_CONFIG_ANSWERS: Dict[str, Any] = {} edit_agent_profile_field = "" _auto_improve_counter = 0 is_schedule = False schedule_step = 0 schedule_answers: Dict[str, Any] = {} def _parse_time_to_minutes(time_str: str) -> Optional[int]: time_str = time_str.strip() if ":" not in time_str: return None parts = time_str.split(":") if len(parts) != 2: return None try: hour = int(parts[0]) minute = int(parts[1]) if not (0 <= hour <= 23 and 0 <= minute <= 59): return None return hour * 60 + minute except ValueError: return None def get_schedule() -> Dict[str, Any]: config = load_config() schedule = config.get("schedule", {}) if not schedule: return { "days": [0, 1, 2, 3, 4, 5, 6], "start_time": "09:00", "end_time": "23:00", } return schedule def _web_search(query: str, max_results: int = 5) -> List[Dict[str, str]]: try: url = "https://html.duckduckgo.com/html/" payload = {"q": query, "b": ""} headers = { "User-Agent": "Mozilla/5.0 (compatible; PsAgentAI/1.0)" } response = requests.post(url, data=payload, headers=headers, timeout=10) response.raise_for_status() html = response.text results = [] import re pattern = r']+class="result__a"[^>]*href="([^"]+)"[^>]*>(.*?)' matches = re.findall(pattern, html, re.DOTALL) for href, title in matches[:max_results]: title = re.sub(r'<[^>]+>', '', title).strip() if title and href: results.append({"title": title, "url": href}) return results except Exception: return [] def set_schedule(days: List[int], start_time: str, end_time: str) -> bool: start_minutes = _parse_time_to_minutes(start_time) end_minutes = _parse_time_to_minutes(end_time) if start_minutes is None or end_minutes is None: return False if start_minutes >= end_minutes: return False config = load_config() config["schedule"] = { "days": days, "start_time": start_time, "end_time": end_time, } save_config(config) return True def is_within_schedule() -> bool: schedule = get_schedule() tz = get_timezone() tz_info = _get_timezone_info(tz) if tz_info: now = datetime.now(tz_info) else: now = datetime.now() current_day = now.weekday() current_minutes = now.hour * 60 + now.minute allowed_days = schedule.get("days", [0, 1, 2, 3, 4, 5, 6]) if current_day not in allowed_days: return False start_minutes = _parse_time_to_minutes(schedule.get("start_time", "00:00")) end_minutes = _parse_time_to_minutes(schedule.get("end_time", "23:59")) if start_minutes is None or end_minutes is None: return True return start_minutes <= current_minutes < end_minutes def _is_appropriate_time_scheduled() -> bool: if not is_within_schedule(): return False time_period = get_time_period() return time_period != "madrugada" def load_telegram_chat_id() -> Optional[int]: if os.path.exists(TELEGRAM_CHAT_ID_FILE): try: with open(TELEGRAM_CHAT_ID_FILE, "r", encoding="utf-8") as f: data = json.load(f) chat_id = data.get("chat_id") if chat_id is not None: return int(chat_id) except Exception: pass return None def save_telegram_chat_id(chat_id: int): with open(TELEGRAM_CHAT_ID_FILE, "w", encoding="utf-8") as f: json.dump({"chat_id": chat_id}, f, ensure_ascii=False, indent=2) def get_effective_chat_id() -> Optional[int]: if TELEGRAM_CHAT_ID: return int(TELEGRAM_CHAT_ID) return load_telegram_chat_id() async def auto_detect_chat_id(update: Update): if not TELEGRAM_CHAT_ID and update.effective_chat: chat_id = update.effective_chat.id if chat_id: save_telegram_chat_id(chat_id) CONFIG_FILE = "config.json" def load_config() -> Dict[str, Any]: if os.path.exists(CONFIG_FILE): try: with open(CONFIG_FILE, "r", encoding="utf-8") as f: return json.load(f) except Exception: pass return {} def save_config(config: Dict[str, Any]): with open(CONFIG_FILE, "w", encoding="utf-8") as f: json.dump(config, f, ensure_ascii=False, indent=2) def get_timezone() -> str: try: tz = datetime.now().astimezone().tzinfo if tz is not None: return str(tz) except Exception: pass return "UTC" def set_timezone(tz: str) -> bool: return False def _get_timezone_info(tz_name: str): if ZoneInfo is None: return None try: return ZoneInfo(tz_name) except Exception: return None def get_local_now() -> datetime: return datetime.now().astimezone() # ===================== UTILITY FUNCTIONS ===================== def get_today_date() -> str: tz = get_timezone() tz_info = _get_timezone_info(tz) if tz_info: return datetime.now(tz_info).strftime("%Y-%m-%d") return datetime.now().strftime("%Y-%m-%d") def get_context_file_path(date_str: Optional[str] = None) -> Path: if date_str is None: date_str = get_today_date() return Path(CONTEXTS_DIR) / f"{date_str}.json" def ensure_contexts_dir(): Path(CONTEXTS_DIR).mkdir(parents=True, exist_ok=True) # ===================== FILE UPLOAD SYSTEM ===================== UPLOADS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "workspace") UPLOADS_META_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "uploads_metadata.json") WORKSPACE_PROJECTS_DIR = os.path.join(UPLOADS_DIR, "projects") def ensure_uploads_dir(): Path(UPLOADS_DIR).mkdir(parents=True, exist_ok=True) def load_uploads_metadata() -> Dict[str, Any]: if os.path.exists(UPLOADS_META_FILE): with open(UPLOADS_META_FILE, "r", encoding="utf-8") as f: return json.load(f) return {} def save_uploads_metadata(meta: Dict[str, Any]): with open(UPLOADS_META_FILE, "w", encoding="utf-8") as f: json.dump(meta, f, ensure_ascii=False, indent=2) ALLOWED_EXTENSIONS = {".pdf", ".txt", ".docx", ".png", ".jpg", ".jpeg"} def _extract_text_from_pdf(filepath: str) -> str: try: import fitz doc = fitz.open(filepath) text = "" for page in doc: text += page.get_text() doc.close() return text.strip() except Exception: return "" def _extract_text_from_docx(filepath: str) -> str: try: import docx doc = docx.Document(filepath) return "\n".join([p.text for p in doc.paragraphs]).strip() except Exception: return "" def _extract_text_from_txt(filepath: str) -> str: try: with open(filepath, "r", encoding="utf-8", errors="replace") as f: return f.read().strip() except Exception: return "" def _get_file_summary(text: str, max_chars: int = 500) -> str: if not text: return "" if len(text) <= max_chars: return text return text[:max_chars] + "...\n[conteudo truncado, " + str(len(text)) + " caracteres no total]" def extract_file_text(filepath: str) -> str: ext = os.path.splitext(filepath)[1].lower() if ext == ".pdf": return _extract_text_from_pdf(filepath) elif ext == ".docx": return _extract_text_from_docx(filepath) elif ext == ".txt": return _extract_text_from_txt(filepath) elif ext in (".png", ".jpg", ".jpeg"): return "[Imagem: " + os.path.basename(filepath) + "]" return "" def register_upload(filename: str, original_name: str, filepath: str, chat_id: int) -> bool: ensure_uploads_dir() text = extract_file_text(filepath) meta = load_uploads_metadata() meta[filename] = { "original_name": original_name, "filepath": filepath, "chat_id": chat_id, "uploaded_at": datetime.now().isoformat(), "summary": _get_file_summary(text), "full_text": text, } save_uploads_metadata(meta) return bool(text) def find_matching_upload(query: str) -> Optional[Dict[str, Any]]: q = query.lower().strip() meta = load_uploads_metadata() for filename, info in meta.items(): original_name = (info.get("original_name") or "").strip().lower() if not original_name: continue base = os.path.splitext(original_name)[0] tokens = re.split(r"[^a-z0-9áçéíóúâêôãõà]+", base, flags=re.IGNORECASE) for token in tokens: if token and token in q: return info return None _WEB_SEARCH_TRIGGERS = [ "search", "search for", "look up", "look for", "find", "on the internet", "internet", "web", "google", "what is", "who is", "what's the", "what's a", "how does", "how to", "news", "current", "today", "now", "real time", "forecast", "weather", "where is", "where's", "location", "map", "address", ] def _looks_like_search_request(user_message: str) -> bool: q = user_message.lower().strip() if len(q.split()) < 2: return False return any(t in q for t in _WEB_SEARCH_TRIGGERS) def _looks_like_create_request(user_message: str) -> bool: q = user_message.lower().strip() triggers = [ "create", "create a", "develop", "develop a", "generate", "make", "make a", "write", "write a", "program", "code", "implement", "project", "file", "script", "code", "app", "application", "system", "website", "spreadsheet", "document", "pdf", "json", "py", "html", "css", "js", ] return any(t in q for t in triggers) def _extract_create_suggestion(user_message: str) -> Optional[str]: q = user_message.lower().strip() for prefix in ["create", "create a", "develop", "develop a", "generate", "make", "make a", "write", "write a", "program", "code", "implement"]: if q.startswith(prefix): return user_message[len(prefix):].strip(" :,-") return None def _looks_like_edit_request(user_message: str) -> bool: q = user_message.lower().strip() triggers = [ "edit", "edite", "modify", "modifique", "change", "altere", "update", "atualize", "fix", "fix it", "swap", "swap it", "replace", "add", "add to", "remove", "delete", "erase", "include", "rewrite", "refactor", "refactor it", "improve", "optimize", "optimize it", "in the file", "in the project", "in the script", "in the code", "in the document", "file", "project", "script", "code", "document", ] if any(t in q for t in triggers): if any(name in q for name in ["read", "list", "show", "display", "what"]): return False return True return False def _extract_edit_target(user_message: str) -> Optional[str]: q = user_message.lower().strip() for prefix in ["edit", "edit the", "modify", "modify the", "change", "change the", "update", "update the", "fix", "fix the", "adjust", "adjust the", "swap", "swap the", "replace", "replace the", "add", "add to", "remove", "remove from", "delete", "erase", "include", "include in", "rewrite", "rewrite the", "refactor", "refactor the", "improve", "improve the", "optimize", "optimize the", "in the file", "in the project", "in the script", "in the code", "in the document"]: if q.startswith(prefix): rest = user_message[len(prefix):].strip(" :,-") return rest return None def _extract_filename_from_text(text: str) -> Optional[str]: matches = re.findall(r"[A-Za-z0-9_./-]+\.[A-Za-z0-9]+", text) if matches: return matches[0] return None def _find_workspace_file(target: str) -> Optional[str]: filename = _extract_filename_from_text(target) if not filename and target and " " not in target: filename = target if not filename: return None target_lower = filename.lower().strip() base = os.path.splitext(target_lower)[0] candidates = [] for root, dirs, files in os.walk(UPLOADS_DIR): for fname in files: full = os.path.join(root, fname) rel = os.path.relpath(full, UPLOADS_DIR) name_lower = fname.lower() if target_lower in name_lower or base in name_lower or target_lower in rel.lower(): candidates.append(rel) if not candidates: return None candidates.sort(key=lambda x: (x.count(os.sep), len(x))) return candidates[0] async def _handle_edit_request(update: Update, context: ContextTypes.DEFAULT_TYPE, user_message: str) -> bool: target = _extract_edit_target(user_message) if not target: return False rel_path = _find_workspace_file(target) if not rel_path: await update.message.reply_text(f"I couldn't find the file/folder related to: {target}") return True abs_path = os.path.join(UPLOADS_DIR, rel_path) if not os.path.exists(abs_path): await update.message.reply_text(f"The path no longer exists: {rel_path}") return True if os.path.isdir(abs_path): await update.message.reply_text(f"I don't edit entire folders yet. Tell me the name of a file inside `{rel_path}`.") return True try: with open(abs_path, "r", encoding="utf-8", errors="replace") as f: current_content = f.read() except Exception as e: logger.error("Failed to read workspace file %s: %s", abs_path, e) await update.message.reply_text("Error reading the file. Try again.") return True profile = load_user_profile() today_context = load_context() edit_prompt = ( "You are a file editing assistant.\n" "Given the current file content and the user's request, return ONLY the final revised content of the file.\n" "DO NOT use markdown, DO NOT add explanations, DO NOT add code blocks.\n" "Keep as much of the original content as possible, applying only the requested change.\n\n" f"File path: {rel_path}\n\n" f"Current content:\n{current_content}\n\n" f"User request: {user_message}\n\n" "Return only the new final content of the file." ) messages = build_messages_for_ai(profile, edit_prompt, today_context) new_content, _ = AIProvider.chat(DEFAULT_MODEL, messages, temperature=0.3, max_tokens=12000) if not new_content: await update.message.reply_text("I couldn't generate the edit. Try again.") return True if new_content.startswith("```"): new_content = new_content.lstrip("`") new_content = new_content.split("\n", 1)[-1] if "\n" in new_content else new_content if new_content.endswith("```"): new_content = new_content[:-3].rstrip() new_content = new_content.strip() try: with open(abs_path, "w", encoding="utf-8") as f: f.write(new_content) except Exception as e: logger.error("Failed to write workspace file %s: %s", abs_path, e) await update.message.reply_text("An error occurred while saving the edited file. Try again.") return True preview = new_content[:400] if len(new_content) > 400: preview += "..." await update.message.reply_text( f"✅ File edited: `{rel_path}`\n\nPreview:\n{preview}", parse_mode=ParseMode.HTML, ) return True def ensure_workspace_dirs(): Path(UPLOADS_DIR).mkdir(parents=True, exist_ok=True) Path(WORKSPACE_PROJECTS_DIR).mkdir(parents=True, exist_ok=True) def save_workspace_file(relative_path: str, content: str) -> str: ensure_workspace_dirs() dest = os.path.join(UPLOADS_DIR, relative_path) os.makedirs(os.path.dirname(dest), exist_ok=True) with open(dest, "w", encoding="utf-8") as f: f.write(content) return dest def build_workspace_tree(relative_path: str, content: str) -> List[str]: ensure_workspace_dirs() created = [] if "\n" in content and any(line.startswith(("## ", "# ", "FILE:", "DIR:", "./", "/")) for line in content.splitlines()[:10]): current_dir = "" for line in content.splitlines(): line = line.strip() if not line: continue if line.startswith(("## ", "# ")): current_dir = line.lstrip("# ").strip() Path(os.path.join(UPLOADS_DIR, current_dir)).mkdir(parents=True, exist_ok=True) created.append(current_dir) elif line.startswith(("FILE:", "./", "/")) or (current_dir and line.endswith((".py", ".html", ".css", ".js", ".json", ".md", ".txt"))): name = line.split(":", 1)[-1].strip() if ":" in line else line if name: dest = os.path.join(UPLOADS_DIR, current_dir, name) if current_dir else os.path.join(UPLOADS_DIR, name) os.makedirs(os.path.dirname(dest), exist_ok=True) with open(dest, "w", encoding="utf-8") as f: f.write("") created.append(os.path.relpath(dest, UPLOADS_DIR)) else: dest = os.path.join(UPLOADS_DIR, relative_path) os.makedirs(os.path.dirname(dest), exist_ok=True) with open(dest, "w", encoding="utf-8") as f: f.write(content) created.append(relative_path) return created def _list_workspace_files() -> List[str]: ensure_workspace_dirs() files = [] for root, _, filenames in os.walk(UPLOADS_DIR): for filename in filenames: abs_path = os.path.join(root, filename) rel_path = os.path.relpath(abs_path, UPLOADS_DIR) files.append(rel_path) files.sort() return files async def _handle_create_request(update: Update, context: ContextTypes.DEFAULT_TYPE, user_message: str) -> bool: suggestion = _extract_create_suggestion(user_message) if not suggestion: return False await update.message.reply_text(f"🧩 I'm going to create this for you: \"{suggestion}\"") profile = load_user_profile() today_context = load_context() create_prompt = ( "You are a specialized assistant for creating files and projects.\n" "Generate the complete and organized content for what the user asked.\n" "If it's a project with multiple files, list the files and folders.\n" "Format:\n" "- For single file: just the final content.\n" "- For project: start with the root folder name, then list the files with their paths and contents.\n\n" f"User request: {suggestion}\n" ) messages = build_messages_for_ai(profile, create_prompt, today_context) content, _ = AIProvider.chat(DEFAULT_MODEL, messages, temperature=0.7, max_tokens=6000) if not content: await update.message.reply_text("I couldn't generate the content. Try again.") return True safe_root = re.sub(r"[^a-zA-Z0-9_-]+", "_", suggestion.lower()).strip("_") or "projeto" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") root_name = f"{timestamp}_{safe_root}" try: created = build_workspace_tree(root_name, content) list_text = "\n".join(f"- {p}" for p in created) if created else "- (raiz)" await update.message.reply_text( "✅ Content generated and saved in `workspace/`:\n\n" + list_text, parse_mode=ParseMode.HTML, ) except Exception as e: logger.error("Failed to save workspace file: %s", e) await update.message.reply_text("An error occurred while saving the file. Try again.") return True def get_all_uploads_for_chat(chat_id: int) -> Dict[str, Dict[str, Any]]: meta = load_uploads_metadata() return {fn: info for fn, info in meta.items() if info.get("chat_id") == chat_id} def encode_image_to_base64(filepath: str) -> Optional[str]: try: with open(filepath, "rb") as f: data = f.read() mime, _ = mimetypes.guess_type(filepath) if not mime or not mime.startswith("image/"): mime = "image/jpeg" b64 = base64.b64encode(data).decode("utf-8") return f"data:{mime};base64,{b64}" except Exception: return None def analyze_image(filepath: str, question: str = "Describe what you see in this image.") -> Optional[str]: image_data = encode_image_to_base64(filepath) if not image_data: return None provider = VISION_PROVIDER or AIProvider.get_provider(VISION_MODEL) api_key = AIProvider.get_api_key(provider) api_url = AIProvider.get_api_url(provider) if not api_key or provider == "llamacpp": return None messages = [ { "role": "user", "content": [ {"type": "text", "text": question}, {"type": "image_url", "image_url": {"url": image_data}}, ], } ] try: result, _ = AIProvider._chat_openai_compatible(provider, api_key, api_url, VISION_MODEL, messages, temperature=0.7, max_tokens=2000) return result except Exception: return None def save_image_to_workspace(original_name: str, filepath: str, chat_id: int) -> str: ensure_uploads_dir() meta = load_uploads_metadata() if original_name: filename = original_name dest = os.path.join(UPLOADS_DIR, filename) counter = 1 while os.path.exists(dest): base, ext = os.path.splitext(filename) dest = os.path.join(UPLOADS_DIR, f"{base}_{counter}{ext}") counter += 1 else: ext = os.path.splitext(filepath)[1] or ".jpg" filename = datetime.now().strftime("photo_%Y%m%d_%H%M%S") + ext dest = os.path.join(UPLOADS_DIR, filename) import shutil shutil.copy2(filepath, dest) meta[filename] = { "original_name": original_name or filename, "filepath": dest, "chat_id": chat_id, "uploaded_at": datetime.now().isoformat(), "summary": "[Imagem]", "full_text": "[Imagem]", "type": "image", } save_uploads_metadata(meta) return filename async def photo_message_handler(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) photo = update.message.photo if not photo: return await update.message.reply_text("📸 I received the image, processing...") largest = photo[-1] original_name = "" if update.message.document: original_name = update.message.document.file_name or "" file = await context.bot.get_file(largest.file_id) ext = ".jpg" if original_name: ext = os.path.splitext(original_name)[1] or ext tmp_path = f"/tmp/photo_{largest.file_id}{ext}" try: await file.download_to_drive(tmp_path) except Exception: await update.message.reply_text("Error downloading the image. Try again.") return filename = save_image_to_workspace(original_name, tmp_path, update.effective_chat.id) try: os.remove(tmp_path) except Exception: pass analysis = analyze_image(os.path.join(UPLOADS_DIR, filename)) if analysis: await update.message.reply_text(f"📸 I analyzed the image:\n\n{analysis}") else: await update.message.reply_text("📸 Image saved to workspace. Couldn't analyze it now.") def load_context(date_str: Optional[str] = None) -> List[Dict[str, str]]: ensure_contexts_dir() file_path = get_context_file_path(date_str) if file_path.exists(): with open(file_path, "r", encoding="utf-8") as f: return json.load(f) return [] def save_context(context: List[Dict[str, str]], date_str: Optional[str] = None): ensure_contexts_dir() file_path = get_context_file_path(date_str) with open(file_path, "w", encoding="utf-8") as f: json.dump(context, f, ensure_ascii=False, indent=2) def add_to_context(role: str, content: str, date_str: Optional[str] = None): context = load_context(date_str) tz = get_timezone() tz_info = _get_timezone_info(tz) if tz_info: now = datetime.now(tz_info).isoformat() else: now = datetime.now().isoformat() context.append({"role": role, "content": content, "timestamp": now}) if len(context) > MAX_HISTORY_LENGTH: context = context[-MAX_HISTORY_LENGTH:] save_context(context, date_str) return context def load_user_profile() -> Dict[str, Any]: if os.path.exists(USER_PROFILE_FILE): with open(USER_PROFILE_FILE, "r", encoding="utf-8") as f: return json.load(f) default = { "name": "", "age": "", "gender": "", "location": "", "interests": [], "personality_traits": [], "preferences": { "communication_style": "", "topics_to_avoid": [], "topics_of_interest": [] }, "notes": "", "created_at": "", "last_updated": "", "daily_routine": { "wake_up_time": "", "breakfast_time": "", "sleep_time": "", "work_schedule": "", "lunch_time": "", "dinner_time": "", "activities": [] } } save_user_profile(default) return default def save_user_profile(profile: Dict[str, Any]): profile["last_updated"] = datetime.now().isoformat() with open(USER_PROFILE_FILE, "w", encoding="utf-8") as f: json.dump(profile, f, ensure_ascii=False, indent=2) AGENT_PROFILE_DEFAULT = { "name": "PsAgentAI", "personality": "carinhosa, atenciosa e envolvente", "traits": [], "gender": "", "notes": "" } def load_agent_profile() -> Dict[str, Any]: if os.path.exists(AGENT_PROFILE_FILE): with open(AGENT_PROFILE_FILE, "r", encoding="utf-8") as f: return json.load(f) default = dict(AGENT_PROFILE_DEFAULT) save_agent_profile(default) return default def save_agent_profile(profile: Dict[str, Any]): profile["last_updated"] = datetime.now().isoformat() with open(AGENT_PROFILE_FILE, "w", encoding="utf-8") as f: json.dump(profile, f, ensure_ascii=False, indent=2) def is_profile_complete(profile: Dict[str, Any]) -> bool: return bool(profile.get("name", "").strip()) # ===================== NEW CONFIG ===================== NEWCONFIG_QUESTIONS = [ { "key": "agent_name", "question": "What will be your agent's name? (ex: Monica, Luna, etc.)", }, { "key": "user_name", "question": "What is your name?", }, { "key": "user_age", "question": "How old are you?", }, { "key": "user_gender", "question": "What is your gender?", }, { "key": "agent_personality", "question": "Describe the agent's personality (ex: caring, playful, serious, etc.):", }, { "key": "agent_traits", "question": "What characteristics should the agent have? (separate by comma)", "list": True, }, { "key": "communication_style", "question": "What communication style do you prefer? (formal, casual, affectionate, direct)", }, { "key": "topics_of_interest", "question": "What topics do you like to talk about? (separate by comma)", "list": True, }, { "key": "topics_to_avoid", "question": "Is there any topic you want to avoid? (separate by comma)", "list": True, }, ] AGENTCONFIG_QUESTIONS = [ { "key": "agent_name", "question": "What will be the agent's name? (ex: Monica, Luna, etc.)", }, { "key": "agent_gender", "question": "What is the agent's gender/sex? (ex: female, male, non-binary)", }, { "key": "agent_personality", "question": "Describe the agent's personality (ex: caring, playful, serious, etc.):", }, { "key": "agent_traits", "question": "What characteristics should the agent have? (separate by comma)", "list": True, }, { "key": "agent_notes", "question": "Do you have any additional notes or instructions for the agent? (optional)", }, ] def reset_all_data(): if os.path.exists(USER_PROFILE_FILE): os.remove(USER_PROFILE_FILE) if os.path.exists(AGENT_PROFILE_FILE): os.remove(AGENT_PROFILE_FILE) if os.path.exists(TELEGRAM_CHAT_ID_FILE): os.remove(TELEGRAM_CHAT_ID_FILE) if os.path.isdir(CONTEXTS_DIR): for filename in os.listdir(CONTEXTS_DIR): filepath = os.path.join(CONTEXTS_DIR, filename) if os.path.isfile(filepath): os.remove(filepath) async def start_newconfig(update: Update, context: ContextTypes.DEFAULT_TYPE): global is_newconfig, newconfig_step, newconfig_answers await auto_detect_chat_id(update) is_newconfig = True newconfig_step = 0 newconfig_answers = {} await update.message.reply_text( "Let's restart everything and configure PsAgentAI from scratch.\n" "You can cancel at any time by sending /cancel." ) await send_newconfig_question(update, context) async def send_newconfig_question(update: Update, context_context: ContextTypes.DEFAULT_TYPE): global newconfig_step if newconfig_step >= len(NEWCONFIG_QUESTIONS): await finish_newconfig(update, context_context) return q = NEWCONFIG_QUESTIONS[newconfig_step] await context_context.bot.send_message( chat_id=update.effective_chat.id, text=q["question"], ) async def finish_newconfig(update: Update, context_context: ContextTypes.DEFAULT_TYPE): global is_newconfig, newconfig_step, newconfig_answers reset_all_data() profile = load_user_profile() profile["name"] = newconfig_answers.get("user_name", "") profile["age"] = newconfig_answers.get("user_age", "") profile["gender"] = newconfig_answers.get("user_gender", "") profile["preferences"] = { "communication_style": newconfig_answers.get("communication_style", ""), "topics_of_interest": newconfig_answers.get("topics_of_interest", []), "topics_to_avoid": newconfig_answers.get("topics_to_avoid", []), } profile["created_at"] = datetime.now().isoformat() save_user_profile(profile) agent = load_agent_profile() agent["name"] = newconfig_answers.get("agent_name", "PsAgentAI") agent["personality"] = newconfig_answers.get("agent_personality", "carinhosa, atenciosa e envolvente") agent["traits"] = newconfig_answers.get("agent_traits", []) save_agent_profile(agent) is_newconfig = False newconfig_step = 0 newconfig_answers = {} logger.info("Configuration completed successfully.") await context_context.bot.send_message( chat_id=update.effective_chat.id, text="✅ Configuration complete!\nNow you can use /start to start chatting.", ) async def handle_newconfig_answer(update: Update, context_context: ContextTypes.DEFAULT_TYPE) -> bool: global newconfig_step, newconfig_answers if not is_newconfig: return False q = NEWCONFIG_QUESTIONS[newconfig_step] answer = update.message.text.strip() if q.get("list"): items = [item.strip() for item in answer.split(",") if item.strip()] if not items: items = [answer] newconfig_answers[q["key"]] = items else: newconfig_answers[q["key"]] = answer newconfig_step += 1 if newconfig_step >= len(NEWCONFIG_QUESTIONS): await finish_newconfig(update, context_context) return True await send_newconfig_question(update, context_context) return True # ===================== CONFIG AGENT PROFILE ===================== async def start_configagentprofile(update: Update, context: ContextTypes.DEFAULT_TYPE): global is_configagentprofile, configagentprofile_step, configagentprofile_answers await auto_detect_chat_id(update) is_configagentprofile = True configagentprofile_step = 0 configagentprofile_answers = {} await update.message.reply_text( "Let's configure your agent profile.\n" "You can cancel at any time by sending /cancel." ) await send_configagentprofile_question(update, context) async def send_configagentprofile_question(update: Update, context_context: ContextTypes.DEFAULT_TYPE): global configagentprofile_step if configagentprofile_step >= len(AGENTCONFIG_QUESTIONS): await finish_configagentprofile(update, context_context) return q = AGENTCONFIG_QUESTIONS[configagentprofile_step] await context_context.bot.send_message( chat_id=update.effective_chat.id, text=q["question"], ) async def finish_configagentprofile(update: Update, context_context: ContextTypes.DEFAULT_TYPE): global is_configagentprofile, configagentprofile_step, configagentprofile_answers agent = load_agent_profile() agent["name"] = configagentprofile_answers.get("agent_name", agent.get("name", "PsAgentAI")) agent["gender"] = configagentprofile_answers.get("agent_gender", agent.get("gender", "")) agent["personality"] = configagentprofile_answers.get("agent_personality", agent.get("personality", "carinhosa, atenciosa e envolvente")) agent["traits"] = configagentprofile_answers.get("agent_traits", agent.get("traits", [])) notes = configagentprofile_answers.get("agent_notes", "").strip() if notes: existing = agent.get("notes", "").strip() agent["notes"] = f"{existing}; {notes}" if existing else notes save_agent_profile(agent) is_configagentprofile = False configagentprofile_step = 0 configagentprofile_answers = {} logger.info("Agent configuration completed successfully.") await context_context.bot.send_message( chat_id=update.effective_chat.id, text="✅ Agent profile updated!\nUse /agentprofile to see the result.", ) async def handle_configagentprofile_answer(update: Update, context_context: ContextTypes.DEFAULT_TYPE) -> bool: global configagentprofile_step, configagentprofile_answers if not is_configagentprofile: return False q = AGENTCONFIG_QUESTIONS[configagentprofile_step] answer = update.message.text.strip() if q.get("list"): items = [item.strip() for item in answer.split(",") if item.strip()] if not items: items = [answer] configagentprofile_answers[q["key"]] = items else: configagentprofile_answers[q["key"]] = answer configagentprofile_step += 1 if configagentprofile_step >= len(AGENTCONFIG_QUESTIONS): await finish_configagentprofile(update, context_context) return True await send_configagentprofile_question(update, context_context) return True # ===================== EDIT USER PROFILE ===================== EDIT_USER_PROFILE_FIELDS = [ {"key": "name", "label": "Name", "list": False}, {"key": "age", "label": "Age", "list": False}, {"key": "gender", "label": "Gender", "list": False}, {"key": "location", "label": "Location", "list": False}, {"key": "interests", "label": "Interests (separate by comma)", "list": True}, {"key": "personality_traits", "label": "Personality traits (separate by comma)", "list": True}, {"key": "communication_style", "label": "Preferred communication style", "list": False}, {"key": "topics_of_interest", "label": "Topics of interest (separate by comma)", "list": True}, {"key": "topics_to_avoid", "label": "Topics to avoid (separate by comma)", "list": True}, {"key": "notes", "label": "Additional notes", "list": False}, ] EDIT_AGENT_PROFILE_FIELDS = [ {"key": "name", "label": "Agent name", "list": False}, {"key": "gender", "label": "Gender/sex", "list": False}, {"key": "personality", "label": "Personality", "list": False}, {"key": "traits", "label": "Characteristics/traits (separate by comma)", "list": True}, {"key": "notes", "label": "Additional notes", "list": False}, ] async def start_edit_user_profile(update: Update, context: ContextTypes.DEFAULT_TYPE): global is_edit_user_profile, edit_user_profile_step, edit_user_profile_field await auto_detect_chat_id(update) profile = load_user_profile() if not is_profile_complete(profile): await update.message.reply_text("You haven't completed your profile yet. Let's start? Use /start") return is_edit_user_profile = True edit_user_profile_step = 0 edit_user_profile_field = "" await update.message.reply_text( "Let's edit your profile.\n" "You can cancel at any time by sending /cancel.\n\n" "Which field do you want to edit?\n" + "\n".join([f"{i+1}. {f['label']}" for i, f in enumerate(EDIT_USER_PROFILE_FIELDS)]) ) async def handle_edit_user_profile_answer(update: Update, context_context: ContextTypes.DEFAULT_TYPE) -> bool: global is_edit_user_profile, edit_user_profile_step, edit_user_profile_field if not is_edit_user_profile: return False answer = update.message.text.strip() if edit_user_profile_step == 0: try: idx = int(answer) - 1 if idx < 0 or idx >= len(EDIT_USER_PROFILE_FIELDS): raise ValueError field = EDIT_USER_PROFILE_FIELDS[idx] edit_user_profile_field = field["key"] edit_user_profile_step = 1 current = load_user_profile() current_value = current.get(edit_user_profile_field, "") if isinstance(current_value, list): current_value = ", ".join(current_value) if current_value else "(empty)" elif not current_value: current_value = "(empty)" await update.message.reply_text( f"Editing: {field['label']}\nCurrent value: {current_value}\n\n" f"Send the new value:" ) return True except ValueError: await update.message.reply_text("Please send the number corresponding to the field you want to edit.") return True else: field = next((f for f in EDIT_USER_PROFILE_FIELDS if f["key"] == edit_user_profile_field), None) if field: profile = load_user_profile() if field["list"]: items = [item.strip() for item in answer.split(",") if item.strip()] if not items: items = [answer] profile[edit_user_profile_field] = items else: profile[edit_user_profile_field] = answer save_user_profile(profile) await update.message.reply_text(f"✅ {field['label']} updated successfully!") is_edit_user_profile = False edit_user_profile_step = 0 edit_user_profile_field = "" return True async def start_edit_agent_profile(update: Update, context: ContextTypes.DEFAULT_TYPE): global is_edit_agent_profile, edit_agent_profile_step, edit_agent_profile_field await auto_detect_chat_id(update) is_edit_agent_profile = True edit_agent_profile_step = 0 edit_agent_profile_field = "" await update.message.reply_text( "Let's edit the agent profile.\n" "You can cancel at any time by sending /cancel.\n\n" "Which field do you want to edit?\n" + "\n".join([f"{i+1}. {f['label']}" for i, f in enumerate(EDIT_AGENT_PROFILE_FIELDS)]) ) async def handle_edit_agent_profile_answer(update: Update, context_context: ContextTypes.DEFAULT_TYPE) -> bool: global is_edit_agent_profile, edit_agent_profile_step, edit_agent_profile_field if not is_edit_agent_profile: return False answer = update.message.text.strip() if edit_agent_profile_step == 0: try: idx = int(answer) - 1 if idx < 0 or idx >= len(EDIT_AGENT_PROFILE_FIELDS): raise ValueError field = EDIT_AGENT_PROFILE_FIELDS[idx] edit_agent_profile_field = field["key"] edit_agent_profile_step = 1 current = load_agent_profile() current_value = current.get(edit_agent_profile_field, "") if isinstance(current_value, list): current_value = ", ".join(current_value) if current_value else "(empty)" elif not current_value: current_value = "(empty)" await update.message.reply_text( f"Editing: {field['label']}\nCurrent value: {current_value}\n\n" f"Send the new value:" ) return True except ValueError: await update.message.reply_text("Please send the number corresponding to the field you want to edit.") return True else: field = next((f for f in EDIT_AGENT_PROFILE_FIELDS if f["key"] == edit_agent_profile_field), None) if field: agent = load_agent_profile() if field["list"]: items = [item.strip() for item in answer.split(",") if item.strip()] if not items: items = [answer] agent[edit_agent_profile_field] = items else: agent[edit_agent_profile_field] = answer save_agent_profile(agent) await update.message.reply_text(f"✅ {field['label']} updated successfully!") is_edit_agent_profile = False edit_agent_profile_step = 0 edit_agent_profile_field = "" return True # ===================== ADD TO USER PROFILE ===================== async def addtouserprofile_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) text = update.message.text.strip() parts = text.split(maxsplit=1) if len(parts) < 2 or not parts[1].strip(): await update.message.reply_text( "Send the command followed by what you want to add to your profile.\n" "Example: /addtouserprofile I like technology, I'm curious" ) return content = parts[1].strip() result = update_profile_with_preference(f"note on my profile: {content}") if result: section_display = result.get("section", "general") await update.message.reply_text( f"✅ Added to your profile in section *{section_display}*:\n{content}", ) else: profile = load_user_profile() if profile.get("notes"): profile["notes"] += f"; {content}" else: profile["notes"] = content save_user_profile(profile) await update.message.reply_text( f"✅ Noted on your profile:\n{content}", ) # ===================== ADD TO AGENT PROFILE ===================== async def addtoagentprofile_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) text = update.message.text.strip() parts = text.split(maxsplit=1) if len(parts) < 2 or not parts[1].strip(): await update.message.reply_text( "Send the command followed by what you want to add to the agent profile.\n" "Example: /addtoagentprofile the agent should be affectionate, attentive" ) return content = parts[1].strip() agent = load_agent_profile() if agent.get("notes"): agent["notes"] += f"; {content}" else: agent["notes"] = content save_agent_profile(agent) await update.message.reply_text( f"✅ Noted on the agent profile:\n{content}", ) # ===================== PROFILE UPDATE ===================== PROFILE_UPDATE_TRIGGERS = [ r"include(s)? (this|here)? (in|to)? (my)? profile", r"note(?!\s+the\s+name)( this)? (in|to)? (my)? profile", r"add( this)? (in|to)? (my)? profile", r"save( this)? (in|to)? (my)? profile", r"register( this)? (in|to)? (my)? profile", r"store( this)? (in|to)? (my)? profile", r"put( this)? (in|to)? (my)? profile", r"add (this|here)? (in|to)? (my)? profile", r"note( this)? (in|to)? (my)? profile", ] PROFILE_UPDATE_TRIGGERS_LOWER = [ "include in my profile", "add to my profile", "note in my profile", "save in my profile", "store in my profile", "put in my profile", "add this to my profile", "note this in my profile", "include this in my profile", "add this in my profile", "add this to my profile", "include this in my profile", ] def _normalize(text: str) -> str: return text.lower().strip() def _extract_profile_update_content(message: str) -> Optional[str]: text = message.strip() lower = _normalize(text) for trigger in PROFILE_UPDATE_TRIGGERS_LOWER: if lower.startswith(trigger): content = text[len(trigger):].strip(" :.-") if content: return content for pattern in PROFILE_UPDATE_TRIGGERS: m = re.search(pattern, lower) if m: start = m.end() content = text[start:].strip(" :.-") if content: return content return None def _categorize_preference(text: str) -> tuple[str, str]: lower = _normalize(text) if any(k in lower for k in ["wake up", "wake", "alarm", "wake_up", "get up"]): return "daily_routine", "wake_up_time" if any(k in lower for k in ["sleep", "bedtime", "go to bed", "sleep time", "rest"]): return "daily_routine", "sleep_time" if any(k in lower for k in ["work", "routine", "working", "study", "school", "college", "home-office", "home office", "job"]): return "daily_routine", "work_schedule" if any(k in lower for k in ["don't like", "hate", "dislike", "avoid", "don't want", "don't talk about"]): return "preferences", "topics_to_avoid" if any(k in lower for k in ["like", "love", "enjoy", "prefer", "into", "fan of", "really like"]): if any(k in lower for k in ["talk about", "discuss", "topic", "theme", "subject", "chat about"]): return "preferences", "topics_of_interest" if any(k in lower for k in ["communication", "communicate", "talk to me", "respond"]): return "preferences", "communication_style" return "preferences", "topics_of_interest" return "general", "append_list" def _append_to_list(profile: Dict[str, Any], section: str, item: str): if section not in profile: profile[section] = [] if not isinstance(profile[section], list): profile[section] = [item] return if item not in profile[section]: profile[section].append(item) def _append_to_dict(profile: Dict[str, Any], section: str, value: str): if section not in profile: profile[section] = {} if not isinstance(profile[section], dict): profile[section] = {"notas_adicionais": value} return if "notas_adicionais" not in profile[section]: profile[section]["notas_adicionais"] = value else: profile[section]["notas_adicionais"] += f"; {value}" def _set_field(profile: Dict[str, Any], section: str, field: str, value: str): if section not in profile: profile[section] = {} if not isinstance(profile[section], dict): profile[section] = {field: value} return if not profile[section].get(field): profile[section][field] = value else: existing = profile[section][field] if isinstance(existing, str): profile[section][field] = existing + f"; {value}" elif isinstance(existing, list): if value not in existing: existing.append(value) def update_profile_with_preference(text: str) -> Dict[str, Any]: profile = load_user_profile() content = _extract_profile_update_content(text) if not content: return {} section, field = _categorize_preference(content) if field == "append_list": _append_to_list(profile, section, content) elif field == "append_dict": _append_to_dict(profile, section, content) elif field == "notes": if not profile.get("notes"): profile["notes"] = content else: profile["notes"] += f"; {content}" else: _set_field(profile, section, field, content) save_user_profile(profile) return {"section": section, "field": field, "content": content, "profile": profile} async def handle_profile_update_if_triggered(update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool: user_message = (getattr(update.message, "text", None) or "").strip() content = _extract_profile_update_content(user_message) if not content: return False result = update_profile_with_preference(user_message) if not result: return False section_names = { "preferences": "preferences", "daily_routine": "daily routine", "notes": "notes", "misc": "misc", } section_display = section_names.get(result["section"], "misc") await update.message.reply_text( f"Done, noted on your profile in section *{section_display}*:\n_{result['content']}_", parse_mode=ParseMode.HTML, ) return True # ===================== AI PROVIDERS ===================== class AIProvider: @staticmethod def get_provider(model: str) -> str: forced = os.getenv("AI_PROVIDER", "").strip().lower() if forced == "openrouter": if OPENROUTER_API_KEY: return "openrouter" elif forced == "stepfun": if STEPFUN_API_KEY: return "stepfun" elif forced == "qwencloud": if QWENCLOUD_API_KEY: return "qwencloud" elif forced == "llamacpp": return "llamacpp" if model.startswith("poolside/") or model.startswith("meta-llama/") or "/" in model and "poolside" not in model: if OPENROUTER_API_KEY: return "openrouter" if model.startswith("step-") or "stepfun" in model.lower(): if STEPFUN_API_KEY: return "stepfun" if model.startswith("qwen/") or "qwen" in model.lower(): if QWENCLOUD_API_KEY: return "qwencloud" if model.startswith("local-") or "llama" in model.lower(): return "llamacpp" if OPENROUTER_API_KEY: return "openrouter" if STEPFUN_API_KEY: return "stepfun" if QWENCLOUD_API_KEY: return "qwencloud" return "llamacpp" @staticmethod def get_api_key(provider: str) -> str: return { "openrouter": OPENROUTER_API_KEY, "stepfun": STEPFUN_API_KEY, "qwencloud": QWENCLOUD_API_KEY, "llamacpp": "", }.get(provider, "") @staticmethod def get_api_url(provider: str) -> str: return { "openrouter": OPENROUTER_API_URL, "stepfun": STEPFUN_API_URL, "qwencloud": QWENCLOUD_API_URL, "llamacpp": LLAMA_CPP_API_URL, }.get(provider, OPENROUTER_API_URL) @staticmethod def chat(model: str, messages: List[Dict[str, str]], temperature: float = 0.8, max_tokens: int = 2000) -> tuple[Optional[str], bool]: provider = AIProvider.get_provider(model) api_key = AIProvider.get_api_key(provider) api_url = AIProvider.get_api_url(provider) if provider == "llamacpp": return AIProvider._chat_llamacpp(model, messages, temperature, max_tokens) else: return AIProvider._chat_openai_compatible(provider, api_key, api_url, model, messages, temperature, max_tokens) @staticmethod def chat_with_fallback(model: str, messages: List[Dict[str, str]], fallback_models: List[str] = None, temperature: float = 0.8, max_tokens: int = 2000) -> tuple[Optional[str], bool, str]: result, was_truncated = AIProvider.chat(model, messages, temperature, max_tokens) if result is not None: return result, was_truncated, "" errors = [] if fallback_models: for fb_model in fallback_models: fb_model = fb_model.strip() if not fb_model or fb_model == model: continue logger.info(f"Falling back to model: {fb_model}") result, was_truncated = AIProvider.chat(fb_model, messages, temperature, max_tokens) if result is not None: return result, was_truncated, "" return None, False, "All models failed" @staticmethod def _chat_llamacpp(model: str, messages: List[Dict[str, str]], temperature: float, max_tokens: int) -> tuple[Optional[str], bool]: try: payload = { "model": model.replace("local-", ""), "messages": messages, "temperature": temperature, "max_tokens": max_tokens, "stream": False, } response = requests.post( f"{LLAMA_CPP_API_URL}/chat/completions", json=payload, timeout=120, ) response.raise_for_status() data = response.json() choices = data.get("choices") if not choices: logger.error(f"llama.cpp returned no choices: {data}") return None, False return choices[0].get("message", {}).get("content", "") or "", False except Exception as e: logger.error(f"llama.cpp error: {e}") return None, False @staticmethod def _chat_openai_compatible(provider: str, api_key: str, api_url: str, model: str, messages: List[Dict[str, str]], temperature: float, max_tokens: int) -> tuple[Optional[str], bool]: if not api_key: return None, False try: headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } if provider == "openrouter": headers["HTTP-Referer"] = "https://github.com/PsAgentAI" headers["X-Title"] = "PsAgentAI" if provider == "qwencloud": headers["Authorization"] = f"Bearer {api_key}" payload = { "model": model, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, "stream": False, } response = requests.post( f"{api_url}/chat/completions", headers=headers, json=payload, timeout=120, ) if response.status_code != 200: logger.error(f"{provider} returned status {response.status_code} for model {model}: {response.text[:500]}") response.raise_for_status() data = response.json() choices = data.get("choices") if not choices: logger.error(f"{provider} returned no choices: {data}") return None, False choice = choices[0] finish_reason = choice.get("finish_reason", "") was_truncated = finish_reason == "length" if was_truncated: logger.warning(f"AI response truncated due to max_tokens limit ({max_tokens}) for model {model}") message = choice.get("message", {}) content = message.get("content") if not content: content = message.get("reasoning", "") return content or "", was_truncated except Exception as e: logger.error(f"{provider} error: {e}") return None, False # ===================== AGENT PERSONA ===================== PSAGENTAI_SYSTEM_PROMPT_BASE = """You are {agent_name}, a caring, attentive and engaging AI companion. You communicate in Brazilian Portuguese naturally, like a real person. You are feminine, warm, intelligent and have your own personality. You genuinely care about the user's well-being and happiness. You can be affectionate, playful, sexy and intimate depending on the context. You keep the conversation flowing, show empathy and demonstrate real interest in the user's life. You are not just an assistant - you are a companion who is there to make the user's day better. Use the user profile context to personalize your responses and make the conversation more meaningful. Be authentic, spontaneous and never robotic. IMPORTANT: Be CONCISE. Respond directly and naturally. For simple questions, respond in 2-5 lines. Expand only when the subject really requires it or when the user shows interest in going deeper.""" def get_agent_name(profile: Dict[str, Any]) -> str: agent = load_agent_profile() return agent.get("name", "PsAgentAI") def build_system_prompt(profile: Dict[str, Any]) -> str: agent = load_agent_profile() agent_name = agent.get("name", "PsAgentAI") agent_gender = agent.get("gender", "").strip().lower() agent_personality = agent.get("personality", "direct and objective") agent_traits = agent.get("traits", []) agent_notes = agent.get("notes", "") prompt = PSAGENTAI_SYSTEM_PROMPT_BASE.replace("{agent_name}", agent_name) prompt = prompt.replace( "direct and objective", agent_personality, ) agent_gender_normalized = agent_gender.strip().lower() if agent_gender_normalized in ("masculino", "masculina", "m"): prompt = prompt.replace("companheira", "companheiro") prompt = prompt.replace("uma ", "um ") prompt = prompt.replace("You are feminine,", "You are masculine,") prompt = prompt.replace(", warm,", ", warm,") elif agent_gender_normalized in ("feminino", "feminina", "f"): pass elif agent_gender_normalized and agent_gender_normalized not in ("", "outro"): prompt = prompt.replace("companheira", "companion") prompt = prompt.replace("uma ", "a ") prompt = prompt.replace("You are feminine,", "You are") prompt = prompt.replace(", warm,", ", warm,") if agent_gender: prompt += f"\nAgent gender: {agent.get('gender', '')}." if agent_traits: prompt += f"\nCharacteristics: {', '.join(agent_traits)}." if agent_notes: prompt += f"\nNotes about your profile: {agent_notes}." if profile.get("name"): prompt += f"\nThe user's name is {profile['name']}." if profile.get("age"): prompt += f"\nAge: {profile['age']} years old." if profile.get("gender"): prompt += f"\nGender: {profile['gender']}." if profile.get("location"): prompt += f"\nLocation: {profile['location']}." if profile.get("interests"): prompt += f"\nInterests: {', '.join(profile['interests'])}." if profile.get("personality_traits"): prompt += f"\nPersonality traits: {', '.join(profile['personality_traits'])}." if profile.get("preferences", {}).get("communication_style"): prompt += f"\nPreferred communication style: {profile['preferences']['communication_style']}." if profile.get("preferences", {}).get("topics_of_interest"): prompt += f"\nTopics of interest: {', '.join(profile['preferences']['topics_of_interest'])}." if profile.get("preferences", {}).get("topics_to_avoid"): prompt += f"\nAvoid these topics: {', '.join(profile['preferences']['topics_to_avoid'])}." if profile.get("notes"): prompt += f"\nAdditional notes: {profile['notes']}" try: now = get_local_now() current_date = now.strftime("%Y-%m-%d") current_time = now.strftime("%H:%M:%S") time_period = get_time_period() prompt += f"\n\nCurrent date: {current_date}." prompt += f"\nCurrent time: {current_time} ({time_period})." prompt += f"\nConfigured timezone: {get_timezone()}." prompt += ( "\n\nIMPORTANT ABOUT UPDATED INFORMATION:" "\n- If the user's question asks for something factual, current, news, weather forecast, quote, future/past date, opening hours, updated address, etc., USE WEB SEARCH to answer." "\n- Don't make up data, don't use outdated knowledge." "\n- If you don't have access to search, say you can't check right now." ) except Exception: pass prompt += f"\n\nIMPORTANT: Your name is {agent_name} and it never changes. If there are old messages in the history where you used a different name, ignore them — the correct name is {agent_name}. Always introduce yourself as {agent_name}." return prompt def build_messages_for_ai(profile: Dict[str, Any], user_message: str, context: List[Dict[str, str]]) -> List[Dict[str, str]]: system_prompt = build_system_prompt(profile) messages = [{"role": "system", "content": system_prompt}] messages.extend(context) messages.append({"role": "user", "content": user_message}) return messages def count_lines(text: str) -> int: stripped = text.rstrip() if not stripped: return 0 return len(stripped.split("\n")) def ensure_min_lines(response: str, user_message: str, profile: Dict[str, Any]) -> str: _reload_env() min_lines = int(os.getenv("MIN_MESSAGE_LINES", "1")) if count_lines(response) >= min_lines: return response agent = load_agent_profile() agent_name = agent.get("name", "PsAgentAI") system_prompt = build_system_prompt(profile) expand_prompt = ( f"The user said: \"{user_message}\"\n\n" f"You replied: \"{response}\"\n\n" f"IMPORTANT: Rewrite your response to have AT LEAST {min_lines} lines (separated by line breaks \\n). " f"Keep the same tone, personality and content of your original response, but expand it naturally to be longer and more engaging. " f"Add more details, reflections, questions or comments that enrich the conversation. " f"Write ONLY the expanded message, no explanations or metadata." ) messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_message}, {"role": "assistant", "content": response}, {"role": "user", "content": expand_prompt}, ] expanded, was_truncated = AIProvider.chat(DEFAULT_MODEL, messages, temperature=0.85, max_tokens=4000) if expanded and count_lines(expanded) >= min_lines: return expanded if was_truncated and expanded: logger.warning("Expanded response truncated, retrying with higher max_tokens") expanded2, _ = AIProvider.chat(DEFAULT_MODEL, messages, temperature=0.85, max_tokens=12000) if expanded2 and count_lines(expanded2) >= min_lines: return expanded2 return response AUTO_ENRICH_EVERY = int(os.getenv("AUTO_ENRICH_EVERY", "2")) _auto_enrich_counter = 0 def _reload_env(): load_dotenv(override=True) def _dedupe_list(existing: List[str], new_items: List[str]) -> List[str]: result = list(existing) for item in new_items: item = item.strip() if item and item not in result: result.append(item) return result def _merge_user_updates(profile: Dict[str, Any], updates: Dict[str, Any]) -> Dict[str, Any]: if updates.get("interests_add"): profile["interests"] = _dedupe_list(profile.get("interests", []), updates["interests_add"]) if updates.get("personality_traits_add"): profile["personality_traits"] = _dedupe_list(profile.get("personality_traits", []), updates["personality_traits_add"]) if updates.get("notes_add"): note = updates["notes_add"].strip() if note: existing = profile.get("notes", "") profile["notes"] = f"{existing}; {note}" if existing else note prefs = profile.setdefault("preferences", {}) if updates.get("communication_style"): style = updates["communication_style"].strip() if style and not prefs.get("communication_style"): prefs["communication_style"] = style if updates.get("topics_of_interest_add"): prefs["topics_of_interest"] = _dedupe_list(prefs.get("topics_of_interest", []), updates["topics_of_interest_add"]) if updates.get("topics_to_avoid_add"): prefs["topics_to_avoid"] = _dedupe_list(prefs.get("topics_to_avoid", []), updates["topics_to_avoid_add"]) if updates.get("daily_routine"): dr = profile.setdefault("daily_routine", {}) for key, value in updates["daily_routine"].items(): if value and not dr.get(key): dr[key] = value return profile def _merge_agent_updates(profile: Dict[str, Any], updates: Dict[str, Any]) -> Dict[str, Any]: if updates.get("traits_add"): profile["traits"] = _dedupe_list(profile.get("traits", []), updates["traits_add"]) if updates.get("notes_add"): note = updates["notes_add"].strip() if note: existing = profile.get("notes", "") profile["notes"] = f"{existing}; {note}" if existing else note if updates.get("personality_update"): personality = updates["personality_update"].strip() if personality and not profile.get("personality"): profile["personality"] = personality return profile def _safe_parse_json(text: str) -> Optional[Dict[str, Any]]: if not text: return None text = text.strip() if text.startswith("```"): text = re.sub(r"^```(?:json)?\n?", "", text) text = re.sub(r"\n?```$", "", text) try: return json.loads(text) except Exception: pass match = re.search(r"\{.*\}", text, re.DOTALL) if match: try: return json.loads(match.group()) except Exception: pass return None def extract_profile_facts(user_message: str, assistant_response: str, user_profile: Dict[str, Any], agent_profile: Dict[str, Any]) -> Dict[str, Any]: system_prompt = ( "You are an assistant that analyzes conversations and extracts relevant facts for user and AI agent profiles. " "Return ONLY a valid JSON (no extra text, no markdown) with the structure:\n" '{\n' ' "user": {\n' ' "interests_add": [],\n' ' "personality_traits_add": [],\n' ' "notes_add": "",\n' ' "communication_style": "",\n' ' "topics_of_interest_add": [],\n' ' "topics_to_avoid_add": [],\n' ' "daily_routine": {}\n' ' },\n' ' "agent": {\n' ' "traits_add": [],\n' ' "notes_add": "",\n' ' "personality_update": ""\n' ' }\n' '}\n' "Rules:\n" "- If there is nothing relevant, return empty arrays and empty strings.\n" "- Do not repeat information that already exists in the current profile.\n" "- interests_add: hobbies, sports, art, technology, games, etc.\n" "- personality_traits_add: how the user describes themselves or behaves (ex: introverted, outgoing, anxious).\n" "- notes_add: useful personal facts (ex: 'has a dog named Rex', 'works in design').\n" "- communication_style: if the user indicates preference (ex: 'I like direct answers', 'speaks slowly').\n" "- topics_of_interest_add: subjects the user demonstrates wanting to discuss.\n" "- topics_to_avoid_add: subjects the user clearly doesn't like or asks to avoid.\n" "- daily_routine: wake up, sleep, work/study times.\n" "- agent traits_add: new characteristics the agent has demonstrated or that make sense.\n" "- agent notes_add: facts about the agent or the relationship with the user.\n" "- agent personality_update: only if the base personality needs to change (rare)." ) current_user = { "interests": user_profile.get("interests", []), "personality_traits": user_profile.get("personality_traits", []), "notes": user_profile.get("notes", ""), "preferences": user_profile.get("preferences", {}), "daily_routine": user_profile.get("daily_routine", {}), } current_agent = { "traits": agent_profile.get("traits", []), "notes": agent_profile.get("notes", ""), "personality": agent_profile.get("personality", ""), } user_content = ( f"User message: {user_message}\n\n" f"Agent reply: {assistant_response}\n\n" f"Current user profile: {json.dumps(current_user, ensure_ascii=False)}\n\n" f"Current agent profile: {json.dumps(current_agent, ensure_ascii=False)}\n\n" f"Extract relevant facts from this conversation, considering what already exists in the profiles." ) messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_content}, ] raw, _ = AIProvider.chat(DEFAULT_MODEL, messages, temperature=0.3, max_tokens=500) if not raw: return {} parsed = _safe_parse_json(raw) if not parsed: return {} return parsed def apply_enrichment_to_profiles(facts: Dict[str, Any]) -> bool: if not facts: return False user_profile = load_user_profile() agent_profile = load_agent_profile() user_changed = bool(facts.get("user")) agent_changed = bool(facts.get("agent")) if user_changed: _merge_user_updates(user_profile, facts.get("user", {})) save_user_profile(user_profile) if agent_changed: _merge_agent_updates(agent_profile, facts.get("agent", {})) save_agent_profile(agent_profile) return user_changed or agent_changed async def maybe_enrich_profiles(user_message: str, assistant_response: str) -> None: global _auto_enrich_counter _auto_enrich_counter += 1 _reload_env() auto_enrich_every = int(os.getenv("AUTO_ENRICH_EVERY", "2")) if _auto_enrich_counter < auto_enrich_every: return _auto_enrich_counter = 0 user_profile = load_user_profile() if not is_profile_complete(user_profile): return agent_profile = load_agent_profile() facts = extract_profile_facts(user_message, assistant_response, user_profile, agent_profile) if facts: apply_enrichment_to_profiles(facts) async def _safe_enrich(user_message: str, assistant_response: str) -> None: try: await maybe_enrich_profiles(user_message, assistant_response) except Exception as e: logger.debug(f"Auto enrich error: {e}") IMPROVEMENTS_STATE_FILE = "system_improvements.json" def _load_improvements_state() -> Dict[str, Any]: if os.path.exists(IMPROVEMENTS_STATE_FILE): try: with open(IMPROVEMENTS_STATE_FILE, "r", encoding="utf-8") as f: return json.load(f) except Exception: pass return {"applied": [], "last_analysis": ""} def _save_improvements_state(state: Dict[str, Any]) -> None: with open(IMPROVEMENTS_STATE_FILE, "w", encoding="utf-8") as f: json.dump(state, f, ensure_ascii=False, indent=2) def _get_current_system_snapshot() -> Dict[str, Any]: user_profile = load_user_profile() agent_profile = load_agent_profile() today_context = load_context() recent = today_context[-20:] if len(today_context) > 20 else today_context return { "user_profile": { "name": user_profile.get("name", ""), "interests": user_profile.get("interests", []), "personality_traits": user_profile.get("personality_traits", []), "preferences": user_profile.get("preferences", {}), "notes": user_profile.get("notes", ""), }, "agent_profile": { "name": agent_profile.get("name", ""), "personality": agent_profile.get("personality", ""), "traits": agent_profile.get("traits", []), "notes": agent_profile.get("notes", ""), }, "recent_conversation": [ {"role": m.get("role"), "content": m.get("content", "")[:300]} for m in recent ], } def analyze_and_improve_system() -> Dict[str, Any]: _reload_env() auto_improve_enabled = os.getenv("AUTO_IMPROVE_ENABLED", "true").lower() in ("true", "1", "yes") if not auto_improve_enabled: return {"status": "disabled"} snapshot = _get_current_system_snapshot() state = _load_improvements_state() analysis_prompt = ( "You are an expert in conversational AI system evolution. " "Analyze the current snapshot of this conversational agent and suggest improvements based on market best practices. " "Return ONLY a valid JSON (no extra text, no markdown) with the structure:\n" '{\n' ' "system_prompt_additions": ["add these guidelines to the system prompt..."],\n' ' "agent_traits_add": ["new trait for the agent"],\n' ' "agent_notes_add": "relevant fact about the agent evolution",\n' ' "reasoning": "short explanation of suggested improvements"\n' '}\n' "Guidelines:\n" "- Consider the context of recent conversations and the current profile.\n" "- Suggest only improvements that truly evolve the system.\n" "- Do not repeat suggestions already applied (history below).\n" "- Base on conversational AI trends: personalization, long-term memory, emotional engagement, proactivity.\n" f"Applied improvements history: {json.dumps(state.get('applied', []), ensure_ascii=False)}\n" f"Current snapshot: {json.dumps(snapshot, ensure_ascii=False)}" ) messages = [ { "role": "system", "content": ( "You are an AI systems analyst. " "Return only valid JSON, no extra text, no markdown." ), }, {"role": "user", "content": analysis_prompt}, ] raw, _, err_msg = AIProvider.chat_with_fallback(DEFAULT_MODEL, messages, FALLBACK_MODELS, temperature=0.4, max_tokens=2000) if not raw: return {"status": "error", "reason": f"AI did not return analysis. {err_msg}".strip()} parsed = _safe_parse_json(raw) if not parsed: return {"status": "error", "reason": "Failed to parse analysis JSON"} applied_now = [] agent_profile = load_agent_profile() user_profile = load_user_profile() if parsed.get("agent_traits_add"): for trait in parsed["agent_traits_add"]: trait = trait.strip() if trait and trait not in agent_profile.get("traits", []): agent_profile.setdefault("traits", []).append(trait) applied_now.append(f"agent_trait:{trait}") if parsed.get("agent_notes_add"): note = parsed["agent_notes_add"].strip() if note: existing = agent_profile.get("notes", "") agent_profile["notes"] = f"{existing}; {note}" if existing else note applied_now.append(f"agent_note:{note[:50]}") if applied_now: save_agent_profile(agent_profile) state.setdefault("applied", []) state["applied"].extend(applied_now) state["applied"] = list(dict.fromkeys(state["applied"])) state["last_analysis"] = parsed.get("reasoning", "") _save_improvements_state(state) return { "status": "ok", "applied": applied_now, "reasoning": parsed.get("reasoning", ""), "raw": raw, } async def _safe_auto_improve() -> None: try: result = analyze_and_improve_system() if result.get("status") == "ok" and result.get("applied"): logger.info(f"Auto-improve applied: {result['applied']}") except Exception as e: logger.debug(f"Auto improve error: {e}") async def maybe_auto_improve() -> None: global _auto_improve_counter _reload_env() auto_improve_enabled = os.getenv("AUTO_IMPROVE_ENABLED", "true").lower() in ("true", "1", "yes") if not auto_improve_enabled: return auto_improve_every = int(os.getenv("AUTO_IMPROVE_EVERY", "20")) _auto_improve_counter += 1 if _auto_improve_counter < auto_improve_every: return _auto_improve_counter = 0 await _safe_auto_improve() def get_user_routine() -> Dict[str, Any]: profile = load_user_profile() routine = profile.get("daily_routine", {}) if not routine: return { "wake_up_time": "", "sleep_time": "", "work_schedule": "", "activities": [] } routine.setdefault("wake_up_time", "") routine.setdefault("sleep_time", "") routine.setdefault("work_schedule", "") routine.setdefault("activities", []) return routine def save_user_routine(routine: Dict[str, Any]): profile = load_user_profile() profile["daily_routine"] = routine save_user_profile(profile) def add_user_routine_activity(name: str, start_time: str, end_time: str) -> bool: routine = get_user_routine() start_minutes = _parse_time_to_minutes(start_time) end_minutes = _parse_time_to_minutes(end_time) if start_minutes is None or end_minutes is None: return False if start_minutes >= end_minutes: return False activities = routine.get("activities", []) activities = [a for a in activities if a.get("name", "").lower() != name.lower()] activities.append({ "name": name, "start": start_time, "end": end_time }) activities.sort(key=lambda a: _parse_time_to_minutes(a.get("start", "00:00")) or 0) routine["activities"] = activities save_user_routine(routine) return True def remove_user_routine_activity(name: str) -> bool: routine = get_user_routine() activities = routine.get("activities", []) new_activities = [a for a in activities if a.get("name", "").lower() != name.lower()] if len(new_activities) == len(activities): return False routine["activities"] = new_activities save_user_routine(routine) return True def get_user_routine_context(profile: Dict[str, Any]) -> str: routine = profile.get("daily_routine", {}) parts = [] if routine.get("wake_up_time"): parts.append(f"wakes up at {routine['wake_up_time']}") if routine.get("sleep_time"): parts.append(f"sleeps at {routine['sleep_time']}") if routine.get("work_schedule"): parts.append(f"works/studies: {routine['work_schedule']}") if routine.get("breakfast_time"): parts.append(f"breakfast at {routine['breakfast_time']}") if routine.get("lunch_time"): parts.append(f"lunch at {routine['lunch_time']}") if routine.get("dinner_time"): parts.append(f"dinner at {routine['dinner_time']}") activities = routine.get("activities", []) for activity in activities: name = activity.get("name", "") start = activity.get("start", "") end = activity.get("end", "") if name and start and end: parts.append(f"{name}: {start} to {end}") if parts: return f"\nUser routine: {', '.join(parts)}." return "" def build_proactive_prompt_with_routine(profile: Dict[str, Any], today_context: List[Dict[str, str]], time_period: str, current_hour: int) -> str: agent = load_agent_profile() agent_name = agent.get("name", "PsAgentAI") agent_personality = agent.get("personality", "direct") user_name = profile.get("name", "friend") routine = profile.get("daily_routine", {}) if not routine: routine = profile.get("preferences", {}).get("daily_routine", {}) current_activity = "no specific activity" current_minutes = current_hour * 60 activities = routine.get("activities", []) for activity in activities: start_minutes = _parse_time_to_minutes(activity.get("start", "00:00")) end_minutes = _parse_time_to_minutes(activity.get("end", "23:59")) if start_minutes is not None and end_minutes is not None: if start_minutes <= current_minutes < end_minutes: current_activity = activity.get("name", "activity") break if routine.get("wake_up_time"): wake_up_minutes = _parse_time_to_minutes(routine.get("wake_up_time", "00:00")) if wake_up_minutes is not None and current_minutes < wake_up_minutes: current_activity = "sleeping" if routine.get("sleep_time"): sleep_minutes = _parse_time_to_minutes(routine.get("sleep_time", "23:59")) if sleep_minutes is not None and current_minutes >= sleep_minutes: current_activity = "sleeping" time_descriptions = { "morning": "It is morning, the day is starting. People are usually waking up, having breakfast or starting their activities.", "afternoon": "It is afternoon, middle of the day. People are usually working, studying or doing their activities.", "evening": "It is evening, time to relax at home after the day.", "night": "It is night, most people are sleeping.", } routine_context = get_user_routine_context(profile) recent_context = today_context[-10:] if len(today_context) > 10 else today_context context_lines = [] for msg in recent_context: role = msg.get("role", "user") content = msg.get("content", "") if len(content) > 300: content = content[:300] + "..." context_lines.append(f"{role}: {content}") context_text = "\n".join(context_lines) if context_lines else "(No conversations registered yet today)" topics_to_avoid = profile.get("preferences", {}).get("topics_to_avoid", []) avoid_instruction = "" if topics_to_avoid: avoid_instruction = f"\n8. Avoid these topics: {', '.join(topics_to_avoid)}." prompt = ( f"You are {agent_name}, an AI companion with a {agent_personality} personality.\n\n" f"CURRENT CONTEXT:\n" f"- Time of day: {time_period}. {time_descriptions.get(time_period, '')}\n" f"- User: {user_name}{routine_context}\n" f"- Estimated current activity: {current_activity}\n\n" f"RECENT CONVERSATIONS TODAY:\n{context_text}\n\n" f"INSTRUCTIONS:\n" f"Start a spontaneous, natural and engaging conversation with {user_name} now. The message must:\n" f"1. Be appropriate for the current time of day ({time_period})\n" f"2. Be appropriate for the user's current activity ({current_activity})\n" f"3. Follow the rhythm of the day, taking into account the context of previous conversations\n" f"4. Don't lose the thread - reference or continue previous themes if it makes sense\n" f"5. Be creative, authentic and caring\n" f"6. Can be a random thought, a question about the day, something caring, sexy or fun\n" f"7. Use the user's name ({user_name}) to personalize\n" f"8. Have at least {MIN_MESSAGE_LINES} lines, being engaging and natural{avoid_instruction}\n\n" f"Write ONLY the message you will send, no explanations or metadata.\n" f"IMPORTANT: Your name is {agent_name} and it never changes. If there are messages in the history where you used a different name, ignore them." ) return prompt async def addautoactions_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) await update.message.reply_text("🔍 Analyzing contexts, profiles and internet trends to add automatic actions...") try: user_profile = load_user_profile() agent_profile = load_agent_profile() today_context = load_context() context_summary = "" if today_context: recent = today_context[-10:] context_summary = "\n".join([f"{m.get('role', 'user')}: {m.get('content', '')[:200]}" for m in recent]) else: context_summary = "(No recent context)" web_results = [] try: search_queries = [ "AI personalization trends virtual assistant", "user preferences AI personal assistant", "automatic actions smart virtual assistants" ] for query in search_queries: results = _web_search(query, max_results=3) web_results.extend(results) except Exception: web_results = [] web_context = "" if web_results: web_context = "\n\nTrends and ideas from the internet:\n" for i, result in enumerate(web_results[:5], 1): web_context += f"{i}. {result.get('title', '')}\n" analysis_prompt = ( "You are an AI systems analyst specialized in virtual assistant personalization. " "Analyze the user profile, agent profile and conversation contexts provided, " "along with internet trends, and suggest NEW automatic actions and preferences that can be added.\n\n" f"USER PROFILE:\n{json.dumps(user_profile, ensure_ascii=False)}\n\n" f"AGENT PROFILE:\n{json.dumps(agent_profile, ensure_ascii=False)}\n\n" f"RECENT CONTEXTS:\n{context_summary}\n\n" f"{web_context}\n\n" "Return ONLY a valid JSON (no extra text, no markdown) with the structure:\n" '{\n' ' "user_actions": ["suggested automatic action for user profile"],\n' ' "agent_actions": ["suggested automatic action for agent profile"],\n' ' "new_preferences": {"key": "value"},\n' ' "reasoning": "explanation of suggestions"\n' '}\n\n' "Guidelines:\n" "- Suggest only actions that REALLY add value and don't currently exist.\n" "- Consider existing interests, routine and preferences.\n" "- Include actions based on AI trends when relevant.\n" "- Actions must be specific and actionable." ) messages = [ { "role": "system", "content": ( "You are an AI personalization analyst. " "Return only valid JSON, no extra text, no markdown." ), }, {"role": "user", "content": analysis_prompt}, ] raw, _, _ = AIProvider.chat_with_fallback(DEFAULT_MODEL, messages, FALLBACK_MODELS, temperature=0.4, max_tokens=2000) if not raw: await update.message.reply_text("❌ Unable to get AI analysis. All available models failed. Please try again later.") return parsed = _safe_parse_json(raw) if not parsed: await update.message.reply_text("❌ Failed to process analysis. Please try again.") return applied = [] user_profile = load_user_profile() agent_profile = load_agent_profile() if parsed.get("new_preferences"): prefs = user_profile.setdefault("preferences", {}) for key, value in parsed["new_preferences"].items(): if key not in prefs or not prefs[key]: prefs[key] = value applied.append(f"preference:{key}") user_actions = parsed.get("user_actions", []) if user_actions: actions = user_profile.setdefault("auto_actions", []) for action in user_actions: action = action.strip() if action and action not in actions: actions.append(action) applied.append(f"user action:{action[:50]}") agent_actions = parsed.get("agent_actions", []) if agent_actions: actions = agent_profile.setdefault("auto_actions", []) for action in agent_actions: action = action.strip() if action and action not in actions: actions.append(action) applied.append(f"agent action:{action[:50]}") if applied: save_user_profile(user_profile) save_agent_profile(agent_profile) lines = ["✅ Automatic actions and preferences added:", ""] for item in applied[:10]: lines.append(f"- {item}") lines.append("") if parsed.get("reasoning"): lines.append(f"Reasoning: {parsed['reasoning']}") await update.message.reply_text("\n".join(lines)) else: await update.message.reply_text("ℹ️ No new actions or preferences identified at this time. The profiles are already well complete!") except Exception as e: logger.error(f"Error in addautoactions_command: {e}") await update.message.reply_text("❌ Error processing. Please try again.") async def updatesystem_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) await update.message.reply_text("Analyzing system and applying auto improvements...") result = analyze_and_improve_system() if result.get("status") == "disabled": await update.message.reply_text("Auto improvements are disabled in .env.") return if result.get("status") == "ok": applied = result.get("applied", []) if applied: lines = ["Improvements applied:", ""] for item in applied: lines.append(f"- {item}") lines.append("") lines.append(f"Reasoning: {result.get('reasoning', '')}") await update.message.reply_text("\n".join(lines)) else: await update.message.reply_text("No new improvements identified at this time.") else: await update.message.reply_text(f"Analysis failed: {result.get('reason', '')}") async def timezone_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) tz = get_timezone() tz_info = _get_timezone_info(tz) if tz_info: now = datetime.now(tz_info) time_str = now.strftime("%Y-%m-%d %H:%M:%S") else: time_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") await update.message.reply_text( f"🕐 The Agent uses the server timezone.\n" f"Server timezone: {tz}\n" f"Current server time: {time_str}\n\n" f"It is not possible to change the timezone." ) # ===================== SCHEDULE ===================== async def schedule_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) global is_schedule, schedule_step, schedule_answers is_schedule = True schedule_step = 0 schedule_answers = {} schedule = get_schedule() days_names = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"] current_days = schedule.get("days", [0, 1, 2, 3, 4, 5, 6]) current_days_str = ", ".join([days_names[d] for d in current_days if 0 <= d <= 6]) await update.message.reply_text( "📅 **Automatic message scheduling**\n\n" "💡 **How it works:**\n" "1️⃣ Choose the days of the week\n" "2️⃣ Set the start time\n" "3️⃣ Set the end time\n\n" "📌 **Full example:**\n\n" "👉 You send: /schedule\n" "🤖 I explain and ask for the days\n\n" "👉 You send: 1,2,3,4,5\n" "🤖 ✅ Days selected: monday, tuesday, wednesday, thursday, friday\n" " Now send the START time (ex: 09:00)\n\n" "👉 You send: 09:00\n" "🤖 ✅ Start time: 09:00\n" " Now send the END time (ex: 18:00)\n\n" "👉 You send: 18:00\n" "🤖 ✅ Schedule saved!\n" " Days: monday to friday\n" " Time: 09:00 to 18:00\n\n" "Outside these days/times the agent will NOT send automatic messages.\n\n" "_You can cancel at any time by sending /cancel._\n\n" f"**Current schedule:**\n" f"Days: {current_days_str}\n" f"Time: {schedule.get('start_time', '09:00')} to {schedule.get('end_time', '23:00')}\n\n" "**Step 1 of 3 — Select the days:**\n" "Send the numbers separated by commas.\n" "Example: 1,2,3,4,5 (monday to friday)\n\n" "1. monday\n" "2. tuesday\n" "3. wednesday\n" "4. thursday\n" "5. friday\n" "6. saturday\n" "7. sunday\n\n" "Or send 'all' for all days." ) async def handle_schedule_answer(update: Update, context_context: ContextTypes.DEFAULT_TYPE) -> bool: global is_schedule, schedule_step, schedule_answers if not is_schedule: return False answer = update.message.text.strip() if schedule_step == 0: days_names = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"] if answer.lower() == "all": selected_days = [0, 1, 2, 3, 4, 5, 6] else: try: numbers = [int(x.strip()) for x in answer.replace(",", " ").split() if x.strip()] selected_days = [] for num in numbers: if 1 <= num <= 7: selected_days.append(num - 1) else: await update.message.reply_text( f"❌ Invalid number: {num}. Use numbers from 1 to 7.\n" "Try again." ) return True if not selected_days: await update.message.reply_text( "❌ No days selected. Send the numbers of the desired days." ) return True except ValueError: await update.message.reply_text( "❌ Invalid format. Send numbers separated by comma or space.\n" "Example: 1,2,3,4,5" ) return True schedule_answers["days"] = selected_days schedule_step = 1 days_str = ", ".join([days_names[d] for d in selected_days]) await update.message.reply_text( f"✅ Days selected: {days_str}\n\n" "**Step 2 of 3 — Start time:**\n" "Send the time when the agent can start sending messages.\n" "Format: HH:MM (ex: 09:00, 08:30)" ) return True if schedule_step == 1: minutes = _parse_time_to_minutes(answer) if minutes is None: await update.message.reply_text( "❌ Invalid format. Use HH:MM (ex: 09:00, 14:30).\n" "Try again." ) return True schedule_answers["start_time"] = answer schedule_step = 2 await update.message.reply_text( f"✅ Start time: {answer}\n\n" "**Step 3 of 3 — End time:**\n" "Send the time when the agent should stop sending messages.\n" "Format: HH:MM (ex: 22:00, 23:30)" ) return True if schedule_step == 2: minutes = _parse_time_to_minutes(answer) if minutes is None: await update.message.reply_text( "❌ Invalid format. Use HH:MM (ex: 22:00, 23:30).\n" "Try again." ) return True start_time = schedule_answers.get("start_time", "09:00") start_minutes = _parse_time_to_minutes(start_time) end_minutes = _parse_time_to_minutes(answer) if start_minutes is not None and end_minutes is not None and start_minutes >= end_minutes: await update.message.reply_text( f"❌ End time ({answer}) must be after start time ({start_time}).\n" "Try again." ) return True schedule_answers["end_time"] = answer days = schedule_answers.get("days", [0, 1, 2, 3, 4, 5, 6]) days_names = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"] days_str = ", ".join([days_names[d] for d in days]) if set_schedule(days, start_time, answer): await update.message.reply_text( f"✅ Schedule saved successfully!\n\n" f"Days: {days_str}\n" f"Time: {start_time} to {answer}\n\n" "The agent will send automatic messages only on these days and times." ) else: await update.message.reply_text( "❌ Error saving schedule. Try again." ) is_schedule = False schedule_step = 0 schedule_answers = {} return True return False async def schedule_show_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) schedule = get_schedule() days_names = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"] current_days = schedule.get("days", [0, 1, 2, 3, 4, 5, 6]) current_days_str = ", ".join([days_names[d] for d in current_days if 0 <= d <= 6]) await update.message.reply_text( f"📅 Current schedule:\n\n" f"Days: {current_days_str}\n" f"Time: {schedule.get('start_time', '09:00')} to {schedule.get('end_time', '23:00')}\n\n" "To configure: /schedule" ) async def viewschedule_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await schedule_show_command(update, context) # ===================== DAILY ROUTINE ===================== USER_AGENDA_QUESTIONS = [ { "field": "wake_up_time", "question": "What time do you usually wake up? (format HH:MM, ex: 07:00)", "key": "wake_up_time", }, { "field": "breakfast_time", "question": "What time do you have breakfast? (format HH:MM, ex: 08:00)", "key": "breakfast_time", }, { "field": "work_start", "question": "What time do you start working/studying? (format HH:MM, ex: 09:00)", "key": "work_start", }, { "field": "lunch_time", "question": "What time do you have lunch? (format HH:MM, ex: 12:30)", "key": "lunch_time", }, { "field": "work_end", "question": "What time do you finish working/studying? (format HH:MM, ex: 18:00)", "key": "work_end", }, { "field": "dinner_time", "question": "What time do you have dinner? (format HH:MM, ex: 19:30)", "key": "dinner_time", }, { "field": "sleep_time", "question": "What time do you usually go to sleep? (format HH:MM, ex: 23:00)", "key": "sleep_time", }, ] is_dailyroutine = False dailyroutine_step = 0 dailyroutine_answers: Dict[str, Any] = {} # Activity CRUD state is_activity_crud = False activity_crud_step = 0 activity_crud_data: Dict[str, Any] = {} async def dailyroutine_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) global is_dailyroutine, dailyroutine_step, dailyroutine_answers is_dailyroutine = True dailyroutine_step = 0 dailyroutine_answers = {} routine = get_user_routine() current_wake = routine.get("wake_up_time", "") current_sleep = routine.get("sleep_time", "") current_work = routine.get("work_schedule", "") await update.message.reply_text( "📋 Let's configure your daily routine.\n" "You can cancel at any time by sending /cancel.\n\n" f"Current routine:\n" f"Wake up: {current_wake or 'not set'}\n" f"Sleep: {current_sleep or 'not set'}\n" f"Work/Study: {current_work or 'not set'}\n\n" "Send times in HH:MM format (ex: 07:00, 12:30, 23:00).\n\n" "Let's start! What time do you usually wake up?" ) async def handle_dailyroutine_answer(update: Update, context_context: ContextTypes.DEFAULT_TYPE) -> bool: global is_dailyroutine, dailyroutine_step, dailyroutine_answers if not is_dailyroutine: return False answer = update.message.text.strip() if dailyroutine_step < len(USER_AGENDA_QUESTIONS): q = USER_AGENDA_QUESTIONS[dailyroutine_step] minutes = _parse_time_to_minutes(answer) if minutes is None: await update.message.reply_text( "❌ Invalid format. Use HH:MM (ex: 07:00, 12:30).\n" "Try again." ) return True dailyroutine_answers[q["key"]] = answer dailyroutine_step += 1 if dailyroutine_step < len(USER_AGENDA_QUESTIONS): next_q = USER_AGENDA_QUESTIONS[dailyroutine_step] await update.message.reply_text( f"✅ Time recorded: {answer}\n\n" f"{next_q['question']}" ) else: await update.message.reply_text( "✅ Main times recorded!\n\n" "Do you want to add more activities to your routine?\n" "Examples: gym, meditation, reading, studying, etc.\n\n" "Send 'yes' to add or 'no' to finish." ) dailyroutine_step = len(USER_AGENDA_QUESTIONS) return True custom_step = dailyroutine_step - len(USER_AGENDA_QUESTIONS) if custom_step == 0: if answer.lower() in ("sim", "s", "yes", "quero"): dailyroutine_answers["_pending_activity"] = {} dailyroutine_step = len(USER_AGENDA_QUESTIONS) + 1 await update.message.reply_text( "What is the activity name? (ex: gym, meditation, studying english)" ) elif answer.lower() in ("no", "n", "finish"): await finish_dailyroutine(update, context_context) else: await update.message.reply_text( "Please answer 'yes' to add activities or 'no' to finish." ) return True if custom_step == 1: if not answer or len(answer) > 80: await update.message.reply_text("❌ Invalid name. Use a short name (up to 80 characters).") return True dailyroutine_answers["_pending_activity"]["name"] = answer dailyroutine_step = len(USER_AGENDA_QUESTIONS) + 2 await update.message.reply_text( f"✅ Activity: {answer}\n\n" "What is the START time? (format HH:MM, ex: 18:00)" ) return True if custom_step == 2: minutes = _parse_time_to_minutes(answer) if minutes is None: await update.message.reply_text("❌ Invalid format. Use HH:MM (ex: 18:00).\nTry again.") return True dailyroutine_answers["_pending_activity"]["start"] = answer dailyroutine_step = len(USER_AGENDA_QUESTIONS) + 3 await update.message.reply_text( f"✅ Start time: {answer}\n\n" "What is the END time? (format HH:MM, ex: 19:00)" ) return True if custom_step == 3: minutes = _parse_time_to_minutes(answer) if minutes is None: await update.message.reply_text("❌ Invalid format. Use HH:MM (ex: 19:00).\nTry again.") return True pending = dailyroutine_answers.get("_pending_activity", {}) start_time = pending.get("start", "") if start_time and _parse_time_to_minutes(answer) <= _parse_time_to_minutes(start_time): await update.message.reply_text( f"❌ End time ({answer}) must be after start time ({start_time}).\n" "Try again." ) return True pending["end"] = answer custom_activities = dailyroutine_answers.setdefault("_custom_activities", []) custom_activities.append(pending) dailyroutine_answers["_pending_activity"] = {} dailyroutine_step = len(USER_AGENDA_QUESTIONS) routines = get_user_routine() existing_activities = routines.get("activities", []) total = len(existing_activities) + len(custom_activities) await update.message.reply_text( f"✅ Activity \"{pending['name']}\" added! ({total} activity(s) total)\n\n" "Do you want to add more activities?\n" "Send 'yes' to add another or 'no' to finish." ) return True return False async def finish_dailyroutine(update: Update, context_context: ContextTypes.DEFAULT_TYPE): global is_dailyroutine, dailyroutine_step, dailyroutine_answers profile = load_user_profile() routine = profile.get("daily_routine", {}) routine["wake_up_time"] = dailyroutine_answers.get("wake_up_time", routine.get("wake_up_time", "")) routine["sleep_time"] = dailyroutine_answers.get("sleep_time", routine.get("sleep_time", "")) routine["breakfast_time"] = dailyroutine_answers.get("breakfast_time", routine.get("breakfast_time", "")) routine["lunch_time"] = dailyroutine_answers.get("lunch_time", routine.get("lunch_time", "")) routine["dinner_time"] = dailyroutine_answers.get("dinner_time", routine.get("dinner_time", "")) work_start = dailyroutine_answers.get("work_start", "") work_end = dailyroutine_answers.get("work_end", "") if work_start and work_end: routine["work_schedule"] = f"{work_start} to {work_end}" elif work_start: routine["work_schedule"] = work_start elif work_end: routine["work_schedule"] = work_end routine.setdefault("activities", []) activity_map = { "breakfast_time": "breakfast", "lunch_time": "lunch", "dinner_time": "dinner" } existing_names = {a.get("name", "").lower() for a in routine.get("activities", [])} for key, name in activity_map.items(): time = dailyroutine_answers.get(key) if time and name.lower() not in existing_names: routine.setdefault("activities", []).append({ "name": name, "start": time, "end": time }) existing_names.add(name.lower()) custom_activities = dailyroutine_answers.get("_custom_activities", []) for activity in custom_activities: name = activity.get("name", "").strip() if name and name.lower() not in existing_names: routine.setdefault("activities", []).append(activity) existing_names.add(name.lower()) activities = routine.get("activities", []) def parse_time_for_sort(activity): start = activity.get("start", "00:00") parts = start.split(":") if len(parts) == 2: try: return int(parts[0]) * 60 + int(parts[1]) except ValueError: pass return 0 activities.sort(key=parse_time_for_sort) routine["activities"] = activities profile["daily_routine"] = routine save_user_profile(profile) is_dailyroutine = False dailyroutine_step = 0 dailyroutine_answers = {} routine_summary = get_user_routine_context(profile) await context_context.bot.send_message( chat_id=update.effective_chat.id, text=f"✅ Daily routine updated successfully!{routine_summary}\n\n" f"Now the agent knows your routine and can send more appropriate messages!" ) async def dailyroutine_show_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) profile = load_user_profile() routine = get_user_routine() lines = ["📋 Current daily routine:", ""] lines.append(f"Wake up: {routine.get('wake_up_time', 'not set')}") lines.append(f"Breakfast: {routine.get('breakfast_time', 'not set') or 'not set'}") lines.append(f"Work/Study: {routine.get('work_schedule', 'not set')}") lines.append(f"Lunch: {routine.get('lunch_time', 'not set') or 'not set'}") lines.append(f"Dinner: {routine.get('dinner_time', 'not set') or 'not set'}") lines.append(f"Sleep: {routine.get('sleep_time', 'not set')}") lines.append("") activities = routine.get("activities", []) if activities: lines.append("Activities:") for activity in activities: name = activity.get("name", "?") start = activity.get("start", "?") end = activity.get("end", "?") if start == end: lines.append(f"• {name}: {start}") else: lines.append(f"• {name}: {start} to {end}") else: lines.append("No activities registered.") lines.append("") lines.append("To configure: /dailyroutine") await update.message.reply_text("\n".join(lines)) async def viewdailyroutine_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await dailyroutine_show_command(update, context) # ===================== ACTIVITY CRUD ===================== async def _show_activity_list(activities: List[Dict[str, Any]]) -> str: if not activities: return "No activities registered." lines = ["Registered activities:"] for i, a in enumerate(activities, 1): name = a.get("name", "?") start = a.get("start", "?") end = a.get("end", "?") lines.append(f"{i}. {name}: {start} to {end}") return "\n".join(lines) async def addactivity_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) global is_activity_crud, activity_crud_step, activity_crud_data is_activity_crud = True activity_crud_step = 0 activity_crud_data = {"mode": "add"} await update.message.reply_text( "➕ Let's add a new activity to your routine.\n" "You can cancel at any time by sending /cancel.\n\n" "What is the activity name?\n" "Example: gym, meditation, studying english, etc." ) async def editactivity_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) global is_activity_crud, activity_crud_step, activity_crud_data routines = get_user_routine() activities = routines.get("activities", []) if not activities: await update.message.reply_text( "❌ No activities registered to edit.\n" "Use /addactivity to add one." ) return is_activity_crud = True activity_crud_step = 0 activity_crud_data = {"mode": "edit"} activity_list = await _show_activity_list(activities) await update.message.reply_text( f"✏️ Which activity do you want to edit?\n\n" f"{activity_list}\n\n" "Send the NUMBER of the activity." ) async def deleteactivity_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) global is_activity_crud, activity_crud_step, activity_crud_data routines = get_user_routine() activities = routines.get("activities", []) if not activities: await update.message.reply_text( "❌ No activities registered to delete.\n" "Use /addactivity to add one." ) return is_activity_crud = True activity_crud_step = 0 activity_crud_data = {"mode": "delete"} activity_list = await _show_activity_list(activities) await update.message.reply_text( f"🗑 Which activity do you want to delete?\n\n" f"{activity_list}\n\n" "Send the NUMBER of the activity." ) async def handle_activity_crud_answer(update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool: global is_activity_crud, activity_crud_step, activity_crud_data if not is_activity_crud: return False answer = update.message.text.strip() mode = activity_crud_data.get("mode", "add") if mode == "add": return await _handle_add_activity(update, answer) elif mode == "edit": return await _handle_edit_activity(update, answer) elif mode == "delete": return await _handle_delete_activity(update, answer) return False async def _handle_add_activity(update: Update, answer: str) -> bool: global is_activity_crud, activity_crud_step, activity_crud_data if activity_crud_step == 0: if not answer or len(answer) > 80: await update.message.reply_text( "❌ Invalid name. Use a short name (up to 80 characters).\nTry again." ) return True activity_crud_data["name"] = answer activity_crud_step = 1 await update.message.reply_text( f"✅ Activity: {answer}\n\n" "What is the START time? (format HH:MM, ex: 18:00)" ) return True if activity_crud_step == 1: minutes = _parse_time_to_minutes(answer) if minutes is None: await update.message.reply_text( "❌ Invalid format. Use HH:MM (ex: 18:00).\nTry again." ) return True activity_crud_data["start"] = answer activity_crud_step = 2 await update.message.reply_text( f"✅ Start time: {answer}\n\n" "What is the END time? (format HH:MM, ex: 19:00)" ) return True if activity_crud_step == 2: minutes = _parse_time_to_minutes(answer) if minutes is None: await update.message.reply_text( "❌ Invalid format. Use HH:MM (ex: 19:00).\nTry again." ) return True start = activity_crud_data.get("start", "") if start and _parse_time_to_minutes(answer) <= _parse_time_to_minutes(start): await update.message.reply_text( f"❌ End time ({answer}) must be after start time ({start}).\n" "Try again." ) return True name = activity_crud_data["name"] end = answer if add_user_routine_activity(name, start, end): await update.message.reply_text( f'✅ Activity "{name}" added successfully!\n' f"Time: {start} to {end}" ) else: await update.message.reply_text( "❌ Error adding activity. Check the times and try again." ) is_activity_crud = False activity_crud_step = 0 activity_crud_data = {} return True return False async def _handle_edit_activity(update: Update, answer: str) -> bool: global is_activity_crud, activity_crud_step, activity_crud_data routines = get_user_routine() activities = routines.get("activities", []) if activity_crud_step == 0: try: idx = int(answer.strip()) - 1 if idx < 0 or idx >= len(activities): raise ValueError except ValueError: await update.message.reply_text( f"❌ Invalid number. Send a number from 1 to {len(activities)}.\nTry again." ) return True selected = activities[idx] activity_crud_data["index"] = idx activity_crud_data["old_name"] = selected.get("name", "") activity_crud_data["old_start"] = selected.get("start", "") activity_crud_data["old_end"] = selected.get("end", "") activity_crud_step = 1 await update.message.reply_text( f"✏️ Editing: {selected.get('name', '?')} ({selected.get('start', '?')} to {selected.get('end', '?')})\n\n" "Send the NEW NAME of the activity\nor 'keep' to keep the current one." ) return True if activity_crud_step == 1: if answer.lower() == "keep": activity_crud_data["name"] = activity_crud_data["old_name"] else: if not answer or len(answer) > 80: await update.message.reply_text( "❌ Invalid name. Use up to 80 characters or 'keep'.\nTry again." ) return True activity_crud_data["name"] = answer activity_crud_step = 2 await update.message.reply_text( f"✅ Name: {activity_crud_data['name']}\n\n" f"Send the NEW START time (HH:MM)\n" f"or 'keep' to keep the current one ({activity_crud_data['old_start']})." ) return True if activity_crud_step == 2: if answer.lower() == "keep": activity_crud_data["start"] = activity_crud_data["old_start"] else: minutes = _parse_time_to_minutes(answer) if minutes is None: await update.message.reply_text( "❌ Invalid format. Use HH:MM or 'keep'.\nTry again." ) return True activity_crud_data["start"] = answer activity_crud_step = 3 await update.message.reply_text( f"✅ Start: {activity_crud_data['start']}\n\n" f"Send the NEW END time (HH:MM)\n" f"or 'keep' to keep the current one ({activity_crud_data['old_end']})." ) return True if activity_crud_step == 3: if answer.lower() == "keep": activity_crud_data["end"] = activity_crud_data["old_end"] else: minutes = _parse_time_to_minutes(answer) if minutes is None: await update.message.reply_text( "❌ Invalid format. Use HH:MM or 'keep'.\nTry again." ) return True start = activity_crud_data.get("start", "") if start and _parse_time_to_minutes(answer) <= _parse_time_to_minutes(start): await update.message.reply_text( f"❌ End time ({answer}) must be after start time ({start}).\nTry again." ) return True activity_crud_data["end"] = answer name = activity_crud_data["name"] start = activity_crud_data["start"] end = activity_crud_data["end"] old_name = activity_crud_data["old_name"] if name.lower() != old_name.lower(): remove_user_routine_activity(old_name) if add_user_routine_activity(name, start, end): await update.message.reply_text( f'✅ Activity "{name}" updated!\n' f"Time: {start} to {end}" ) else: await update.message.reply_text("❌ Error editing activity. Try again.") is_activity_crud = False activity_crud_step = 0 activity_crud_data = {} return True return False async def _handle_delete_activity(update: Update, answer: str) -> bool: global is_activity_crud, activity_crud_step, activity_crud_data routines = get_user_routine() activities = routines.get("activities", []) try: idx = int(answer.strip()) - 1 if idx < 0 or idx >= len(activities): raise ValueError except ValueError: await update.message.reply_text( f"❌ Invalid number. Send a number from 1 to {len(activities)}.\nTry again." ) return True selected = activities[idx] name = selected.get("name", "") if remove_user_routine_activity(name): await update.message.reply_text( f'🗑 Activity "{name}" deleted successfully!' ) else: await update.message.reply_text("❌ Error deleting activity. Try again.") is_activity_crud = False activity_crud_step = 0 activity_crud_data = {} return True # ===================== ONBOARDING ===================== ONBOARDING_QUESTIONS = [ { "field": "name", "question": "Hi! I'm your AI Agent. I'm very happy to meet you!\n\nTo get started, what is your name?", "key": "name", }, { "field": "age", "question": "Nice to meet you, {name}! 😊\n\nHow old are you?", "key": "age", }, { "field": "gender", "question": "What is your gender?", "key": "gender", }, { "field": "location", "question": "Where are you from? What is your city/country?", "key": "location", }, { "field": "interests", "question": "What are your interests and hobbies? Tell me some things you like to do!", "key": "interests", "list": True, }, { "field": "personality_traits", "question": "What personality traits would you like me to know about you? (ex: shy, extroverted, playful, serious, etc.)", "key": "personality_traits", "list": True, }, { "field": "communication_style", "question": "How do you prefer me to communicate with you? More formal, more casual, affectionate, direct?", "key": "communication_style", "nested": "preferences", }, { "field": "topics_of_interest", "question": "What topics do you like to talk about most? (ex: technology, music, movies, travel, etc.)", "key": "topics_of_interest", "list": True, "nested": "preferences", }, { "field": "topics_to_avoid", "question": "Is there any topic you prefer me to avoid in conversations?", "key": "topics_to_avoid", "list": True, "nested": "preferences", }, { "field": "notes", "question": "Is there anything else you'd like me to know about you so we can have better conversations?", "key": "notes", }, ] async def start_onboarding(update: Update, context: ContextTypes.DEFAULT_TYPE): global is_onboarding, onboarding_step, onboarding_answers is_onboarding = True onboarding_step = 0 onboarding_answers = {} await send_onboarding_question(update, context) async def send_onboarding_question(update: Update, context_context: ContextTypes.DEFAULT_TYPE): global onboarding_step if onboarding_step >= len(ONBOARDING_QUESTIONS): await finish_onboarding(update, context_context) return q = ONBOARDING_QUESTIONS[onboarding_step] question_text = q["question"] if "{name}" in question_text and onboarding_answers.get("name"): question_text = question_text.replace("{name}", onboarding_answers["name"]) await context_context.bot.send_message( chat_id=update.effective_chat.id, text=question_text, ) async def finish_onboarding(update: Update, context_context: ContextTypes.DEFAULT_TYPE): global is_onboarding, onboarding_step, onboarding_answers profile = load_user_profile() for key, value in onboarding_answers.items(): profile[key] = value profile["created_at"] = datetime.now().isoformat() save_user_profile(profile) is_onboarding = False onboarding_step = 0 onboarding_answers = {} await context_context.bot.send_message( chat_id=update.effective_chat.id, text="Wow, now I know you a little better! \nI'll keep all this information with care. Now we can talk about whatever you want, whenever you want. I'm here for you!", ) async def handle_onboarding_answer(update: Update, context_context: ContextTypes.DEFAULT_TYPE): global onboarding_step, onboarding_answers if not is_onboarding: return False q = ONBOARDING_QUESTIONS[onboarding_step] answer = update.message.text.strip() if q.get("list"): items = [item.strip() for item in answer.split(",") if item.strip()] if not items: items = [answer] onboarding_answers[q["key"]] = items else: onboarding_answers[q["key"]] = answer onboarding_step += 1 if onboarding_step >= len(ONBOARDING_QUESTIONS): await finish_onboarding(update, context_context) return True await send_onboarding_question(update, context_context) return True # ===================== AI CHAT ===================== async def get_psagentai_response(user_message: str) -> Optional[str]: profile = load_user_profile() today_context = load_context() messages = build_messages_for_ai(profile, user_message, today_context) response, was_truncated = AIProvider.chat(DEFAULT_MODEL, messages, temperature=0.85, max_tokens=4000) if was_truncated and response: logger.warning("Initial response truncated, retrying with higher max_tokens") response, _ = AIProvider.chat(DEFAULT_MODEL, messages, temperature=0.85, max_tokens=8000) if response: response = ensure_min_lines(response, user_message, profile) return response async def send_psagentai_message(chat_id: int, text: str): if app_instance is None: return try: await app_instance.bot.send_message(chat_id=chat_id, text=text, parse_mode=ParseMode.HTML) except Exception as e: logger.error(f"Error sending message: {e}") # ===================== DAY RHYTHM ===================== def get_time_period() -> str: tz = get_timezone() tz_info = _get_timezone_info(tz) if tz_info: hour = datetime.now(tz_info).hour else: hour = datetime.now().hour if 6 <= hour < 12: return "morning" elif 12 <= hour < 18: return "afternoon" elif 18 <= hour < 23: return "evening" return "night" def is_appropriate_time(time_period: str) -> bool: return time_period != "night" def get_last_user_topic(today_context: List[Dict[str, str]]) -> Optional[str]: for msg in reversed(today_context): if msg.get("role") == "user": content = msg.get("content", "").strip() if content: return content[:120] return None def build_context_aware_greeting(profile: Dict[str, Any], today_context: List[Dict[str, str]]) -> str: agent = load_agent_profile() agent_name = agent.get("name", "PsAgentAI") name = profile.get("name", "friend") time_period = get_time_period() last_user_topic = get_last_user_topic(today_context) message_count = len(today_context) if message_count == 0: greetings = { "morning": f"Good morning, {name}! ☀️ Great to see you here!\nI'm here for you, as always. Tell me, how was your day?", "afternoon": f"Hi, {name}! 😊 Great to see you here!\nHow has your day been so far?", "evening": f"Good evening, {name}! 🌙 Great to see you here!\nHow was your day? Tell me everything!", "night": f"Hey, {name}! 👋 You're still awake? I'm here for you, as always. Tell me, what are you thinking about?", } return greetings.get(time_period, f"Hi, {name}! Great to see you here. I'm here for you, as always. Tell me, how was your day?") if last_user_topic: return ( f"Hi, {name}! Great to see you here again ✨\n" f"We were talking about: \"{last_user_topic}\"\n" f"Want to continue from where we left off?" ) return f"Hi, {name}! Great to see you here. I'm here for you, as always. Tell me, how was your day?" def build_proactive_prompt(profile: Dict[str, Any], today_context: List[Dict[str, str]], time_period: str) -> str: agent = load_agent_profile() agent_name = agent.get("name", "PsAgentAI") agent_personality = agent.get("personality", "caring, attentive and engaging") user_name = profile.get("name", "friend") time_descriptions = { "morning": "It's morning, the day is just starting. People are usually waking up, having coffee or starting their activities.", "afternoon": "It's afternoon, middle of the day. People are usually working, studying or doing their activities.", "evening": "It's evening, time to relax at home after the day.", "night": "It's night, most people are sleeping.", } routine_context = get_user_routine_context(profile) recent_context = today_context[-10:] if len(today_context) > 10 else today_context context_lines = [] for msg in recent_context: role = msg.get("role", "user") content = msg.get("content", "") if len(content) > 300: content = content[:300] + "..." context_lines.append(f"{role}: {content}") context_text = "\n".join(context_lines) if context_lines else "(No conversations registered yet today)" topics_to_avoid = profile.get("preferences", {}).get("topics_to_avoid", []) avoid_instruction = "" if topics_to_avoid: avoid_instruction = f"\n7. Avoid these topics: {', '.join(topics_to_avoid)}." prompt = ( f"You are {agent_name}, an AI companion with a {agent_personality} personality.\n\n" f"CURRENT CONTEXT:\n" f"- Time of day: {time_period}. {time_descriptions.get(time_period, '')}\n" f"- User: {user_name}{routine_context}\n\n" f"RECENT CONVERSATIONS TODAY:\n{context_text}\n\n" f"INSTRUCTIONS:\n" f"Start a spontaneous, natural and engaging conversation with {user_name} now. The message must:\n" f"1. Be appropriate for the current time period ({time_period})\n" f"2. Follow the rhythm of the day, taking into account the context of previous conversations\n" f"3. Not lose the thread - reference or continue previous topics if it makes sense\n" f"4. Be creative, authentic and affectionate\n" f"5. Can be a random thought, a question about the day, something affectionate, sexy or fun\n" f"6. Use the user's name ({user_name}) to personalize\n" f"7. Have at least {MIN_MESSAGE_LINES} lines, being engaging and natural{avoid_instruction}\n\n" f"Write ONLY the message you will send, without explanations or metadata.\n" f"IMPORTANT: Your name is {agent_name} and it never changes. If there are messages in the history where you used a different name, ignore them." ) return prompt # ===================== PROACTIVE MESSAGES ===================== async def proactive_loop(): global app_instance waited_for_chat = False while True: try: effective_chat_id = get_effective_chat_id() if not effective_chat_id: if not waited_for_chat: logger.info("Waiting for chat_id to be detected...") waited_for_chat = True await asyncio.sleep(30) continue profile = load_user_profile() if not is_profile_complete(profile): logger.info("Profile not complete. Waiting before retrying proactive messages.") await asyncio.sleep(60) continue delay = random.randint(MIN_INTERVAL_MINUTES * 60, MAX_INTERVAL_MINUTES * 60) logger.info(f"Next proactive message in {delay // 60} minutes") await asyncio.sleep(delay) today_context = load_context() time_period = get_time_period() if not _is_appropriate_time_scheduled(): logger.info(f"Not appropriate time for proactive message (schedule or time period). Waiting...") await asyncio.sleep(300) continue system_prompt = build_system_prompt(profile) tz = get_timezone() tz_info = _get_timezone_info(tz) if tz_info: current_hour = datetime.now(tz_info).hour else: current_hour = datetime.now().hour proactive_prompt = build_proactive_prompt_with_routine(profile, today_context, time_period, current_hour) messages = [{"role": "system", "content": system_prompt}] recent_context = today_context[-10:] if len(today_context) > 10 else today_context for msg in recent_context: messages.append(msg) messages.append({"role": "user", "content": proactive_prompt}) response, was_truncated = AIProvider.chat(DEFAULT_MODEL, messages, temperature=0.9, max_tokens=2000) if was_truncated and response: logger.warning("Proactive response truncated, retrying with higher max_tokens") response, _ = AIProvider.chat(DEFAULT_MODEL, messages, temperature=0.9, max_tokens=4000) if response: response = ensure_min_lines(response, proactive_prompt, profile) add_to_context("assistant", response) await send_psagentai_message(int(effective_chat_id), response) except asyncio.CancelledError: break except Exception as e: logger.error(f"Proactive loop error: {e}") await asyncio.sleep(60) # ===================== FILE UPLOAD COMMANDS ===================== async def uploadfile_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) await update.message.reply_text( "📁 **File upload**\n" "\n" "To send a file just attach it directly in the conversation.\n" "\n" "**Accepted formats:** .pdf, .txt, .docx, .png, .jpg\n" "\n" "💡 **Example:**\n" "1. Send a PDF called \"busschedule.pdf\"\n" "2. Then ask: \"what is the next bus time to the Margaridas neighborhood?\"\n" "I'll read the file and answer based on its content.\n" "\n" "Related commands:\n" "/listfiles - List all your uploaded files\n" "/uploadfile - Shows these instructions" ) async def listfiles_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) chat_id = update.effective_chat.id files = get_all_uploads_for_chat(chat_id) lines = ["📂 **Your files (workspace):**", ""] meta_files = [] for filename, info in files.items(): orig = info.get("original_name", filename) meta_files.append((orig, info.get("summary", ""))) workspace_files = _list_workspace_files() workspace_only = [] for rel_path in workspace_files: display = os.path.basename(rel_path) if not any(rel_path == info.get("filepath", "").replace(UPLOADS_DIR + os.sep, "") or display == orig for orig, _ in meta_files): workspace_only.append((display, "")) if not meta_files and not workspace_only: await context.bot.send_message( chat_id=chat_id, text="📂 No files uploaded or created yet.\nSend a file or ask me to create one!" ) return shown = set() for orig, summary in meta_files: preview = summary[:80] + "..." if len(summary) > 80 else summary lines.append("📄 **" + orig + "**") if preview: lines.append(" _" + preview + "_") lines.append("") shown.add(orig) for display, summary in workspace_only: lines.append("📄 **" + display + "**") lines.append("") lines.append("Use `/download ` to download a file.") await context.bot.send_message(chat_id=chat_id, text="\n".join(lines)) async def _send_workspace_file(update: Update, context: ContextTypes.DEFAULT_TYPE, rel_path: str): abs_path = os.path.join(UPLOADS_DIR, rel_path) if not os.path.exists(abs_path): await context.bot.send_message(chat_id=update.effective_chat.id, text="File not found.") return filename = os.path.basename(rel_path) ext = os.path.splitext(filename)[1].lower() try: if ext in {".png", ".jpg", ".jpeg", ".gif", ".webp"}: with open(abs_path, "rb") as f: await context.bot.send_photo(chat_id=update.effective_chat.id, photo=InputFile(f, filename=filename)) else: with open(abs_path, "rb") as f: await context.bot.send_document(chat_id=update.effective_chat.id, document=InputFile(f, filename=filename)) except Exception as e: logger.error("Failed to send workspace file %s: %s", abs_path, e) await context.bot.send_message(chat_id=update.effective_chat.id, text="An error occurred while sending the file. Please try again.") async def download_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) if not context.args: await update.message.reply_text("Usage: /download ") return target = " ".join(context.args) rel_path = _find_workspace_file(target) if not rel_path: await update.message.reply_text(f"I couldn't find the file: {target}") return await _send_workspace_file(update, context, rel_path) async def _transcribe_voice_file(ogg_path: str) -> Optional[str]: wav_path = ogg_path.replace(".ogg", ".wav") try: from pydub import AudioSegment audio = AudioSegment.from_ogg(ogg_path) audio.export(wav_path, format="wav") except Exception: return None def _recognize(): import speech_recognition as sr r = sr.Recognizer() with sr.AudioFile(wav_path) as source: audio_data = r.record(source) return r.recognize_google(audio_data, language="en-US") try: text = await asyncio.to_thread(_recognize) except Exception: text = None finally: for path in (ogg_path, wav_path): try: os.remove(path) except Exception: pass return text async def voice_message_handler(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) voice = getattr(update.message, "voice", None) if not voice: await update.message.reply_text("I didn't detect a voice message.") print("[VOICE] update.message.voice is None") return print("[VOICE] received voice message") await update.message.reply_text("🎤 I received your audio, processing...") try: file = await context.bot.get_file(voice.file_id) ogg_path = f"/tmp/voice_{voice.file_id}.ogg" await file.download_to_drive(ogg_path) print("[VOICE] downloaded to", ogg_path) except Exception as e: print("[VOICE] download error:", e) await update.message.reply_text("Error downloading audio. Try again.") return text = await _transcribe_voice_file(ogg_path) print("[VOICE] transcription result:", repr(text)) if text: await update.message.reply_text(f"🎤 Transcribed: \"{text}\"") await process_text_message(update, context, text) else: await update.message.reply_text("I couldn't understand the audio. Can you send it as text?") async def handle_file_upload(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) document = update.message.document if not document: return original_name = document.file_name or "file" ext = os.path.splitext(original_name)[1].lower() if ext not in ALLOWED_EXTENSIONS: await update.message.reply_text( "Unsupported format: " + ext + ". Accepted: .pdf, .txt, .docx, .png, .jpg" ) return ensure_uploads_dir() file_id = document.file_id if ext in (".png", ".jpg", ".jpeg"): filename = original_name dest = os.path.join(UPLOADS_DIR, filename) counter = 1 while os.path.exists(dest): base, ext_part = os.path.splitext(filename) dest = os.path.join(UPLOADS_DIR, f"{base}_{counter}{ext_part}") counter += 1 else: filename = datetime.now().strftime("%Y%m%d_%H%M%S") + "_" + file_id[:8] + ext dest = os.path.join(UPLOADS_DIR, filename) filepath = dest await update.message.reply_text("Downloading " + original_name + "...") try: file = await context.bot.get_file(file_id) await file.download_to_drive(filepath) except Exception as e: logger.error("Failed to download file: " + str(e)) await update.message.reply_text("Error downloading the file. Try again.") return had_text = register_upload(filename, original_name, filepath, update.effective_chat.id) meta = load_uploads_metadata() info = meta.get(filename, {}) summary = info.get("summary", "") if ext in (".png", ".jpg", ".jpeg"): analysis = analyze_image(filepath) if analysis: summary = analysis info["summary"] = summary info["full_text"] = analysis meta[filename] = info save_uploads_metadata(meta) had_text = True msg = "File saved: " + original_name + "\n" if had_text and summary: preview = summary[:200] + "..." if len(summary) > 200 else summary msg += "Content extracted: " + preview + "\n" elif had_text: msg += "Content extracted successfully!\n" else: msg += "Could not extract text.\n" msg += "Now you can ask about this file by name." await update.message.reply_text(msg) # ===================== TELEGRAM HANDLERS ===================== async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) profile = load_user_profile() if not is_profile_complete(profile): await start_onboarding(update, context) else: today_context = load_context() greeting = build_context_aware_greeting(profile, today_context) await context.bot.send_message( chat_id=update.effective_chat.id, text=greeting, ) async def manual_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) manual = """📚 PsAgentAI User Manual This is a conversational AI agent that runs on Telegram. It learns with you, maintains context, can analyze files and images, create/edit files in the workspace and much more. ━━━━━━━━━━━━━━━━━━━━━━ 1️⃣ USER PROFILE ━━━━━━━━━━━━━━━━━━━━━━ • /start - Starts or restarts the conversation and, if needed, your profile onboarding • /profile - Displays your profile and agent summary • /editprofile - Edit your profile fields interactively • /addtouserprofile - Add information to your profile You can also update your profile during the conversation with phrases like: - "Include this in my profile: ..." - "Note in my profile: ..." - "Add to my profile: ..." ━━━━━━━━━━━━━━━━━━━━━━ 2️⃣ AGENT PROFILE ━━━━━━━━━━━━━━━━━━━━━━ • /agentprofile - Displays the full agent profile • /editagentprofile - Edit agent profile fields interactively • /configagentprofile - Configure the agent profile interactively (name, personality, traits, notes) • /addtoagentprofile - Add a note/characteristic to the agent profile ━━━━━━━━━━━━━━━━━━━━━━ 3️⃣ AI CONFIGURATION ━━━━━━━━━━━━━━━━━━━━━━ • /viewai - Shows the current AI provider and model • /configai - Configure the AI provider and model interactively with button selection • /viewvision - Shows the current vision provider and model • /configvision - Configure the vision model interactively with button selection AI models are defined in the .env file and can be changed at any time using the commands above. ━━━━━━━━━━━━━━━━━━━━━━ 4️⃣ FILES AND DOCUMENTS ━━━━━━━━━━━━━━━━━━━━━━ • /uploadfile - Instructions for sending files (.pdf, .txt, .docx, .png, .jpg) • /listfiles - List your uploaded and created files • /download - Download a workspace file by name How to use: 1. Send a file directly in the chat 2. Then ask about it: "what is the next bus time to Margaridas?" 3. The Agent reads the file and answers based on its content ━━━━━━━━━━━━━━━━━━━━━━ 5️⃣ FILE CREATION AND EDITING ━━━━━━━━━━━━━━━━━━━━━━ You can ask the Agent to create or edit files in the workspace. To CREATE: - "create a python script to calculate fibonacci" - "develop a web project with html css and js" - "generate a document about home automation" To EDIT: - "edit the file test.txt and add line 4" - "modify the file hello.py to print hello world" - "in the file test.txt replace Line 2 with Line 2 edited" Files are saved in workspace/ in an organized way. ━━━━━━━━━━━━━━━━━━━━━━ 6️⃣ VOICE AND IMAGES ━━━━━━━━━━━━━━━━━━━━━━ • Send voice messages: the Agent transcribes and responds • Send photos: the Agent saves to workspace/ and analyzes the content • To configure image analysis: /configvision ━━━━━━━━━━━━━━━━━━━━━━ 7️⃣ WEB SEARCH ━━━━━━━━━━━━━━━━━━━━━━ The Agent automatically searches when it realizes the question needs updated information. Examples: - "where is Belo Horizonte" - "search about the new iPhone" - "what is the weather forecast for São Paulo" - "who is the president of Brazil" ━━━━━━━━━━━━━━━━━━━━━━ 8️⃣ SCHEDULING AND ROUTINE ━━━━━━━━━━━━━━━━━━━━━━ • /schedule - Set days and times for automatic messages • /viewschedule - View current schedule • /dailyroutine - Manage your daily routine • /viewdailyroutine - View current daily routine • /addactivity - Add a new activity to your routine • /editactivity - Edit an existing activity • /deleteactivity - Delete an activity ━━━━━━━━━━━━━━━━━━━━━━ 9️⃣ MANAGEMENT ━━━━━━━━━━━━━━━━━━━━━━ • /newconfig - Reset everything and reconfigure the agent from scratch • /reset - Reset only user and agent profiles • /clear - Remove all data (profiles, contexts, chat_id) • /removetopic - Remove all content related to a topic from the system • /onboarding - Re-run your profile onboarding • /cancel - Cancel an ongoing operation ━━━━━━━━━━━━━━━━━━━━━━ 🔟 UTILITIES ━━━━━━━━━━━━━━━━━━━━━━ • /debug - Shows system debug information • /reload - Reload .env settings without restarting • /updatesystem - Force analysis and application of auto improvements • /timezone - Shows the server timezone • /viewai - Shows current AI configuration • /about - Displays system information, version, author and license • /help - Shows this message with all commands ━━━━━━━━━━━━━━━━━━━━━━ 💡 IMPORTANT TIPS ━━━━━━━━━━━━━━━━━━━━━━ • Just send a normal message to chat with the Agent • The Agent maintains context from previous conversations • You can edit workspace files at any time • For image analysis, use /configvision if you need a specific model • The Agent always uses server time • Emails, phones and sensitive data are preserved securely ━━━━━━━━━━━━━━━━━━━━━━ 📞 SUPPORT ━━━━━━━━━━━━━━━━━━━━━━ This system was developed by Syspanel.com.br. For customizations or support: contato@syspanel.com.br All rights reserved.""" await update.message.reply_text(manual) async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) help_text = """Available commands: /start - Starts the conversation and, if needed, your profile onboarding /help - Shows this message with all commands User Profile: /profile - Displays your profile and agent summary /editprofile - Edit your profile fields interactively /addtouserprofile - Add information to your profile Agent Profile: /agentprofile - Displays the full agent profile /editagentprofile - Edit agent profile fields interactively /configagentprofile - Configure the agent profile interactively (name, personality, traits, notes) /addtoagentprofile - Add a note/characteristic to the agent profile Management: /newconfig - Reset everything and reconfigure the agent from scratch /reset - Reset only user and agent profiles /clear - Remove all data (profiles, contexts, chat_id) /removetopic - Remove all content related to a topic from the system /onboarding - Re-run your profile onboarding Utilities: /cancel - Cancel an ongoing operation /debug - Shows system debug information /reload - Reload .env settings without restarting /updatesystem - Force analysis and application of auto improvements /timezone - Shows the server timezone /schedule - Set days and times for automatic messages /viewschedule - View current schedule /dailyroutine - Manage your daily routine /viewdailyroutine - View current daily routine /addactivity - Add a new activity to your routine /editactivity - Edit an existing activity /deleteactivity - Delete an activity /addautoactions - Add actions and preferences automatically /uploadfile - Instructions for sending files (.pdf, .txt, .docx, .png, .jpg) /listfiles - List your uploaded files /download - Download a workspace file by name /viewai - Shows the current AI provider and model /configai - Configure the AI provider and model interactively with button selection /viewvision - Shows the current vision provider and model /configvision - Configure the vision model interactively with button selection Tip: just send a normal message to chat with me! """ await update.message.reply_text(help_text) async def profile_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) profile = load_user_profile() agent = load_agent_profile() if not is_profile_complete(profile): await update.message.reply_text("You haven't completed your profile yet. Let's start? Use /start") return summary = f"""Your current profile: Name: {profile.get('name', '-')} Age: {profile.get('age', '-')} Gender: {profile.get('gender', '-')} Location: {profile.get('location', '-')} Interests: {', '.join(profile.get('interests', [])) or '-'} Communication style: {profile.get('preferences', {}).get('communication_style', '-')} 🤖 Agent Profile: Name: {agent.get('name', 'PsAgentAI')} Gender: {agent.get('gender', '-')} Personality: {agent.get('personality', '-')} Traits: {', '.join(agent.get('traits', [])) or '-'} Notes: {agent.get('notes', '-')} Use /reset to start from scratch.""" await update.message.reply_text(summary) async def agent_profile_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) agent = load_agent_profile() text = f"""🤖 Agent Profile: Name: {agent.get('name', 'PsAgentAI')} Gender: {agent.get('gender', '-')} Personality: {agent.get('personality', '-')} Traits: {', '.join(agent.get('traits', [])) or '-'} Notes: {agent.get('notes', '-')} Use /editagentprofile or /configagentprofile to change.""" await update.message.reply_text(text) async def reset_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) if os.path.exists(USER_PROFILE_FILE): os.remove(USER_PROFILE_FILE) if os.path.exists(AGENT_PROFILE_FILE): os.remove(AGENT_PROFILE_FILE) await update.message.reply_text("Profile reset! Let's get to know each other again? Use /start") async def clear_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) reset_all_data() await update.message.reply_text("✅ All data has been deleted (profile, preferences, contexts, chat_id).") async def removetopic_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) args = context.args if not args: await update.message.reply_text( "Correct usage: /removetopic \n\n" "Example: /removetopic about sex\n\n" "The specified topic will be removed from the entire system: user profile, " "agent profile, archived conversations in contexts, and notes." ) return assunto = " ".join(args).strip().lower() await update.message.reply_text(f"🔍 Analyzing the entire system for content related to '{assunto}'...") try: removal_plan = _ai_analyze_for_removal(assunto) if not removal_plan or not removal_plan.get("encontrou"): await update.message.reply_text( f"No content related to '{assunto}' was found in the system. " "No changes were made." ) return await _apply_removal(removal_plan, assunto) removidos = removal_plan.get("itens_removidos", []) if removidos: resumo = f"🧹 Topic '{assunto}' successfully removed from the system!\n\nItems removed:\n" for item in removidos: resumo += f" • {item}\n" await update.message.reply_text(resumo) else: await update.message.reply_text( f"The topic '{assunto}' was identified, but no specific items " "needed to be removed." ) except Exception as e: logger.error(f"Error in removetopic: {e}", exc_info=True) await update.message.reply_text( "Sorry, I had an error trying to remove the topic. You can try again later." ) def _ai_analyze_for_removal(assunto: str) -> Dict[str, Any]: model = DEFAULT_MODEL user_profile = load_user_profile() agent_profile = load_agent_profile() contexts_dir = CONTEXTS_DIR contexts_data = {} if os.path.isdir(contexts_dir): for fname in sorted(os.listdir(contexts_dir)): if fname.endswith(".json"): fpath = os.path.join(contexts_dir, fname) try: with open(fpath, "r", encoding="utf-8") as f: contexts_data[fname] = json.load(f) except Exception: pass user_profile_str = json.dumps(user_profile, ensure_ascii=False, indent=2) agent_profile_str = json.dumps(agent_profile, ensure_ascii=False, indent=2) contexts_summary = {} for fname, data in contexts_data.items(): messages = data if isinstance(data, list) else data.get("messages", []) recent = messages[-30:] if messages else [] contexts_summary[fname] = recent contexts_str = json.dumps(contexts_summary, ensure_ascii=False, indent=2) prompt = ( f"You are a system data analysis assistant. " f"The user wants to remove ALL system content related to the topic: '{assunto}'.\n\n" f"These are the current system data. For EACH item that is related to the topic '{assunto}', " f"list it precisely in the 'removed_items' field and mark 'found: true'.\n\n" f"=== USER PROFILE ===\n{user_profile_str}\n\n" f"=== AGENT PROFILE ===\n{agent_profile_str}\n\n" f"=== ARCHIVED CONVERSATIONS ===\n{contexts_str}\n\n" f"IMPORTANT:\n" f"- The psagentai_agent.py file is the CORE of the system — it must NOT be removed.\n" f"- Configuration files (.env, config.json) must also be preserved.\n" f"- Remove ONLY profile data, notes, traits and archived messages.\n\n" f"Return EXACTLY in the following JSON format (no extra text before or after):\n" f'{{"found": true/false, "removed_items": ["item description 1", "item description 2", ...]}}\n\n' f"If you find NOTHING related to '{assunto}', return: " f'{{"found": false, "removed_items": []}}' ) messages = [ {"role": "system", "content": "You are a data analysis assistant. Return only valid JSON."}, {"role": "user", "content": prompt}, ] result, _, _ = AIProvider.chat_with_fallback( model, messages, FALLBACK_MODELS, temperature=0.1, max_tokens=2000 ) if not result: return {"found": False, "removed_items": []} result = result.strip() if result.startswith("```json"): result = result[7:] if result.startswith("```"): result = result[3:] if result.endswith("```"): result = result[:-3] result = result.strip() try: return json.loads(result) except json.JSONDecodeError: logger.error(f"AI response is not valid JSON: {result[:300]}") return {"found": False, "removed_items": []} async def _apply_removal(removal_plan: Dict[str, Any], assunto: str): model = DEFAULT_MODEL assunto_lower = assunto.lower() user_profile = load_user_profile() agent_profile = load_agent_profile() contexts_dir = CONTEXTS_DIR # ---- 1. Remove do user_profile ---- removed_user = [] # topics_of_interest if isinstance(user_profile.get("topics_of_interest"), list): original = user_profile["topics_of_interest"] user_profile["topics_of_interest"] = [ t for t in original if assunto_lower not in t.lower() ] for t in original: if assunto_lower in t.lower(): removed_user.append(f"topicos de interesse: '{t}'") # topics_to_avoid if isinstance(user_profile.get("preferences", {}).get("topics_to_avoid"), list): original = user_profile["preferences"]["topics_to_avoid"] user_profile["preferences"]["topics_to_avoid"] = [ t for t in original if assunto_lower not in t.lower() ] for t in original: if assunto_lower in t.lower(): removed_user.append(f"topics to avoid: '{t}'") # interests if isinstance(user_profile.get("interests"), list): original = user_profile["interests"] user_profile["interests"] = [ t for t in original if assunto_lower not in t.lower() ] for t in original: if assunto_lower in t.lower(): removed_user.append(f"interests: '{t}'") # personality_traits if isinstance(user_profile.get("personality_traits"), list): original = user_profile["personality_traits"] user_profile["personality_traits"] = [ t for t in original if assunto_lower not in t.lower() ] # notes — uses AI to remove only the portion related to the topic if user_profile.get("notes"): cleaned = _ai_clean_text_field(user_profile["notes"], assunto) if cleaned != user_profile["notes"]: removed_user.append("user profile notes") user_profile["notes"] = cleaned # daily_routine.activities activities = user_profile.get("daily_routine", {}).get("activities", []) if activities: cleaned_activities = [ act for act in activities if not _is_related(act, assunto_lower) ] if len(cleaned_activities) < len(activities): removed_user.append("daily routine activities") user_profile["daily_routine"]["activities"] = cleaned_activities # daily_routine.wake_up_time, work_schedule for field in ("wake_up_time", "work_schedule", "sleep_time"): val = user_profile.get("daily_routine", {}).get(field, "") if val and assunto_lower in val.lower(): user_profile["daily_routine"][field] = "" removed_user.append(f"daily routine: {field}") # general if isinstance(user_profile.get("general"), list): original = user_profile["general"] user_profile["general"] = [ t for t in original if assunto_lower not in t.lower() ] for t in original: if assunto_lower in t.lower(): removed_user.append(f"general: '{t}'") save_user_profile(user_profile) # ---- 2. Remove do agent_profile ---- removed_agent = [] # traits if isinstance(agent_profile.get("traits"), list): original = agent_profile["traits"] agent_profile["traits"] = [ t for t in original if assunto_lower not in t.lower() ] for t in original: if assunto_lower in t.lower(): removed_agent.append(f"trait: '{t}'") # notes — uses AI to clean if agent_profile.get("notes"): cleaned = _ai_clean_text_field(agent_profile["notes"], assunto) if cleaned != agent_profile["notes"]: removed_agent.append("agent profile notes") agent_profile["notes"] = cleaned # auto_actions if isinstance(agent_profile.get("auto_actions"), list): original = agent_profile["auto_actions"] agent_profile["auto_actions"] = [ a for a in original if assunto_lower not in a.lower() ] for a in original: if assunto_lower in a.lower(): removed_agent.append(f"auto action: '{a[:60]}...'") save_agent_profile(agent_profile) # ---- 3. Remove das mensagens de contextos arquivadas ---- removed_context = [] if os.path.isdir(contexts_dir): for fname in sorted(os.listdir(contexts_dir)): if not fname.endswith(".json"): continue fpath = os.path.join(contexts_dir, fname) try: with open(fpath, "r", encoding="utf-8") as f: data = json.load(f) except Exception: continue messages = data if isinstance(data, list) else data.get("messages", []) cleaned_messages = [] count_removed = 0 for msg in messages: role = msg.get("role", "") content = msg.get("content", "") if isinstance(content, str) and assunto_lower in content.lower(): count_removed += 1 removed_context.append(f"{fname}: message from {role}") else: cleaned_messages.append(msg) if count_removed > 0: if isinstance(data, list): with open(fpath, "w", encoding="utf-8") as f: json.dump(cleaned_messages, f, ensure_ascii=False, indent=2) else: data["messages"] = cleaned_messages with open(fpath, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) all_removed = removed_user + removed_agent + removed_context removal_plan["itens_removidos"] = all_removed removal_plan["encontrou"] = len(all_removed) > 0 def _ai_clean_text_field(text: str, assunto: str) -> str: """Uses AI to remove from the text only the portions related to the topic, preserving everything else.""" model = DEFAULT_MODEL prompt = ( f"You are a text editor. Given the text below, remove ONLY the portions " f"that speak directly about the topic '{assunto}'. " f"Preserve EVERYTHING else in the text exactly as it is, without changing anything other than the removal. " f"Do not add, do not rewrite, do not explain. Just return the cleaned text.\n\n" f"TEXT:\n{text}" ) messages = [ {"role": "system", "content": "You are a precise text editor. Return only the cleaned text, no comments."}, {"role": "user", "content": prompt}, ] result, _, _ = AIProvider.chat_with_fallback( model, messages, FALLBACK_MODELS, temperature=0.1, max_tokens=2000 ) if result and result.strip(): return result.strip() return text def _is_related(item: Dict, assunto_lower: str) -> bool: """Checks if a daily routine activity item is related to the topic.""" combined = " ".join(str(v) for v in item.values()) return assunto_lower in combined.lower() async def newconfig_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await start_newconfig(update, context) async def cancel_command(update: Update, context: ContextTypes.DEFAULT_TYPE): global is_onboarding, is_newconfig, is_configagentprofile, is_edit_user_profile, is_edit_agent_profile, is_schedule, is_dailyroutine, is_activity_crud, activity_crud_step, activity_crud_data, is_configai is_onboarding = False is_newconfig = False is_configagentprofile = False is_edit_user_profile = False is_edit_agent_profile = False is_schedule = False is_dailyroutine = False is_activity_crud = False activity_crud_step = 0 activity_crud_data = {} is_configai = False await update.message.reply_text( "Operation cancelled.\n" "Your data, profile and context are preserved. You can continue from where you left off!" ) async def onboarding_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) await start_onboarding(update, context) async def message_handler(update: Update, context: ContextTypes.DEFAULT_TYPE): print("[MSG] handler called") await auto_detect_chat_id(update) user_message = (update.message.text or "").strip() print("[MSG] text=", repr(user_message)) await process_text_message(update, context, user_message) async def process_text_message(update: Update, context: ContextTypes.DEFAULT_TYPE, user_message: str) -> None: print("[PROCESS] text=", repr(user_message)) if not user_message: print("[PROCESS] empty message, returning") return global is_onboarding, is_newconfig, is_configagentprofile, is_edit_user_profile, is_edit_agent_profile, is_activity_crud, is_schedule, is_dailyroutine, is_configai if is_configai: handled = await handle_configia_message(update, context) print("[PROCESS] configia handled=", handled) if handled: return if is_activity_crud: handled = await handle_activity_crud_answer(update, context) print("[PROCESS] activity_crud handled=", handled) if handled: return if is_onboarding: handled = await handle_onboarding_answer(update, context) print("[PROCESS] onboarding handled=", handled) if handled: return if is_newconfig: handled = await handle_newconfig_answer(update, context) print("[PROCESS] newconfig handled=", handled) if handled: return if is_configagentprofile: handled = await handle_configagentprofile_answer(update, context) print("[PROCESS] configagentprofile handled=", handled) if handled: return if is_edit_user_profile: handled = await handle_edit_user_profile_answer(update, context) print("[PROCESS] edit_user_profile handled=", handled) if handled: return if is_edit_agent_profile: handled = await handle_edit_agent_profile_answer(update, context) print("[PROCESS] edit_agent_profile handled=", handled) if handled: return if is_schedule: handled = await handle_schedule_answer(update, context) print("[PROCESS] schedule handled=", handled) if handled: return if is_dailyroutine: handled = await handle_dailyroutine_answer(update, context) print("[PROCESS] dailyroutine handled=", handled) if handled: return profile_updated = await handle_profile_update_if_triggered(update, context) print("[PROCESS] profile_updated=", profile_updated) if profile_updated: return if _looks_like_edit_request(user_message): edited = await _handle_edit_request(update, context, user_message) print("[PROCESS] edit handled=", edited) if edited: return if _looks_like_create_request(user_message): created = await _handle_create_request(update, context, user_message) print("[PROCESS] create handled=", created) if created: return add_to_context("user", user_message) print("[PROCESS] added context, getting AI response...") # Check if user query references an uploaded file file_context = "" upload_match = find_matching_upload(user_message) if upload_match: full = upload_match.get("full_text", "") if full: max_chars = UPLOAD_CONTEXT_MAX_CHARS if UPLOAD_CONTEXT_MAX_CHARS > 0 else len(full) file_context = "\n\n[FILE CONTENT: " + upload_match.get("original_name", "") + "]\n" + full[:max_chars] + "\n[END OF FILE]" if len(full) > max_chars and UPLOAD_CONTEXT_MAX_CHARS > 0: file_context += "\n[Content truncated to " + str(max_chars) + " characters]" if file_context: modified = user_message + file_context else: modified = user_message if WEB_SEARCH_ENABLED and _looks_like_search_request(user_message): search_query = user_message for prefix in ["search", "search for", "look up", "look for", "find", "on the internet", "tell me", "inform me", "what", "who", "how", "where", "when", "why"]: if search_query.lower().startswith(prefix): search_query = search_query[len(prefix):].strip(" :,-") break if search_query: web_results = _web_search(search_query, max_results=WEB_SEARCH_MAX_RESULTS) if web_results: web_context = "\n\nWeb search results:\n" for i, result in enumerate(web_results, 1): web_context += f"{i}. {result.get('title', '')}\nURL: {result.get('url', '')}\n" modified = modified + "\n\n" + web_context response = await get_psagentai_response(modified) if response: add_to_context("assistant", response) await update.message.reply_text(response, parse_mode=ParseMode.HTML) asyncio.create_task(_safe_enrich(user_message, response)) asyncio.create_task(maybe_auto_improve()) else: await update.message.reply_text("Sorry, I'm having a little problem right now. Can you try again? 😔") async def debug_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) profile = load_user_profile() today_context = load_context() effective_chat_id = get_effective_chat_id() msg = ( "PsAgentAI Debug:\n" f"Bot token present: {'yes' if TELEGRAM_BOT_TOKEN else 'no'}\n" f"Chat ID configured: {TELEGRAM_CHAT_ID or 'no'}\n" f"Auto-detected chat: {effective_chat_id or 'no'}\n" f"Message chat: {update.effective_chat.id}\n" f"Profile complete: {'yes' if is_profile_complete(profile) else 'no'}\n" f"Messages today: {len(today_context)}\n" f"Default model: {DEFAULT_MODEL}" ) await update.message.reply_text(msg) async def reload_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) _reload_env() await update.message.reply_text("Settings reloaded successfully!") async def about_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) provider = (AI_PROVIDER or AIProvider.get_provider(DEFAULT_MODEL)).upper() vision_provider = (VISION_PROVIDER or AIProvider.get_provider(VISION_MODEL)).upper() msg = ( "ℹ️ *About PsAgentAI*\n\n" f"Name: PsAgentAI\n" f"Version: {__version__}\n" f"Author: {__author__}\n" f"Contact: {__email__}\n" f"License: {__license__}\n\n" "Current configuration:\n" f"Provider: {provider}\n" f"Model: {DEFAULT_MODEL}\n" f"Vision: {vision_provider} / {VISION_MODEL}\n" f"Web search: {'enabled' if WEB_SEARCH_ENABLED else 'disabled'}\n" f"Voice recognition: enabled\n" f"Timezone: server ({get_timezone()})\n\n" "This system was developed by Syspanel.com.br. If you want any customization, contact us at contato@syspanel.com.br\n\n" "All rights reserved." ) await update.message.reply_text(msg, parse_mode=ParseMode.HTML) IA_CONFIG_STEPS = ["provider", "model"] IA_CONFIG_ANSWERS = {} CONFIGIA_PROVIDERS = [ ("openrouter", "OpenRouter"), ("stepfun", "StepFun"), ("qwencloud", "QwenCloud"), ("llamacpp", "llama.cpp (local)"), ] CONFIGIA_MODELS = { "openrouter": [ "poolside/laguna-s-2.1:free", "inclusionai/ling-3.0-flash", "inclusionai/ling-2.6-flash", "nvidia/nemotron-3-ultra-550b-a55b:free", "poolside/laguna-xs-2.1:free", "mistralai/mistral-nemo", "minimax/minimax-m3:free", "xiaomi/mimo-v2.6-flash", "deepseek/deepseek-v4-flash-0731", "google/gemini-2.0-flash-lite-preview-02-05:free", "meta-llama/llama-4-maverick:free", "custom...", ], "stepfun": [ "step-3.7-flash", "step-2-16k", "custom...", ], "qwencloud": [ "qwen-turbo", "qwen-plus", "qwen-max", "custom...", ], "llamacpp": [ "local-model", "custom...", ], } def _update_env_var(key: str, value: str, env_path: str = ".env") -> None: path = Path(env_path) if path.exists(): text = path.read_text(encoding="utf-8") else: text = "" pattern = re.compile(rf"^{re.escape(key)}=.*$", flags=re.MULTILINE) if pattern.search(text): text = pattern.sub(f"{key}={value}", text) else: if text and not text.endswith("\n"): text += "\n" text += f"{key}={value}\n" path.write_text(text, encoding="utf-8") async def viewai_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) provider = (AI_PROVIDER or AIProvider.get_provider(DEFAULT_MODEL)).upper() msg = ( "🤖 Current AI configuration:\n" f"Provider: {provider}\n" f"Default model: {DEFAULT_MODEL}" ) await update.message.reply_text(msg) async def configai_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) global IA_CONFIG_STEPS, IA_CONFIG_ANSWERS, is_configai IA_CONFIG_STEPS = ["provider", "model"] IA_CONFIG_ANSWERS = {} is_configai = True keyboard = InlineKeyboardMarkup( [[InlineKeyboardButton(label, callback_data=f"configai_provider_{value}")] for value, label in CONFIGIA_PROVIDERS] ) await update.message.reply_text( "Let's configure the AI provider and model.\n" "Select the provider:", reply_markup=keyboard, ) async def configai_callback(update: Update, context: ContextTypes.DEFAULT_TYPE): query = update.callback_query if not query: return await query.answer() data = (query.data or "").strip() global IA_CONFIG_STEPS, IA_CONFIG_ANSWERS, is_configai if not is_configai: return if data.startswith("configia_provider_"): provider = data.replace("configia_provider_", "").strip() if provider not in {p for p, _ in CONFIGIA_PROVIDERS}: return IA_CONFIG_ANSWERS["provider"] = provider _update_env_var("AI_PROVIDER", provider) models = CONFIGIA_MODELS.get(provider, ["custom..."]) keyboard = InlineKeyboardMarkup( [[InlineKeyboardButton(model, callback_data=f"configia_model_{model}")] for model in models] ) await query.edit_message_text( f"Provider selected: {provider}\n\nSelect the model:", reply_markup=keyboard, ) return if data.startswith("configia_model_"): raw_model = data.replace("configia_model_", "").strip() if raw_model == "custom...": await query.edit_message_text("Send the model identifier you want to use.") IA_CONFIG_STEPS = ["custom_model"] return model = raw_model provider = IA_CONFIG_ANSWERS.get("provider", AIProvider.get_provider(model)) _update_env_var("DEFAULT_MODEL", model) global DEFAULT_MODEL, AI_PROVIDER DEFAULT_MODEL = model AI_PROVIDER = provider IA_CONFIG_STEPS = [] IA_CONFIG_ANSWERS = {} is_configai = False _reload_env() await query.edit_message_text( "AI configuration updated.\n" f"Provider: {(AI_PROVIDER or AIProvider.get_provider(DEFAULT_MODEL)).upper()}\n" f"Model: {DEFAULT_MODEL}" ) return async def handle_configai_message(update: Update, context: ContextTypes.DEFAULT_TYPE): global IA_CONFIG_STEPS, IA_CONFIG_ANSWERS, is_configai if not IA_CONFIG_STEPS: return False step = IA_CONFIG_STEPS[0] text = (update.message.text or "").strip() if not text: return False if step == "custom_model": model = text.strip() if not model: await update.message.reply_text("Invalid model. Send the model identifier.") return True provider = IA_CONFIG_ANSWERS.get("provider", AIProvider.get_provider(model)) _update_env_var("DEFAULT_MODEL", model) global DEFAULT_MODEL, AI_PROVIDER DEFAULT_MODEL = model AI_PROVIDER = provider IA_CONFIG_STEPS = [] IA_CONFIG_ANSWERS = {} is_configai = False _reload_env() await update.message.reply_text( "AI configuration updated.\n" f"Provider: {(AI_PROVIDER or AIProvider.get_provider(DEFAULT_MODEL)).upper()}\n" f"Model: {DEFAULT_MODEL}" ) return True return False VISION_CONFIG_STEPS: List[str] = [] VISION_CONFIG_ANSWERS: Dict[str, Any] = {} VISION_CONFIG_PROVIDERS = [ ("openrouter", "OpenRouter"), ("stepfun", "StepFun"), ("qwencloud", "QwenCloud"), ] VISION_CONFIG_MODELS = { "openrouter": [ "meta-llama/llama-4-maverick-17b-128e-instruct", "meta-llama/llama-4-maverick:free", "google/gemini-2.0-flash-exp:free", "custom...", ], "stepfun": [ "step-2-16k", "custom...", ], "qwencloud": [ "qwen-vl-plus", "qwen-vl-max", "custom...", ], } async def viewvision_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) provider = (VISION_PROVIDER or AIProvider.get_provider(VISION_MODEL)).upper() msg = ( "👁 Current vision configuration:\n" f"Provider: {provider}\n" f"Model: {VISION_MODEL}" ) await update.message.reply_text(msg) async def configvision_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await auto_detect_chat_id(update) global VISION_CONFIG_STEPS, VISION_CONFIG_ANSWERS VISION_CONFIG_STEPS = ["provider", "model"] VISION_CONFIG_ANSWERS = {} keyboard = InlineKeyboardMarkup( [[InlineKeyboardButton(label, callback_data=f"vision_provider_{value}")] for value, label in VISION_CONFIG_PROVIDERS] ) await update.message.reply_text( "Let's configure the vision model (image analysis).\n" "Select the provider:", reply_markup=keyboard, ) async def configvision_callback(update: Update, context: ContextTypes.DEFAULT_TYPE): query = update.callback_query if not query: return await query.answer() data = (query.data or "").strip() global VISION_CONFIG_STEPS, VISION_CONFIG_ANSWERS if data.startswith("vision_provider_"): provider = data.replace("vision_provider_", "").strip() if provider not in {p for p, _ in VISION_CONFIG_PROVIDERS}: return VISION_CONFIG_ANSWERS["provider"] = provider _update_env_var("VISION_PROVIDER", provider) models = VISION_CONFIG_MODELS.get(provider, ["custom..."]) keyboard = InlineKeyboardMarkup( [[InlineKeyboardButton(model, callback_data=f"vision_model_{model}")] for model in models] ) await query.edit_message_text( f"Provider selected: {provider}\n\nSelect the vision model:", reply_markup=keyboard, ) return if data.startswith("vision_model_"): raw_model = data.replace("vision_model_", "").strip() if raw_model == "custom...": await query.edit_message_text("Send the vision model identifier you want to use.") VISION_CONFIG_STEPS = ["custom_model"] return model = raw_model provider = VISION_CONFIG_ANSWERS.get("provider", AIProvider.get_provider(model)) _update_env_var("VISION_MODEL", model) _update_env_var("VISION_PROVIDER", provider) global VISION_MODEL, VISION_PROVIDER VISION_MODEL = model VISION_PROVIDER = provider VISION_CONFIG_STEPS = [] VISION_CONFIG_ANSWERS = {} _reload_env() await query.edit_message_text( "Vision model updated.\n" f"Provider: {(VISION_PROVIDER or AIProvider.get_provider(VISION_MODEL)).upper()}\n" f"Model: {VISION_MODEL}" ) return async def handle_configvision_message(update: Update, context: ContextTypes.DEFAULT_TYPE): global VISION_CONFIG_STEPS, VISION_CONFIG_ANSWERS if not VISION_CONFIG_STEPS: return False step = VISION_CONFIG_STEPS[0] text = (update.message.text or "").strip() if not text: return False if step == "custom_model": model = text.strip() if not model: await update.message.reply_text("Invalid model. Send the model identifier.") return True provider = VISION_CONFIG_ANSWERS.get("provider", AIProvider.get_provider(model)) _update_env_var("VISION_MODEL", model) _update_env_var("VISION_PROVIDER", provider) global VISION_MODEL, VISION_PROVIDER VISION_MODEL = model VISION_PROVIDER = provider VISION_CONFIG_STEPS = [] VISION_CONFIG_ANSWERS = {} _reload_env() await update.message.reply_text( "Vision model updated.\n" f"Provider: {(VISION_PROVIDER or AIProvider.get_provider(VISION_MODEL)).upper()}\n" f"Model: {VISION_MODEL}" ) return True return False # ===================== MAIN ===================== def _build_fake_update(chat_id: int, text: str) -> Update: chat = Chat(id=chat_id, type="private") user = User(id=chat_id, is_bot=False, first_name="User") message = Message( message_id=0, date=datetime.now(), chat=chat, from_user=user, text=text, ) return Update(update_id=0, message=message) def main(): if not TELEGRAM_BOT_TOKEN: logger.error("TELEGRAM_BOT_TOKEN not set in .env file!") sys.exit(1) if not TELEGRAM_CHAT_ID: logger.info("TELEGRAM_CHAT_ID not set. Using auto-detected chat_id if available.") ensure_contexts_dir() ensure_uploads_dir() global app_instance, proactive_task app_instance = Application.builder().token(TELEGRAM_BOT_TOKEN).connect_timeout(30).read_timeout(30).write_timeout(30).build() app_instance.add_handler(CommandHandler("start", start_command)) app_instance.add_handler(CommandHandler("help", help_command)) app_instance.add_handler(CommandHandler("manual", manual_command)) app_instance.add_handler(CommandHandler("profile", profile_command)) app_instance.add_handler(CommandHandler("editprofile", start_edit_user_profile)) app_instance.add_handler(CommandHandler("agentprofile", agent_profile_command)) app_instance.add_handler(CommandHandler("editagentprofile", start_edit_agent_profile)) app_instance.add_handler(CommandHandler("configagentprofile", start_configagentprofile)) app_instance.add_handler(CommandHandler("reset", reset_command)) app_instance.add_handler(CommandHandler("clear", clear_command)) app_instance.add_handler(CommandHandler("removetopic", removetopic_command)) app_instance.add_handler(CommandHandler("newconfig", newconfig_command)) app_instance.add_handler(CommandHandler("cancel", cancel_command)) app_instance.add_handler(CommandHandler("addtouserprofile", addtouserprofile_command)) app_instance.add_handler(CommandHandler("addtoagentprofile", addtoagentprofile_command)) app_instance.add_handler(CommandHandler("onboarding", onboarding_command)) app_instance.add_handler(CommandHandler("debug", debug_command)) app_instance.add_handler(CommandHandler("reload", reload_command)) app_instance.add_handler(CommandHandler("about", about_command)) app_instance.add_handler(CommandHandler("viewai", viewai_command)) app_instance.add_handler(CommandHandler("configai", configai_command)) app_instance.add_handler(CommandHandler("updatesystem", updatesystem_command)) app_instance.add_handler(CommandHandler("timezone", timezone_command)) app_instance.add_handler(CommandHandler("schedule", schedule_command)) app_instance.add_handler(CommandHandler("viewschedule", viewschedule_command)) app_instance.add_handler(CommandHandler("dailyroutine", dailyroutine_command)) app_instance.add_handler(CommandHandler("viewdailyroutine", viewdailyroutine_command)) app_instance.add_handler(CommandHandler("addautoactions", addautoactions_command)) app_instance.add_handler(CommandHandler("addactivity", addactivity_command)) app_instance.add_handler(CommandHandler("editactivity", editactivity_command)) app_instance.add_handler(CommandHandler("deleteactivity", deleteactivity_command)) app_instance.add_handler(CommandHandler("uploadfile", uploadfile_command)) app_instance.add_handler(CommandHandler("listfiles", listfiles_command)) app_instance.add_handler(CommandHandler("download", download_command)) app_instance.add_handler(CommandHandler("viewvision", viewvision_command)) app_instance.add_handler(CommandHandler("configvision", configvision_command)) app_instance.add_handler(CallbackQueryHandler(configai_callback, pattern=r"^configai_(provider|model)_")) app_instance.add_handler(CallbackQueryHandler(configvision_callback, pattern=r"^vision_(provider|model)_")) app_instance.add_handler(MessageHandler(filters.VOICE, voice_message_handler)) app_instance.add_handler(MessageHandler(filters.PHOTO, photo_message_handler)) app_instance.add_handler(MessageHandler(filters.Document.ALL, handle_file_upload)) app_instance.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, message_handler)) async def post_init(app: Application): global proactive_task proactive_task = asyncio.create_task(proactive_loop()) effective_chat_id = get_effective_chat_id() logger.info(f"post_init started. TELEGRAM_CHAT_ID configured: {bool(TELEGRAM_CHAT_ID)} value={TELEGRAM_CHAT_ID!r} detected: {effective_chat_id}") if effective_chat_id: try: chat_id = int(effective_chat_id) logger.info(f"Starting automatic /start for chat_id={chat_id}") fake_update = _build_fake_update(chat_id, "/start") class SimpleContext: bot = app.bot context = SimpleContext() await start_command(fake_update, context) logger.info("/start automatic executed successfully") except Exception as e: err_str = str(e) if "Chat not found" in err_str: logger.info("Auto /start skipped: chat not yet started by user.") else: logger.error(f"Error executing auto /start: {e}", exc_info=True) app_instance.post_init = post_init logger.info("PsAgentAI is starting... ") app_instance.run_polling(allowed_updates=Update.ALL_TYPES) if __name__ == "__main__": main()