import os
import uuid
import time
import shutil
import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI, UploadFile, File, Form, HTTPException
from fastapi.responses import FileResponse, HTMLResponse
# Configuration
UPLOAD_DIR = "temp_storage"
PORT = 7860
os.makedirs(UPLOAD_DIR, exist_ok=True)
# Background task to clean up expired files
async def cleanup_task():
while True:
now = time.time()
if os.path.exists(UPLOAD_DIR):
for subdir in os.listdir(UPLOAD_DIR):
path = os.path.join(UPLOAD_DIR, subdir)
if os.path.isdir(path):
meta_path = os.path.join(path, "expiry.txt")
if os.path.exists(meta_path):
with open(meta_path, "r") as f:
expiry_time = float(f.read())
if now > expiry_time:
shutil.rmtree(path)
print(f"[*] Expired and deleted: {subdir}")
await asyncio.sleep(60)
# Lifespan handler (Replaces on_event)
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup logic
loop = asyncio.get_event_loop()
task = loop.create_task(cleanup_task())
yield
# Shutdown logic
task.cancel()
app = FastAPI(lifespan=lifespan)
@app.post("/upload")
async def upload_file(
file: UploadFile = File(...),
expiry_hours: float = Form(2.0),
base_url: str = Form(...)
):
# Enforce constraints: max 24 hours
expiry_hours = min(max(expiry_hours, 0.1), 24.0)
file_id = str(uuid.uuid4())
save_path = os.path.join(UPLOAD_DIR, file_id)
os.makedirs(save_path, exist_ok=True)
file_location = os.path.join(save_path, file.filename)
with open(file_location, "wb+") as file_object:
shutil.copyfileobj(file.file, file_object)
# Store expiry timestamp
expiry_timestamp = time.time() + (expiry_hours * 3600)
with open(os.path.join(save_path, "expiry.txt"), "w") as f:
f.write(str(expiry_timestamp))
# Generate direct link (TinyURL removed)
direct_link = f"{base_url.rstrip('/')}/download/{file_id}/{file.filename}"
return {
"status": "success",
"link": direct_link,
"expires_in_hours": expiry_hours
}
@app.get("/download/{uid}/{filename}")
async def download_file(uid: str, filename: str):
file_path = os.path.join(UPLOAD_DIR, uid, filename)
if os.path.exists(file_path):
return FileResponse(file_path)
raise HTTPException(status_code=404, detail="File not found or expired.")
@app.get("/", response_class=HTMLResponse)
async def main_page():
return """
File Tunnel
"""
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860)