"""Fluxos criticos de escrita (POST) executados na base (7.4) e no alvo (8.3), comparando o efeito no banco. Cada fluxo devolve {"fluxo", "status_http", "db_antes", "db_depois", "email", "ok", "detalhe"}. Os registros criados levam o prefixo MIGR83-- para serem identificados na copia do banco. """ import argparse import json import re import subprocess import time import urllib.request from pathlib import Path from config import AMBIENTES, BANCO, LOGIN_SENHA, LOGIN_USUARIO, MYSQL_CONTAINER, SAIDA MAILPIT = "http://localhost:8025/api/v1" FIXTURES = Path(__file__).resolve().parent / "fixtures" # ---------- infra ---------- def sql(consulta: str) -> list: r = subprocess.run( ["docker", "exec", "-i", MYSQL_CONTAINER, "sh", "-c", f'mysql -uroot -p"$MYSQL_ROOT_PASSWORD" -N -B --default-character-set=utf8mb4 {BANCO} 2>/dev/null'], input=consulta, capture_output=True, text=True, encoding="utf8", errors="replace") return [l.split("\t") for l in r.stdout.splitlines() if l] def um(consulta: str): linhas = sql(consulta) return linhas[0][0] if linhas and linhas[0][0] != "NULL" else None def mailpit_mensagens() -> list: with urllib.request.urlopen(MAILPIT + "/messages?limit=500", timeout=20) as r: return json.load(r).get("messages", []) def base_url(rotulo: str) -> str: return f"http://localhost:{AMBIENTES[rotulo][0]}" def login(page, url_base: str): page.goto(url_base + "/site/login", wait_until="load", timeout=60000) # a tela tem dois formularios (desktop e mobile): usa o visivel form = page.locator("form").filter(has=page.locator("input[name='LoginForm[username]']:visible")).first campo = form.locator("input[name='LoginForm[username]']") campo.click() campo.press_sequentially(re.sub(r"\D", "", LOGIN_USUARIO), delay=30) form.locator("input[name='LoginForm[password]']").fill(LOGIN_SENHA) with page.expect_navigation(timeout=60000): form.locator("button[type=submit], input[type=submit]").first.click() return "/site/login" not in page.url # ---------- fluxos ---------- def fluxo_login_sessao(page, rotulo: str) -> dict: """Login pelo formulario (mascara de CPF) grava sessao no DbSession com o id do usuario.""" antes = um("select count(*) from session where data like '%__id|i:4;%';") logou = login(page, base_url(rotulo)) depois = um("select count(*) from session where data like '%__id|i:4;%';") # a contagem global oscila (GC do DbSession apaga sessoes expiradas); o criterio e a linha da sessao do navegador sess = sessao_id(page) propria = um(f"select count(*) from session where id = '{sess}' and data like '%__id|i:4;%';") return {"fluxo": "login + DbSession", "status_http": 200 if logou else 0, "db_antes": {"sessoes_usuario_4": antes}, "db_depois": {"sessoes_usuario_4": depois}, "efeito": {"linha_session_do_navegador_com_id_4": propria}, "email": None, "ok": logou and propria == "1", "detalhe": page.url} def fluxo_email_inventario_scope(page, rotulo: str) -> dict: """SiteController::actionEditaInventarioScope envia 'Responda o Inventario de Dados!' ao trocar o e-mail.""" linha = sql("select i.id, s.token from inventario_scope i join scope s on s.id = i.scope_id " "where s.status = 1 and s.token is not null and s.token <> '' order by i.id limit 1;") if not linha: return {"fluxo": "e-mail inventario scope", "ok": False, "detalhe": "sem scope com token na copia"} id_, token = linha[0] destino = f"migr83-{rotulo}-{int(time.time())}@teste.local" antes_msgs = {m["ID"] for m in mailpit_mensagens()} antes = um(f"select concat_ws('|', email_responsavel, length(token)) from inventario_scope where id = {id_};") resp = page.goto(f"{base_url(rotulo)}/site/edita-inventario-scope?id={id_}&token={token}", wait_until="load") page.fill("input[name='InventarioScope[email_responsavel]']", destino) with page.expect_navigation(timeout=60000): page.locator("form button[type=submit], form input[type=submit]").first.click() depois = um(f"select concat_ws('|', email_responsavel, length(token)) from inventario_scope where id = {id_};") novas = [m for m in mailpit_mensagens() if m["ID"] not in antes_msgs] email = None if novas: m = novas[0] email = {"para": [t["Address"] for t in m.get("To", [])], "assunto": m.get("Subject")} ok = (depois or "").startswith(destino) and email is not None and email["para"] == [destino] \ and email["assunto"] == "Responda o Inventário de Dados!" return {"fluxo": "e-mail inventario scope", "status_http": resp.status if resp else None, "db_antes": {"inventario_scope": antes}, "db_depois": {"inventario_scope": depois}, "email": email, "ok": ok, "detalhe": f"inventario_scope.id={id_}"} # ---------- helpers de formulario / http ---------- def prefixo(rotulo: str) -> str: return f"MIGR83-{rotulo}-" def normaliza_texto(v, rotulo: str): if isinstance(v, str): return v.replace(prefixo(rotulo), "MIGR83-*-") if isinstance(v, list): return [normaliza_texto(x, rotulo) for x in v] if isinstance(v, dict): return {k: normaliza_texto(x, rotulo) for k, x in v.items()} return v JS_SERIALIZA = """(modelo) => { const alvo = document.querySelector(`[name^="${modelo}["]`); if (!alvo) return null; const form = alvo.closest('form'); if (!form) return null; const pares = []; for (const el of form.elements) { if (!el.name || el.disabled) continue; const t = (el.type || '').toLowerCase(); if (['file', 'submit', 'button', 'reset', 'image'].includes(t)) continue; if ((t === 'checkbox' || t === 'radio') && !el.checked) continue; if (el.tagName === 'SELECT') { if (el.multiple) { for (const o of el.selectedOptions) pares.push([el.name, o.value]); continue; } let v = el.value; if (v === '') { const o = [...el.options].find(o => o.value !== ''); v = o ? o.value : ''; } pares.push([el.name, v]); continue; } pares.push([el.name, el.value]); } return {action: form.getAttribute('action'), pares}; }""" def serializa_form(page, modelo: str) -> list: r = page.evaluate(JS_SERIALIZA, modelo) if r is None: raise RuntimeError(f"formulario com campos {modelo}[...] nao encontrado em {page.url}") return [list(p) for p in r["pares"]] def define(pares: list, nome: str, valor) -> list: """Substitui (ou acrescenta) o campo `nome` na lista de pares do formulario.""" novos = [p for p in pares if p[0] != nome] novos.append([nome, "" if valor is None else str(valor)]) return novos def post_form(page, url: str, pares: list, multipart: dict = None) -> dict: """POST com os cookies da sessao do navegador, sem seguir redirecionamento.""" from urllib.parse import urlencode if multipart is not None: dados = {k: v for k, v in pares} dados.update(multipart) r = page.request.post(url, multipart=dados, max_redirects=0, timeout=180000) else: r = page.request.post(url, data=urlencode([tuple(p) for p in pares]), max_redirects=0, timeout=180000, headers={"Content-Type": "application/x-www-form-urlencoded"}) corpo = r.text() if "text" in r.headers.get("content-type", "") or "json" in r.headers.get("content-type", "") else "" return {"status": r.status, "location": r.headers.get("location"), "corpo": corpo} def resumo_resposta(resp: dict, rotulo: str) -> dict: """Parte comparavel da resposta: status, destino do redirect sem ids, erros de validacao/excecao visiveis.""" loc = resp.get("location") or "" loc = re.sub(r"https?://[^/]+", "", loc) loc = re.sub(r"\d+", "{n}", loc) corpo = resp.get("corpo") or "" erros = re.findall(r'class="help-block[^"]*"[^>]*>([^<]+)<', corpo) erros += re.findall(r'
]*>(.*?)
', corpo, re.S) excecao = re.findall(r"

