File size: 3,459 Bytes
52dc5f7 |
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 |
import os
import asyncio
from telethon.sync import TelegramClient
from telethon.sessions import StringSession
from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import FileResponse
import uvicorn
import socks
# Configuration from environment variables
SESSION_STRING = os.environ.get('SESSION_STRING')
API_ID = os.environ.get('API_ID') # You need to get your API_ID from my.telegram.org
API_HASH = os.environ.get('API_HASH') # You need to get your API_HASH from my.telegram.org
PROXY_IP = os.environ.get('PROXY_IP')
PROXY_PORT = os.environ.get('PROXY_PORT')
PROXY_TYPE = os.environ.get('PROXY_TYPE', 'socks5').lower() # Default to socks5
# The Telegram channel you want to upload files to
TARGET_CHANNEL = 'https://t.me/TestNAI01'
# Directory to store uploaded files temporarily
UPLOAD_DIR = "uploads"
os.makedirs(UPLOAD_DIR, exist_ok=True)
app = FastAPI()
def get_proxy():
"""Returns proxy settings if configured."""
if PROXY_IP and PROXY_PORT:
proxy_type_map = {
'socks4': socks.SOCKS4,
'socks5': socks.SOCKS5,
'http': socks.HTTP,
}
return (proxy_type_map.get(PROXY_TYPE, socks.SOCKS5), PROXY_IP, int(PROXY_PORT))
return None
async def upload_to_telegram(file_path: str):
"""
Uploads a file to the specified Telegram channel using a session string.
"""
if not SESSION_STRING or not API_ID or not API_HASH:
raise ValueError("SESSION_STRING, API_ID, and API_HASH must be set in environment variables.")
proxy = get_proxy()
async with TelegramClient(StringSession(SESSION_STRING), API_ID, API_HASH, proxy=proxy) as client:
try:
await client.send_file(TARGET_CHANNEL, file_path)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to upload to Telegram: {e}")
@app.post("/v1/upload")
async def upload_file(file: UploadFile = File(...)):
"""
Receives a file and uploads it to the Telegram channel.
"""
file_location = os.path.join(UPLOAD_DIR, file.filename)
with open(file_location, "wb+") as file_object:
file_object.write(file.file.read())
try:
await upload_to_telegram(file_location)
return {"message": f"File '{file.filename}' uploaded successfully to Telegram."}
finally:
# Clean up the local file after uploading
if os.path.exists(file_location):
os.remove(file_location)
@app.get("/v1/download/{filename}")
async def download_file(filename: str):
"""
Serves a file for download.
Note: This endpoint serves files from the local 'uploads' directory.
For a production system, you would need a more robust way to retrieve files,
possibly by storing a mapping of filenames to Telegram message IDs.
"""
file_path = os.path.join(UPLOAD_DIR, filename)
if os.path.exists(file_path):
return FileResponse(file_path)
raise HTTPException(status_code=404, detail="File not found. This is a simple server; for production, a proper file retrieval from Telegram would be needed.")
if __name__ == "__main__":
# This block is for local testing without Docker
# To run with Docker, the CMD in the Dockerfile will be used
if not all([SESSION_STRING, API_ID, API_HASH]):
print("Error: SESSION_STRING, API_ID, and API_HASH environment variables must be set.")
else:
uvicorn.run(app, host="0.0.0.0", port=8000) |