"""Interface web local em HTML para o planejador diário.""" from __future__ import annotations import json import mimetypes import os import threading import webbrowser from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import urlparse import pandas as pd from src.planejador_missao.main import executar_planejamento from src.planejador_missao.quadrinhos import quadrinhos_para_api from src.planejador_missao.utils import normalizar_texto, parse_bool BASE_DIR = Path(__file__).resolve().parent DADOS_DIR = BASE_DIR / "dados" HOST = "127.0.0.1" PORT = 8050 def read_csv(path: Path, default: pd.DataFrame) -> pd.DataFrame: if not path.exists(): return default.copy() return pd.read_csv(path).fillna("") def load_state() -> dict: parametros_default = pd.DataFrame( { "parametro": ["data_planejamento", "criterio_missao", "criterio_sbv"], "valor": ["hoje", "meta_50", "equalizar_quadrinhos"], } ) parametros_df = read_csv(DADOS_DIR / "parametros_missao.csv", parametros_default) parametros = dict(zip(parametros_df["parametro"], parametros_df["valor"])) aeronaves_default = pd.DataFrame( { "aeronave": ["C98", "C97", "C95"], "disponivel_sede": [True, True, True], "em_pane": [False, False, False], "em_missao_rota": [False, False, False], "hora_livre": ["08:00", "08:00", "08:00"], } ) aeronaves = read_csv(DADOS_DIR / "aeronaves_disponiveis.csv", aeronaves_default) for col in ["disponivel_sede", "em_pane", "em_missao_rota"]: aeronaves[col] = aeronaves[col].map(parse_bool) rota_default = pd.DataFrame( { "origem": ["NT"], "destino": [""], "inicio": [pd.Timestamp.today().strftime("%Y-%m-%d 08:00")], "fim": [pd.Timestamp.today().strftime("%Y-%m-%d 17:00")], "tev_horas": [0], "pernoite_dias": [0], "rota": [""], } ) rotas = read_csv(DADOS_DIR / "rotas_acionadas.csv", rota_default) rota = rota_default.iloc[0].to_dict() if rotas.empty else rotas.iloc[0].to_dict() trips_default = pd.DataFrame({"tripulante": [], "disponivel": []}) tripulantes = read_csv(DADOS_DIR / "tripulantes_disponiveis.csv", trips_default) if not tripulantes.empty: tripulantes["tripulante"] = tripulantes["tripulante"].map(normalizar_texto) tripulantes["disponivel"] = tripulantes["disponivel"].map(parse_bool) tripulantes = tripulantes.sort_values("tripulante") return { "parametros": { "data_planejamento": str(parametros.get("data_planejamento", "hoje")), "criterio_missao": str(parametros.get("criterio_missao", "meta_50")), "criterio_sbv": str(parametros.get("criterio_sbv", "equalizar_quadrinhos")), }, "aeronaves": aeronaves.to_dict(orient="records"), "rota": rota, "tem_rota": str(rota.get("destino", "")).strip() != "", "tripulantes": tripulantes.to_dict(orient="records"), } def save_state(payload: dict) -> None: parametros = payload.get("parametros", {}) pd.DataFrame( { "parametro": ["data_planejamento", "criterio_missao", "criterio_sbv"], "valor": [ parametros.get("data_planejamento", "hoje"), parametros.get("criterio_missao", "meta_50"), parametros.get("criterio_sbv", "equalizar_quadrinhos"), ], } ).to_csv(DADOS_DIR / "parametros_missao.csv", index=False) pd.DataFrame(payload.get("aeronaves", [])).to_csv(DADOS_DIR / "aeronaves_disponiveis.csv", index=False) if payload.get("tem_rota", False): rota = payload.get("rota", {}) pd.DataFrame( { "origem": [normalizar_texto(rota.get("origem", "NT"))], "destino": [normalizar_texto(rota.get("destino", ""))], "inicio": [str(rota.get("inicio", "")).strip()], "fim": [str(rota.get("fim", "")).strip()], "tev_horas": [str(rota.get("tev_horas", "0")).strip()], "pernoite_dias": [str(rota.get("pernoite_dias", "0")).strip()], "rota": [normalizar_texto(rota.get("rota", ""))], } ).to_csv(DADOS_DIR / "rotas_acionadas.csv", index=False) else: pd.DataFrame(columns=["origem", "destino", "inicio", "fim", "tev_horas", "pernoite_dias", "rota"]).to_csv( DADOS_DIR / "rotas_acionadas.csv", index=False ) tripulantes = pd.DataFrame(payload.get("tripulantes", [])) if not tripulantes.empty: tripulantes["tripulante"] = tripulantes["tripulante"].map(normalizar_texto) tripulantes.to_csv(DADOS_DIR / "tripulantes_disponiveis.csv", index=False) class Handler(BaseHTTPRequestHandler): def do_GET(self) -> None: path = urlparse(self.path).path if path == "/": self._send_html(INDEX_HTML) return if path == "/api/state": self._send_json(load_state()) return if path == "/api/quadrinhos": self._send_json(quadrinhos_para_api(BASE_DIR)) return if path.startswith("/assets/"): self._send_file(DADOS_DIR / "assets" / path.removeprefix("/assets/")) return self.send_error(404) def do_POST(self) -> None: path = urlparse(self.path).path payload = self._read_json() if path == "/api/save": save_state(payload) self._send_json({"ok": True, "message": "Entradas salvas em dados/."}) return if path == "/api/run": save_state(payload) resultado = executar_planejamento(BASE_DIR) self._send_json( { "ok": True, "message": "Escala gerada com sucesso.", "data_planejamento": str(resultado["data_planejamento"]), "arquivo_saida": str(resultado["arquivo_saida"]), "total_candidatas": resultado["total_candidatas"], "total_selecionadas": resultado["total_selecionadas"], } ) return if path == "/api/open-latest": arquivos = sorted((BASE_DIR / "resultados").glob("planejamento_diario_*.xlsx"), key=lambda p: p.stat().st_mtime) if not arquivos: self._send_json({"ok": False, "message": "Nenhuma planilha encontrada."}, status=404) return os.startfile(arquivos[-1]) self._send_json({"ok": True, "message": f"Abrindo {arquivos[-1]}"}) return self.send_error(404) def _read_json(self) -> dict: length = int(self.headers.get("Content-Length", "0")) if length == 0: return {} return json.loads(self.rfile.read(length).decode("utf-8")) def _send_json(self, payload: dict, status: int = 200) -> None: data = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def _send_html(self, html: str) -> None: data = 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(data))) self.end_headers() self.wfile.write(data) def _send_file(self, path: Path) -> None: try: resolved = path.resolve() assets_dir = (DADOS_DIR / "assets").resolve() if not resolved.is_file() or assets_dir not in resolved.parents: self.send_error(404) return data = resolved.read_bytes() content_type = mimetypes.guess_type(str(resolved))[0] or "application/octet-stream" self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) except OSError: self.send_error(404) def log_message(self, format: str, *args) -> None: return INDEX_HTML = r""" Planejador Diário de Missões

Planejador Diário de Missões

Escala, sobreaviso e acompanhamento operacional

Parâmetros

Aeronaves

Missão acionada

Tripulantes disponíveis

Status

Carregando dados...

Quadrinho operacional

Tripulantes-
Quadrinhos abaixo de 50h-
Média de horas no ano-
TripAeronaveQualificaçãoPróxima OI Horas baseHoras anoSBVFaltam 50hÚltimo vooObs
Carregando...
""" def main() -> None: server = ThreadingHTTPServer((HOST, PORT), Handler) url = f"http://{HOST}:{PORT}" threading.Timer(0.5, lambda: webbrowser.open(url)).start() print(f"Interface web aberta em {url}") print("Pressione Ctrl+C para encerrar.") server.serve_forever() if __name__ == "__main__": main()