File size: 3,011 Bytes
3bc4fcb |
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 |
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import httpx
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes
import os
app = FastAPI()
# Replace this with your Hugging Face Space URL
HUGGING_FACE_SPACE_URL = "https://demaking-decision-helper-bot.hf.space"
# Get Telegram bot token from environment variables
TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
if not TOKEN:
raise ValueError("Missing Telegram Bot Token. Please set TELEGRAM_BOT_TOKEN environment variable.")
async def call_hugging_face_space(input_data: str):
async with httpx.AsyncClient(timeout=45.0) as client:
try:
response = await client.post(HUGGING_FACE_SPACE_URL, json={"input": input_data})
response.raise_for_status() # Raise an error for bad responses
return response.json()
except httpx.HTTPStatusError as e:
print(f"HTTP error occurred: {e}")
return {"error": str(e)}
except httpx.ConnectError as e:
print(f"Connection error: {e}")
return {"error": "Could not connect to the Hugging Face Space"}
except Exception as e:
print(f"An error occurred: {e}")
return {"error": str(e)}
@app.post("/webhook/{token}")
async def webhook(token: str, request: Request):
if token != TOKEN:
return JSONResponse(status_code=403, content={"message": "Forbidden"})
update = Update.de_json(await request.json(), None)
message_text = update.message.text
result = await call_hugging_face_space(message_text)
return JSONResponse(content=result)
def start_telegram_bot():
application = ApplicationBuilder().token(TOKEN).build()
# Set up a command handler
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text("Hello! Tell me your decision-making issue, and I'll try to help.")
print("Start command received.")
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_text = update.message.text
print(f"User message: {user_text}")
# Send the user text to the FastAPI server and get the response.
result = await call_hugging_face_space(user_text)
response_text = result.get("response", "Error generating response.")
print(f"API Response: {response_text}")
await update.message.reply_text(response_text)
application.add_handler(CommandHandler("start", start))
application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
# Start the bot
application.run_polling()
if __name__ == "__main__":
import threading
# Start the Telegram bot in a separate thread
threading.Thread(target=start_telegram_bot).start()
# Start the FastAPI app
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860)
|