wander/Exporta-SD
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Commit inicial: app de exportação da SD + deploy Docker/Coolify

Browse Source 
Aplicação Flask (enviar SD.md -> corrigir pendências -> exportar JSON),
scripts de contrato (scripts/) e infraestrutura de deploy (Dockerfile,
docker-compose.yml) para VPS via Coolify.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
wander
2026-09-04 14:07:38 -03:00
co-authored by Claude Sonnet 5
commit 0d4fdd2533
29 changed files with 4268 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/servidor.py
+474
View File
@@ -0,0 +1,474 @@
#!/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)