([^<]*(?:Exception|Error)[^<]*)

", corpo) return {"status": resp["status"], "redirect": loc or None, "erros_form": [normaliza_texto(e.strip(), rotulo) for e in erros if e.strip()][:5], "excecao": excecao[:1]} def sessao_id(page) -> str: for c in page.context.cookies(): if c["name"] == "PHPSESSID": return c["value"] return "" def linhas_tabela(tabela: str, colunas: list, onde: str, ordem: str = "id") -> list: cols = ", ".join(f"ifnull(`{c}`, 'NULL')" for c in colunas) return sql(f"select {cols} from {tabela} where {onde} order by {ordem};") # ---------- create + update por modulo ---------- def crud(page, rotulo: str, spec: dict) -> dict: """Cria um registro pelo formulario de create e altera pelo de update; compara o efeito no banco.""" url = base_url(rotulo) p = prefixo(rotulo) modelo, campo, tabela, col = spec["modelo"], spec["campo"], spec["tabela"], spec["coluna"] # so as linhas criadas nesta execucao (rodadas anteriores deixam registros com o mesmo prefixo) id_antes = um(f"select ifnull(max(id), 0) from {tabela};") or "0" filtro = f"`{col}` like '{p}%' and id > {id_antes}" antes = um(f"select count(*) from {tabela} where {filtro};") page.goto(url + spec["create"], wait_until="load", timeout=90000) pares = serializa_form(page, modelo) for k, v in spec.get("fixos", {}).items(): pares = define(pares, k, v) pares = define(pares, f"{modelo}[{campo}]", p + spec["nome"]) r_create = post_form(page, url + spec["create"], pares) id_ = um(f"select max(id) from {tabela} where `{col}` = '{p + spec['nome']}';") r_update = None if id_ and spec.get("update"): page.goto(url + spec["update"].format(id=id_), wait_until="load", timeout=90000) pares = serializa_form(page, modelo) for k, v in spec.get("fixos_update", {}).items(): pares = define(pares, k, v) pares = define(pares, f"{modelo}[{campo}]", p + spec["nome"] + "-alterado") r_update = post_form(page, url + spec["update"].format(id=id_), pares) linhas = linhas_tabela(tabela, spec["colunas"], filtro) final = um(f"select `{col}` from {tabela} where id = {id_ or 0};") ok = id_ is not None and (not spec.get("update") or final == p + spec["nome"] + "-alterado") return {"fluxo": spec["fluxo"], "status_http": r_create["status"], "resposta": {"create": resumo_resposta(r_create, rotulo), "update": resumo_resposta(r_update, rotulo) if r_update else None}, "db_antes": {tabela: antes}, "db_depois": {tabela: len(linhas)}, "efeito": normaliza_texto(linhas, rotulo), "email": None, "ok": ok, "detalhe": f"{tabela}.id={id_} sessao={sessao_id(page)}"} def _contexto_financeiro(): # a copia nao tem cf_fc_tipo; sem tipo existente a lista de contextos permitidos e vazia -> 403 esperado return um("select id from contexto where active = 1 and rgt - lft = 1 order by id limit 1;") or um("select t.contexto_id from cf_fc_tipo t join contexto c on c.id = t.contexto_id " "join cliente cl on cl.id = c.cliente_id where t.status = 1 and c.active = 1 and cl.status = 1 " "order by t.contexto_id limit 1;") def _scope_ref(): # a copia nao tem cm_provider/cm_task e o usuario 4 nao tem escopos: usa o primeiro scope ativo return um("select min(id) from scope where status = 1;") def fluxo_performance_arvore(page, rotulo): return crud(page, rotulo, { "fluxo": "performance: arvore create+update", "modelo": "Arvore", "campo": "nome", "nome": "arvore", "create": "/performance/arvore/create", "update": "/performance/arvore/update?id={id}", "tabela": "ca_arvore", "coluna": "nome", "fixos": {"Arvore[descricao]": "descricao migr83"}, "colunas": ["contexto_id", "nome", "status", "descricao", "codigo_fim_resposta", "msg_fim", "cadastrado_por"]}) def fluxo_financial_tipo(page, rotulo): ctx = _contexto_financeiro() return crud(page, rotulo, { "fluxo": "financial: tipo (cf_fc_tipo) create+update", "modelo": "Tipo", "campo": "nome", "nome": "tipo", "create": "/financial/tipo/create", "update": "/financial/tipo/update?id={id}", "tabela": "cf_fc_tipo", "coluna": "nome", "fixos": {"Tipo[contexto_id]": ctx}, "fixos_update": {"Tipo[contexto_id]": ctx}, "colunas": ["contexto_id", "nome", "status"]}) def fluxo_contract_provider(page, rotulo): return crud(page, rotulo, { "fluxo": "contract: provider (cm_provider) create+update", "modelo": "Provider", "campo": "name", "nome": "fornecedor", "create": "/contract/provider/create", "update": "/contract/provider/update?id={id}", "tabela": "cm_provider", "coluna": "name", "fixos_update": {"Provider[target_id]": _scope_ref()}, "fixos": {"Provider[target_id]": _scope_ref(), "Provider[contact_name]": "Contato Ação", "Provider[contact_email]": "contato@teste.local"}, "colunas": ["target_id", "name", "identifier", "contact_name", "contact_number", "contact_email", "active"]}) def fluxo_risk_modelo(page, rotulo): ref = sql("select cliente_id, vertical_vinculo, referencia_vertical from risco_modelo where status = 1 " "order by id limit 1;") fixos = {} if ref: fixos = {"RiscoModelo[cliente_id]": ref[0][0], "RiscoModelo[vertical_vinculo]": ref[0][1], "RiscoModelo[referencia_vertical]": ref[0][2]} return crud(page, rotulo, { "fluxo": "risk: modelo de risco create+update", "modelo": "RiscoModelo", "campo": "nome", "nome": "modelo", "create": "/risk/risco-modelo/create", "update": "/risk/risco-modelo/update?id={id}", "tabela": "risco_modelo", "coluna": "nome", "fixos": dict(fixos, **{"RiscoModelo[descricao]": "desc migr83"}), "colunas": ["nome", "descricao", "vertical_vinculo", "status", "created_by", "cliente_id", "referencia_vertical"]}) def fluxo_task(page, rotulo): return crud(page, rotulo, { "fluxo": "task: tarefa (cm_task) create+update", "modelo": "Task", "campo": "name", "nome": "tarefa", "create": "/task/task/create", "update": "/task/task/update?id={id}", "tabela": "cm_task", "coluna": "name", # usuario 4 nao tem paineis (Logged::getMyEditorPanelsIds vazio): validacao de panel_id deve recusar nos dois "fixos": {"Task[description]": "descrição migr83", "Task[panel_id]": _scope_ref(), "Task[to_panel_id][]": _scope_ref()}, "colunas": ["module", "target_id", "panel_id", "to_panel_id", "name", "description", "priority", "status", "created_by", "updated_by", "parent_id", "balance"]}) def fluxo_operational_ocorrencia(page, rotulo): """Cria ocorrencia (resposta JSON) copiando os vinculos de uma ocorrencia existente com subtipo sem destaque.""" url = base_url(rotulo) p = prefixo(rotulo) ref = sql("select o.cliente_id, o.contexto_id, o.co_ocorrencia_origem_id, o.co_ocorrencia_tipo_id, " "o.co_ocorrencia_subtipo_id, o.co_ativo_id, o.co_local_ocorrencia_id, o.co_local_id, a.id " "from co_ocorrencia o join co_ocorrencia_subtipo s on s.id = o.co_ocorrencia_subtipo_id " "join aplicacao a on a.status = 1 " "where o.isDeleted = 0 and ifnull(s.destaque, 0) = 0 and o.co_ativo_id is not null " "and o.co_local_ocorrencia_id is not null order by o.id desc limit 1;") if not ref: return {"fluxo": "operational: ocorrencia create+update", "ok": False, "detalhe": "sem ocorrencia de referencia"} cli, ctx, origem, tipo, subtipo, ativo, local_oc, local, apl = ref[0] id_antes = um("select ifnull(max(id), 0) from co_ocorrencia;") or "0" filtro = f"observacao like '{p}%' and id > {id_antes}" antes = um(f"select count(*) from co_ocorrencia where {filtro};") rota = f"/operational/ocorrencias/create?contexto_id={ctx}&aplicacao_id={apl}" page.goto(url + rota, wait_until="load", timeout=90000) pares = serializa_form(page, "Ocorrencia") for k, v in {"cliente_id": cli, "contexto_id": ctx, "co_ocorrencia_origem_id": origem, "co_ocorrencia_tipo_id": tipo, "co_ocorrencia_subtipo_id": subtipo, "co_ativo_id": ativo, "co_local_ocorrencia_id": local_oc, "co_local_id": local, "observacao": p + "ocorrencia", "created_at": "22/10/2026 10:00:00"}.items(): if v != "NULL": pares = define(pares, f"Ocorrencia[{k}]", v) r = post_form(page, url + rota, pares) try: js = json.loads(r["corpo"]) except Exception: js = {"nao_json": r["corpo"][:300]} id_ = js.get("id") if isinstance(js, dict) else None colunas = ["cliente_id", "contexto_id", "co_ocorrencia_origem_id", "co_ocorrencia_tipo_id", "co_ocorrencia_subtipo_id", "co_ativo_id", "co_local_ocorrencia_id", "co_local_id", "status", "observacao", "created_by", "updated_by", "isDeleted"] linhas = linhas_tabela("co_ocorrencia", colunas, filtro) resp = {"status": r["status"], "json": {k: v for k, v in js.items() if k != "id"} if isinstance(js, dict) else js} return {"fluxo": "operational: ocorrencia create (JSON)", "status_http": r["status"], "resposta": resp, "db_antes": {"co_ocorrencia": antes}, "db_depois": {"co_ocorrencia": len(linhas)}, "efeito": normaliza_texto(linhas, rotulo), "email": None, "ok": bool(id_) and len(linhas) == 1, "detalhe": f"co_ocorrencia.id={id_} sessao={sessao_id(page)}", "_id": id_} # ---------- importacao GPP ---------- GPP_TABELAS = { "gpp_temp": None, "gpp_causa_provavel": None, "gpp_acao_proposta": None, "gpp_tarefa": None, } GPP_IGNORAR = re.compile(r"^(id|.*_id_fk|causa_provavel_id|acao_proposta_id|created_at|updated_at|data_upload)$") def _colunas(tabela: str) -> list: return [l[0] for l in sql(f"select column_name from information_schema.columns where table_schema = " f"'{BANCO}' and table_name = '{tabela}' order by ordinal_position;")] def _snapshot_gpp() -> dict: out = {} for t in GPP_TABELAS: cols = [c for c in _colunas(t) if not GPP_IGNORAR.match(c)] out[t] = {"colunas": cols, "linhas": linhas_tabela(t, cols, "1=1")} # vinculos por texto (os ids mudam a cada upload) out["vinculo_acao_causa"] = sql("select a.acao_proposta, c.causa_provavel from gpp_acao_proposta a " "left join gpp_causa_provavel c on c.id = a.causa_provavel_id order by a.id;") out["vinculo_tarefa_acao"] = sql("select t.tarefa, a.acao_proposta from gpp_tarefa t " "left join gpp_acao_proposta a on a.id = t.acao_proposta_id order by t.id;") return out def _upload_gpp(page, rotulo: str, arquivo: str) -> dict: url = base_url(rotulo) + "/performance/gpp-temp/upload" page.goto(url, wait_until="load", timeout=90000) pares = [p for p in serializa_form(page, "GppTemp") if p[0] != "GppTemp[excelFile]"] caminho = FIXTURES / arquivo mime = ("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" if arquivo.endswith("xlsx") else "application/vnd.ms-excel") r = post_form(page, url, pares, multipart={ "GppTemp[excelFile]": {"name": arquivo, "mimeType": mime, "buffer": caminho.read_bytes()}}) return r def fluxo_gpp(page, rotulo: str, arquivo: str) -> dict: antes = {t: um(f"select count(*) from {t};") for t in GPP_TABELAS} r = _upload_gpp(page, rotulo, arquivo) snap = _snapshot_gpp() depois = {t: len(snap[t]["linhas"]) for t in GPP_TABELAS} ok = r["status"] in (301, 302) and depois["gpp_tarefa"] > 0 return {"fluxo": f"GPP upload {arquivo}", "status_http": r["status"], "resposta": resumo_resposta(r, rotulo), "db_antes": antes, "db_depois": depois, "efeito": snap, "email": None, "ok": ok, "detalhe": f"sessao={sessao_id(page)}"} def fluxo_gpp_xlsx(page, rotulo): return fluxo_gpp(page, rotulo, "gpp.xlsx") def fluxo_gpp_xls(page, rotulo): return fluxo_gpp(page, rotulo, "gpp.xls") def fluxo_gpp_valido_xlsx(page, rotulo): return fluxo_gpp(page, rotulo, "gpp-valido.xlsx") def fluxo_gpp_valido_xls(page, rotulo): return fluxo_gpp(page, rotulo, "gpp-valido.xls") # ---------- PDF / Word ---------- def _pdf_texto(conteudo: bytes, container: str = "php83_fluxos") -> str: tmp = SAIDA / "capturas" / "_tmp_fluxo.pdf" tmp.parent.mkdir(parents=True, exist_ok=True) tmp.write_bytes(conteudo) subprocess.run(["docker", "cp", str(tmp), f"{container}:/tmp/_fluxo.pdf"], capture_output=True) r = subprocess.run(["docker", "exec", container, "pdftotext", "-layout", "/tmp/_fluxo.pdf", "-"], capture_output=True, text=True, encoding="utf8", errors="replace") tmp.unlink(missing_ok=True) return r.stdout if r.returncode == 0 else "ERRO pdftotext: " + r.stderr[:200] def _normaliza_pdf(txt: str) -> list: # remove datas/horas de geracao e espacos de layout txt = re.sub(r"\d{2}/\d{2}/\d{4}( \d{2}:\d{2}(:\d{2})?)?", "{data}", txt) return [re.sub(r"\s+", " ", l).strip() for l in txt.splitlines() if l.strip()] def _baixa(page, url: str): r = page.request.get(url, timeout=180000) return r.status, r.headers.get("content-type", ""), r.headers.get("content-disposition", ""), r.body() def fluxo_pdf_ocorrencia(page, rotulo): """OcorrenciasController::actionImprimir (mpdf, download) para uma ocorrencia fixa da copia.""" # so ocorrencias finalizadas (status 3) podem ser impressas (OcorrenciasController::actionImprimir) id_ = um("select max(id) from co_ocorrencia where isDeleted = 0 and status = 3;") st, ct, cd, corpo = _baixa(page, f"{base_url(rotulo)}/operational/ocorrencias/imprimir?id={id_}") eh_pdf = corpo[:4] == b"%PDF" texto = _normaliza_pdf(_pdf_texto(corpo)) if eh_pdf else [] return {"fluxo": "PDF mpdf: ocorrencias/imprimir", "status_http": st, "resposta": {"status": st, "content_type": ct.split(";")[0], "disposition": cd, "eh_pdf": eh_pdf}, "efeito": {"texto": texto}, "ok": eh_pdf and len(texto) > 0, "detalhe": f"co_ocorrencia.id={id_} bytes={len(corpo)} linhas_texto={len(texto)}"} def fluxo_pdf_versionamento(page, rotulo): """VersionamentoController::actionExportarPdf (kartik Pdf -> mpdf).""" id_ = um("select max(id) from versionamento where status = 1;") st, ct, cd, corpo = _baixa(page, f"{base_url(rotulo)}/versionamento/exportar-pdf?id={id_}") eh_pdf = corpo[:4] == b"%PDF" texto = _normaliza_pdf(_pdf_texto(corpo)) if eh_pdf else [] return {"fluxo": "PDF kartik/mpdf: versionamento/exportar-pdf", "status_http": st, "resposta": {"status": st, "content_type": ct.split(";")[0], "disposition": cd, "eh_pdf": eh_pdf}, "efeito": {"texto": texto}, "ok": eh_pdf and len(texto) > 0, "detalhe": f"versionamento.id={id_} bytes={len(corpo)} linhas_texto={len(texto)}"} def fluxo_pdf_mpdf_cli(page, rotulo): """mpdf com as chamadas de OcorrenciasController::actionImprimir, via CLI no container do ambiente (tempDir=/tmp).""" container = AMBIENTES[rotulo][1] raiz = "/var/www/html/" + AMBIENTES[rotulo][2].name r = subprocess.run(["docker", "exec", container, "php", "-d", "display_errors=stderr", "/var/www/html/accia-php83/tools/migracao-php83/php/mpdf_texto.php", raiz, f"/tmp/mpdf-{rotulo}.pdf"], capture_output=True, text=True, encoding="utf8", errors="replace") b = subprocess.run(["docker", "exec", container, "cat", f"/tmp/mpdf-{rotulo}.pdf"], capture_output=True).stdout eh_pdf = b[:4] == b"%PDF" texto = _normaliza_pdf(_pdf_texto(b)) if eh_pdf else [] return {"fluxo": "PDF mpdf: chamadas de ocorrencias/imprimir (CLI)", "status_http": None, "resposta": {"rc": r.returncode, "stderr": r.stderr.strip()[-300:], "eh_pdf": eh_pdf}, "efeito": {"texto": texto}, "ok": eh_pdf and len(texto) > 0, "detalhe": f"{container} {raiz} bytes={len(b)}"} def _docx_xml(conteudo: bytes) -> str: import io import zipfile with zipfile.ZipFile(io.BytesIO(conteudo)) as z: xml = z.read("word/document.xml").decode("utf8") xml = re.sub(r'w:rsid\w*="[^"]*"', "", xml) return re.sub(r">\s+<", "><", xml) def fluxo_word_relatorio(page, rotulo): """AnexoController::actionCreateRelatorio (PhpWord TemplateProcessor).""" ref = sql("select c.id, a.id from contexto c join aplicacao a on a.status = 1 where c.active = 1 " "and c.rgt - c.lft = 1 order by c.id limit 1;") ctx, apl = ref[0] if ref else (0, 0) st, ct, cd, corpo = _baixa(page, f"{base_url(rotulo)}/performance/anexo/create-relatorio?contexto_id={ctx}" f"&aplicacao_id={apl}") eh_zip = corpo[:2] == b"PK" efeito = {"document_xml": _docx_xml(corpo)} if eh_zip else {"corpo": _texto_visivel(corpo)} return {"fluxo": "Word PhpWord: anexo/create-relatorio", "status_http": st, "resposta": {"status": st, "content_type": ct.split(";")[0], "eh_docx": eh_zip}, "efeito": efeito, "ok": eh_zip, "detalhe": f"contexto_id={ctx} aplicacao_id={apl} bytes={len(corpo)}"} def _texto_visivel(corpo: bytes) -> str: t = corpo.decode("utf8", "replace") t = re.sub(r"(?s)<(script|style)[^>]*>.*?", "", t) t = re.sub(r"<[^>]+>", " ", t) return re.sub(r"\s+", " ", t).strip()[:400] def fluxo_word_template_cli(page, rotulo): """PhpWord TemplateProcessor (mesmas chamadas do laudo) via CLI no container do ambiente.""" container = AMBIENTES[rotulo][1] raiz = "/var/www/html/" + AMBIENTES[rotulo][2].name r = subprocess.run(["docker", "exec", container, "php", "-d", "display_errors=stderr", "/var/www/html/accia-php83/tools/migracao-php83/php/word_template.php", raiz, f"/tmp/laudo-{rotulo}.docx"], capture_output=True, text=True, encoding="utf8", errors="replace") b = subprocess.run(["docker", "exec", container, "cat", f"/tmp/laudo-{rotulo}.docx"], capture_output=True) ok = r.returncode == 0 and b.stdout[:2] == b"PK" return {"fluxo": "Word PhpWord: TemplateProcessor (CLI, chamadas do laudo)", "status_http": None, "resposta": {"rc": r.returncode, "stderr": r.stderr.strip()[-300:]}, "efeito": {"document_xml": _docx_xml(b.stdout)} if ok else {}, "ok": ok, "detalhe": f"{container} {raiz}"} # ---------- API V1 ---------- API_GETS = [ "lista-cliente-contexto", "lista-coleta", "cliente?id={cliente}", "lista-aplicacao?id={contexto}", "coleta?coleta={coleta}", "organograma?id={contexto}", "pesquisa-estrutura?id={contexto}", ] def fluxo_api_v1(page, rotulo): """GETs autenticados por QueryParamAuth (?access-token=user.auth_key) e Bearer; compara o JSON.""" token = um("select auth_key from user_ceon where id = 4;") # User::tableName() = user_ceon ids = sql("select c.cliente_id, c.id, (select max(id) from coleta where contexto_id = c.id) from contexto c " "where c.active = 1 and exists (select 1 from coleta k where k.contexto_id = c.id) order by c.id limit 1;") cliente, contexto, coleta = ids[0] if ids else (0, 0, 0) saida = {} for rota in API_GETS: r = rota.format(cliente=cliente, contexto=contexto, coleta=coleta) sep = "&" if "?" in r else "?" resp = page.request.get(f"{base_url(rotulo)}/api/v1/{r}{sep}access-token={token}", timeout=120000, headers={"Accept": "application/json"}) try: corpo = resp.json() except Exception: corpo = {"nao_json": resp.text()[:300]} saida[rota] = {"status": resp.status, "json": corpo} sem = page.request.get(f"{base_url(rotulo)}/api/v1/lista-coleta", headers={"Accept": "application/json"}) bearer = page.request.get(f"{base_url(rotulo)}/api/v1/lista-coleta", headers={"Accept": "application/json", "Authorization": f"Bearer {token}"}) saida["sem token (lista-coleta)"] = {"status": sem.status} saida["bearer (lista-coleta)"] = {"status": bearer.status} ok = all(v["status"] == 200 for k, v in saida.items() if k != "sem token (lista-coleta)") and sem.status == 401 return {"fluxo": "API v1 GET autenticado", "status_http": saida["lista-coleta"]["status"], "resposta": {k: v["status"] for k, v in saida.items()}, "efeito": saida, "ok": ok, "detalhe": f"cliente={cliente} contexto={contexto} coleta={coleta}"} FLUXOS = [fluxo_login_sessao, fluxo_email_inventario_scope, fluxo_performance_arvore, fluxo_financial_tipo, fluxo_contract_provider, fluxo_risk_modelo, fluxo_task, fluxo_operational_ocorrencia, fluxo_gpp_xlsx, fluxo_gpp_xls, fluxo_gpp_valido_xlsx, fluxo_gpp_valido_xls, fluxo_pdf_ocorrencia, fluxo_pdf_mpdf_cli, fluxo_pdf_versionamento, fluxo_word_relatorio, fluxo_word_template_cli, fluxo_api_v1] def rodar_todos(rotulo: str) -> list: from playwright.sync_api import sync_playwright resultados = [] with sync_playwright() as p: nav = p.chromium.launch() page = nav.new_context(viewport={"width": 1366, "height": 900}).new_page() page.on("dialog", lambda d: d.dismiss()) for fluxo in FLUXOS: log_antes = um("select ifnull(max(id), 0) from cm_log;") or "0" try: r = fluxo(page, rotulo) except Exception as e: # registra e segue para o proximo fluxo r = {"fluxo": fluxo.__name__, "ok": False, "detalhe": "excecao: " + str(e)[:300]} r["funcao"] = fluxo.__name__ sess = sessao_id(page) # erros (level 1) gravados pelo Yii no cm_log durante o fluxo, na sessao deste navegador r["erros_cm_log"] = [f"{c} | {m}" for c, m in sql( f"select category, left(replace(message, '\\n', ' '), 400) from cm_log where id > {log_antes} " f"and level = 1 and prefix like '%[{sess}]' order by id;")] if sess else [] resultados.append(r) nav.close() saida = SAIDA / f"fluxos-{rotulo}.json" saida.write_text(json.dumps(resultados, ensure_ascii=False, indent=1), encoding="utf8") return resultados def _norm_comp(v): """Normaliza valores dependentes do ambiente (rotulo, timestamp do e-mail, PHPSESSID) para comparar.""" if isinstance(v, str): v = re.sub(r"(?i)migr83-fluxo\d\d-", "MIGR83-*-", v) v = re.sub(r"(?i)MIGR83-\*-\d{9,}", "MIGR83-*-{ts}", v) v = v.replace("/accia-php83/", "/accia/") v = re.sub(r"sessao=\w+", "sessao=*", v) return v if isinstance(v, list): return [_norm_comp(x) for x in v] if isinstance(v, dict): return {k: _norm_comp(x) for k, x in v.items()} return v def comparar(base: str = "fluxo74", alvo: str = "fluxo83") -> list: a = {r.get("funcao", r["fluxo"]): r for r in json.loads((SAIDA / f"fluxos-{base}.json").read_text(encoding="utf8"))} b = {r.get("funcao", r["fluxo"]): r for r in json.loads((SAIDA / f"fluxos-{alvo}.json").read_text(encoding="utf8"))} saida = [] for nome in list(dict.fromkeys(list(a) + list(b))): ra, rb = a.get(nome, {}), b.get(nome, {}) for r in (ra, rb): if r.get("erros_cm_log"): # compara a mensagem, sem caminho:linha e stack (versoes de vendor diferem) r["erros_cm_log"] = [re.sub(r" in /\S+:\d+ Stack trace:.*", "", e) for e in r["erros_cm_log"]] dif = [c for c in ("status_http", "resposta", "efeito", "email", "erros_cm_log") if _norm_comp(ra.get(c)) != _norm_comp(rb.get(c))] saida.append({"funcao": nome, "fluxo": (ra or rb).get("fluxo"), base: ra.get("ok"), alvo: rb.get("ok"), "igual": not dif, "diferencas": dif, "detalhe_base": ra.get("detalhe"), "detalhe_alvo": rb.get("detalhe")}) (SAIDA / "fluxos-comparacao.json").write_text(json.dumps(saida, ensure_ascii=False, indent=1), encoding="utf8") return saida if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--rotulo", choices=list(AMBIENTES)) ap.add_argument("--comparar", action="store_true") ap.add_argument("--so", help="nomes de funcao de fluxo separados por virgula (login sempre roda)") args = ap.parse_args() if args.comparar: for r in comparar(): print(("IGUAL " if r["igual"] else "DIFERE") + f" 74={r['fluxo74']} 83={r['fluxo83']} | {r['fluxo']} | " + ",".join(r["diferencas"])) else: if args.so: nomes = set(args.so.split(",")) FLUXOS[:] = [f for f in FLUXOS if f.__name__ in nomes or f is fluxo_login_sessao] for r in rodar_todos(args.rotulo): print(("OK " if r.get("ok") else "FALHA ") + r["fluxo"], "|", r.get("detalhe", ""))