Something went wrong. Try again.
Hermes WebUI: The best way to use Hermes Agent from the web or from your phone!
Something went wrong. Try again.
5.5 kB · 131 lines
Python
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132"""Hermes Web UI -- File upload: multipart parser and upload handler."""import re as _reimport email.parserimport tempfilefrom pathlib import Path
from api.config import MAX_UPLOAD_BYTESfrom api.helpers import j, badfrom api.models import get_sessionfrom api.workspace import safe_resolve_ws
def parse_multipart(rfile, content_type, content_length) -> tuple: import re as _re, email.parser as _ep m = _re.search(r'boundary=([^;\s]+)', content_type) if not m: raise ValueError('No boundary in Content-Type') boundary = m.group(1).strip('"').encode() raw = rfile.read(content_length) fields = {} files = {} delimiter = b'--' + boundary end_marker = b'--' + boundary + b'--' parts = raw.split(delimiter) for part in parts[1:]: stripped = part.lstrip(b'\r\n') if stripped.startswith(b'--'): break sep = b'\r\n\r\n' if b'\r\n\r\n' in part else b'\n\n' if sep not in part: continue header_raw, body = part.split(sep, 1) if body.endswith(b'\r\n'): body = body[:-2] elif body.endswith(b'\n'): body = body[:-1] header_text = header_raw.lstrip(b'\r\n').decode('utf-8', errors='replace') msg = _ep.HeaderParser().parsestr(header_text) disp = msg.get('Content-Disposition', '') name_m = _re.search(r'name="([^"]*)"', disp) file_m = _re.search(r'filename="([^"]*)"', disp) if not name_m: continue name = name_m.group(1) if file_m: files[name] = (file_m.group(1), body) else: fields[name] = body.decode('utf-8', errors='replace') return fields, files
def _sanitize_upload_name(filename: str) -> str: safe_name = _re.sub(r'[^\w.\-]', '_', Path(filename).name)[:200] if not safe_name or safe_name.strip('.') == '': raise ValueError('Invalid filename') return safe_name
def handle_upload(handler): import traceback as _tb try: content_type = handler.headers.get('Content-Type', '') content_length = int(handler.headers.get('Content-Length', 0) or 0) if content_length > MAX_UPLOAD_BYTES: return j(handler, {'error': f'File too large (max {MAX_UPLOAD_BYTES//1024//1024}MB)'}, status=413) fields, files = parse_multipart(handler.rfile, content_type, content_length) session_id = fields.get('session_id', '') if 'file' not in files: return j(handler, {'error': 'No file field in request'}, status=400) filename, file_bytes = files['file'] if not filename: return j(handler, {'error': 'No filename in upload'}, status=400) try: s = get_session(session_id) except KeyError: return j(handler, {'error': 'Session not found'}, status=404) workspace = Path(s.workspace) safe_name = _sanitize_upload_name(filename) dest = safe_resolve_ws(workspace, safe_name) dest.write_bytes(file_bytes) return j(handler, {'filename': safe_name, 'path': str(dest), 'size': dest.stat().st_size}) except ValueError as e: return j(handler, {'error': str(e)}, status=400) except Exception: print('[webui] upload error: ' + _tb.format_exc(), flush=True) return j(handler, {'error': 'Upload failed'}, status=500)
def handle_transcribe(handler): import traceback as _tb temp_path = None try: content_type = handler.headers.get('Content-Type', '') content_length = int(handler.headers.get('Content-Length', 0) or 0) if content_length > MAX_UPLOAD_BYTES: return j(handler, {'error': f'File too large (max {MAX_UPLOAD_BYTES//1024//1024}MB)'}, status=413) fields, files = parse_multipart(handler.rfile, content_type, content_length) if 'file' not in files: return j(handler, {'error': 'No file field in request'}, status=400) filename, file_bytes = files['file'] if not filename: return j(handler, {'error': 'No filename in upload'}, status=400) safe_name = _sanitize_upload_name(filename) suffix = Path(safe_name).suffix or '.webm' with tempfile.NamedTemporaryFile(prefix='webui-stt-', suffix=suffix, delete=False) as tmp: temp_path = tmp.name tmp.write(file_bytes) try: from tools.transcription_tools import transcribe_audio except ImportError: return j(handler, {'error': 'Speech-to-text is unavailable on this server'}, status=503) result = transcribe_audio(temp_path) if not result.get('success'): msg = str(result.get('error') or 'Transcription failed') status = 503 if 'unavailable' in msg.lower() or 'not configured' in msg.lower() else 400 return j(handler, {'error': msg}, status=status) transcript = str(result.get('transcript') or '').strip() return j(handler, {'ok': True, 'transcript': transcript}) except ValueError as e: return j(handler, {'error': str(e)}, status=400) except Exception: print('[webui] transcribe error: ' + _tb.format_exc(), flush=True) return j(handler, {'error': 'Transcription failed'}, status=500) finally: if temp_path: try: Path(temp_path).unlink(missing_ok=True) except Exception: pass