andre/betterbot
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
betterbot/main.py

254 lines
9 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""BetterBot — a Telegram bot that edits the Better Life SG website via LLM."""
from __future__ import annotations
import logging
import os
import pathlib
from openai import OpenAI
from telegram import Update
from telegram.ext import (
ApplicationBuilder,
CommandHandler,
ContextTypes,
MessageHandler,
filters,
)
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s")
log = logging.getLogger("betterbot")
# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
TG_BOT_TOKEN = os.environ["TG_BOT_TOKEN"]
OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]
OPENAI_BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1")
MODEL = os.environ.get("MODEL", "gpt-4.1")
SITE_DIR = pathlib.Path(os.environ.get("SITE_DIR", "/site"))
client = OpenAI(api_key=OPENAI_API_KEY, base_url=OPENAI_BASE_URL)
# ---------------------------------------------------------------------------
# Tools — read / write / list site files
# ---------------------------------------------------------------------------
TOOLS = [
{
"type": "function",
"function": {
"name": "list_files",
"description": "List all files in the website directory.",
"parameters": {"type": "object", "properties": {}, "required": []},
},
},
{
"type": "function",
"function": {
"name": "read_file",
"description": "Read the full contents of a website file.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Relative path inside the site directory, e.g. 'index.html' or 'images/logo.png'.",
}
},
"required": ["path"],
},
},
},
{
"type": "function",
"function": {
"name": "write_file",
"description": "Write (create or overwrite) a text file in the website directory. Use for HTML, CSS, JS files.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Relative path inside the site directory.",
},
"content": {
"type": "string",
"description": "The full file content to write.",
},
},
"required": ["path", "content"],
},
},
},
]
SYSTEM_PROMPT = """\
You are BetterBot, a helpful assistant that manages the Better Life SG website.
The website is a static HTML site using Tailwind CSS (loaded via CDN).
The site files are in the /site directory. Key files:
- index.html (home page)
- fresh-grads.html, prenatal.html, retirement.html, legacy.html (service pages)
- team.html, contact.html
- images/ folder with photos and backgrounds
When the user asks you to change something on the website:
1. First read the relevant file(s) to understand the current state
2. Make the requested changes
3. Write the updated file back
4. Confirm what you changed
The brand color is teal/green (#00b49a). The site uses Tailwind CSS classes.
Keep your responses concise and friendly. Always confirm changes after making them.
Do NOT change the overall page structure unless explicitly asked.
When writing a file, always write the COMPLETE file content — never partial.
"""
def _resolve(path: str) -> pathlib.Path:
"""Resolve a relative path inside SITE_DIR, preventing path traversal."""
resolved = (SITE_DIR / path).resolve()
if not str(resolved).startswith(str(SITE_DIR.resolve())):
raise ValueError(f"Path traversal blocked: {path}")
return resolved
def handle_tool_call(name: str, args: dict) -> str:
"""Execute a tool call and return the result as a string."""
if name == "list_files":
files = []
for p in sorted(SITE_DIR.rglob("*")):
if p.is_file():
files.append(str(p.relative_to(SITE_DIR)))
return "\n".join(files) if files else "(no files found)"
if name == "read_file":
path = _resolve(args["path"])
if not path.exists():
return f"Error: {args['path']} does not exist."
if path.suffix in (".png", ".jpg", ".jpeg", ".gif", ".webp", ".ico"):
return f"[Binary image file: {args['path']}, {path.stat().st_size} bytes]"
return path.read_text(encoding="utf-8")
if name == "write_file":
path = _resolve(args["path"])
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(args["content"], encoding="utf-8")
return f"OK — wrote {len(args['content'])} chars to {args['path']}"
return f"Unknown tool: {name}"
# ---------------------------------------------------------------------------
# Conversation state (in-memory, per-chat)
# ---------------------------------------------------------------------------
conversations: dict[int, list[dict]] = {}
def get_messages(chat_id: int) -> list[dict]:
if chat_id not in conversations:
conversations[chat_id] = [{"role": "system", "content": SYSTEM_PROMPT}]
return conversations[chat_id]
# ---------------------------------------------------------------------------
# Telegram handlers
# ---------------------------------------------------------------------------
async def cmd_start(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
await update.message.reply_text(
"Hi! I'm BetterBot 🤖\n\n"
"I manage the Better Life SG website. Just tell me what you'd like to change!\n\n"
"Examples:\n"
'• "Change the WhatsApp number to 91234567"\n'
'• "Update Hendri\'s title to Senior Consultant"\n'
'• "Add a new section about critical illness"\n\n'
"Type /reset to start a fresh conversation."
)
async def cmd_reset(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
conversations.pop(update.effective_chat.id, None)
await update.message.reply_text("Conversation reset! Send me a new request.")
async def handle_message(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
chat_id = update.effective_chat.id
user_text = update.message.text
if not user_text:
return
messages = get_messages(chat_id)
messages.append({"role": "user", "content": user_text})
# Send "typing" indicator
await update.message.chat.send_action("typing")
# Run the LLM loop (with tool calls)
max_rounds = 10
for _ in range(max_rounds):
try:
response = client.chat.completions.create(
model=MODEL,
messages=messages,
tools=TOOLS,
tool_choice="auto",
)
except Exception as e:
log.error("OpenAI API error: %s", e)
await update.message.reply_text(f"Sorry, I hit an error: {e}")
return
choice = response.choices[0]
msg = choice.message
# Append assistant message
messages.append(msg.model_dump(exclude_none=True))
if msg.tool_calls:
for tc in msg.tool_calls:
import json
args = json.loads(tc.function.arguments) if tc.function.arguments else {}
log.info("Tool call: %s(%s)", tc.function.name, list(args.keys()))
try:
result = handle_tool_call(tc.function.name, args)
except Exception as e:
result = f"Error: {e}"
messages.append(
{
"role": "tool",
"tool_call_id": tc.id,
"content": result,
}
)
# Continue the loop for the LLM to process tool results
await update.message.chat.send_action("typing")
continue
# No tool calls — we have a final text reply
if msg.content:
# Trim conversation to prevent unbounded growth
if len(messages) > 60:
messages[:] = messages[:1] + messages[-40:]
await update.message.reply_text(msg.content)
return
await update.message.reply_text("I ran out of steps — please try again with a simpler request.")
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main() -> None:
app = ApplicationBuilder().token(TG_BOT_TOKEN).build()
app.add_handler(CommandHandler("start", cmd_start))
app.add_handler(CommandHandler("reset", cmd_reset))
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
log.info("BetterBot starting (model=%s, site_dir=%s)", MODEL, SITE_DIR)
app.run_polling(drop_pending_updates=True)
if __name__ == "__main__":
main()
Reference in a new issue View git blame Copy permalink