import os import http.server import socketserver import urllib.parse import zipfile import io import shutil import json import re from datetime import datetime # --- Funções auxiliares --- def format_file_size(size_bytes): if size_bytes < 1024: return f"{size_bytes} B" elif size_bytes < 1024**2: return f"{size_bytes/1024:.1f} KB" elif size_bytes < 1024**3: return f"{size_bytes/1024**2:.1f} MB" else: return f"{size_bytes/1024**3:.1f} GB" def load_file_content(filename): try: folder = "oplaboptionvols" file_path = os.path.join(folder, f"{filename}.json") if os.path.exists(file_path): with open(file_path, 'r', encoding='utf-8') as file: return file.read() return None except Exception: return None def log(tag, msg, ok=True): ts = datetime.now().strftime('%Y-%m-%d %H:%M:%S') status = "OK " if ok else "ERRO" print(f"[{ts}] {status} {tag:<8} {msg}") # --- Handler do Servidor --- class MyAdvancedHttpRequestHandler(http.server.SimpleHTTPRequestHandler): def send_cors_headers(self): """Adiciona headers CORS em todas as respostas.""" self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type, Content-Disposition") def do_OPTIONS(self): """Responde ao preflight CORS do navegador/app.""" self.send_response(204) self.send_cors_headers() self.end_headers() def end_headers(self): """Injeta CORS em todas as respostas automaticamente.""" self.send_cors_headers() super().end_headers() def do_GET(self): parsed_url = urllib.parse.urlparse(self.path) query_params = urllib.parse.parse_qs(parsed_url.query) # 1. Rota JSON para o Delphi if 'list' in query_params and query_params['list'][0] == 'json': path_to_list = self.translate_path(parsed_url.path) if os.path.isdir(path_to_list): self.serve_json_directory_list(path_to_list) else: self.send_error(404, "Diretorio nao encontrado") return # 2. Rota original de JSON por nome if 'filename' in query_params: fname = query_params['filename'][0] log("JSON", f"{fname} ← {self.client_address[0]}") self.handle_json_request(fname) return # 3. Download em ZIP if 'download' in query_params and query_params['download'][0] == 'zip': path_to_zip = self.translate_path(parsed_url.path) if os.path.isdir(path_to_zip): log("ZIP", f"{parsed_url.path} ← {self.client_address[0]}") self.serve_directory_as_zip(path_to_zip) else: log("ZIP", f"{parsed_url.path} ← {self.client_address[0]} (diretório não encontrado)", ok=False) self.send_error(404, "Diretório não encontrado") return super().do_GET() def serve_json_directory_list(self, path): try: parsed_url = urllib.parse.urlparse(self.path) clean_base_path = parsed_url.path raw_list = os.listdir(path) items = [] for name in raw_list: fullname = os.path.join(path, name) is_dir = os.path.isdir(fullname) relative_path = clean_base_path.rstrip('/') + '/' + name info = { "name": name, "is_dir": is_dir, "path": relative_path, "size_bytes": os.path.getsize(fullname) if not is_dir else 0, "mtime": datetime.fromtimestamp(os.path.getmtime(fullname)).isoformat() } items.append(info) items.sort(key=lambda x: (not x['is_dir'], x['name'].lower())) response_json = json.dumps(items, indent=4).encode('utf-8') self.send_response(200) self.send_header("Content-type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(response_json))) self.end_headers() self.wfile.write(response_json) except Exception as e: self.send_error(500, f"Erro ao listar: {e}") def do_POST(self): content_type = self.headers.get('Content-Type', '') # LÓGICA DE UPLOAD if 'multipart/form-data' in content_type: result, message = self.handle_upload() if result: log("UPLOAD", f"{message} ← {self.client_address[0]}") self.send_response(201) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"status": "success", "message": message}).encode()) else: log("UPLOAD", f"{message} ← {self.client_address[0]}", ok=False) self.send_error(400, message) return # LÓGICA DE DELETE try: content_length = int(self.headers['Content-Length']) post_data = self.rfile.read(content_length) params = urllib.parse.parse_qs(post_data.decode('utf-8')) if params.get('action', [None])[0] == 'delete': path_to_delete_str = params.get('path', [None])[0] if not path_to_delete_str: raise ValueError("Caminho não especificado.") path_to_delete = self.translate_path(path_to_delete_str) if os.path.isdir(path_to_delete): log("DELETE", f"pasta {path_to_delete_str} ← {self.client_address[0]}") shutil.rmtree(path_to_delete) elif os.path.isfile(path_to_delete): size = format_file_size(os.path.getsize(path_to_delete)) log("DELETE", f"arquivo {path_to_delete_str} ({size}) ← {self.client_address[0]}") os.remove(path_to_delete) else: log("DELETE", f"{path_to_delete_str} não encontrado ← {self.client_address[0]}", ok=False) self.send_error(404, "Arquivo ou diretório não encontrado para deletar.") return parent_dir = os.path.dirname(path_to_delete_str) or '/' if not parent_dir.endswith('/'): parent_dir += '/' self.send_response(303) self.send_header('Location', parent_dir) self.end_headers() else: self.send_error(400, "Ação POST inválida.") except Exception as e: log("POST", str(e), ok=False) self.send_error(500, str(e)) def handle_upload(self): """Processa o upload multipart/form-data usando cgi.FieldStorage (eficiente para arquivos grandes).""" import cgi try: form = cgi.FieldStorage( fp=self.rfile, headers=self.headers, environ={ 'REQUEST_METHOD': 'POST', 'CONTENT_TYPE': self.headers['Content-Type'], 'CONTENT_LENGTH': self.headers.get('Content-Length', '0'), } ) # Determina o diretório destino dest_dir = self.translate_path(".") # Default: raiz if 'directory' in form: val = form['directory'].value.strip() if val: dest_dir = self.translate_path(val) # Processa o arquivo if 'file' not in form: return False, "Campo 'file' não encontrado no formulário" file_item = form['file'] if not file_item.filename: return False, "Nome do arquivo não enviado" filename = os.path.basename(file_item.filename) if not os.path.exists(dest_dir): os.makedirs(dest_dir) out_path = os.path.join(dest_dir, filename) # Grava em chunks de 1MB — eficiente para arquivos grandes CHUNK = 1024 * 1024 with open(out_path, 'wb') as out_file: while True: chunk = file_item.file.read(CHUNK) if not chunk: break out_file.write(chunk) return True, f"Salvo em: {os.path.relpath(out_path, os.getcwd())}" except Exception as e: return False, f"Erro no processamento: {str(e)}" def handle_json_request(self, filename): json_content = load_file_content(filename) if json_content: response_bytes = json_content.encode('utf-8') self.send_response(200) self.send_header("Content-type", "application/json") else: error_msg = '{"error": "File not found"}' response_bytes = error_msg.encode('utf-8') self.send_response(404) self.send_header("Content-type", "application/json") self.send_header("Content-Length", str(len(response_bytes))) self.end_headers() self.wfile.write(response_bytes) def serve_directory_as_zip(self, dir_path): dir_name = os.path.basename(os.path.normpath(dir_path)) zip_filename = f"{dir_name}.zip" memory_file = io.BytesIO() log("ZIP", f"compactando '{dir_name}'...") with zipfile.ZipFile(memory_file, 'w', zipfile.ZIP_DEFLATED) as zf: for root, dirs, files in os.walk(dir_path): for file in files: fp = os.path.join(root, file) archive_name = os.path.relpath(fp, start=dir_path) zf.write(fp, archive_name) if not files and not dirs: archive_name = os.path.relpath(root, start=dir_path) zf.writestr(archive_name + '/', '') memory_file.seek(0) size = memory_file.getbuffer().nbytes log("ZIP", f"{zip_filename} ({format_file_size(size)})") self.send_response(200) self.send_header('Content-Type', 'application/zip') self.send_header('Content-Disposition', f'attachment; filename="{zip_filename}"') self.send_header('Content-Length', str(size)) self.end_headers() self.wfile.write(memory_file.getvalue()) # --- INTERFACE PRINCIPAL --- def list_directory(self, path): try: raw_list = os.listdir(path) except OSError: self.send_error(404, "Permissão negada para listar o diretório") return None display_path = urllib.parse.unquote(self.path, errors='surrogatepass').split('?')[0] directories, files = [], [] for name in raw_list: if os.path.isdir(os.path.join(path, name)): directories.append(name) else: files.append(name) directories.sort(key=str.lower) files.sort(key=str.lower) sorted_list = directories + files # --- SVG Icons --- ico_download = '' ico_delete = '' ico_folder = '' ico_file = '' ico_upload = '' ico_zip = '' rows_html = [] if display_path != '/': rows_html.append(''' ⬆ .. —— ''') for name in sorted_list: fullname = os.path.join(path, name) link_name = urllib.parse.quote(name, errors='surrogatepass') current_path = (display_path.rstrip('/') + '/' + link_name) try: mod_time_str = datetime.fromtimestamp(os.path.getmtime(fullname)).strftime('%Y-%m-%d %H:%M') file_size_str = format_file_size(os.path.getsize(fullname)) if os.path.isfile(fullname) else '—' except OSError: mod_time_str, file_size_str = '—', '—' if os.path.isdir(fullname): type_icon = ico_folder name_cell = f'{name}/' dl_link = f'{ico_zip}' else: type_icon = ico_file name_cell = f'{name}' dl_link = f'{ico_download}' del_form = f'''
''' rows_html.append(f''' {dl_link} {del_form} {type_icon} {name_cell} {file_size_str} {mod_time_str} ''') # Constrói o HTML completo da página html = f''' {display_path}
File Server · porta {PORT}
{self._render_path_html(display_path)}
{self._render_breadcrumb(display_path)}
{''.join(rows_html)}
DL Del Tipo Nome Tamanho Modificado em
📁 {len(directories)} pasta{'s' if len(directories) != 1 else ''} 📄 {len(files)} arquivo{'s' if len(files) != 1 else ''}
''' encoded_html = html.encode('utf-8') self.send_response(200) self.send_header("Content-type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(encoded_html))) self.end_headers() self.wfile.write(encoded_html) def _render_path_html(self, path): """Renderiza o path com spans coloridos.""" parts = [p for p in path.split('/') if p] if not parts: return '/' out = '/' for i, seg in enumerate(parts): out += f'{seg}' if i < len(parts) - 1: out += '/' return out def _render_breadcrumb(self, path): """Gera links de navegação em breadcrumb.""" parts = [p for p in path.split('/') if p] crumbs = ['raiz'] for i, seg in enumerate(parts): href = '/' + '/'.join(parts[:i+1]) + '/' if i == len(parts) - 1: crumbs.append(f'{seg}') else: crumbs.append(f'{seg}') return ' › '.join(crumbs) def translate_path(self, path): """Proteção contra Directory Traversal.""" path = path.split('?', 1)[0].split('#', 1)[0] path = os.path.normpath(urllib.parse.unquote(path)) words = path.split(os.sep) result_path = os.getcwd() for word in words: if word in (os.curdir, os.pardir, ''): continue result_path = os.path.join(result_path, word) # Garante que o caminho resultante está dentro do cwd base_dir = os.path.abspath(os.getcwd()) requested = os.path.abspath(result_path) if os.path.commonpath([base_dir, requested]) != base_dir: raise PermissionError("Tentativa de acesso fora do diretório raiz.") return result_path # --- Configuração e Inicialização --- PORT = 8000 with socketserver.ThreadingTCPServer(("", PORT), MyAdvancedHttpRequestHandler) as httpd: print(f"╔══════════════════════════════════╗") print(f"║ Servidor rodando na porta {PORT} ║") print(f"║ http://localhost:{PORT} ║") print(f"╚══════════════════════════════════╝") try: httpd.serve_forever() except KeyboardInterrupt: print("\nDesligando...") httpd.shutdown()