File size: 5,236 Bytes
5e0e3eb |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 |
import os
import logging
import asyncio
import nest_asyncio
# Import the Hugging Face Inference Client and login function
from huggingface_hub import InferenceClient, login
import langid
# Import Telegram bot components
from telegram import Update, Bot
from telegram.ext import Application, CommandHandler, MessageHandler, filters, CallbackContext
# -------------------------
# Configure logging
# -------------------------
logging.basicConfig(
format="%(asctime)s - %(levelname)s - %(message)s", level=logging.INFO
)
logger = logging.getLogger(__name__)
# -------------------------
# Environment variables and login to Hugging Face
# -------------------------
HF_HUB_TOKEN = os.getenv("HUGGINGFACEHUB_API_TOKEN")
if not HF_HUB_TOKEN:
raise ValueError("Missing Hugging Face API token. Please set HUGGINGFACEHUB_API_TOKEN.")
login(token=HF_HUB_TOKEN)
client = InferenceClient(api_key=HF_HUB_TOKEN)
# Get Telegram Bot Token
TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
if not TOKEN:
raise ValueError("Missing Telegram Bot Token. Please set TELEGRAM_BOT_TOKEN.")
# Get Webhook Domain (e.g., your-space.hf.space)
WEBHOOK_DOMAIN = os.getenv("WEBHOOK_DOMAIN")
if not WEBHOOK_DOMAIN:
raise ValueError("Missing Webhook Domain. Please set WEBHOOK_DOMAIN.")
# Construct the full webhook URL.
# We use the Telegram bot token as a unique path.
WEBHOOK_URL = f"https://{WEBHOOK_DOMAIN}/{TOKEN}"
# -------------------------
# Function to detect language
# -------------------------
def detect_language(user_input: str):
try:
lang, _ = langid.classify(user_input)
if lang == "he":
return "hebrew"
elif lang == "en":
return "english"
else:
return "unsupported"
except Exception as e:
logger.error(f"Language detection error: {e}")
return "unsupported"
# -------------------------
# Function to generate response using Hugging Face Chat models
# -------------------------
def generate_response(text: str) -> str:
language = detect_language(text)
if language == "hebrew":
# For Hebrew, prompt the model accordingly.
content = "转注谞讛 讘拽爪专讛 讗讘诇 转砖转祝 讗转 转讛诇讬讱 拽讘诇转 讛讛讞诇讟讜转 砖诇讱, " + text
model = "microsoft/Phi-3.5-mini-instruct"
elif language == "english":
# For English, use a different prompt.
content = "keep it short but tell your decision making process, " + text
model = "mistralai/Mistral-Nemo-Instruct-2407"
else:
return "Sorry, I only support Hebrew and English."
messages = [{"role": "user", "content": content}]
try:
# Call the chat completion API
completion = client.chat.completions.create(
model=model,
messages=messages,
max_tokens=2048,
temperature=0.5,
top_p=0.7
)
return completion.choices[0].message.content
except Exception as e:
logger.error(f"Error generating response: {e}")
return "Error: Could not generate response."
# -------------------------
# Telegram Bot Handlers
# -------------------------
async def start(update: Update, context: CallbackContext):
# Respond to the /start command.
await update.message.reply_text("Hello! Tell me your decision-making issue, and I'll try to help.")
logger.info("Start command received.")
async def handle_message(update: Update, context: CallbackContext):
user_text = update.message.text
logger.info(f"User message: {user_text}")
# Generate a response using our Hugging Face chat logic.
response_text = generate_response(user_text)
logger.info(f"Generated response: {response_text}")
await update.message.reply_text(response_text)
# -------------------------
# Main function to run the Telegram Bot using Webhook mode
# -------------------------
async def main():
# Build the Application with the Telegram Bot Token.
application = Application.builder().token(TOKEN).build()
# Add command and message handlers.
application.add_handler(CommandHandler("start", start))
application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
logger.info("Starting bot in webhook mode...")
print("Starting bot in webhook mode...")
# Run the bot in webhook mode.
# The bot will listen on all interfaces (0.0.0.0) at port 7860.
# Telegram will send updates to WEBHOOK_URL.
await application.run_webhook(
listen="0.0.0.0", # Listen on all interfaces.
port=7860, # Port to listen on.
url_path=TOKEN, # Use the bot token as the URL path.
webhook_url=WEBHOOK_URL # Full webhook URL.
)
# -------------------------
# Run the main function without closing a running event loop
# -------------------------
if __name__ == "__main__":
# Apply nest_asyncio to allow nested event loops (needed in some HF Space environments).
nest_asyncio.apply()
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(main())
except Exception as e:
logger.error(f"Error in main loop: {e}")
print(f"Error in main loop: {e}")
|