File size: 3,070 Bytes
52dc5f7 51919c9 52dc5f7 eb586a3 52dc5f7 eb586a3 52dc5f7 51919c9 52dc5f7 eb586a3 52dc5f7 51919c9 eb586a3 52dc5f7 51919c9 52dc5f7 51919c9 52dc5f7 eb586a3 52dc5f7 eb586a3 52dc5f7 51919c9 52dc5f7 51919c9 52dc5f7 eb586a3 52dc5f7 11ddc70 |
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 |
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 tempfile # Import the tempfile module
# --- Configuration from environment variables ---
# Your Telegram session string
SESSION_STRING = os.environ.get('SESSION_STRING')
# You MUST get your own API ID and HASH from my.telegram.org
API_ID = os.environ.get('API_ID')
API_HASH = os.environ.get('API_HASH')
# The Telegram channel you want to upload files to
TARGET_CHANNEL = 'https://t.me/TestNAI01'
# --- FIX: Use the system's temporary directory ---
# This directory is guaranteed to be writable and avoids permission errors.
UPLOAD_DIR = tempfile.gettempdir()
app = FastAPI()
async def upload_to_telegram(file_path: str):
"""
Uploads a file to the specified Telegram channel using a session string.
"""
if not all([SESSION_STRING, API_ID, API_HASH]):
raise ValueError("SESSION_STRING, API_ID, and API_HASH must be set as environment variables.")
# Initialize the client
async with TelegramClient(StringSession(SESSION_STRING), API_ID, API_HASH) 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, saves it to the temp directory, and uploads it to the Telegram channel.
"""
# Create a secure path in the temporary directory
file_location = os.path.join(UPLOAD_DIR, file.filename)
try:
# Save the uploaded file temporarily
with open(file_location, "wb+") as file_object:
file_object.write(file.file.read())
# Upload the file to Telegram
await upload_to_telegram(file_location)
return {"message": f"File '{file.filename}' uploaded successfully to Telegram."}
finally:
# Clean up the local file after uploading to save space
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 is for demonstration. It looks for files in the temp
directory, but files uploaded in previous requests will have been deleted.
A production system would need a proper way to retrieve files from Telegram.
"""
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. Files are only stored temporarily during upload.")
if __name__ == "__main__":
if not all([SESSION_STRING, API_ID, API_HASH]):
print("FATAL ERROR: The environment variables SESSION_STRING, API_ID, and API_HASH must be set.")
else:
uvicorn.run(app, host="0.0.0.0", port=8000) |