Spaces:
Paused
Paused
File size: 4,067 Bytes
b2f507e 143a8e7 45bc669 143a8e7 45bc669 143a8e7 b2f507e 45bc669 143a8e7 45bc669 8174193 45bc669 b2f507e 143a8e7 45bc669 143a8e7 b2f507e 143a8e7 45bc669 143a8e7 45bc669 143a8e7 45bc669 143a8e7 45bc669 7c74069 45bc669 b2f507e 8174193 b2f507e 8174193 b2f507e 8174193 b2f507e 8174193 b2f507e 45bc669 b2f507e 143a8e7 45bc669 143a8e7 |
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 |
import socket
import httpx
from fastapi import FastAPI, HTTPException
from telegram import Bot
from telegram.error import TelegramError
from pydantic import BaseModel
from datetime import datetime, timezone
import os
from dotenv import load_dotenv
import logging
app = FastAPI()
load_dotenv()
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID")
# Hardcode Telegram API IP (replace with the IP you get from nslookup)
TELEGRAM_API_IP = "149.154.167.220" # Example IP, update with actual IP
# Set up logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
class MessageData(BaseModel):
name: str
email: str
message: str
@app.get("/health")
async def health_check():
return {
"status": "healthy",
"message": "Telegram backend is running",
"timestamp": datetime.now(timezone.utc).isoformat() + " UTC"
}
@app.get("/test-dns")
async def test_dns():
try:
ip = socket.gethostbyname("api.telegram.org")
logger.info(f"DNS resolved: api.telegram.org -> {ip}")
return {"status": "success", "ip": ip}
except socket.gaierror as e:
logger.error(f"DNS resolution failed: {str(e)}")
return {"status": "error", "details": str(e)}
@app.get("/test-network")
async def test_network():
try:
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.get("https://api.ipify.org?format=json")
logger.info(f"Network test successful: {response.json()}")
return {"status": "success", "data": response.json()}
except httpx.HTTPError as e:
logger.error(f"Network test failed: {str(e)}")
return {"status": "error", "details": str(e)}
@app.post("/send-message")
async def send_message(data: MessageData):
try:
name = data.name.strip()
email = data.email.strip()
message = data.message.strip()
if not all([name, email, message]):
raise HTTPException(status_code=400, detail="All fields must not be empty")
if len(message) > 4096:
raise HTTPException(status_code=400, detail="Message exceeds 4096 character limit")
if '@' not in email or '.' not in email.split('@')[1]:
raise HTTPException(status_code=400, detail="Invalid email format")
timestamp = datetime.now(timezone.utc).isoformat().split('.')[0] + " UTC"
telegram_message = (
f"*New Message*\n"
f"_Received on: {timestamp}_\n"
f"---\n"
f"*Sender Details:*\n"
f"- *Name:* {name}\n"
f"- *Email:* {email}\n"
f"---\n"
f"*Message:*\n"
f"{message}"
)
logger.debug(f"Sending message to chat_id: {TELEGRAM_CHAT_ID}")
# Use custom httpx client with hardcoded IP
async with httpx.AsyncClient(timeout=30.0) as client:
bot = Bot(
token=TELEGRAM_BOT_TOKEN,
base_url=f"https://{TELEGRAM_API_IP}",
client=client
)
try:
# Set Host header for HTTPS
await bot.send_message(
chat_id=TELEGRAM_CHAT_ID,
text=telegram_message,
parse_mode="Markdown",
disable_notification=True,
request_kwargs={"headers": {"Host": "api.telegram.org"}}
)
logger.info("Message sent successfully")
return {"success": True, "message": "Message sent to Telegram"}
except TelegramError as e:
logger.error(f"TelegramError: {str(e)}")
raise HTTPException(status_code=500, detail=f"Failed to send message: {str(e)}")
except Exception as e:
logger.error(f"Unexpected error: {str(e)}")
raise HTTPException(status_code=500, detail=f"Server error: {str(e)}")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860) |