Telegram Bot
This commit is contained in:
Justin Oros
+163
@@ -0,0 +1,163 @@
|
||||
import os
|
||||
import sys
|
||||
import asyncio
|
||||
import logging
|
||||
import random
|
||||
from pathlib import Path
|
||||
from dotenv import load_dotenv
|
||||
from telegram.error import NetworkError
|
||||
|
||||
ENV_FILE = Path(".env")
|
||||
PERSONALITY_FILE = Path("personality.txt")
|
||||
LOG_FILE = Path("telegram-bot.log")
|
||||
WORDS_PER_MINUTE = 30
|
||||
CHARS_PER_SECOND = (WORDS_PER_MINUTE * 5) / 60
|
||||
MAX_HISTORY = 20
|
||||
READ_DELAY_MIN = 1
|
||||
READ_DELAY_MAX = 10
|
||||
TYPING_ACTION_INTERVAL = 4
|
||||
|
||||
logging.basicConfig(level=logging.WARNING)
|
||||
for noisy in ("httpx", "httpcore", "telegram", "apscheduler"):
|
||||
logging.getLogger(noisy).setLevel(logging.ERROR)
|
||||
|
||||
logger = logging.getLogger("telegramGPT")
|
||||
logger.setLevel(logging.INFO)
|
||||
logger.propagate = False
|
||||
|
||||
_formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S")
|
||||
_file_handler = logging.FileHandler(LOG_FILE)
|
||||
_file_handler.setFormatter(_formatter)
|
||||
_console_handler = logging.StreamHandler(sys.stdout)
|
||||
_console_handler.setFormatter(_formatter)
|
||||
logger.addHandler(_file_handler)
|
||||
logger.addHandler(_console_handler)
|
||||
|
||||
def create_env_file():
|
||||
print("\n🚀 Welcome to TelegramGPT! First-time setup...\n")
|
||||
openai_key = input("Enter your OpenAI API Key: ").strip()
|
||||
telegram_token = input("Enter your Telegram Bot Token: ").strip()
|
||||
|
||||
with open(ENV_FILE, "w") as f:
|
||||
f.write(f"OPENAI_API_KEY={openai_key}\n")
|
||||
f.write(f"TELEGRAM_BOT_TOKEN={telegram_token}\n")
|
||||
|
||||
logger.info("Created .env file with user credentials")
|
||||
print("\n✅ .env file created and saved!\n")
|
||||
|
||||
def load_personality():
|
||||
if PERSONALITY_FILE.exists():
|
||||
return PERSONALITY_FILE.read_text().strip()
|
||||
default = (
|
||||
"You are a chill, helpful friend responding via text message. "
|
||||
"Keep ALL responses very short — 1 to 3 sentences max. "
|
||||
"No long explanations unless explicitly asked. "
|
||||
"Be casual, friendly, and to the point. "
|
||||
"Write like you're texting a friend, not writing an essay."
|
||||
)
|
||||
PERSONALITY_FILE.write_text(default)
|
||||
logger.info("personality.txt not found — created with default personality")
|
||||
return default
|
||||
|
||||
def typing_delay(text):
|
||||
return max(1.0, len(text) / CHARS_PER_SECOND)
|
||||
|
||||
def main():
|
||||
if not ENV_FILE.exists():
|
||||
create_env_file()
|
||||
|
||||
load_dotenv()
|
||||
|
||||
openai_api_key = os.getenv("OPENAI_API_KEY")
|
||||
telegram_token = os.getenv("TELEGRAM_BOT_TOKEN")
|
||||
|
||||
if not all([openai_api_key, telegram_token]):
|
||||
logger.error("Missing values in .env — delete it and re-run to set up again.")
|
||||
sys.exit(1)
|
||||
|
||||
try:
|
||||
from openai import OpenAI
|
||||
from telegram import Update
|
||||
from telegram.ext import ApplicationBuilder, MessageHandler, filters, ContextTypes
|
||||
except ImportError as e:
|
||||
logger.error(f"Missing dependency: {e}")
|
||||
print("Run: pip install openai python-telegram-bot python-dotenv")
|
||||
sys.exit(1)
|
||||
|
||||
openai_client = OpenAI(api_key=openai_api_key)
|
||||
personality = load_personality()
|
||||
conversation_histories = {}
|
||||
|
||||
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||||
chat_id = update.effective_chat.id
|
||||
username = update.effective_user.username or update.effective_user.first_name
|
||||
user_text = update.message.text
|
||||
|
||||
if chat_id not in conversation_histories:
|
||||
conversation_histories[chat_id] = []
|
||||
|
||||
history = conversation_histories[chat_id]
|
||||
history.append({"role": "user", "content": user_text})
|
||||
|
||||
logger.info(f"[{username}] {user_text}")
|
||||
|
||||
trimmed_history = history[-MAX_HISTORY:]
|
||||
|
||||
mimic_instruction = (
|
||||
f"\n\nThe person just sent you this message: {repr(user_text)}\n"
|
||||
"You must mirror their exact texting style in your reply. Follow these rules strictly:\n"
|
||||
"- Capitalization: If their message starts with a capital letter, start yours with one too. If they use all lowercase, you use all lowercase.\n"
|
||||
"- Punctuation: If they use no punctuation, use none. If they use periods, commas, question marks, use them the same way.\n"
|
||||
"- Grammar: If they write in proper grammatical sentences, you do too. If they write loosely or skip words, match that.\n"
|
||||
"- Shorthand: If they use shorthand like 'u' for 'you', 'r' for 'are', 'ur' for 'your', 'tbh', 'ngl', 'imo', 'idk', 'lmk', 'rn', 'omg', 'wtf', etc — use the same shorthand naturally in your reply.\n"
|
||||
"- Length: Match the approximate length and energy of their message.\n"
|
||||
"Do not explain or comment on their style. Just mirror it naturally."
|
||||
)
|
||||
|
||||
response = openai_client.chat.completions.create(
|
||||
model="gpt-4o",
|
||||
messages=[{"role": "system", "content": personality + mimic_instruction}] + trimmed_history
|
||||
)
|
||||
|
||||
reply = response.choices[0].message.content.strip()
|
||||
history.append({"role": "assistant", "content": reply})
|
||||
|
||||
logger.info(f"[GPT -> {username}] {reply}")
|
||||
|
||||
read_delay = random.uniform(READ_DELAY_MIN, READ_DELAY_MAX)
|
||||
await asyncio.sleep(read_delay)
|
||||
|
||||
typing_duration = typing_delay(reply)
|
||||
elapsed = 0
|
||||
while elapsed < typing_duration:
|
||||
await context.bot.send_chat_action(chat_id=chat_id, action="typing")
|
||||
chunk = min(TYPING_ACTION_INTERVAL, typing_duration - elapsed)
|
||||
await asyncio.sleep(chunk)
|
||||
elapsed += chunk
|
||||
|
||||
await update.message.reply_text(reply)
|
||||
|
||||
async def error_handler(update, context):
|
||||
err = context.error
|
||||
if isinstance(err, NetworkError):
|
||||
if "Bad Gateway" in str(err) or "nodename nor servname" in str(err):
|
||||
logger.error(f"❌ Network error: {err}")
|
||||
else:
|
||||
logger.error(f"❌ Network error: {err}")
|
||||
else:
|
||||
logger.error(f"❌ Unexpected error: {err}")
|
||||
|
||||
app = ApplicationBuilder().token(telegram_token).build()
|
||||
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
|
||||
app.add_error_handler(error_handler)
|
||||
|
||||
logger.info("TelegramGPT started")
|
||||
print("✅ TelegramGPT is running! Open Telegram and message your bot. (Ctrl+C to stop)\n")
|
||||
try:
|
||||
app.run_polling()
|
||||
except NetworkError:
|
||||
logger.error("❌ Could not connect to api.telegram.org — check your internet connection and try again.")
|
||||
sys.exit(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user