#!/usr/bin/env python3 """ servidor.py — a interface do fluxo enviar SD.md → corrigir o que falta → exportar. python3 app/servidor.py # http://127.0.0.1:5000 O fluxo é curto de propósito: o upload lê o SD.md, o exportador o analisa contra o contrato da carga (CONTRATO-JSON-V2.md) e a tela mostra o resumo do que vai sair. Se falta algo, aparece um formulário SÓ com os campos pendentes; corrigido, o botão Exportar libera. Nada mais. A INVARIANTE desta aplicação, e o critério pelo qual ela deve ser julgada: nenhuma rota entrega um caminho de arquivo ao exporta_sd.py, e nenhuma chama o main() dele. A exportação chama montar() com o dicionário DA SESSÃO — o mesmo objeto que o formulário editou. Duas consequências. Primeira: "upload → download automático" é impossível por construção — não existe caminho de código do /upload até o send_file; o único que gera JSON é o /exportar, e ele só roda depois de um clique do usuário. Segunda: o /exportar não confia no botão nem no estado da sessão — chama montar(), que reprova sozinho se houver pendência. Um POST direto não passa por botão nenhum. Estado em memória de processo, uma SD por vez: é ferramenta local de uso individual. Reiniciar o servidor descarta a sessão — o SD.md de origem continua intacto no disco, e nada é gravado sem ação explícita (Salvar SD.md). """ from __future__ import annotations import io import sys from datetime import date from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts")) from flask import Flask, redirect, render_template, request, send_file, url_for import escrita_sd import sessao as S from exporta_sd import ExportacaoReprovada, Pendencia, montar, serializar from regras_sd import (ESTADOS_SD, TIPOS_IMPORTAVEIS, carregar_canonico, definir_secao, ler_sd_texto, secao) app = Flask(__name__) # Chave YAML que existe com valor nulo — `processo_sei: null`, `linha_os: null` — # renderizava como o texto "None" no campo do formulário, e voltava do POST como # a STRING "None". Um "Salvar SD.md" depois disso gravaria `linha_os: "None"`. # Vale para todo o template de uma vez, e não campo a campo. app.jinja_env.finalize = lambda v: "" if v is None else v # Dias da janela, para a tabela de prévia — a mesma subtração que a carga faz # para derivar semanas (e, a partir delas, a UST). app.jinja_env.globals["dias"] = lambda a, b: (date.fromisoformat(b) - date.fromisoformat(a)).days # Uma SD por vez, em memória de processo. Ver docstring. ATUAL: dict = {"sessao": None, "aviso": None, "erro": None} def canonico(): return carregar_canonico() # --------------------------------------------------------------------------- # Conversão formulário → dados # --------------------------------------------------------------------------- # Tipo YAML de cada chave do frontmatter que o formulário pode editar. O que não # está aqui é texto. TIPO_DO_CAMPO = { "n": "int", "semanas": "int", "ust": "int", "inicio": "data", "fim": "data", "data_abertura": "data", } def _limpo(v) -> str: return (v or "").strip() def _como_valor(texto: str, tipo: str): """Converte o texto do formulário para o tipo do YAML. Quando não converte, devolve o texto cru DE PROPÓSITO: quem reclama de data ilegível ou semana não-inteira é o exportador, com mensagem própria e apontando o entregável. Silenciar aqui — virando None, ou recusando o post — tiraria do usuário a única explicação que ele ia receber. """ if tipo == "int": try: return int(texto) except ValueError: return texto if tipo == "data": try: return date.fromisoformat(texto) except ValueError: return texto return texto def _aplicar(mapa, chave: str, texto: str, tipo: str = "texto") -> None: """Grava a chave no CommentedMap, ou a remove quando o campo vem vazio. Campo esvaziado vira chave AUSENTE, e não string vazia: é assim que o exportador entende "campo faltante" — e, para `semanas`/`ust`/`linha_os`, é como o usuário diz "apague a declaração e derive". Só escreve quando o valor mudou de fato — reatribuir um valor idêntico é oportunidade gratuita de perder formatação e comentário de linha no writeback. """ atual = mapa.get(chave) if not texto: if chave in mapa and atual is not None: del mapa[chave] return if isinstance(atual, str) and atual.strip() == texto.strip(): return novo = _como_valor(texto, tipo) if atual != novo or chave not in mapa: mapa[chave] = novo def _linhas(texto: str) -> list[str]: return [l.strip() for l in (texto or "").splitlines() if l.strip()] def _aplicar_lista(mapa, chave: str, texto: str) -> None: itens = _linhas(texto) atual = mapa.get(chave) if not itens: if chave in mapa and atual: del mapa[chave] return if isinstance(atual, list) and [str(x).strip() for x in atual] == itens: return mapa[chave] = itens def _artefatos_de_texto(texto: str) -> list[dict]: """Uma linha por artefato, no formato `nome | AAAA-MM-DD`.""" saida = [] for linha in _linhas(texto): nome, sep, data = linha.partition("|") art = {"nome": nome.strip()} if sep and data.strip(): art["data"] = _como_valor(data.strip(), "data") saida.append(art) return saida def _texto_de_artefatos(artefatos) -> str: linhas = [] for a in (artefatos or []): if isinstance(a, dict): d = a.get("data") linhas.append(f"{a.get('nome', '')} | {d}" if d else str(a.get("nome", ""))) else: linhas.append(str(a)) return "\n".join(linhas) def aplicar_formulario(sess: S.Sessao, form) -> None: """Escreve as edições NO OBJETO DA SESSÃO. É a única mutação dos dados. Percorre as chaves que VIERAM no POST, e só elas. O formulário só traz os campos pendentes; um laço sobre uma lista fixa de campos leria os ausentes como "apagados" e o primeiro Salvar removeria do YAML tudo o que não estava pendente. A chave segue Pendencia.chave: `sd__campo`, `e{i}__campo`, `corpo__campo`. """ d = sess.dados for chave in form.keys(): escopo, sep, campo = chave.partition("__") if not sep or not campo: continue bruto = form.get(chave, "") texto = _limpo(bruto) if escopo == "sd": _aplicar(d, campo, texto, TIPO_DO_CAMPO.get(campo, "texto")) elif escopo == "corpo": if campo == "objetivo": sess.corpo = definir_secao(sess.corpo, "1", texto.replace("\r\n", "\n")) elif escopo.startswith("e") and escopo[1:].isdigit(): i = int(escopo[1:]) ents = d.get("entregaveis") or [] if i >= len(ents) or not isinstance(ents[i], dict): continue e = ents[i] if campo in ("atividades", "criterios_aceite"): _aplicar_lista(e, campo, bruto) elif campo == "artefatos": novos = _artefatos_de_texto(bruto) if novos: if _texto_de_artefatos(e.get("artefatos")).strip() != _texto_de_artefatos(novos).strip(): e["artefatos"] = novos elif "artefatos" in e and e.get("artefatos"): del e["artefatos"] else: _aplicar(e, campo, texto, TIPO_DO_CAMPO.get(campo, "texto")) # --------------------------------------------------------------------------- # Da pendência ao controle do formulário # --------------------------------------------------------------------------- # Decisão de APRESENTAÇÃO, e por isso mora aqui e não no exportador: ele diz # qual campo falhou, não com que controle se edita. ROTULO = { "sd": "Identificador da SD", "titulo": "Título", "po_responsavel": "PO responsável", "versao": "Versão do documento", "estado": "Estado da SD", "data_abertura": "Data de abertura", "objetivo": "Objetivo (Seção 1 do SD.md)", "n": "Número (n)", "nome": "Nome", "tipo": "Tipo", "item": "Item", "inicio": "Início", "fim": "Fim", "semanas": "Semanas declaradas", "ust": "UST declarada", "linha_os": "Linha de OS — copie da tela da OS no sistema, formato {OS}-L{n}", "atividades": "Backlog — um por linha", "criterios_aceite": "Critérios de aceite — um por linha", "artefatos": "Artefatos — nome | AAAA-MM-DD, um por linha", } TEXTAREA = {"objetivo", "atividades", "criterios_aceite", "artefatos"} # Sem controle na tela: o conserto é estrutural e se faz no SD.md. SO_MENSAGEM = {"entregaveis", "dependencias", "secao5", "alocacoes_desvio"} PLACEHOLDER = {"inicio": "AAAA-MM-DD", "fim": "AAAA-MM-DD", "data_abertura": "AAAA-MM-DD", "versao": "V1", "sd": "P2·SD8", "linha_os": "1090-L1"} def _valor_atual(sess: S.Sessao, p: Pendencia): if p.escopo == "corpo": return secao(sess.corpo or "", "1") or "" if p.campo == "objetivo" else "" fonte = sess.dados if p.escopo == "sd" else (sess.dados.get("entregaveis") or [])[p.indice] if not isinstance(fonte, dict): return "" v = fonte.get(p.campo) if p.campo == "artefatos": return _texto_de_artefatos(v) if p.campo in ("atividades", "criterios_aceite"): return "\n".join(str(x) for x in (v or [])) return "" if v is None else v def _opcoes(p: Pendencia) -> list | None: if p.campo == "estado": return list(ESTADOS_SD) if p.campo == "tipo": return list(TIPOS_IMPORTAVEIS) if p.campo == "item": return list(p.opcoes or []) return None def montar_formulario(sess: S.Sessao) -> list[dict]: """Agrupa as pendências por alvo (SD, cada entregável, corpo), um controle por campo. Uma pendência de `semanas` traz junto `inicio` e `fim`: a correção pode ser em qualquer um dos três, e a pessoa decide qual.""" grupos: dict[tuple, dict] = {} for p in sess.pendencias: if p.escopo == "entregavel": gk = ("entregavel", p.indice) titulo = f"Entregável {p.n}" if p.n not in (None, "") else f"Entregável #{p.indice + 1}" elif p.escopo == "corpo": gk, titulo = ("corpo", None), "Corpo do documento" else: gk, titulo = ("sd", None), "Cabeçalho da SD" g = grupos.setdefault(gk, {"titulo": titulo, "campos": {}}) # Semanas, início e fim são um só problema visto de três lados: a janela # que dá as semanas. Pendência em qualquer um traz os outros para a tela. campos = [p.campo] if p.campo in ("semanas", "inicio", "fim"): campos = [c for c in ("semanas", "inicio", "fim") if c == p.campo or c != "semanas"] for campo in campos: c = g["campos"].get(campo) if c is None: chave = Pendencia(p.escopo, campo, "", indice=p.indice, n=p.n).chave c = g["campos"][campo] = { "chave": chave, "campo": campo, "rotulo": ROTULO.get(campo, campo), "controle": ("mensagem" if campo in SO_MENSAGEM else "textarea" if campo in TEXTAREA else "select" if _opcoes(Pendencia(p.escopo, campo, "")) is not None else "texto"), "valor": _valor_atual(sess, Pendencia(p.escopo, campo, "", indice=p.indice)), "opcoes": _opcoes(p) if campo == p.campo else _opcoes(Pendencia(p.escopo, campo, "")), "placeholder": PLACEHOLDER.get(campo, ""), "mensagens": [], "derivado": None, } if campo == p.campo: c["mensagens"].append(p.mensagem) if p.derivado is not None: c["derivado"] = p.derivado return [{"titulo": g["titulo"], "campos": list(g["campos"].values())} for g in grupos.values()] # --------------------------------------------------------------------------- # Rotas # --------------------------------------------------------------------------- @app.get("/") def inicio(): if ATUAL["sessao"] is None: return render_template("upload.html", estado=S.SEM_ARQUIVO, rotulo=S.ROTULO_ESTADO[S.SEM_ARQUIVO], erro=ATUAL.pop("erro", None)) return redirect(url_for("revisar")) @app.post("/upload") def upload(): arq = request.files.get("arquivo") if arq is None or not arq.filename: ATUAL["erro"] = "Nenhum arquivo selecionado." return redirect(url_for("inicio")) if not arq.filename.lower().endswith(".md"): ATUAL["erro"] = (f"“{arq.filename}” não é um arquivo .md. " "A SD é um Markdown com frontmatter YAML.") return redirect(url_for("inicio")) bruto = arq.read() if not bruto.strip(): ATUAL["erro"] = f"“{arq.filename}” está vazio." return redirect(url_for("inicio")) try: texto = bruto.decode("utf-8") except UnicodeDecodeError: ATUAL["erro"] = (f"“{arq.filename}” não está em UTF-8. " "Salve o arquivo como UTF-8 e envie de novo.") return redirect(url_for("inicio")) try: # preservar_comentarios=True: é este objeto que o writeback regrava, e os # comentários do frontmatter carregam decisão de contrato. dados, corpo = ler_sd_texto(texto, preservar_comentarios=True) except Exception as exc: ATUAL["erro"] = f"Não consegui ler “{arq.filename}”: {exc}" return redirect(url_for("inicio")) sess = S.Sessao(nome_arquivo=arq.filename, dados=dados, corpo=corpo) sess.caminho_origem = escrita_sd.destino_provavel(str(dados.get("sd", ""))) ATUAL["sessao"] = sess ATUAL["erro"] = None reavaliar(sess) # Redireciona para a tela de revisão — NUNCA para um download. É a única # saída desta rota. return redirect(url_for("revisar")) def reavaliar(sess: S.Sessao) -> None: try: sess.reavaliar(canonico()) except Exception as exc: sess.ultimo_erro = f"A análise falhou: {type(exc).__name__} — {exc}" sess.pendencias = [Pendencia("sd", "entregaveis", f"a análise estourou: {exc}")] sess.estado = S.COM_PENDENCIAS @app.get("/revisar") def revisar(): sess = ATUAL["sessao"] if sess is None: return redirect(url_for("inicio")) return render_template( "revisar.html", s=sess, rotulo=sess.rotulo_estado(), formulario=montar_formulario(sess), previa=sess.previa or {}, aviso=ATUAL.pop("aviso", None), ) @app.post("/revisar") def salvar_revisao(): """Aplica as correções e reavalia. Não exporta — só o /exportar exporta.""" sess = ATUAL["sessao"] if sess is None: return redirect(url_for("inicio")) sess.ultimo_erro = None try: aplicar_formulario(sess, request.form) except Exception as exc: sess.ultimo_erro = f"Não consegui aplicar as correções: {type(exc).__name__} — {exc}" return redirect(url_for("revisar")) reavaliar(sess) return redirect(url_for("revisar")) @app.post("/exportar") def exportar(): """A ÚNICA rota que gera JSON — e montar() reprova sozinho se houver pendência.""" sess = ATUAL["sessao"] if sess is None: return redirect(url_for("inicio")) sess.ultimo_erro = None try: # montar() recebe o DICIONÁRIO DA SESSÃO — o mesmo que o formulário # editou. Nenhum caminho de arquivo, nenhuma releitura do SD.md. payload = montar(sess.dados, sess.corpo, canonico()) texto = serializar(payload) except ExportacaoReprovada as exc: # Reavaliado no servidor: um POST direto não passa por botão nenhum. sess.pendencias = exc.pendencias sess.estado = S.COM_PENDENCIAS sess.ultimo_erro = (f"Exportação recusada — {len(exc.pendencias)} pendência(s). " "Nada foi baixado: um arquivo que o importador rejeita não é " "uma exportação.") return redirect(url_for("revisar")) except Exception as exc: sess.ultimo_erro = f"A geração do JSON falhou: {type(exc).__name__} — {exc}" return redirect(url_for("revisar")) sess.marcar_exportada() nome = f"{str(sess.dados.get('sd', 'sd')).replace('·', '-')}.json" return send_file(io.BytesIO(texto.encode("utf-8")), mimetype="application/json", as_attachment=True, download_name=nome) @app.get("/salvar-sd") def salvar_sd_preview(): sess = ATUAL["sessao"] if sess is None: return redirect(url_for("inicio")) destino = Path(request.args.get("destino") or (sess.caminho_origem or "")) try: texto = escrita_sd.render_sd(sess.dados, sess.corpo) except Exception as exc: sess.ultimo_erro = f"Não consegui montar o SD.md: {type(exc).__name__} — {exc}" return redirect(url_for("revisar")) return render_template( "salvar.html", s=sess, destino=str(destino) if str(destino) else "", diff=escrita_sd.diff_contra(destino, texto) if str(destino) else "", no_repositorio=bool(str(destino)) and escrita_sd.dentro_do_repositorio(destino), rotulo=sess.rotulo_estado(), ) @app.post("/salvar-sd") def salvar_sd(): sess = ATUAL["sessao"] if sess is None: return redirect(url_for("inicio")) destino = _limpo(request.form.get("destino")) if not destino: sess.ultimo_erro = "Informe o caminho de destino para salvar o SD.md." return redirect(url_for("revisar")) if request.form.get("confirmo") != "sim": sess.ultimo_erro = "Gravação cancelada: a confirmação não foi marcada." return redirect(url_for("revisar")) try: escrita_sd.gravar(Path(destino), escrita_sd.render_sd(sess.dados, sess.corpo)) except Exception as exc: sess.ultimo_erro = f"Não consegui gravar {destino}: {type(exc).__name__} — {exc}" return redirect(url_for("revisar")) ATUAL["aviso"] = f"SD.md gravado em {destino}." return redirect(url_for("revisar")) @app.post("/reiniciar") def reiniciar(): ATUAL["sessao"] = None ATUAL["aviso"] = None ATUAL["erro"] = None return redirect(url_for("inicio")) if __name__ == "__main__": for stream in (sys.stdout, sys.stderr): if hasattr(stream, "reconfigure"): stream.reconfigure(encoding="utf-8", errors="replace") print(" Exportação da SD em http://127.0.0.1:5000") app.run(host="127.0.0.1", port=5000, debug=False)