forked from crony/UpFast
114 lines
4.3 KiB
Python
114 lines
4.3 KiB
Python
import shutil, magic, re
|
|
from typing import Annotated, Union
|
|
from os import listdir, remove
|
|
from os.path import abspath, dirname, exists
|
|
from fastapi import FastAPI, Request, UploadFile, File, Header
|
|
from fastapi.responses import HTMLResponse, PlainTextResponse, RedirectResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
|
|
|
|
# file class
|
|
class _File:
|
|
def __init__(self, name: str, fileType: str, content=None):
|
|
self.name = name
|
|
self.fileType = fileType
|
|
self.content = content if content is not None else ''
|
|
|
|
|
|
# File list generator with filetype assignemt and content reading for previews
|
|
def file_list_generator(path: str, file_list: list[str]):
|
|
files = []
|
|
for file in file_list:
|
|
file_type = magic.from_file(f"{path}{file}", mime=True)
|
|
if re.search("^image/.*", file_type):
|
|
_file = _File(file, "image")
|
|
elif re.search("^video/.*", file_type):
|
|
_file = _File(file, "video")
|
|
elif re.search("^text/.*", file_type):
|
|
with open(f"{path}{file}") as f:
|
|
content = f.read()
|
|
_file = _File(file, "text", content)
|
|
elif file_type == 'application/json':
|
|
with open(f"{path}{file}") as f:
|
|
content = f.read()
|
|
_file = _File(file, "text", content)
|
|
else:
|
|
_file = _File(file, "else")
|
|
files.append(_file)
|
|
return files
|
|
|
|
|
|
# Load jinja2 for templating
|
|
templates = Jinja2Templates(directory="templates")
|
|
|
|
|
|
# Create fastapi template
|
|
app = FastAPI()
|
|
|
|
|
|
# Mount all uploaded files at the /files path
|
|
app.mount("/files", StaticFiles(directory="upload"), name='upload')
|
|
|
|
|
|
# Mount static files like css
|
|
app.mount("/static", StaticFiles(directory="static"), name='static')
|
|
|
|
|
|
# show the homepage when just getting the website
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def index(request: Request, user_agent: Annotated[Union[str, None], Header()] = None):
|
|
if re.search("^curl/.*", str(user_agent)):
|
|
return PlainTextResponse("It fucking works!\n")
|
|
else:
|
|
context = {'request': request, "user_agent": user_agent}
|
|
return templates.TemplateResponse("index.html", context)
|
|
|
|
|
|
# get the file user want's to upload and save it
|
|
@app.post("/")
|
|
async def upload( request: Request, file: UploadFile = File(...), user_agent: Annotated[Union[str, None], Header()] = None):
|
|
file_location = f"upload/{file.filename}"
|
|
if exists(file_location):
|
|
context = f'''File: "{request.url._url}files/{file.filename}" already exists on the server!
|
|
Please rename the file or delete the one on the server\n'''
|
|
return PlainTextResponse(context)
|
|
with open(file_location, "wb+") as file_object:
|
|
shutil.copyfileobj(file.file, file_object)
|
|
if re.search("^curl/.*", str(user_agent)):
|
|
return PlainTextResponse(f"{request.url._url}files/{file.filename}\n")
|
|
else:
|
|
return {"filename": file.filename, "path": f"{request.url._url}files/{file.filename}"}
|
|
|
|
|
|
# show the files page with the list of all files or if run with curl list all files in url format
|
|
@app.get("/files")
|
|
async def files( request: Request, user_agent: Annotated[Union[str, None], Header()] = None):
|
|
path = f"{dirname(abspath(__file__))}/upload/"
|
|
file_list = listdir(path)
|
|
files = file_list_generator(path, file_list)
|
|
context = { 'request': request, "files": files}
|
|
if re.search("^curl/.*", str(user_agent)):
|
|
context = ""
|
|
for file in files:
|
|
context += f"{request.url._url}/{file.name}\n"
|
|
return PlainTextResponse(f"{context}")
|
|
else:
|
|
return templates.TemplateResponse("files.html", context)
|
|
|
|
|
|
# delete specific file when getting this request
|
|
@app.get("/delete/{file}")
|
|
async def delete(request: Request, file: str, user_agent: Annotated[Union[str, None], Header()] = None):
|
|
file_path = f"{dirname(abspath(__file__))}/upload/{file}"
|
|
if exists(file_path):
|
|
remove(file_path)
|
|
if re.search("^curl/.*", str(user_agent)):
|
|
return PlainTextResponse(f"file {file} deleted from the server\n")
|
|
else:
|
|
return RedirectResponse(request.url_for('files'))
|
|
if re.search("^curl/.*", str(user_agent)):
|
|
return PlainTextResponse(f"file {file} doesn't exist on the server\n")
|
|
else:
|
|
return RedirectResponse(request.url_for('files'))
|