import os import http.server import socketserver import urllib.parse import zipfile import io import shutil # Importado para deletar diretórios recursivamente # --- Sua função original para carregar JSON --- def load_file_content(filename): # ... (o código da sua função permanece o mesmo) ... 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 # --- Handler do Servidor Aprimorado --- class MyAdvancedHttpRequestHandler(http.server.SimpleHTTPRequestHandler): # --- MÉTODO GET: Lida com visualização, download e sua rota JSON --- def do_GET(self): parsed_url = urllib.parse.urlparse(self.path) query_params = urllib.parse.parse_qs(parsed_url.query) # Rota para sua lógica JSON if 'filename' in query_params: self.handle_json_request(query_params['filename'][0]) return # Rota para baixar diretório como .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): self.serve_directory_as_zip(path_to_zip) else: self.send_error(404, "Diretório não encontrado") return # Lógica padrão (servir arquivo ou listar diretório) super().do_GET() # --- MÉTODO POST: Lida com ações, como deletar --- def do_POST(self): 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 para deletar não especificado.") # Traduz e VALIDA o caminho para segurança path_to_delete = self.translate_path(path_to_delete_str) print(f"Tentando deletar: {path_to_delete}") if os.path.isdir(path_to_delete): shutil.rmtree(path_to_delete) print(f"Diretório deletado: {path_to_delete}") elif os.path.isfile(path_to_delete): os.remove(path_to_delete) print(f"Arquivo deletado: {path_to_delete}") else: self.send_error(404, "Arquivo ou diretório não encontrado para deletar.") return # Redireciona para o diretório pai após a exclusão parent_dir = os.path.dirname(path_to_delete_str) self.send_response(303) # 303 See Other é apropriado para redirecionamento pós-POST self.send_header('Location', parent_dir) self.end_headers() else: self.send_error(400, "Ação POST inválida.") except Exception as e: print(f"Erro no POST: {e}") self.send_error(500, f"Erro interno do servidor: {e}") # --- Métodos auxiliares para GET --- def handle_json_request(self, filename): # ... (seu código aqui, sem alterações) ... 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): # ... (seu código aqui, sem alterações) ... dir_name = os.path.basename(os.path.normpath(dir_path)) memory_file = io.BytesIO() with zipfile.ZipFile(memory_file, 'w', zipfile.ZIP_DEFLATED) as zf: for root, dirs, files in os.walk(dir_path): for file in files: zf.write(os.path.join(root, file), os.path.relpath(os.path.join(root, file), dir_path)) memory_file.seek(0) self.send_response(200) self.send_header('Content-Type', 'application/zip') self.send_header('Content-Disposition', f'attachment; filename="{dir_name}.zip"') self.send_header('Content-Length', str(memory_file.getbuffer().nbytes)) self.end_headers() self.wfile.write(memory_file.getvalue()) # --- MÉTODO PRINCIPAL DA INTERFACE --- def list_directory(self, path): try: list = os.listdir(path) except OSError: self.send_error(404, "Permissão negada para listar o diretório") return None list.sort(key=lambda a: a.lower()) display_path = urllib.parse.unquote(self.path, errors='surrogatepass') # --- Ícones SVG --- download_icon = '' delete_icon = '' folder_icon = '' file_icon = '' html_parts = [ '', '', f'Índice de {display_path}', '', '', '', f'

Índice de {display_path}

', '', '', '', '', '', '' ] if display_path != '/': html_parts.append('') for name in list: fullname = os.path.join(path, name) display_name = name link_name = urllib.parse.quote(name, errors='surrogatepass') current_path = os.path.join(display_path, link_name).replace("\\", "/") # Coluna 4: Ícone de Tipo e Nome if os.path.isdir(fullname): type_icon = folder_icon name_cell = f'' else: type_icon = file_icon name_cell = f'' # Coluna 1: Download if os.path.isdir(fullname): download_link = f'{download_icon}' else: download_link = f'{download_icon}' # Coluna 2: Deletar (com formulário e confirmação) delete_form = f''' ''' html_parts.append('') html_parts.append(f'') html_parts.append(f'') html_parts.append(f'') html_parts.append(name_cell) html_parts.append('') html_parts.extend(['
BaixarDeletarTipoNome
⬆️Diretório Pai
{display_name}/{display_name}
{download_link}{delete_form}{type_icon}
']) encoded_html = '\n'.join(html_parts).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) # Sobrescreve para garantir que o path é seguro def translate_path(self, path): path = super().translate_path(path) base_dir = os.path.abspath(os.getcwd()) requested_path = os.path.abspath(path) if not os.path.commonpath([base_dir, requested_path]) == base_dir: raise PermissionError("Tentativa de acesso a um caminho fora do diretório raiz.") return requested_path # --- Configuração e Inicialização do Servidor --- PORT = 8000 handler_object = MyAdvancedHttpRequestHandler my_server = socketserver.TCPServer(("", PORT), handler_object) print(f"Servidor iniciado na porta {PORT}. Acesse http://localhost:{PORT}") print("Para parar o servidor, pressione Ctrl+C") try: my_server.serve_forever() except KeyboardInterrupt: print("\nServidor sendo desligado.") my_server.shutdown()