""" Telegram AI Chatbot — Multi-persona, owner-protected, semantic memory Powered by Groq API + PostgreSQL 16 + pgvector + Ollama embeddings Config files: .env — persona, bot name, trigger, rules (different per deployment) .env_credentials — secret keys, DB and Ollama credentials """ import os import re import json import random import logging import asyncio import aiohttp from urllib.parse import quote_plus import datetime import zoneinfo import traceback from functools import wraps from dotenv import load_dotenv load_dotenv(".env_credentials") load_dotenv(".env") import psycopg2 from psycopg2.extras import RealDictCursor from psycopg2 import pool as pg_pool from telegram import Update from telegram.ext import ( Application, CommandHandler, MessageHandler, filters, ContextTypes ) from groq import Groq try: from openai import OpenAI except ImportError: OpenAI = None # ══════════════════════════════════════════════ # CONFIG # ══════════════════════════════════════════════ # ── Credentials (.env_credentials) ──────────── TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN") GROQ_API_KEY = os.getenv("GROQ_API_KEY", "") OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY", "") LLM_PROVIDER = os.getenv("LLM_PROVIDER", "groq") GROQ_MODEL = os.getenv("GROQ_MODEL", "llama-3.3-70b-versatile") OPENROUTER_MODEL = os.getenv("OPENROUTER_MODEL", "deepseek/deepseek-chat-v3-0324") PG_CONFIG = { "host": os.getenv("PG_HOST", "localhost"), "port": int(os.getenv("PG_PORT", "5432")), "dbname": os.getenv("PG_DB", "manel_bot"), "user": os.getenv("PG_USER", "postgres"), "password": os.getenv("PG_PASSWORD", ""), } OWNER_ID = int(os.getenv("OWNER_TELEGRAM_ID", "0")) # NewsData.io NEWSDATA_API_KEY = os.getenv("NEWSDATA_API_KEY", "") BRAVE_API_KEY = os.getenv("BRAVE_API_KEY", "") # real web search; DDG fallback if unset # Ollama endpoint and embedding model OLLAMA_URL = os.getenv("OLLAMA_URL", "http://localhost:11434") EMBED_MODEL = os.getenv("EMBED_MODEL", "nomic-embed-text-v2-moe") # nomic-embed-text-v2-moe output dimension — update if you switch models EMBED_DIM = int(os.getenv("EMBED_DIM", "768")) # How many semantically similar results to return per search_memory call VECTOR_TOP_K = int(os.getenv("VECTOR_TOP_K", "8")) # Minimum cosine similarity for search_memory results (nomic-embed scores # for unrelated text usually sit around 0.3-0.5, so 0.45 is a sane floor) MEM_SIM_THRESHOLD = float(os.getenv("MEM_SIM_THRESHOLD", "0.45")) # ── Persona (.env) ───────────────────────────── BOT_NAME = os.getenv("BOT_NAME", "Manel") BOT_TRIGGER = os.getenv("BOT_TRIGGER", "manel") BOT_MAX_CONTEXT = int(os.getenv("BOT_MAX_CONTEXT", "40")) BOT_HISTORY_SAVE = int(os.getenv("BOT_HISTORY_SAVE", "500")) BOT_ROLE = os.getenv("BOT_ROLE", "") BOT_RULES = os.getenv("BOT_RULES", "") BOT_TOOL_RULES = os.getenv("BOT_TOOL_RULES", "") BOT_MEMORY_RULES = os.getenv("BOT_MEMORY_RULES", "") BOT_EXAMPLES = os.getenv("BOT_EXAMPLES", "") BOT_TIMEZONE = os.getenv("BOT_TIMEZONE", "Europe/Lisbon") # home timezone for date/time display # ───────────────────────────────────────────── logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") # httpx logs every Telegram API URL at INFO — including the bot token. # Keep it quiet so tokens never land in logfiles. logging.getLogger("httpx").setLevel(logging.WARNING) log = logging.getLogger(__name__) if LLM_PROVIDER == "openrouter": if OpenAI is None: raise ImportError("openai package not installed. Run: pip install openai") llm_client = OpenAI( api_key=OPENROUTER_API_KEY, base_url="https://openrouter.ai/api/v1", default_headers={"HTTP-Referer": "https://github.com/manel-bot"} ) LLM_MODEL = OPENROUTER_MODEL else: llm_client = Groq(api_key=GROQ_API_KEY) LLM_MODEL = GROQ_MODEL pg_pool_conn: pg_pool.ThreadedConnectionPool | None = None # Keep references to fire-and-forget tasks so they aren't garbage-collected _bg_tasks: set = set() def fire_and_forget(coro): """Schedule a coroutine in the background without awaiting it.""" task = asyncio.create_task(coro) _bg_tasks.add(task) task.add_done_callback(_bg_tasks.discard) async def llm_create(**kwargs): """Run the (synchronous) LLM client call in a worker thread so the Telegram event loop is never blocked while waiting on the API.""" return await asyncio.to_thread(lambda: llm_client.chat.completions.create(**kwargs)) # ══════════════════════════════════════════════ # SYSTEM PROMPT BUILDER # ══════════════════════════════════════════════ def build_system_prompt(user_first_name: str, memories: list[str]) -> str: def env_to_bullets(raw: str) -> str: return "\n".join("- " + l.strip() for l in raw.splitlines() if l.strip()) memory_block = "" if memories: mem_lines = "\n".join("- " + m for m in memories) memory_block = "\n\n[LONG-TERM MEMORY about users in this group]\n" + mem_lines sections = [] if BOT_ROLE: sections.append("# Role\n" + BOT_ROLE) if BOT_RULES: sections.append("# Rules\n" + env_to_bullets(BOT_RULES)) if BOT_TOOL_RULES: sections.append("# Tool Selection Strategy\n"+ env_to_bullets(BOT_TOOL_RULES)) if BOT_MEMORY_RULES: sections.append("# Memory Rules\n" + env_to_bullets(BOT_MEMORY_RULES)) if BOT_EXAMPLES: sections.append("# Examples\n" + BOT_EXAMPLES) try: home_tz = zoneinfo.ZoneInfo(BOT_TIMEZONE) except Exception: home_tz = datetime.timezone.utc now_home = datetime.datetime.now(datetime.timezone.utc).astimezone(home_tz) date_str = now_home.strftime("%A, %d %B %Y") time_str = now_home.strftime("%H:%M") time_block = ( f"Today is: {date_str}\n" f"Current time ({BOT_TIMEZONE}): {time_str}\n" f"For other timezones, use the web_search tool.\n" f"IMPORTANT — your training data predates today. For sports results, " f"competitions, news or any current events: (1) ALWAYS include the year " f"{now_home.year} in your search query — e.g. 'mundial {now_home.year}', " f"never a different year; (2) base your answer ONLY on the search results, " f"never on what you remember about similar past events; (3) if the results " f"don't contain the answer, say you don't know — do not fill gaps from memory." ) sections.append("# Current Date and Time\n" + time_block) sections.append( "# Current User\n" "The user who mentioned you is: " + user_first_name + memory_block ) return "\n\n".join(sections) # ══════════════════════════════════════════════ # OLLAMA EMBEDDINGS # ══════════════════════════════════════════════ async def get_embedding(text: str) -> list[float] | None: """Call local Ollama to embed text. Returns list of floats or None on error.""" url = OLLAMA_URL.rstrip("/") + "/api/embed" payload = {"model": EMBED_MODEL, "input": text} try: async with aiohttp.ClientSession() as session: async with session.post(url, json=payload, timeout=aiohttp.ClientTimeout(total=10)) as resp: if resp.status != 200: log.warning(f"Ollama embed HTTP {resp.status}: {await resp.text()}") return None data = await resp.json() # Ollama /api/embed returns {"embeddings": [[...]]} embeddings = data.get("embeddings") if embeddings and len(embeddings) > 0: return embeddings[0] log.warning(f"Unexpected Ollama response: {data}") return None except Exception as e: log.error(f"Ollama embedding error: {e}") return None # ══════════════════════════════════════════════ # DATABASE — PostgreSQL 16 + pgvector # ══════════════════════════════════════════════ def init_db(): global pg_pool_conn pg_pool_conn = pg_pool.ThreadedConnectionPool(1, 10, **PG_CONFIG) with get_db() as con: with con.cursor() as cur: # Enable pgvector extension cur.execute("CREATE EXTENSION IF NOT EXISTS vector;") cur.execute(""" CREATE TABLE IF NOT EXISTS users ( user_id TEXT PRIMARY KEY, username TEXT, full_name TEXT, first_seen TIMESTAMPTZ DEFAULT NOW() ); """) cur.execute(""" CREATE TABLE IF NOT EXISTS memories ( id SERIAL PRIMARY KEY, bot_name TEXT NOT NULL DEFAULT 'default', user_id TEXT NOT NULL, memory TEXT NOT NULL, created TIMESTAMPTZ DEFAULT NOW(), FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE CASCADE ); CREATE INDEX IF NOT EXISTS idx_memories_bot_user ON memories(bot_name, user_id); """) # History table with vector column for semantic search cur.execute(f""" CREATE TABLE IF NOT EXISTS history ( id SERIAL PRIMARY KEY, bot_name TEXT NOT NULL DEFAULT 'default', chat_id TEXT NOT NULL, user_id TEXT, username TEXT, role TEXT NOT NULL, content TEXT NOT NULL, embedding vector({EMBED_DIM}), created TIMESTAMPTZ DEFAULT NOW() ); CREATE INDEX IF NOT EXISTS idx_history_bot_chat ON history(bot_name, chat_id, created DESC); """) # IVFFlat index for fast vector search (built lazily once rows exist) # We create it only if it doesn't exist yet cur.execute(""" SELECT 1 FROM pg_indexes WHERE tablename = 'history' AND indexname = 'idx_history_embedding'; """) if not cur.fetchone(): # Only create after enough rows accumulate — skip silently for now # The bot will still work via sequential scan until you run: # CREATE INDEX idx_history_embedding ON history # USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100); log.info("Vector index not yet created — will use seq scan until you create it manually.") con.commit() log.info(f"PostgreSQL + pgvector ready. Bot={BOT_NAME} dim={EMBED_DIM}") class get_db: def __enter__(self): self.con = pg_pool_conn.getconn() return self.con def __exit__(self, exc_type, exc_val, exc_tb): try: if exc_type is not None: # A failed statement leaves the connection in an aborted # transaction — roll back before returning it to the pool. try: self.con.rollback() except Exception: pass finally: pg_pool_conn.putconn(self.con) return False # don't suppress exceptions # ── Users ────────────────────────────────────── def upsert_user(user_id: str, username: str, full_name: str): with get_db() as con: with con.cursor() as cur: cur.execute(""" INSERT INTO users (user_id, username, full_name) VALUES (%s, %s, %s) ON CONFLICT (user_id) DO UPDATE SET username = EXCLUDED.username, full_name = EXCLUDED.full_name """, (user_id, username, full_name)) con.commit() def user_exists(user_id: str) -> bool: with get_db() as con: with con.cursor() as cur: cur.execute("SELECT 1 FROM users WHERE user_id = %s", (user_id,)) return cur.fetchone() is not None # ── History ──────────────────────────────────── def store_message_sync(chat_id: str, user_id: str | None, username: str, role: str, content: str, embedding: list[float] | None): """Synchronous DB write — called from async context via asyncio.to_thread.""" with get_db() as con: with con.cursor() as cur: cur.execute(""" INSERT INTO history (bot_name, chat_id, user_id, username, role, content, embedding) VALUES (%s, %s, %s, %s, %s, %s, %s) """, (BOT_NAME, chat_id, user_id, username, role, content, embedding)) # Trim oldest beyond BOT_HISTORY_SAVE. # Probabilistic (~1 in 50 inserts): the subquery sorts the whole # chat history, which is too expensive to run on every message. # Ordered by id — created has second resolution and ties are # nondeterministic. if random.random() < 0.02: cur.execute(""" DELETE FROM history WHERE bot_name = %s AND chat_id = %s AND id NOT IN ( SELECT id FROM history WHERE bot_name = %s AND chat_id = %s ORDER BY id DESC LIMIT %s ) """, (BOT_NAME, chat_id, BOT_NAME, chat_id, BOT_HISTORY_SAVE)) con.commit() async def store_message(chat_id: str, user_id: str | None, username: str, role: str, content: str): """Embed the message then write to DB. Non-blocking — runs embedding + DB in thread.""" embedding = await get_embedding(content) if embedding is None: log.warning("Storing message without embedding (Ollama unavailable).") await asyncio.to_thread(store_message_sync, chat_id, user_id, username, role, content, embedding) def get_group_context(chat_id: str, limit: int = BOT_MAX_CONTEXT) -> list[dict]: """Fetch most recent messages for LLM context window.""" with get_db() as con: with con.cursor(cursor_factory=RealDictCursor) as cur: cur.execute(""" SELECT role, username, content, created FROM history WHERE bot_name = %s AND chat_id = %s ORDER BY id DESC LIMIT %s """, (BOT_NAME, chat_id, limit)) rows = cur.fetchall() return list(reversed(rows)) def get_history_for_llm(chat_id: str) -> list[dict]: rows = get_group_context(chat_id, BOT_MAX_CONTEXT) result = [] for r in rows: if r["role"] == "user": # Label each user message with their name so the LLM knows who said what prefix = f"[{r['username']}] " if r["username"] else "" result.append({"role": "user", "content": prefix + r["content"]}) else: # Assistant messages: no prefix — prevents model from echoing "[Manel]" in replies result.append({"role": "assistant", "content": r["content"]}) return result def vector_search_sync(chat_id: str, query_embedding: list[float], top_k: int) -> list[dict]: """Cosine similarity search against history embeddings.""" with get_db() as con: with con.cursor(cursor_factory=RealDictCursor) as cur: cur.execute(""" SELECT username, content, created, 1 - (embedding <=> %s::vector) AS similarity FROM history WHERE bot_name = %s AND chat_id = %s AND embedding IS NOT NULL ORDER BY embedding <=> %s::vector LIMIT %s """, (query_embedding, BOT_NAME, chat_id, query_embedding, top_k)) return cur.fetchall() def clear_history(chat_id: str): with get_db() as con: with con.cursor() as cur: cur.execute( "DELETE FROM history WHERE bot_name = %s AND chat_id = %s", (BOT_NAME, chat_id) ) con.commit() # ── Memories ─────────────────────────────────── def add_memory(user_id: str, memory: str): with get_db() as con: with con.cursor() as cur: cur.execute( "INSERT INTO memories (bot_name, user_id, memory) VALUES (%s, %s, %s)", (BOT_NAME, user_id, memory) ) con.commit() def get_memories(user_id: str) -> list[str]: with get_db() as con: with con.cursor() as cur: cur.execute(""" SELECT memory FROM memories WHERE bot_name = %s AND user_id = %s ORDER BY created DESC LIMIT 50 """, (BOT_NAME, user_id)) return [r[0] for r in cur.fetchall()] def get_all_memories_indexed(user_id: str) -> list[tuple]: with get_db() as con: with con.cursor() as cur: cur.execute(""" SELECT id, memory FROM memories WHERE bot_name = %s AND user_id = %s ORDER BY created DESC LIMIT 50 """, (BOT_NAME, user_id)) return cur.fetchall() def delete_memory_by_id(memory_id: int): with get_db() as con: with con.cursor() as cur: cur.execute( "DELETE FROM memories WHERE id = %s AND bot_name = %s", (memory_id, BOT_NAME) ) con.commit() def clear_memories(user_id: str): with get_db() as con: with con.cursor() as cur: cur.execute( "DELETE FROM memories WHERE bot_name = %s AND user_id = %s", (BOT_NAME, user_id) ) con.commit() def clear_all_memories(): with get_db() as con: with con.cursor() as cur: cur.execute("DELETE FROM memories WHERE bot_name = %s", (BOT_NAME,)) con.commit() # ══════════════════════════════════════════════ # OWNER GUARD # ══════════════════════════════════════════════ def owner_only(func): @wraps(func) async def wrapper(update: Update, ctx: ContextTypes.DEFAULT_TYPE): if update.effective_user.id != OWNER_ID: log.warning(f"Blocked admin command from user_id={update.effective_user.id}") await update.message.reply_text("⛔ Não tens permissão para usar este comando.") return return await func(update, ctx) return wrapper def is_owner(update: Update) -> bool: return update.effective_user.id == OWNER_ID # ══════════════════════════════════════════════ # WEB SEARCH — DuckDuckGo + Open-Meteo weather # ══════════════════════════════════════════════ WEATHER_KEYWORDS = [ "tempo", "weather", "temperatura", "chuva", "chover", "sol", "vento", "neve", "frio", "calor", "forecast", "previsão", "graus", "humidade", "nublado", "trovoada", "granizo", "nevoeiro", "céu", "guarda-chuva", "mínima", "máxima", "minima", "maxima", "sensação", "precipitação" ] # Keywords that require a web search for current time in other timezones TIMEZONE_KEYWORDS = [ "que horas são", "que horas", "horas são", "hora é", "que hora", "what time", "horas na", "horas em", "horas no", "horas nos", "horas são na", "horas são em", "horas aí", "horas lá", "fuso horário", "timezone", "time zone" ] # Location words that combined with recent time context suggest a timezone followup TIMEZONE_PLACES = [ "islândia", "iceland", "finlândia", "finland", "portugal", "espanha", "spain", "frança", "france", "alemanha", "germany", "reino unido", "uk", "brasil", "brazil", "estados unidos", "usa", "london", "paris", "lisboa", "madrid", "berlim", "berlin", "reykjavik" ] def _kw_match(text: str, keywords) -> bool: """Whole-word/phrase match. Plain `kw in text` gives false positives: 'sol' matches 'consola', 'vento' matches 'invento', etc. — and those false positives *force* a web_search tool call downstream.""" q = text.lower() return any( re.search(r"(? bool: if _kw_match(query, TIMEZONE_KEYWORDS): return True # Short followup like "e na Islândia?" — place name + no other topic if _kw_match(query, TIMEZONE_PLACES): words = query.lower().split() if len(words) <= 6 and not _kw_match(query, WEATHER_KEYWORDS): return True return False NEWS_KEYWORDS = [ "notícias", "noticias", "noticia", "notícia", "news", "últimas", "ultimas", "hoje no jornal", "manchetes", "atualidade", "atualidades", "pesquisa", "pesquisar", "procura na internet", "procurar na internet", "pesquisa na internet", "o que aconteceu", "o que se passa", "novidades" ] def is_news_query(query: str) -> bool: return _kw_match(query, NEWS_KEYWORDS) # WMO weather code → Portuguese description WMO_CODES = { 0: "céu limpo", 1: "principalmente limpo", 2: "parcialmente nublado", 3: "nublado", 45: "nevoeiro", 48: "nevoeiro com geada", 51: "chuviscos fracos", 53: "chuviscos moderados", 55: "chuviscos densos", 61: "chuva fraca", 63: "chuva moderada", 65: "chuva forte", 71: "neve fraca", 73: "neve moderada", 75: "neve forte", 77: "grãos de neve", 80: "aguaceiros fracos", 81: "aguaceiros moderados", 82: "aguaceiros violentos", 85: "aguaceiros de neve fracos", 86: "aguaceiros de neve fortes", 95: "trovoada", 96: "trovoada com granizo", 99: "trovoada com granizo forte" } def is_weather_query(query: str) -> bool: return _kw_match(query, WEATHER_KEYWORDS) # Nominatim allows max 1 req/s — cache geocode results so repeated # questions about the same place don't hit it again. _geo_cache: dict[str, tuple[float, float, str]] = {} async def weather_search(query: str) -> str | None: """Geocode location then call Open-Meteo for current + forecast. No API key needed.""" # Extract location using word-boundary replacement stop_words = ( WEATHER_KEYWORDS + [BOT_TRIGGER, "hoje", "amanhã", "amanha", "agora", "atual", "atualmente", "esta", "semana", "preciso", "precisas", "precisa", "qual", "como", "está", "vai", "vais", "vou", "janeiro", "fevereiro", "março", "marco", "abril", "maio", "junho", "julho", "agosto", "setembro", "outubro", "novembro", "dezembro", # the model frequently writes tool queries in English "january", "february", "march", "april", "may", "june", "july", "august", "september", "october", "november", "december", "today", "tomorrow", "tonight", "now", "current", "currently", "this", "week", "weekend", "what", "whats", "is", "the", "in", "at", "for", "of", "and", "o", "a", "os", "as", "em", "de", "do", "da", "dos", "das", "para", "por", "com", "sem", "?", "!"] ) location = query.lower() for sw in sorted(stop_words, key=len, reverse=True): location = re.sub(r"\b" + re.escape(sw) + r"\b", " ", location) # Drop stray numbers (day/year the model puts in search queries) — # "faro 17 2026" geocodes to nothing and we'd lose the real weather data. location = " ".join(w for w in location.split() if not w.isdigit()).strip() if not location: return None # No location — let LLM answer from context try: # Step 1: Geocode with Nominatim (cached — usage policy is 1 req/s) if location in _geo_cache: lat, lon, name = _geo_cache[location] else: geo_url = "https://nominatim.openstreetmap.org/search" geo_params = {"q": location, "format": "json", "limit": 1} headers = {"User-Agent": "ManelBot/1.0"} async with aiohttp.ClientSession() as session: async with session.get(geo_url, params=geo_params, headers=headers, timeout=aiohttp.ClientTimeout(total=8)) as r: geo = await r.json() if not geo: # Signal "couldn't geocode" — caller falls back to a normal # web search instead of telling the model to rephrase forever. log.warning(f"Geocode miss for extracted location: '{location}'") return None lat = float(geo[0]["lat"]) lon = float(geo[0]["lon"]) name = geo[0]["display_name"].split(",")[0] _geo_cache[location] = (lat, lon, name) # Step 2: Open-Meteo current + hourly forecast om_url = "https://api.open-meteo.com/v1/forecast" om_params = ( f"latitude={lat}&longitude={lon}" f"¤t=temperature_2m,apparent_temperature,relative_humidity_2m," f"precipitation,weather_code,wind_speed_10m,wind_gusts_10m" f"&hourly=temperature_2m,precipitation_probability,weather_code,wind_gusts_10m" f"&forecast_days=1&timezone=auto&models=ecmwf_ifs025" ) async with aiohttp.ClientSession() as session: async with session.get(f"{om_url}?{om_params}", timeout=aiohttp.ClientTimeout(total=8)) as r: if r.status != 200: return f"Erro ao obter tempo (HTTP {r.status})." data = await r.json() cur = data["current"] temp = cur["temperature_2m"] feels = cur["apparent_temperature"] humidity = cur["relative_humidity_2m"] wind = cur["wind_speed_10m"] gusts = cur["wind_gusts_10m"] wcode = cur["weather_code"] desc = WMO_CODES.get(wcode, f"código {wcode}") result = ( f"🌍 {name}\n" f"🌡️ {temp:.0f}°C (sensação {feels:.0f}°C)\n" f"🌤️ {desc.capitalize()}\n" f"💧 Humidade: {humidity}%\n" f"💨 Vento: {wind:.0f} km/h (rajadas até {gusts:.0f} km/h)" ) # Next few hours hours = data["hourly"]["time"] temps = data["hourly"]["temperature_2m"] probs = data["hourly"]["precipitation_probability"] wcodes = data["hourly"]["weather_code"] # Use UTC offset from Open-Meteo response to get correct local time. # Compare full timestamps — comparing only hour-of-day breaks near # midnight (at 23:00 there would be no "upcoming" hours at all). utc_offset = data.get("utc_offset_seconds", 0) now_local = (datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(seconds=utc_offset)).replace(tzinfo=None) upcoming = [(h, t, p, w) for h, t, p, w in zip(hours, temps, probs, wcodes) if datetime.datetime.fromisoformat(h) > now_local][:4] if upcoming: result += "\n\n📅 Próximas horas:" for h, t, p, w in upcoming: d = WMO_CODES.get(w, "") result += f"\n {h[11:16]} — {t:.0f}°C, {p}% chuva, {d}" return result except Exception as e: log.warning(f"Open-Meteo error: {e}") return "Serviço meteorológico indisponível agora." async def web_search(query: str) -> str: """Route weather queries to Open-Meteo, everything else to DuckDuckGo.""" if is_weather_query(query): result = await weather_search(query) if result is not None: return result # Couldn't extract/geocode a location — fall through to a regular # web search of the original query rather than dead-ending the model. log.info("weather_search found no location — falling back to general search") # Use NewsData.io for news queries, DuckDuckGo for everything else if is_news_query(query) and NEWSDATA_API_KEY: try: # Detect if query is about Portugal specifically pt_keywords = ["portugal", "nacional", "nacionais", "portuguesa", "português", "porto", "lisboa", "governo", "assembleia", "república"] is_pt = any(kw in query.lower() for kw in pt_keywords) if is_pt: # Portuguese national news — Portuguese language, Portugal only params = { "apikey": NEWSDATA_API_KEY, "language": "pt", "country": "pt", "size": 6, } else: # International news — English, no country filter, use query keywords params = { "apikey": NEWSDATA_API_KEY, "language": "en", "size": 6, "q": query, } async with aiohttp.ClientSession() as session: async with session.get("https://newsdata.io/api/1/news", params=params, timeout=aiohttp.ClientTimeout(total=8)) as resp: data = await resp.json() if data.get("status") == "success" and data.get("results"): lines = [] for art in data["results"][:5]: title = art.get("title", "").strip() source = art.get("source_name") or art.get("source_id", "") source = source.strip() if source else "" pubdate = art.get("pubDate", "")[:10] if title: lines.append(f"• {title}" + (f" ({source}, {pubdate})" if source else "")) return "\n".join(lines) if lines else "Sem notícias encontradas." return "Sem notícias encontradas." except Exception as e: log.warning(f"NewsData.io error: {e}") # Brave Search — real web results (titles + snippets), best for current # events, sports, prices. Free tier: ~2000 queries/month. if BRAVE_API_KEY: try: async with aiohttp.ClientSession() as session: async with session.get( "https://api.search.brave.com/res/v1/web/search", params={"q": query, "count": 5}, headers={"X-Subscription-Token": BRAVE_API_KEY, "Accept": "application/json"}, timeout=aiohttp.ClientTimeout(total=8), ) as resp: if resp.status == 429: log.warning("Brave API rate/quota limit hit — falling back to DDG") raise RuntimeError("brave quota") resp.raise_for_status() data = await resp.json() lines = [] for r in (data.get("web", {}).get("results") or [])[:5]: title = (r.get("title") or "").strip() desc = re.sub(r"<[^>]+>", "", r.get("description") or "").strip() url_ = r.get("url", "") age = r.get("age", "") if title: lines.append(f"• {title}" + (f" ({age})" if age else "") + (f"\n {desc}" if desc else "") + (f"\n {url_}" if url_ else "")) if lines: return "\n".join(lines) log.info("Brave returned no results — falling back to DDG") except Exception as e: log.warning(f"Brave search error: {e} — falling back to DDG") # DuckDuckGo Instant Answer fallback (encyclopedic answers only) url = f"https://api.duckduckgo.com/?q={quote_plus(query)}&format=json&no_redirect=1&no_html=1&skip_disambig=1" try: async with aiohttp.ClientSession() as session: async with session.get(url, timeout=aiohttp.ClientTimeout(total=8)) as resp: data = await resp.json(content_type=None) results = [] if data.get("AbstractText"): results.append(data["AbstractText"]) if data.get("AbstractURL"): results.append("Fonte: " + data["AbstractURL"]) for topic in data.get("RelatedTopics", [])[:4]: if isinstance(topic, dict) and topic.get("Text"): results.append("• " + topic["Text"]) return "\n".join(results) if results else "Sem resultados. Responderei com base no meu conhecimento." except Exception as e: log.warning(f"Web search error: {e}") return "Pesquisa web indisponível agora." # ══════════════════════════════════════════════ # TRIGGER CHECK # ══════════════════════════════════════════════ # Some models (e.g. DeepSeek) occasionally emit their native tool-call # markup as plain text instead of structured tool_calls — that must never # be posted to the chat. _MARKUP_PATTERNS = [ r"]*>", # DeepSeek DSML tags: <|DSML|tool_calls> etc. r"<[||][^<>]*[||]>", # generic <|...|> special tokens r"]*>", # style wrappers r"]*>", ] def sanitize_reply(text: str) -> str: """Cut the reply at the first tool-markup tag. Any text after the model starts 'calling a tool' in plain text is machinery, not an answer — a pure-markup reply becomes empty and the caller falls back to '…'.""" if not text or "<" not in text: return text earliest = len(text) for p in _MARKUP_PATTERNS: m = re.search(p, text, flags=re.IGNORECASE) if m: earliest = min(earliest, m.start()) if earliest == len(text): return text log.warning("Model reply contained raw tool-call markup — truncated before it") return text[:earliest].strip() def is_mentioned(text: str) -> bool: trig = re.escape(BOT_TRIGGER) # 1) Whole-word match, any capitalization: "tabernas", "Tabernas,", "TABERNAS!" # — but not inside other words ("tabernasca"). if re.search(r"(? str: """One retry with an explicit plain-text instruction — used whenever the model returns an empty (or markup-only) answer.""" messages.append({ "role": "user", "content": ("[system] Responde AGORA em texto simples, em português, " "com base na conversa acima. Sem chamadas de ferramentas. " "Se não tiveres a informação, di-lo honestamente e com humor.") }) r = await llm_create( model=LLM_MODEL, messages=messages, max_tokens=2048, temperature=0.7 ) return sanitize_reply(r.choices[0].message.content or "") async def chat_with_groq(chat_id: str, user_id: str, user_first_name: str, user_message: str) -> str: # DB reads in a worker thread — never block the Telegram event loop memories = await asyncio.to_thread(get_memories, user_id) history = await asyncio.to_thread(get_history_for_llm, chat_id) system = build_system_prompt(user_first_name, memories) # NOTE: the triggering message was already stored by handle_message, so # it is the last entry of `history` (with its [name] prefix). Do NOT # append it again — that duplicated the current message in every prompt. messages = [{"role": "system", "content": system}] + history # Force web_search tool when weather, timezone or news keywords detected if is_weather_query(user_message): tool_choice = {"type": "function", "function": {"name": "web_search"}} log.info("Forcing web_search for weather query") elif is_timezone_query(user_message): tool_choice = {"type": "function", "function": {"name": "web_search"}} log.info("Forcing web_search for timezone query") elif is_news_query(user_message): tool_choice = {"type": "function", "function": {"name": "web_search"}} log.info("Forcing web_search for news query") # Inject intent into messages so model generates correct query intl_keywords = ["internacionais", "internacional", "mundo", "world", "international"] if any(kw in user_message.lower() for kw in intl_keywords): # Prepend hint to system prompt instead of fake user message messages[0]["content"] += "\n\nNEXT TOOL CALL: Search for INTERNATIONAL world news in English. Do NOT include Portugal in query." else: messages[0]["content"] += "\n\nNEXT TOOL CALL: Search for Portuguese national news. Use query: noticias portugal hoje." else: tool_choice = "auto" forced = tool_choice != "auto" response = await llm_create( model=LLM_MODEL, messages=messages, tools=TOOLS, tool_choice=tool_choice, max_tokens=2048, temperature=0.7, ) msg = response.choices[0].message tool_calls = msg.tool_calls or [] if not tool_calls: reply = msg.content or "" # Model was forced to call a tool but produced neither a tool call # nor usable content. if forced and (not reply.strip() or reply.strip() == "…"): if is_weather_query(user_message): log.info("Model skipped forced tool — calling weather_search directly") weather_result = await weather_search(user_message) if weather_result: messages.append({"role": "user", "content": f"[weather data]\n{weather_result}\n\nAnswer the user naturally in Portuguese based on this data."}) r2 = await llm_create( model=LLM_MODEL, messages=messages, max_tokens=2048, temperature=0.7 ) return sanitize_reply(r2.choices[0].message.content or "") or "…" # Timezone/news (or weather with no direct result): retry once # without forcing so the model can answer normally. log.info("Model skipped forced tool — retrying with tool_choice=auto") r2 = await llm_create( model=LLM_MODEL, messages=messages, tools=TOOLS, tool_choice="auto", max_tokens=2048, temperature=0.7, ) m2 = r2.choices[0].message if not (m2.tool_calls or []): return sanitize_reply(m2.content or "") or "…" msg, tool_calls = m2, m2.tool_calls else: reply = sanitize_reply(reply) if not reply: # No tools, no text — typically the model spent the whole # token budget on reasoning. Nudge once for a plain answer. log.warning( f"Empty direct reply (finish_reason=" f"{response.choices[0].finish_reason}) — retrying with nudge" ) reply = await _nudge_final(messages) return reply or "…" # ── Multi-round tool loop ───────────────────────────────────────── # The model may need to chain tools (search the city, then search its # weather). Previously the post-tool call passed no `tools`, so models # wanting a second search emitted raw tool markup as text. MAX_TOOL_ROUNDS = 3 rounds = 0 seen_calls: dict = {} # (fn, normalized args) -> previous result while tool_calls: rounds += 1 messages.append({ "role": "assistant", "content": msg.content or "", "tool_calls": [ {"id": tc.id, "type": "function", "function": {"name": tc.function.name, "arguments": tc.function.arguments}} for tc in tool_calls ] }) for tc in tool_calls: fn = tc.function.name # Key for dedupe — computed outside the try so it always exists, # even when the arguments fail to parse. call_key = (fn, (tc.function.arguments or "").strip().lower()) # Each tool runs in its own try/except: one bad tool call (bad JSON # args, FK violation, network error) must not kill the whole reply. try: args = json.loads(tc.function.arguments) log.info(f"Tool: {fn}({args})") # Dedupe: the model sometimes re-issues near-identical searches # and burns the whole round budget. Return the cached result # with an explicit instruction to answer. if call_key in seen_calls: log.info("Duplicate tool call detected — returning cached result") result = ("[REPEATED SEARCH — same result as before]\n" + seen_calls[call_key] + "\n\nDo NOT search again. Answer the user now " "in Portuguese using the information above. If the " "information is insufficient, say so honestly.") messages.append({"role": "tool", "content": result, "tool_call_id": tc.id}) continue if fn == "web_search": result = await web_search(args["query"]) # Stamp results with today's date — reminds the model at # the moment of answering that these are current, and its # own memory of similar past events is not. today = datetime.datetime.now().strftime("%d/%m/%Y") result = (f"[Search results — today is {today}. Answer from " f"these results, not from memory.]\n{result}") elif fn == "search_memory": query = args["query"] found = [] # ── 1. Semantic search over group history ────────────── query_emb = await get_embedding(query) if query_emb: rows = await asyncio.to_thread( vector_search_sync, chat_id, query_emb, VECTOR_TOP_K ) for row in rows: sim = float(row["similarity"]) if sim < MEM_SIM_THRESHOLD: # skip low similarity continue ts = row["created"].strftime("%Y-%m-%d %H:%M") if row.get("created") else "?" found.append( f"[{ts}] {row.get('username', '?')} (sim={sim:.2f}): {row['content']}" ) else: log.warning("search_memory: no embedding — skipping vector search") # ── 2. Always include saved long-term memories ───────── for mem in await asyncio.to_thread(get_memories, user_id): found.append(f"[memory] {mem}") if found: result = "\n".join(found) else: result = "Nothing relevant found in memory." elif fn == "save_memory": # Only accept a target user_id the bot has actually seen; # a hallucinated ID would violate the FK on memories.user_id. # Default to the person speaking. target_uid = str(args.get("user_id") or user_id) if target_uid != user_id and not await asyncio.to_thread(user_exists, target_uid): log.warning(f"save_memory: unknown user_id {target_uid}, saving under caller") target_uid = user_id await asyncio.to_thread(add_memory, target_uid, args["memory"]) result = f"Memory saved: {args['memory']}" else: result = "Unknown tool." except Exception as e: log.error(f"Tool {fn} failed: {e}\n{traceback.format_exc()}") result = f"Tool error: {e}" seen_calls[call_key] = result messages.append({"role": "tool", "content": result, "tool_call_id": tc.id}) if rounds < MAX_TOOL_ROUNDS: # Model may chain another tool call if it needs to response = await llm_create( model=LLM_MODEL, messages=messages, tools=TOOLS, tool_choice="auto", max_tokens=2048, temperature=0.7, ) else: # Round budget spent — force a plain-text answer log.warning(f"Tool round limit ({MAX_TOOL_ROUNDS}) reached — forcing final answer") response = await llm_create( model=LLM_MODEL, messages=messages, max_tokens=2048, temperature=0.7, ) msg = response.choices[0].message tool_calls = (msg.tool_calls or []) if rounds < MAX_TOOL_ROUNDS else [] reply = sanitize_reply(msg.content or "") if not reply: # Model produced nothing (or only tool markup). One retry with an # explicit instruction — this is what caused the bare "…" replies. log.warning( f"Empty final answer (finish_reason=" f"{response.choices[0].finish_reason}) — retrying once with nudge" ) reply = await _nudge_final(messages) return reply or "…" # ══════════════════════════════════════════════ # TELEGRAM HANDLERS # ══════════════════════════════════════════════ async def cmd_start(update: Update, ctx: ContextTypes.DEFAULT_TYPE): await asyncio.to_thread( upsert_user, str(update.effective_user.id), update.effective_user.username or "", update.effective_user.full_name, ) owner_hint = "\n\n🔑 És o owner deste bot." if is_owner(update) else "" await update.message.reply_text( f"👋 Olá {update.effective_user.first_name}! Sou o **{BOT_NAME}** 😊\n\n" f"Menciona o meu nome numa mensagem para eu responder!\n\n" f"Comandos:\n" f"/memorias – ver as tuas memórias\n" f"/esquecer `` – apagar memória nº n\n" f"/ajuda – mostrar esta mensagem\n" f"\n*Comandos de owner:*\n" f"/apagarmem `` – apagar memórias de um utilizador\n" f"/apagartudo – apagar TODAS as memórias do bot\n" f"/novachat – limpar histórico deste grupo" + owner_hint, parse_mode="Markdown" ) async def cmd_help(update: Update, ctx: ContextTypes.DEFAULT_TYPE): await cmd_start(update, ctx) async def cmd_memories(update: Update, ctx: ContextTypes.DEFAULT_TYPE): uid = str(update.effective_user.id) rows = await asyncio.to_thread(get_all_memories_indexed, uid) if not rows: await update.message.reply_text("🧠 Ainda não tenho memórias tuas. Fala comigo!") return text = f"🧠 **As tuas memórias ({BOT_NAME}):**\n\n" + "\n".join(f"{i+1}. {m}" for i, (_, m) in enumerate(rows)) text += "\n\nUsa /esquecer `` para apagar uma." await update.message.reply_text(text, parse_mode="Markdown") async def cmd_forget(update: Update, ctx: ContextTypes.DEFAULT_TYPE): uid = str(update.effective_user.id) args = ctx.args if not args or not args[0].isdigit(): await update.message.reply_text("Uso: /esquecer `` (obtém números com /memorias)", parse_mode="Markdown") return rows = await asyncio.to_thread(get_all_memories_indexed, uid) idx = int(args[0]) - 1 if 0 <= idx < len(rows): await asyncio.to_thread(delete_memory_by_id, rows[idx][0]) await update.message.reply_text("✅ Memória apagada.") else: await update.message.reply_text("❌ Número inválido.") @owner_only async def cmd_clear_user_memories(update: Update, ctx: ContextTypes.DEFAULT_TYPE): if not ctx.args: await update.message.reply_text("Uso: /apagarmem ``", parse_mode="Markdown") return await asyncio.to_thread(clear_memories, ctx.args[0]) await update.message.reply_text(f"🗑️ Memórias do utilizador `{ctx.args[0]}` apagadas.", parse_mode="Markdown") @owner_only async def cmd_clearall(update: Update, ctx: ContextTypes.DEFAULT_TYPE): await asyncio.to_thread(clear_all_memories) await update.message.reply_text(f"🗑️ Todas as memórias do {BOT_NAME} foram apagadas.") @owner_only async def cmd_newchat(update: Update, ctx: ContextTypes.DEFAULT_TYPE): await asyncio.to_thread(clear_history, str(update.effective_chat.id)) await update.message.reply_text("🔄 Histórico do grupo limpo! (Memórias de longo prazo mantidas.)") async def handle_message(update: Update, ctx: ContextTypes.DEFAULT_TYPE): if not update.message or not update.message.text: return user = update.effective_user uid = str(user.id) chat_id = str(update.effective_chat.id) text = update.message.text fname = user.first_name or "utilizador" await asyncio.to_thread(upsert_user, uid, user.username or "", user.full_name or fname) if not is_mentioned(text): # Store + embed in the background — no reason to make the handler # wait on Ollama for a message the bot won't answer. fire_and_forget(store_message(chat_id, uid, fname, "user", text)) log.debug(f"[{BOT_NAME}] Storing silently: {text[:60]}") return # Mentioned: store BEFORE generating, and await it — chat_with_groq # reads this message back from the history table (it must not be # appended a second time there). await store_message(chat_id, uid, fname, "user", text) await update.message.chat.send_action("typing") error_reply = False try: reply = await chat_with_groq(chat_id, uid, fname, text) except Exception as e: log.error(f"Groq error: {e}\n{traceback.format_exc()}") reply = "⚠️ Ocorreu um erro. Tenta novamente." error_reply = True if not error_reply: fire_and_forget(store_message(chat_id, None, BOT_NAME, "assistant", reply)) for chunk in [reply[i:i+4096] for i in range(0, len(reply), 4096)]: if not chunk.strip(): continue try: await update.message.chat.send_message(chunk, parse_mode="Markdown") except Exception: try: await update.message.chat.send_message(chunk) except Exception: pass # ══════════════════════════════════════════════ # MAIN # ══════════════════════════════════════════════ def main(): if not TELEGRAM_BOT_TOKEN: raise ValueError("TELEGRAM_BOT_TOKEN not set — check .env_credentials") if LLM_PROVIDER == "groq" and not GROQ_API_KEY: raise ValueError("GROQ_API_KEY not set — check .env_credentials") if LLM_PROVIDER == "openrouter" and not OPENROUTER_API_KEY: raise ValueError("OPENROUTER_API_KEY not set — check .env_credentials") if OWNER_ID == 0: log.warning("OWNER_TELEGRAM_ID not set — admin commands blocked for everyone.") init_db() log.info(f"Starting {BOT_NAME} | trigger='{BOT_TRIGGER}' | provider={LLM_PROVIDER} | model={LLM_MODEL} | embed={EMBED_MODEL}@{OLLAMA_URL}") app = Application.builder().token(TELEGRAM_BOT_TOKEN).build() app.add_handler(CommandHandler("start", cmd_start)) app.add_handler(CommandHandler("ajuda", cmd_help)) app.add_handler(CommandHandler("help", cmd_help)) app.add_handler(CommandHandler("memorias", cmd_memories)) app.add_handler(CommandHandler("esquecer", cmd_forget)) app.add_handler(CommandHandler("apagarmem", cmd_clear_user_memories)) app.add_handler(CommandHandler("apagartudo", cmd_clearall)) app.add_handler(CommandHandler("novachat", cmd_newchat)) app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message)) app.run_polling(allowed_updates=Update.ALL_TYPES) if __name__ == "__main__": main()