#!/usr/bin/env python3 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from datetime import datetime, timezone from urllib.parse import urlparse import html import json import os import re import shutil import tempfile import cgi HOST = '127.0.0.1' PORT = 18950 BASE_DIR = Path('/home/sebas/outputs/inbox/ballbox-dropzone') BASE_DIR.mkdir(parents=True, exist_ok=True) MAX_BYTES = 512 * 1024 * 1024 def safe_name(name: str) -> str: name = os.path.basename(name).strip().replace('\x00', '') name = re.sub(r'[^A-Za-z0-9._ -]+', '-', name) name = re.sub(r'\s+', '-', name) name = name.strip('.-') return name or 'upload.bin' def stamp_name(name: str) -> str: p = Path(name) ts = datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S') stem = safe_name(p.stem) suffix = ''.join(p.suffixes) or '' return f'{ts}-{stem}{suffix}' def file_rows() -> str: rows = [] for p in sorted(BASE_DIR.iterdir(), key=lambda x: x.stat().st_mtime, reverse=True): if not p.is_file(): continue size_mb = p.stat().st_size / (1024 * 1024) mtime = datetime.fromtimestamp(p.stat().st_mtime).strftime('%Y-%m-%d %H:%M') href = f'/files/outputs/inbox/ballbox-dropzone/{p.name}' rows.append(f'{html.escape(p.name)}{size_mb:.1f} MB{mtime}') return '\n'.join(rows) or 'Vacío' def page(message: str = '') -> bytes: msg = f'

{html.escape(message)}

' if message else '' body = f''' Ballbox Dropzone

Ballbox Dropzone

Subí acá los archivos de la máquina.

Acepta uno o varios archivos juntos. Quedan guardados en /home/sebas/outputs/inbox/ballbox-dropzone/

{msg}

Archivos recibidos

{file_rows()}

Archivo	Tamaño	Fecha

''' return body.encode('utf-8') class Handler(BaseHTTPRequestHandler): def do_GET(self): parsed = urlparse(self.path) if parsed.path in ('/', ''): self.respond_html(200, page()) return if parsed.path == '/health': self.respond_json(200, {'ok': True, 'dir': str(BASE_DIR)}) return self.respond_text(404, 'not found') def do_POST(self): parsed = urlparse(self.path) if parsed.path != '/upload': self.respond_text(404, 'not found') return ctype, pdict = cgi.parse_header(self.headers.get('content-type', '')) if ctype != 'multipart/form-data': self.respond_html(400, page('Pedido inválido.')) return form = cgi.FieldStorage(fp=self.rfile, headers=self.headers, environ={'REQUEST_METHOD': 'POST'}, keep_blank_values=True) items = form['file'] if 'file' in form else None if items is None: self.respond_html(400, page('Falta archivo.')) return if not isinstance(items, list): items = [items] uploaded = [] try: for item in items: if getattr(item, 'file', None) is None or not getattr(item, 'filename', None): continue name = stamp_name(item.filename) tmp_fd, tmp_path = tempfile.mkstemp(dir=str(BASE_DIR), prefix='.upload-') total = 0 try: with os.fdopen(tmp_fd, 'wb') as out: while True: chunk = item.file.read(1024 * 1024) if not chunk: break total += len(chunk) if total > MAX_BYTES: raise ValueError(f'Archivo demasiado grande: {item.filename}') out.write(chunk) final_path = BASE_DIR / name os.replace(tmp_path, final_path) uploaded.append(name) except Exception: try: os.unlink(tmp_path) except FileNotFoundError: pass raise if not uploaded: self.respond_html(400, page('Falta archivo.')) return summary = 'Subidos:\n- ' + '\n- '.join(uploaded) self.respond_html(200, page(summary)) except Exception as e: self.respond_html(400, page(str(e))) def log_message(self, fmt, *args): return def respond_html(self, code, payload: bytes): self.send_response(code) self.send_header('Content-Type', 'text/html; charset=utf-8') self.send_header('Content-Length', str(len(payload))) self.end_headers() self.wfile.write(payload) def respond_text(self, code, text): payload = text.encode('utf-8') self.send_response(code) self.send_header('Content-Type', 'text/plain; charset=utf-8') self.send_header('Content-Length', str(len(payload))) self.end_headers() self.wfile.write(payload) def respond_json(self, code, obj): payload = json.dumps(obj).encode('utf-8') self.send_response(code) self.send_header('Content-Type', 'application/json') self.send_header('Content-Length', str(len(payload))) self.end_headers() self.wfile.write(payload) if __name__ == '__main__': httpd = ThreadingHTTPServer((HOST, PORT), Handler) httpd.serve_forever()