Spaces:

DeMaking
/

decision-helper-bot

Paused

Create app.py

bcd1672 verified almost 2 years ago

3.3 kB

import logging
import os
from fastapi import FastAPI, Request
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


# 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()


# Initialize Telegram Bot
telegram_app = Application.builder().token(TOKEN).build()


# 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))


# 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 = WEBHOOK_URL
    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=PORT)
    

	import logging
	import os
	from fastapi import FastAPI, Request
	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


	# 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()


	# Initialize Telegram Bot
	telegram_app = Application.builder().token(TOKEN).build()


	# 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))


	# 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 = WEBHOOK_URL
	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=PORT)