import hashlib import os from pathlib import Path from fastapi import FastAPI, HTTPException, Query from fastapi.responses import HTMLResponse, JSONResponse from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware import uvicorn app = FastAPI() app.add_middleware(ProxyHeadersMiddleware, trusted_hosts="*") SERVER_ROOT = Path(__file__).parent.resolve() BASE_URL = "https://dl-srv.gtk.cl" INDEX_BASE_URL = "https://srv.gtk.cl" HIDDEN_NAMES = { '.env', '.git', '.gitignore', '.htaccess', '.htpasswd', '__pycache__', '.DS_Store', 'Thumbs.db', } def is_hidden(path: Path) -> bool: return path.name.startswith('.') or path.name in HIDDEN_NAMES def get_safe_path(path: str) -> Path: if not path or path in ("", "/"): return SERVER_ROOT full_path = (SERVER_ROOT / path).resolve() try: full_path.relative_to(SERVER_ROOT) except ValueError: raise HTTPException(403, "Access denied") if not full_path.exists(): raise HTTPException(404, "Not found") if is_hidden(full_path): raise HTTPException(403, "Access denied") if full_path.is_file(): raise HTTPException(403, "Access denied") return full_path def get_safe_file_path(path: str) -> Path: """Как get_safe_path, но требует существующий ФАЙЛ (для t=info).""" if not path: raise HTTPException(400, "File path required") full_path = (SERVER_ROOT / path).resolve() try: full_path.relative_to(SERVER_ROOT) except ValueError: raise HTTPException(403, "Access denied") if not full_path.exists() or not full_path.is_file(): raise HTTPException(404, "Not found") if is_hidden(full_path): raise HTTPException(403, "Access denied") return full_path # ── md5 (для сравнения контента при синхронизации) ──────────────────────── # Кэшируем по (mtime, size), чтобы не пересчитывать md5 больших файлов на # каждый запрос листинга — пересчёт только если файл реально изменился. _MD5_CACHE: dict[str, tuple[float, int, str]] = {} def compute_md5(path: Path) -> str: stat = path.stat() key = str(path) cached = _MD5_CACHE.get(key) if cached and cached[0] == stat.st_mtime and cached[1] == stat.st_size: return cached[2] h = hashlib.md5() with open(path, "rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) digest = h.hexdigest() _MD5_CACHE[key] = (stat.st_mtime, stat.st_size, digest) return digest def format_size(size: int) -> str: for unit in ['B', 'KB', 'MB', 'GB', 'TB']: if size < 1024: return f"{size}{unit}" if unit == 'B' else f"{size:.1f}{unit}" size /= 1024 return f"{size:.1f}PB" def make_file_item(path: Path, depth: int = 0, with_md5: bool = False) -> dict: rel = str(path.relative_to(SERVER_ROOT)).replace('\\', '/') size = path.stat().st_size item = { "name": path.name, "type": "file", "path": rel, "depth": depth, "size": size, "size_human": format_size(size), "url": f"{BASE_URL}/{rel}", } if with_md5: item["md5"] = compute_md5(path) return item def make_dir_item(path: Path, depth: int = 0, children: list = None) -> dict: rel = str(path.relative_to(SERVER_ROOT)).replace('\\', '/') item = { "name": path.name, "type": "dir", "path": rel, "depth": depth, "url": f"{INDEX_BASE_URL}/{rel}", } if children is not None: item["children"] = children return item def get_items(path: Path, recursive: int = 0, hidedirs: bool = False, with_md5: bool = False) -> list: if recursive == 2: return [ make_file_item(Path(root) / f, with_md5=with_md5) for root, dirs, files in os.walk(path) for f in files if not is_hidden(Path(f)) and not any(is_hidden(Path(p)) for p in Path(root).parts) ] def build(p: Path, depth: int = 0) -> list: items = [] dirs, files = [], [] for i in sorted(p.iterdir()): if is_hidden(i): continue (dirs if i.is_dir() else files).append(i) if not hidedirs: for d in dirs: children = build(d, depth + 1) if recursive == 1 else None items.append(make_dir_item(d, depth, children)) items.extend(make_file_item(f, depth, with_md5=with_md5) for f in files) return items return build(path) def get_urls_flat(path: Path, recursive: bool = False) -> list: if recursive: return [ f"{BASE_URL}/{str((Path(root) / f).relative_to(SERVER_ROOT)).replace(chr(92), '/')}" for root, dirs, files in os.walk(path) for f in files if not is_hidden(Path(f)) and not any(is_hidden(Path(p)) for p in Path(root).parts) ] return [ f"{BASE_URL}/{str(f.relative_to(SERVER_ROOT)).replace(chr(92), '/')}" for f in sorted(path.iterdir()) if f.is_file() and not is_hidden(f) ] def generate_html(data: dict, recursive: int = 0) -> str: html = f"

Index of {data['path']}

" stripped = data['path'].strip('/') if stripped: parent = '/'.join(stripped.split('/')[:-1]) parent_url = f"{INDEX_BASE_URL}/{parent}" if parent else INDEX_BASE_URL html += f'
..
' def render(items: list) -> str: result = "" for item in items: indent = "  " * item.get('depth', 0) if item['type'] == 'dir': result += f'
{indent}{item["name"]}/
' if 'children' in item: result += render(item['children']) else: result += ( f'
{indent}{item["name"]}' f' ({item["size_human"]})
' ) return result if recursive == 2: for item in data['items']: html += ( f'
{item["path"]}' f' ({item["size_human"]})
' ) else: html += render(data['items']) html += "
" return html @app.get("/{path:path}") async def handle_request( path: str = "", t: str = Query("human"), hd: bool = Query(False), R: int = Query(0, ge=0, le=2), md5: bool = Query(False), ): # t=info — метаданные (в т.ч. md5) ровно одного файла, без обхода # директории. Используется клиентом при открытии конкретной картинки. if t == "info": file_path = get_safe_file_path(path) item = make_file_item(file_path, with_md5=True) item["mtime"] = file_path.stat().st_mtime return JSONResponse(content=item) full_path = get_safe_path(path) display_path = f"/{path}" if path else "/" print(display_path) if t == "mjson": if R == 1: raise HTTPException(400, "R=1 is not compatible with t=mjson") return JSONResponse(content={"items": get_urls_flat(full_path, recursive=(R == 2))}) items = get_items(full_path, recursive=R, hidedirs=hd, with_md5=md5) json_data = { "path": display_path, "index_base_url": INDEX_BASE_URL, "file_base_url": BASE_URL, "items": items, "total": len(items), "recursive": R, "hidedirs": hd, } if t == "json": return JSONResponse(content=json_data) return HTMLResponse(content=generate_html(json_data, R)) if __name__ == "__main__": import multiprocessing workers = min(multiprocessing.cpu_count(), 8) uvicorn.run( app, host="127.0.0.1", port=8000, workers=workers, timeout_keep_alive=3600, limit_concurrency=1000, limit_max_requests=10000, loop="asyncio", )