autoindex/autoindex.py
2023-05-24 18:15:36 +03:00

84 lines
2.5 KiB
Python

import os
import socket
from os import environ
from pathlib import Path
import logging
from sanic import Sanic, HTTPResponse
from sanic.request import Request
from sanic.response import text, html, file_stream, redirect
from sanic.exceptions import NotFound, MethodNotAllowed
from sanic_cors import CORS
from sanic_ext import Extend
from utils import get_j2env, get_sort_icon, get_sort_link, resolve_path, list_dir
DEBUG = Path('.debug').exists()
app = Sanic('autoindex', strict_slashes=True)
Extend(app)
cors = CORS(app)
app.static('/~static/', '~static/', use_content_range=True, stream_large_files=True)
j2env = get_j2env(DEBUG)
@app.get(r'/<path:.*/?>')
async def index(request: Request, path=''):
logging.warning(path)
path = path.replace('%20', ' ')
domain = request.host
query = f'?{request.query_string}' if request.query_string else ''
try:
resolved_path, resolved_query = resolve_path(domain, path)
except ValueError:
return text('GTFO', 400)
if resolved_path.is_dir():
if path and path[-1] != '/':
return redirect(f'/{path}/')
hidden = request.args.get('hidden') is not None
sort = request.args.get('sort', 'name')
if sort not in ['name', '-name', 'size', '-size', 'created', '-created']:
sort = 'name'
return html(j2env.get_template('filelist.tpl').render(
lst=list_dir(resolved_path, sort, hidden, root=not resolved_query),
get_sort_icon=get_sort_icon,
get_sort_link=get_sort_link,
sort=sort,
hidden=hidden,
path=resolved_query,
query=query,
))
elif resolved_path.is_file():
stream = await file_stream(resolved_path)
if request.method == 'HEAD':
stream.headers['Content-Length'] = resolved_path.stat().st_size
return HTTPResponse(
'', status=200, headers=stream.headers, content_type=stream.content_type
)
return stream
raise NotFound('Path was not found')
if __name__ == '__main__':
if DEBUG:
app.run(host='localhost', port=8080, debug=True, auto_reload=True)
else:
socket_address = Path('/tmp/drop.sock')
try:
socket_address.unlink()
except OSError:
if socket_address.exists():
raise
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.bind(str(socket_address))
socket_address.chmod(0o666)
app.run(sock=sock, workers=2)