"""Fluxos de escrita do esg: 7.4 x 8.3, mesmo banco (psec_vialgpd local). Fluxos: 1. login/logout (persistencia de sessao) 2. create de wildcard (POST -> INSERT) 3. upload de planilha no GppTemp (PhpSpreadsheet -> INSERT) 4. relatorio Word (PhpWord 1.4 -> .docx) 5. e-mail (mailer symfony/mailer via console -> Mailpit) """ import json import re import subprocess import time import zipfile from io import BytesIO from pathlib import Path from playwright.sync_api import sync_playwright HOST = "esg.localhost" USUARIO = "99999999999" SENHA = "Gledson@2026" PREFIXO = "MIGR83" FIXTURE = Path(r"C:\Users\gleds\AppData\Local\Temp\opencode\esg-migr\gpp_MIGR83.xlsx") AMBIENTES = [("74", 8098, "esg_php74"), ("83", 8097, "esg_php83")] BANCO = "psec_vialgpd" def sql(query, banco=BANCO): cmd = ["docker", "exec", "mysql57", "mysql", "-uroot", "-prootpassword", "-N", "-B", banco, "-e", query] p = subprocess.run(cmd, capture_output=True, text=True) return p.stdout.strip() if p.returncode == 0 else f"ERRO: {p.stderr.strip()[:200]}" def mailpit(limite=10): try: p = subprocess.run(["docker", "exec", "mailpit_php83", "wget", "-qO-", f"http://localhost:8025/api/v1/messages?limit={limite}"], capture_output=True, text=True, timeout=30) dados = json.loads(p.stdout) return [{"subject": m.get("Subject"), "to": [t.get("Address") for t in m.get("To", [])]} for m in dados.get("messages", [])] except Exception as e: return [{"erro": str(e)}] def login(page, base): page.goto(base + "/site/login", wait_until="domcontentloaded") page.fill("input[name='LoginForm[username]']", USUARIO) page.fill("input[name='LoginForm[password]']", SENHA) page.click("form button[type=submit]") page.wait_for_load_state("networkidle") return "/site/login" not in page.url def fluxo_login_logout(page, base, porta, marca, container): ok = login(page, base) antes = sql("select count(*) from session where user_id=4") page.goto(base + "/site/logout", wait_until="domcontentloaded") page.wait_for_load_state("networkidle") depois = sql("select count(*) from session where user_id=4") return {"login_ok": ok, "sessoes_usuario_antes": antes, "sessoes_usuario_depois": depois, "url_final": page.url.replace(base, "")} def cookies_header(page): return "; ".join(f"{c['name']}={c['value']}" for c in page.context.cookies()) def fluxo_wildcard(page, base, porta, marca, container): login(page, base) tag = f"{PREFIXO}-{marca}-wildcard" sql(f"delete from wildcard where tag='{tag}'") page.goto(base + "/wildcard/create", wait_until="domcontentloaded") csrf = page.eval_on_selector("input[name='_csrf']", "e => e.value") resp = page.request.post( f"http://localhost:{porta}/wildcard/create", headers={"Host": HOST, "Cookie": cookies_header(page)}, form={"_csrf": csrf, "Wildcard[tag]": tag, "Wildcard[tipo]": "1", "Wildcard[status]": "1", "Wildcard[label]": f"label {PREFIXO}", "Wildcard[descricao]": f"descricao {PREFIXO} {marca}"}, max_redirects=0) linhas = sql(f"select tag, descricao, tipo, status from wildcard where tag='{tag}'") return {"status_http": resp.status, "linhas": linhas.split("\n") if linhas else [], "location": resp.headers.get("location", "")} def fluxo_gpp_upload(page, base, porta, marca, container): login(page, base) tabelas = ("gpp_temp", "gpp_causa_provavel", "gpp_acao_proposta", "gpp_tarefa") antes = {t: sql(f"select count(*) from {t}") for t in tabelas} page.goto(base + "/performance/gpp-temp/upload", wait_until="domcontentloaded") csrf = page.eval_on_selector("input[name='_csrf']", "e => e.value") dados = FIXTURE.read_bytes() resp = page.request.post( f"http://localhost:{porta}/performance/gpp-temp/upload", headers={"Host": HOST, "Cookie": cookies_header(page)}, multipart={ "_csrf": csrf, "GppTemp[excelFile]": { "name": "gpp_MIGR83.xlsx", "mimeType": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "buffer": dados, }, }, max_redirects=0) depois = {t: sql(f"select count(*) from {t}") for t in tabelas} causas = sql("select vertical_indicador_id, causa_provavel from gpp_causa_provavel order by id") acoes = sql("select indicador_id, causa_provavel, acao_proposta from gpp_acao_proposta order by id") tarefas = sql("select count(*) from gpp_tarefa") return {"status_http": resp.status, "antes": antes, "depois": depois, "causas": causas.split("\n") if causas else [], "acoes": acoes.split("\n") if acoes else [], "tarefas": tarefas} def fluxo_word(page, base, porta, marca, container): login(page, base) id_dados = sql("select min(id) from inventario_dados").split("\n")[0] resp = page.request.get(f"http://localhost:{porta}/site/word-old?id={id_dados}", headers={"Host": HOST, "Cookie": "; ".join( f"{c['name']}={c['value']}" for c in page.context.cookies())}) corpo = resp.body() info = {"id": id_dados, "status": resp.status, "bytes": len(corpo), "tipo": resp.headers.get("content-type", "")} try: z = zipfile.ZipFile(BytesIO(corpo)) doc = z.read("word/document.xml").decode("utf-8", errors="replace") texto = re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", doc)).strip() info["zip_ok"] = True info["partes"] = len(z.namelist()) info["texto_bytes"] = len(texto) except Exception as e: info["zip_ok"] = False info["erro"] = f"{type(e).__name__}: {e}" info["inicio"] = corpo[:150].decode("utf-8", errors="replace") return info def fluxo_email(page, base, porta, marca, container): destino = f"migr83-{marca}@teste.local" p = subprocess.run(["docker", "exec", container, "php", "/var/www/html/esg-php83/tools/migracao-php83/php/testa_mailer.php", destino], capture_output=True, text=True, timeout=120) return {"saida": p.stdout.strip()[:200], "erro": p.stderr.strip()[:200], "destino": destino} FLUXOS = [ ("login_logout", fluxo_login_logout), ("wildcard_create", fluxo_wildcard), ("gpp_upload", fluxo_gpp_upload), ("relatorio_word", fluxo_word), ("email", fluxo_email), ] MARCAS = {"74": "74", "83": "83"} def normaliza(obj, marca): """remove marcas e versoes de PHP para comparar 7.4 x 8.3""" s = json.dumps(obj, sort_keys=True, ensure_ascii=False) s = s.replace(PREFIXO, "MARCA") s = s.replace(marca, "ROT") s = re.sub(r"PHP \d+\.\d+\.\d+", "PHP X", s) s = re.sub(r'"bytes": \d+', '"bytes": N', s) s = re.sub(r"\d+", "N", s) return s def main(): resultados = {} for rotulo, porta, container in AMBIENTES: resultados[rotulo] = {} with sync_playwright() as p: br = p.chromium.launch(args=[f"--host-resolver-rules=MAP {HOST} 127.0.0.1:{porta}"]) base = f"http://{HOST}" for nome, fn in FLUXOS: page = br.new_context().new_page() try: t0 = time.time() r = fn(page, base, porta, rotulo, container) resultados[rotulo][nome] = r print(f"[{rotulo}] {nome}: {json.dumps(r, ensure_ascii=False)[:400]}", flush=True) except Exception as e: resultados[rotulo][nome] = {"erro": f"{type(e).__name__}: {e}"} print(f"[{rotulo}] {nome}: ERRO {type(e).__name__}: {e}", flush=True) finally: page.close() br.close() time.sleep(2) print("\n=== comparacao 74 x 83 (normalizada) ===") for nome, _ in FLUXOS: a = normaliza(resultados["74"].get(nome), "74") b = normaliza(resultados["83"].get(nome), "83") print(f"{nome}: {'IGUAL' if a == b else 'DIFERENTE'}") if a != b: print(f" 74: {a[:500]}") print(f" 83: {b[:500]}") print("\n=== e-mails recebidos no Mailpit ===") for m in mailpit(): print(" ", m) Path(r"C:\Users\gleds\AppData\Local\Temp\opencode\esg-migr\fluxos_result.json").write_text( json.dumps(resultados, indent=2, ensure_ascii=False), encoding="utf-8") if __name__ == "__main__": main()