File size: 4,899 Bytes
bcd1672 9111a31 bcd1672 9803b4b bcd1672 6b8f3fb e7c5290 e2d29dc bcd1672 e2d29dc bcd1672 75a0cba bcd1672 f552a64 3c3481f bcd1672 9111a31 9803b4b 9111a31 c5edd30 3c3481f 9111a31 75a0cba 9803b4b bcd1672 c5edd30 0093abb f552a64 bcd1672 0093abb e5894e8 1f84520 bcd1672 |
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 157 158 159 160 161 162 163 164 165 166 167 168 169 170 |
import logging
import os
from fastapi import FastAPI, Request
from contextlib import asynccontextmanager
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, CallbackContext
import asyncio
from transformers import pipeline
from langdetect import detect
from huggingface_hub import login
import httpx
# Global variables
TOKEN = os.getenv("TELEGRAM_BOT_TOKEN") # Telegram Token
# WEBHOOK_URL = "https://demaking-decision-helper-bot.hf.space/"
# PORT = int(os.getenv("PORT", 7860)) # HF Spaces listens on port 7860 by default
HF_HUB_TOKEN = os.getenv("HUGGINGFACEHUB_API_TOKEN")
# Verify Hugging Face token
if not HF_HUB_TOKEN:
raise ValueError("Missing Hugging Face API token. Please set HUGGINGFACEHUB_API_TOKEN in environment variables.")
login(token=HF_HUB_TOKEN)
# Load Hebrew text generation model
hebrew_generator = pipeline("text-generation", model="Norod78/hebrew-gpt_neo-small")
# Load English text generation model
english_generator = pipeline("text-generation", model="distilgpt2")
# Function to detect language
def detect_language(user_input):
try:
lang = detect(user_input)
if lang == "he":
return "hebrew"
elif lang == "en":
return "english"
else:
return "unsupported"
except:
return "unsupported"
# Function to generate a response based on detected language
def generate_response(text):
language = detect_language(text)
if language == "hebrew":
response = hebrew_generator(text, max_length=100)[0]["generated_text"]
elif language == "english":
response = english_generator(text, max_length=100)[0]["generated_text"]
else:
response = "Sorry, I only support Hebrew and English."
return response
# Create FastAPI app
app = FastAPI()
# Health check endpoint
@app.get("/")
async def root():
return {"message": "Decision Helper Bot is running!"}
# Initialize Telegram Bot
telegram_app = Application.builder().token(TOKEN).build()
async def init_telegram():
await telegram_app.initialize()
# Configure logging
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)
# Start command handler
async def start(update: Update, context: CallbackContext):
await update.message.reply_text("Hello! Tell me your decision-making issue in HE or EN, and I'll try to help you.")
# Message handler
async def handle_message(update: Update, context: CallbackContext):
user_text = update.message.text
response = generate_response(user_text) # Generate AI-based response
await update.message.reply_text(response)
# Add handlers to the bot
telegram_app.add_handler(CommandHandler("start", start))
telegram_app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
@asynccontextmanager
async def lifespan(app: FastAPI):
print("Starting application...")
# Check internet connection
try:
async with httpx.AsyncClient() as client:
response = await client.get("https://api.telegram.org")
print(f"Network test successful! Status Code: {response.status_code}")
except Exception as e:
print(f"Network test failed: {e}")
# Initialize Telegram
await init_telegram()
await telegram_app.start()
print("Telegram bot initialized successfully!")
yield # Wait until app closes
print("Shutting down application...")
app = FastAPI(lifespan=lifespan)
@app.get("/")
async def root():
return {"message": "Hello, FastAPI with Telegram Bot!"}
# Initialize Telegram Bot properly on startup
# @app.on_event("startup")
# async def test_network():
# try:
# async with httpx.AsyncClient() as client:
# response = await client.get("https://api.telegram.org")
# print(f"Network test successful! Status Code: {response.status_code}")
# except Exception as e:
# print(f"Network test failed: {e}")
# async def startup_event():
# await init_telegram()
# await telegram_app.start()
# print("Telegram bot initialized successfully!")
# API Route to receive updates from Telegram
@app.post("/")
async def receive_update(request: Request):
update = Update.de_json(await request.json(), telegram_app.bot)
await telegram_app.process_update(update)
return {"status": "ok"}
# """# Function to set up webhook
# async def set_webhook():
# webhook_url = "https://demaking-decision-helper-bot.hf.space/"
# await telegram_app.bot.set_webhook(webhook_url)
# logger.info(f"Webhook set to {webhook_url}")
# """
# Run the server
if __name__ == "__main__":
import uvicorn
# loop = asyncio.get_event_loop()
# loop.run_until_complete(set_webhook())
uvicorn.run(app, host="0.0.0.0", port=7860)
|