Skip to content

feat(scripts): grafo de subsistema do ERP padrão + fechamento em N hops #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
adapcon-franklin-satler merged 1 commit into from
Jul 29, 2026
Merged

feat(scripts): grafo de subsistema do ERP padrão + fechamento em N hops #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 34 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,6 +137,40 @@ E o efeito prático: `ObterItemComprado()` deixa de apontar para o stub `CCTCPRG
a apontar para `VerDadosGerais()` em `7.5/csw75/rotinas/CCTCP/CCTCPRG001.mac:L833` — com
precisão de label, não de arquivo.

### Gerar o grafo de um subsistema do padrão

O irmão do script anterior, para o ERP padrão em vez de um cliente — escopo por prefixo de
módulo dentro de uma árvore de versão:

```powershell
.venv\Scripts\python.exe scripts\csw_grafo_modulo.py CCT --versao 8.1 # têxtil
.venv\Scripts\python.exe scripts\csw_grafo_modulo.py CCPV --versao 7.5 # pedido de venda
```

`CCT` pega os 99 diretórios `rotinas/CCT*` da 8.1 (11.386 rotinas). Sai em
`<out>\<VERSÃO>-<PREFIXO>`. Medido no têxtil: 12.511 arquivos com o fechamento → 360.905
nós, 1.144.309 arestas, 1,5% em stub, 315s.

Classes não entram por regra de nome: o mapeamento rotina→pacote é irregular (`CCTCO` tem
`classescls/TCo`, `CCTTGT` não tem `TTgt`), então quem decide é o fechamento — a classe
entra quando o código do módulo a referencia.

### Níveis de fechamento (`--hops`)

Os dois scripts aceitam `--hops N` (padrão 1). O hop 1 traz o que o *seu* código chama; o
hop 2 traz o que essas rotinas chamam. O que melhora é a proporção, não a contagem
absoluta — cada nível resolve o que o anterior pediu, e os arquivos novos têm a própria
fronteira:

| cliente GB | hop 1 | hop 2 |
|---|---|---|
| arquivos | 949 | 2.038 |
| arestas | 93.756 | 221.275 |
| em stub | 4.382 (4,7%) | 6.323 (2,9%) |

O resíduo nunca vai a zero: sobram nomes que não existem em árvore alguma — os placeholders
`*zzz` do framework (`$$ver^CCTCzzz`, 4.767 arestas no têxtil) e referências mortas.

### Consultar

