From 5b904cab086944bc5e531babad5643181e7052a3 Mon Sep 17 00:00:00 2001 From: Jeremias Date: Fri, 17 Jul 2026 21:42:41 +0000 Subject: [PATCH] Latest fixes by Fable 5 --- bot.py | 321 +++++++++++++++++++++++++++++++++---------------- gitignore.txt | 7 ++ pyproject.toml | 22 ++++ 3 files changed, 245 insertions(+), 105 deletions(-) create mode 100644 gitignore.txt create mode 100644 pyproject.toml diff --git a/bot.py b/bot.py index 82d682f..50cc3c4 100644 --- a/bot.py +++ b/bot.py @@ -10,6 +10,7 @@ Config files: import os import re import json +import random import logging import asyncio import aiohttp @@ -72,6 +73,9 @@ EMBED_MODEL = os.getenv("EMBED_MODEL", "nomic-embed-text-v2-moe") 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") @@ -88,6 +92,9 @@ BOT_TIMEZONE = os.getenv("BOT_TIMEZONE", "Europe/Lisbon") # home timezo # ───────────────────────────────────────────── 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": @@ -104,6 +111,21 @@ else: 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 @@ -157,7 +179,7 @@ async def get_embedding(text: str) -> list[float] | None: payload = {"model": EMBED_MODEL, "input": text} try: async with aiohttp.ClientSession() as session: - async with session.post(url, json=payload, timeout=aiohttp.ClientTimeout(total=3)) as resp: + 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 @@ -245,7 +267,16 @@ class get_db: self.con = pg_pool_conn.getconn() return self.con def __exit__(self, exc_type, exc_val, exc_tb): - pg_pool_conn.putconn(self.con) + 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 @@ -263,6 +294,13 @@ def upsert_user(user_id: str, username: str, full_name: str): 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): @@ -275,15 +313,20 @@ def store_message_sync(chat_id: str, user_id: str | None, username: str, VALUES (%s, %s, %s, %s, %s, %s, %s) """, (BOT_NAME, chat_id, user_id, username, role, content, embedding)) - # Trim oldest beyond BOT_HISTORY_SAVE - 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 created DESC LIMIT %s - ) - """, (BOT_NAME, chat_id, BOT_NAME, chat_id, BOT_HISTORY_SAVE)) + # 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() @@ -304,7 +347,7 @@ def get_group_context(chat_id: str, limit: int = BOT_MAX_CONTEXT) -> list[dict]: SELECT role, username, content, created FROM history WHERE bot_name = %s AND chat_id = %s - ORDER BY created DESC LIMIT %s + ORDER BY id DESC LIMIT %s """, (BOT_NAME, chat_id, limit)) rows = cur.fetchall() return list(reversed(rows)) @@ -454,14 +497,24 @@ TIMEZONE_PLACES = [ "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: - q = query.lower() - if any(kw in q for kw in TIMEZONE_KEYWORDS): + if _kw_match(query, TIMEZONE_KEYWORDS): return True # Short followup like "e na Islândia?" — place name + no other topic - if any(place in q for place in TIMEZONE_PLACES): - words = q.split() - if len(words) <= 6 and not any(kw in q for kw in WEATHER_KEYWORDS): + 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 @@ -473,8 +526,7 @@ NEWS_KEYWORDS = [ ] def is_news_query(query: str) -> bool: - q = query.lower() - return any(kw in q for kw in NEWS_KEYWORDS) + return _kw_match(query, NEWS_KEYWORDS) # WMO weather code → Portuguese description WMO_CODES = { @@ -489,46 +541,59 @@ WMO_CODES = { } def is_weather_query(query: str) -> bool: - return any(kw in query.lower() for kw in WEATHER_KEYWORDS) + return _kw_match(query, WEATHER_KEYWORDS) -async def weather_search(query: str) -> str: +# 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ã", "agora", "atual", "atualmente", + + [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", "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) - location = " ".join(location.split()).strip() + # 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 - geo_url = "https://nominatim.openstreetmap.org/search" - geo_params = {"q": location, "format": "json", "limit": 1} - headers = {"User-Agent": "ManelBot/1.0"} + # 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() + 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: - return f"Não encontrei a localização '{location}'." + if not geo: + return f"Não encontrei a localização '{location}'." - lat = float(geo[0]["lat"]) - lon = float(geo[0]["lon"]) - name = geo[0]["display_name"].split(",")[0] + 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" @@ -570,12 +635,14 @@ async def weather_search(query: str) -> str: probs = data["hourly"]["precipitation_probability"] wcodes = data["hourly"]["weather_code"] - # Use UTC offset from Open-Meteo response to get correct local hour + # 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_hour = (datetime.datetime.now(datetime.timezone.utc) + - datetime.timedelta(seconds=utc_offset)).hour + 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 int(h[11:13]) > now_local_hour][:4] + if datetime.datetime.fromisoformat(h) > now_local][:4] if upcoming: result += "\n\n📅 Próximas horas:" @@ -667,7 +734,14 @@ async def web_search(query: str) -> str: # TRIGGER CHECK # ══════════════════════════════════════════════ def is_mentioned(text: str) -> bool: - return bool(re.search(re.escape(BOT_TRIGGER), text, re.IGNORECASE)) + 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: - memories = get_memories(user_id) - history = get_history_for_llm(chat_id) + # 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) - messages = ( - [{"role": "system", "content": system}] - + history - + [{"role": "user", "content": f"[{user_first_name}] {user_message}"}] - ) + # 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): @@ -760,8 +834,9 @@ async def chat_with_groq(chat_id: str, user_id: str, user_first_name: str, user_ 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 = llm_client.chat.completions.create( + response = await llm_create( model=LLM_MODEL, messages=messages, tools=TOOLS, @@ -775,18 +850,31 @@ async def chat_with_groq(chat_id: str, user_id: str, user_first_name: str, user_ if not tool_calls: reply = msg.content or "" - # If forced tool_choice but model returned nothing, call weather directly - if not reply.strip() or reply.strip() == "…": + # 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 = llm_client.chat.completions.create( + r2 = await llm_create( model=LLM_MODEL, messages=messages, max_tokens=1024, temperature=0.7 ) return r2.choices[0].message.content or "…" - return reply 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=1024, temperature=0.7, + ) + m2 = r2.choices[0].message + if not (m2.tool_calls or []): + return m2.content or "…" + msg, tool_calls = m2, m2.tool_calls + else: + return reply or "…" messages.append({ "role": "assistant", @@ -799,54 +887,67 @@ async def chat_with_groq(chat_id: str, user_id: str, user_first_name: str, user_ }) for tc in tool_calls: - fn = tc.function.name - args = json.loads(tc.function.arguments) - log.info(f"Tool: {fn}({args})") + fn = tc.function.name + # 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})") - if fn == "web_search": - result = await web_search(args["query"]) + if fn == "web_search": + result = await web_search(args["query"]) - elif fn == "search_memory": - query = args["query"] - found = [] + 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 < 0.3: # skip very 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']}" + # ── 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: - log.warning("search_memory: no embedding — skipping vector search") + result = "Unknown tool." - # ── 2. Always include saved long-term memories ───────── - for mem in 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": - target_uid = args.get("user_id", user_id) - 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}" messages.append({"role": "tool", "content": result, "tool_call_id": tc.id}) - response2 = llm_client.chat.completions.create( + response2 = await llm_create( model=LLM_MODEL, messages=messages, max_tokens=1024, @@ -859,7 +960,12 @@ async def chat_with_groq(chat_id: str, user_id: str, user_first_name: str, user_ # TELEGRAM HANDLERS # ══════════════════════════════════════════════ async def cmd_start(update: Update, ctx: ContextTypes.DEFAULT_TYPE): - upsert_user(str(update.effective_user.id), update.effective_user.username or "", update.effective_user.full_name) + 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" @@ -881,7 +987,7 @@ async def cmd_help(update: Update, ctx: ContextTypes.DEFAULT_TYPE): async def cmd_memories(update: Update, ctx: ContextTypes.DEFAULT_TYPE): uid = str(update.effective_user.id) - rows = get_all_memories_indexed(uid) + 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 @@ -895,10 +1001,10 @@ async def cmd_forget(update: Update, ctx: ContextTypes.DEFAULT_TYPE): 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 = get_all_memories_indexed(uid) + rows = await asyncio.to_thread(get_all_memories_indexed, uid) idx = int(args[0]) - 1 if 0 <= idx < len(rows): - delete_memory_by_id(rows[idx][0]) + 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.") @@ -908,17 +1014,17 @@ 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 - clear_memories(ctx.args[0]) + 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): - clear_all_memories() + 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): - clear_history(str(update.effective_chat.id)) + 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.)") @@ -932,15 +1038,20 @@ async def handle_message(update: Update, ctx: ContextTypes.DEFAULT_TYPE): text = update.message.text fname = user.first_name or "utilizador" - upsert_user(uid, user.username or "", user.full_name or fname) - - # Always store + embed every message, even when bot stays silent - await store_message(chat_id, uid, fname, "user", text) + await asyncio.to_thread(upsert_user, uid, user.username or "", user.full_name or fname) if not is_mentioned(text): - log.debug(f"[{BOT_NAME}] Stored silently: {text[:60]}") + # 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 @@ -952,7 +1063,7 @@ async def handle_message(update: Update, ctx: ContextTypes.DEFAULT_TYPE): error_reply = True if not error_reply: - await store_message(chat_id, None, BOT_NAME, "assistant", 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(): diff --git a/gitignore.txt b/gitignore.txt new file mode 100644 index 0000000..644baea --- /dev/null +++ b/gitignore.txt @@ -0,0 +1,7 @@ +.env_credentials +new_env_c +*.db +*.log +__pycache__/ +*.pyc +.env.local diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..7471f34 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,22 @@ +[tool.poetry] +name = "tabernas-bot" +version = "0.1.0" +description = "Telegram bot. Tabernas. Maluco e irreverente" +authors = ["Raude"] +license = "Beer ware" +readme = "README.md" +package-mode = false + +[tool.poetry.dependencies] +python = "^3.12" +python-telegram-bot = "21.6" +groq = "^1.1.0" +aiohttp = "^3.9.0" +psycopg2-binary = "^2.9.9" +python-dotenv = "^1.0.0" +pgvector = "^0.4.2" +openai = "^2.29.0" + +[build-system] +requires = ["poetry-core"] +build-backend = "poetry.core.masonry.api"