File size: 4,558 Bytes
ead9d27 c153710 ead9d27 a54dcc3 ead9d27 a54dcc3 a5f5251 a54dcc3 ead9d27 c153710 89962dc c153710 a5f5251 c153710 ead9d27 c153710 ead9d27 03ad962 c153710 03ad962 c153710 03ad962 ead9d27 48450b3 |
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 |
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 httpx
import asyncio
from transformers import pipeline
from langdetect import detect
from huggingface_hub import login
import socket
# Global variables
TOKEN = os.getenv("TELEGRAM_BOT_TOKEN") # Telegram Token
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)
# Verify Telegram token
if not TOKEN:
raise ValueError("Missing Telegram token. Please set TELEGRAM_BOT_TOKEN in environment variables.")
try:
response = httpx.get(f"https://api.telegram.org/bot{TOKEN}/getMe")
print(f"Using TELEGRAM_TOKEN: {TOKEN[:5]}***") # Part of the token
print(response.json())
except httpx.RequestError as e:
print(f"aaRequest failed: {e}")
# Load Hebrew and English text generation models
hebrew_generator = pipeline("text-generation", model="Norod78/hebrew-gpt_neo-small")
english_generator = pipeline("text-generation", model="distilgpt2")
# Function to detect language
def detect_language(user_input):
try:
lang = detect(user_input)
return "hebrew" if lang == "he" else "english" if lang == "en" else "unsupported"
except:
return "unsupported"
# Function to generate a response
def generate_response(text):
language = detect_language(text)
if language == "hebrew":
return hebrew_generator(text, max_length=100)[0]["generated_text"]
elif language == "english":
return english_generator(text, max_length=100)[0]["generated_text"]
return "Sorry, I only support Hebrew and English."
# Create Telegram Bot
telegram_app = Application.builder().token(TOKEN).build()
# Initialize Telegram
async def init_telegram():
await telegram_app.initialize()
await telegram_app.start()
# Configure logging
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)
# Command and Message Handlers
async def start(update: Update, context: CallbackContext):
await update.message.reply_text("Hello! Tell me your decision-making issue in Hebrew or English, and I'll try to help you.")
async def handle_message(update: Update, context: CallbackContext):
user_text = update.message.text
response = generate_response(user_text)
await update.message.reply_text(response)
telegram_app.add_handler(CommandHandler("start", start))
telegram_app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
# FastAPI lifespan event
@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}")
# Start Telegram bot
asyncio.create_task(init_telegram()) # Run in background
yield # Wait until app closes
print("Shutting down application...")
# Create FastAPI app
app = FastAPI(lifespan=lifespan)
@app.get("/")
async def root():
return {"message": "Decision Helper Bot is running!"}
@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"}
# async def check_internet():
# 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}")
# asyncio.run(check_internet())
# Network test
def test_network():
try:
socket.gethostbyname("api.telegram.org")
print("Network test passed!")
except socket.gaierror:
print("Network test failed: No address associated with hostname")
test_network()
# Run the server
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8443)
# uvicorn.run(app, host="0.0.0.0", port=7860)
|