```powershell
Expand Down
204 changes: 51 additions & 153 deletions scripts/csw_grafo_cliente.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,6 @@
from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
Expand All @@ -44,139 +43,49 @@

REPO = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO))
sys.path.insert(0, str(Path(__file__).resolve().parent))

WORKSPACE_PADRAO = Path(os.environ.get("GRAPHIFY_CSW_WORKSPACE", r"C:\workspacecsw\projetos"))
OUT_PADRAO = Path(os.environ.get("GRAPHIFY_CSW_OUT", r"C:\workspacecsw\graphify-csw"))
SUFFIXES = (".mac", ".cls", ".inc")
# Componentes por familia de versao (CLAUDE.md: 7.x -> cswutil70, 8.x -> cswutil80).
COMPONENTES = {"7": "COMP-7.0", "8": "COMP-8.0"}


# ── coleta e indexacao ───────────────────────────────────────────────────────

def coletar(raiz: Path, pular: tuple[str, ...] = ()) -> list[Path]:
"""Todos os fontes IRIS sob `raiz`, ignorando diretorios que casem `pular`."""
achados: list[Path] = []
for dirpath, _dirnames, filenames in os.walk(raiz):
low = dirpath.lower().replace("\\", "/")
if any(x in low for x in pular):
continue
for fn in filenames:
if os.path.splitext(fn)[1].lower() in SUFFIXES:
achados.append(Path(dirpath) / fn)
return achados


def canonico(nome: str) -> str:
"""`%CSW1UTI`, `_CSW1UTI` e `csw1uti` viram uma chave só — o `%` vira `_` no disco."""
return nome.lstrip("%_").casefold()


def indexar(raizes: list[Path], pular: tuple[str, ...] = ()) -> tuple[dict, dict]:
"""(rotinas/includes por nome canonico, classes pelo nome pontuado do path)."""
nomes: dict[str, list[Path]] = {}
classes: dict[str, list[Path]] = {}
for raiz in raizes:
if not raiz.is_dir():
continue
for p in coletar(raiz, pular):
stem, ext = os.path.splitext(p.name)
if ext.lower() == ".cls":
# classescls/Pkg_Sub/Nome.cls -> Pkg.Sub.Nome
pontuado = f"{p.parent.name.replace('_', '.')}.{stem}".casefold()
classes.setdefault(pontuado, []).append(p)
else:
nomes.setdefault(canonico(stem), []).append(p)
return nomes, classes


def resolver(externos: list[str], nomes: dict, classes: dict) -> tuple[set[Path], list[str]]:
"""(arquivos que atendem os nomes externos, nomes que ninguem atende)."""
arquivos: set[Path] = set()
faltam: list[str] = []
for nome in externos:
hits = nomes.get(canonico(nome)) or classes.get(nome.casefold()) or []
if hits:
arquivos.update(hits)
else:
faltam.append(nome)
return arquivos, faltam


def externos_de(resultado: dict) -> list[str]:
"""Nomes que o resolvedor nao achou no escopo — a lista do que falta trazer."""
return [
str(n.get("label", "")) for n in resultado.get("nodes", [])
if str(n.get("id", "")).startswith("ref") and n.get("label")
]


# ── versoes ──────────────────────────────────────────────────────────────────

def versoes_disponiveis(workspace: Path) -> list[str]:
return sorted(
d.name for d in workspace.iterdir()
if d.is_dir() and d.name[:1].isdigit() and "." in d.name
)


def _releases(pom: Path, sistema: str) -> list[tuple[int, ...]]:
"""Releases declarados para um `sistemaId` no pomcs.xml, como tuplas ordenaveis."""
import xml.etree.ElementTree as ET
try:
raiz = ET.parse(pom).getroot()
except (OSError, ET.ParseError):
return []
achados: list[tuple[int, ...]] = []
for dep in raiz.iter("dependencia"):
sid = dep.findtext("sistemaId") or ""
if sid.strip().casefold() != sistema:
continue
for rel in dep.iter("release"):
partes = (rel.text or "").strip().split(".")
if len(partes) >= 2 and all(p.isdigit() for p in partes[:2]):
achados.append(tuple(int(p) for p in partes if p.isdigit()))
return achados


def detectar_versao(cliente_dir: Path, workspace: Path) -> tuple[str, str]:
"""(versao do ERP, diretorio de componentes) declarados no pomcs.xml do cliente.
from csw_grafo_comum import ( # noqa: E402
arvores_padrao,
coletar,
contar_stubs,
escrever_grafo,
externos_de,
fechar,
releases_pomcs,
versoes_disponiveis,
)


def detectar_versao(cliente_dir: Path, workspace: Path) -> str:
"""Versao do ERP declarada no pomcs.xml do cliente.

Cada cliente roda a sua versao, e nada na arvore de fontes diz qual — os
`.vscode/settings.json` ficam nas pastas de VERSAO. Quem declara e o `pomcs.xml` do
cliente: `<sistemaId>csw</sistemaId>` lista os releases do ERP (o maior e o vigente)
e `<sistemaId>cswutil</sistemaId>` o dos componentes, que da a pasta `COMP-x.y`.
cliente: `<sistemaId>csw</sistemaId>` lista os releases do ERP e o maior e o vigente.

Nao tente adivinhar comparando os fontes: nome de rotina e estavel entre versoes, e
as quatro arvores do workspace resolvem exatamente os mesmos nomes — medido, 291 de
291 em todas. Sem o pomcs.xml a resposta e perguntar.
Nao tente adivinhar comparando os fontes: nome de rotina e estavel entre versoes, e as
quatro arvores do workspace resolvem exatamente os mesmos nomes — medido, 291 de 291 em
todas. Sem o pomcs.xml a resposta e perguntar.
"""
pom = cliente_dir / "pomcs.xml"
erp = comp = ""
if pom.is_file():
csw = _releases(pom, "csw")
util = _releases(pom, "cswutil")
if csw:
maior = max(csw)
erp = f"{maior[0]}.{maior[1]}"
if util:
maior_util = max(util)
comp = f"COMP-{maior_util[0]}.{maior_util[1]}"
if erp:
print(f" pomcs.xml declara: ERP {erp}"
+ (f", componentes {comp}" if comp else "")
+ f" (releases csw: {', '.join('.'.join(map(str, r)) for r in sorted(csw))})")
if not erp:
# Sem declaracao: perguntar, porque comparar fontes nao discrimina versao.
csw = releases_pomcs(pom, "csw") if pom.is_file() else []
if csw:
maior = max(csw)
erp = f"{maior[0]}.{maior[1]}"
print(f" pomcs.xml declara ERP {erp} "
f"(releases csw: {', '.join('.'.join(map(str, r)) for r in sorted(csw))})")
else:
disponiveis = versoes_disponiveis(workspace)
print(f" {pom.name} nao declara a versao do ERP.")
escolha = input(f" Versao do ERP {disponiveis} [7.5]: ").strip() or "7.5"
erp = escolha
if not comp:
comp = COMPONENTES.get(erp[:1], "COMP-7.0")
erp = input(f" Versao do ERP {disponiveis} [7.5]: ").strip() or "7.5"
if not (workspace / erp).is_dir():
raise SystemExit(f"erro: arvore da versao {erp} nao existe em {workspace}")
return erp, comp
return erp


# ── principal ────────────────────────────────────────────────────────────────
Expand All @@ -194,6 +103,8 @@ def main() -> int:
help=f"diretorio de saida (padrao: {OUT_PADRAO}\\<CONTA>)")
ap.add_argument("--sem-fechamento", action="store_true",
help="extrai so a customizacao (rapido; deixa ~31%% das arestas em stub)")
ap.add_argument("--hops", type=int, default=1, metavar="N",
help="niveis de fechamento (padrao 1; 2 fecha as dependencias do padrao)")
ap.add_argument("--cluster", action="store_true",
help="roda cluster-only ao final (comunidades + GRAPH_REPORT.md)")
ap.add_argument("--sequencial", action="store_true", help="desliga o pool de processos")
Expand Down Expand Up @@ -239,42 +150,29 @@ def main() -> int:
else:
# FASE 2 — de quem a customizacao depende, e onde isso mora.
print("[2/4] resolvendo o fechamento de dependencia...", flush=True)
if args.versao:
versao = args.versao
comp_nome = COMPONENTES.get(versao[:1], "COMP-7.0")
else:
versao, comp_nome = detectar_versao(cliente_dir, workspace)
erp = workspace / versao / f"csw{versao.replace('.', '')}"
comp = workspace / comp_nome
# Produtos dos customizadores (AD*, CX*, TT*) tambem sao chamados pela
# customizacao; entram excluindo `custom/` para nao arrastar outro cliente.
produtos = workspace / "DESENV"
nomes, classes = indexar([erp, comp, produtos], pular=("desenv/custom",))
fechamento, faltam = resolver(externos, nomes, classes)
print(f" ERP {versao} + {comp.name} + produtos DESENV -> "
f"{len(fechamento)} arquivos ({len(faltam)} nomes sem dono)", flush=True)

# FASE 3 — a unica passada que faz as arestas cruzarem. Os arquivos do cliente
# vem do cache da fase 1, entao aqui se paga so pelo fechamento.
alvos = sorted(set(arquivos_cliente) | fechamento)
print(f"[3/4] extraindo customizacao + fechamento ({len(alvos)} arquivos)...", flush=True)
t0 = time.time()
resultado = extract(alvos, cache_root=out, root=workspace, parallel=not args.sequencial)
print(f" {len(resultado['nodes'])} nos, {len(resultado['edges'])} arestas em "
f"{time.time() - t0:.0f}s", flush=True)
versao = args.versao or detectar_versao(cliente_dir, workspace)
# ERP + core + componentes da versao, mais os produtos dos customizadores
# (AD*, CX*, TT*), que a customizacao tambem chama. `desenv/custom` fica de fora
# para nao arrastar outro cliente para dentro deste grafo.
arvores = (arvores_padrao(workspace, versao, pom_cliente=cliente_dir / "pomcs.xml")
+ [workspace / "DESENV"])
print(" " + ", ".join(a.name for a in arvores if a.is_dir()), flush=True)

def extrair(lista):
print(f"[3/4] extraindo {len(lista)} arquivos...", flush=True)
t0 = time.time()
r = extract(lista, cache_root=out, root=workspace, parallel=not args.sequencial)
print(f" {len(r['nodes'])} nos, {len(r['edges'])} arestas em "
f"{time.time() - t0:.0f}s", flush=True)
return r

resultado, fechamento, faltam = fechar(
arquivos_cliente, parcial, arvores, extrair, hops=args.hops,
pular=("desenv/custom",))

# FASE 4 — grava no formato que o CLI do graphify le.
grafo = out / "graphify-out" / "graph.json"
grafo.write_text(
json.dumps({"nodes": resultado["nodes"], "edges": resultado["edges"],
"hyperedges": [], "input_tokens": 0, "output_tokens": 0},
ensure_ascii=False),
encoding="utf-8")
# Ancora o corpus real: e o que faz os path:linha do relatorio abrirem no workspace.
(out / "graphify-out" / ".graphify_root").write_text(str(workspace), encoding="utf-8")

ref_ids = {n["id"] for n in resultado["nodes"] if str(n["id"]).startswith("ref")}
em_stub = sum(1 for e in resultado["edges"] if e.get("target") in ref_ids)
grafo = escrever_grafo(out, resultado, workspace)
em_stub = contar_stubs(resultado)
mb = grafo.stat().st_size / 1024 / 1024
print(f"[4/4] gravado: {grafo} ({mb:.0f} MB)")

Expand Down
Loading