""" Gera o dossie de contexto de uma aplicacao Oracle APEX, em Markdown. python3.8 contexto_apex.py --app=103 \ --packages=FINANCEIRO,PKG_OC_APROVACAO,PKG_ACESSO \ --out= Os datasets dizem O QUE ACONTECEU. Este script extrai O QUE O SISTEMA FAZ: os modulos que a aplicacao expoe, e as regras de negocio que vivem no PL/SQL e em lugar nenhum do schema. POR QUE ISSO IMPORTA -------------------- Num ERP tipico o schema quase nao tem restricao -- um deles declarava, no proprio codigo, "so 3 CHECK de dominio em 340 tabelas". O dominio de um status, a ordem de um fluxo de aprovacao e a condicao que barra uma operacao moram no codigo. Um agente que so ve linhas nao alcanca nada disso. O QUE ELE NAO FAZ ----------------- Nao interpreta, nao resume regra e nao deduz significado: **cita**. A especificacao de um package e o contrato que o proprio time escreveu, e vai para o documento como esta. O que este script deriva -- contagens, ligacoes entre pagina e package -- aparece separado e rotulado como derivado. Isso e deliberado. Um resumo meu de uma regra de negocio vira, no documento indexado, uma afirmacao com a mesma autoridade do codigo -- e ninguem revisa o que parece confirmado. O dossie e do ERP, nao de um cliente: indexe com `--scope=tenant` para que todos os projetos herdem. """ import argparse import io import os import re import sys def _texto_longo(p_conexao): """APEX guarda o corpo dos processos em LONG; sem isto o driver trunca.""" import oracledb # type: ignore p_conexao.outputtypehandler = ( lambda cur, n, t, s, p, sc: cur.var(oracledb.DB_TYPE_LONG, arraysize=cur.arraysize) if t == oracledb.DB_TYPE_LONG else None ) def _linhas(p_cursor, p_sql, **binds): p_cursor.execute(p_sql, binds) return p_cursor.fetchall() def escrever(p_dir, p_nome, p_conteudo): caminho = os.path.join(p_dir, p_nome) with io.open(caminho, 'w', encoding='utf-8') as f: f.write(p_conteudo) return caminho def doc_aplicacao(cur, p_app): app = _linhas(cur, """SELECT application_name, alias, pages, last_updated_on, version FROM APEX_APPLICATIONS WHERE application_id=:a""", a=p_app) if not app: raise SystemExit('Aplicacao %s nao existe ou nao e visivel para este usuario.' % p_app) nome, alias, pags, atualizada, versao = app[0] grupos = _linhas(cur, """SELECT NVL(page_group,'(sem grupo)'), COUNT(*) FROM APEX_APPLICATION_PAGES WHERE application_id=:a GROUP BY page_group ORDER BY 2 DESC""", a=p_app) proc = _linhas(cur, """SELECT process_type, COUNT(*) FROM APEX_APPLICATION_PAGE_PROC WHERE application_id=:a GROUP BY process_type ORDER BY 2 DESC""", a=p_app) itens = _linhas(cur, """SELECT COUNT(*) FROM APEX_APPLICATION_PAGE_ITEMS WHERE application_id=:a""", a=p_app) s = ['# Aplicação %s — "%s"\n' % (p_app, nome), '> Extraído da metadata do próprio APEX. Descreve **o que o sistema faz**;', '> o que aconteceu nele está nos datasets.\n', '| | |', '|---|---|', '| Aplicação | %s (`%s`), id %s |' % (nome, alias, p_app), '| Páginas | %s |' % pags, '| Itens de tela | %s |' % itens[0][0], '| Última alteração | %s |' % str(atualizada)[:10], '| Versão | %s |' % (versao or '—'), '\n## Módulos, como a própria aplicação os declara\n', 'Os grupos abaixo são o `PAGE_GROUP` do APEX — a taxonomia que o time do ERP', 'escreveu, não uma classificação nossa.\n', '| Módulo | Páginas |', '|---|---|'] s += ['| %s | %s |' % (g, n) for g, n in grupos] # Cobertura: quantas paginas o mapa de modulos realmente alcanca. Sem este # aviso o mapa parece o sistema inteiro, e nao e. sem = next((n for g, n in grupos if g == '(sem grupo)'), 0) total = sum(n for _, n in grupos) if sem: cobertas = total - sem sistema = _linhas(cur, """SELECT COUNT(*) FROM APEX_APPLICATION_PAGES WHERE application_id=:a AND page_group IS NULL AND page_id < 10""", a=p_app)[0][0] s += ['', '> **Este mapa cobre %d das %d páginas (%d%%).** As outras %d não têm ' '`PAGE_GROUP`' % (cobertas, total, round(100.0 * cobertas / total), sem), '> preenchido, e apenas %d delas são páginas de sistema (id < 10). O resto é' % sistema, '> funcionalidade de negócio que simplesmente não foi classificada pelo time —', '> incluindo telas como reajuste, cobrança e conferência. **Não conclua que um', '> assunto não existe no ERP só porque não aparece na tabela acima**; procure', '> em `10-telas.md`, na seção `(sem grupo)`.'] s += ['\n## O que as páginas executam\n', 'Tipo de processo, e quantos existem. `PL/SQL anonymous block` é onde mora', 'lógica escrita à mão — é o número que diz o quanto do comportamento **não**', 'está em componente declarativo.\n', '| Tipo de processo | Quantos |', '|---|---|'] s += ['| %s | %s |' % ((t or '?'), n) for t, n in proc] return '\n'.join(s) + '\n' def doc_modulos(cur, p_app): linhas = _linhas(cur, """SELECT NVL(page_group,'(sem grupo)'), page_id, page_name, page_mode FROM APEX_APPLICATION_PAGES WHERE application_id=:a ORDER BY NVL(page_group,'zzz'), page_id""", a=p_app) s = ['# Mapa das telas — aplicação %s\n' % p_app, 'Para responder "onde isso é feito no sistema" e "que assunto o ERP cobre".', 'Página `Modal Dialog` costuma ser formulário de edição chamado de outra tela.\n', '> A seção **`(sem grupo)`** não é sobra: é quase metade das telas, com', '> funcionalidade de negócio real que o time não classificou. Vale ler junto', '> com os módulos nomeados, nunca no lugar deles.\n'] atual = None for grupo, pid, nome, modo in linhas: if grupo != atual: atual = grupo s.append('\n## %s\n' % grupo) s.append('| Página | Nome | Tipo |') s.append('|---|---|---|') s.append('| %s | %s | %s |' % (pid, (nome or '').replace('|', '\\|'), modo or '')) return '\n'.join(s) + '\n' def doc_package(cur, p_app, p_pkg): def fonte(tipo): cur.execute("SELECT text FROM user_source WHERE name=:n AND type=:t ORDER BY line", {'n': p_pkg, 't': tipo}) return ''.join(r[0] or '' for r in cur) spec, corpo = fonte('PACKAGE'), fonte('PACKAGE BODY') if not spec: return None paginas = _linhas(cur, """SELECT p.page_id, MAX(pg.page_name), MAX(pg.page_group) FROM APEX_APPLICATION_PAGE_PROC p JOIN APEX_APPLICATION_PAGES pg ON pg.application_id=p.application_id AND pg.page_id=p.page_id WHERE p.application_id=:a AND UPPER(p.process_source) LIKE :k GROUP BY p.page_id ORDER BY p.page_id""", a=p_app, k='%%%s.%%' % p_pkg) # Constantes de dominio: `gc_x constant varchar2(n) := 'Valor';` dominio = re.findall(r"(\w+)\s+constant\s+varchar2\s*\(\s*\d+\s*\)\s*:=\s*'([^']+)'", spec, re.I) # Mensagem de erro = a regra dita em portugues, pelo proprio time. # # Muitas sao montadas com `|| variavel ||`, entao o literal termina no meio da # frase. Marcar isso importa: uma frase cortada apresentada como completa faz # o leitor -- humano ou agente -- concluir uma regra que o codigo nao diz. erros = [] for m in re.findall(r"raise_application_error\s*\(\s*-?\d+\s*,\s*'([^']{25,200})'", corpo, re.I): m = m.strip() if m and m[-1] not in '.!?': m += ' …_(continua com um valor da execução)_' erros.append(m) s = ['# Regras de negócio: `%s`\n' % p_pkg, '> **Fonte: o código do ERP.** A especificação abaixo é o contrato que o time', '> do ERP escreveu, reproduzido como está. Não é interpretação nossa.\n', '| | |', '|---|---|', '| Especificação | %d linhas |' % len(spec.splitlines()), '| Corpo | %d linhas |' % len(corpo.splitlines()), '| Telas que chamam | %d |' % len(paginas)] if paginas: s += ['\n## Onde a aplicação usa estas regras\n', '| Página | Nome | Módulo |', '|---|---|---|'] s += ['| %s | %s | %s |' % (pid, (n or ''), g or '—') for pid, n, g in paginas] if dominio: s += ['\n## Domínio de valores declarado no código\n', 'Estes são os valores que o ERP considera válidos. Não estão no schema —', 'e é por isso que o agente não os alcançaria pelos dados.\n', '| Constante | Valor |', '|---|---|'] s += ['| `%s` | `%s` |' % (k, v) for k, v in dominio] if erros: s += ['\n## O que o sistema recusa, nas palavras dele\n', 'Mensagens de erro do próprio código. Cada uma é uma regra de negócio', 'dita em português. As marcadas com **…** são montadas em tempo de execução', 'e continuam com um valor concreto (um número de OC, um nome de status).\n'] s += ['- %s' % e for e in dict.fromkeys(erros)] s += ['\n## Especificação, íntegra\n', 'Reproduzida sem edição: os comentários do time explicam decisões que', 'nenhuma outra fonte registra.\n', '```sql', spec.rstrip(), '```'] return '\n'.join(s) + '\n' def doc_lacunas(cur, p_app, p_pkgs, p_todas_apps): obj = _linhas(cur, """SELECT object_type, COUNT(*) FROM user_objects WHERE object_type IN ('TRIGGER','PROCEDURE','FUNCTION','PACKAGE') GROUP BY object_type ORDER BY 2 DESC""") s = ['# O que este dossiê NÃO cobre\n', 'Registrado de propósito: um documento indexado que parece completo é pior', 'que um que declara o próprio limite.\n', '## Outras aplicações no mesmo APEX\n', '| Id | Nome | Alias | Páginas | Última alteração | Incluída? |', '|---|---|---|---|---|---|'] for aid, nome, alias, pags, atu in p_todas_apps: s.append('| %s | %s | %s | %s | %s | %s |' % (aid, nome, alias, pags, str(atu)[:10], '**sim**' if aid == p_app else 'não')) s += ['\n## Lógica PL/SQL fora dos packages documentados\n', 'O schema tem muito mais código do que os %d packages deste dossiê:\n' % len(p_pkgs), '| Tipo | Objetos |', '|---|---|'] s += ['| %s | %s |' % (t, n) for t, n in obj] s += ['', 'Os **triggers** são a maior lacuna: disparam sem ninguém pedir, e por isso', 'são a regra mais invisível de todas. Um agente que explique por que um valor', 'mudou não tem como saber deles por aqui.', '', 'Frameworks de terceiros (`AOP_*` para impressão, `BNR_*`, `RWD_EMAIL`,', '`UTL_HTTP_MULTIPART`, `XML_IMP`) ficaram fora por não conterem regra de', 'negócio do cliente.'] return '\n'.join(s) + '\n' # --------------------------------------------------------------------------- # Regras que moram no schema, nao na aplicacao # --------------------------------------------------------------------------- # Trigger cujo nome revela encanamento, nao regra de negocio. Documentar as 83 # uma a uma afogaria as poucas que importam. ENCANAMENTO = ( ('_ID_TRIG', 'atribui o ID pela sequence'), ('_AUD', 'trilha de auditoria'), ('BI_', 'valor padrao na inserção'), ) def _classificar(p_nome): """Devolve (e_encanamento, motivo).""" for marca, motivo in ENCANAMENTO: if (p_nome.endswith(marca) if marca.startswith('_') else p_nome.startswith(marca)): # BI_*_TNT parece encanamento pelo prefixo, mas atribui a EMPRESA -- # e a regra que define de quem e cada linha. Nao e encanamento. if p_nome.endswith('_TNT'): return False, None return True, motivo return False, None def doc_triggers(cur, p_tabelas): trg = _linhas(cur, """SELECT table_name, trigger_name, trigger_type, triggering_event, status FROM user_triggers WHERE table_name IN (%s) ORDER BY table_name, trigger_name""" % ','.join(':t%d' % i for i in range(len(p_tabelas))), **{('t%d' % i): t for i, t in enumerate(p_tabelas)}) negocio = [t for t in trg if not _classificar(t[1])[0]] plumbing = [t for t in trg if _classificar(t[1])[0]] desligados = [t for t in trg if (t[4] or '').upper() != 'ENABLED'] s = ['# Regras que disparam sozinhas (triggers)\n', '> Estas regras rodam **sem ninguém pedir**, em toda inserção ou alteração.', '> São a parte mais invisível do ERP: não aparecem em tela, não aparecem no', '> schema, e explicam por que um valor mudou quando nada óbvio o mudou.\n', '| | |', '|---|---|', '| Tabelas que sustentam os datasets | %d |' % len(p_tabelas), '| Triggers nelas | %d |' % len(trg), '| Com regra de negócio | %d |' % len(negocio), '| Desativadas | %d |' % len(desligados)] if desligados: s += ['\n## Desativadas — leia primeiro\n', 'Existem no banco e **não executam**. Quem lê o código conclui que a regra', 'vale; ela não vale. É a divergência mais cara deste documento.\n', '| Tabela | Trigger | Evento |', '|---|---|---|'] s += ['| `%s` | `%s` | %s |' % (t[0], t[1], t[3]) for t in desligados] s += ['\n## Com regra de negócio\n', '| Tabela | Trigger | Quando | Evento | Ativa? |', '|---|---|---|---|---|'] for tab, nome, tipo, ev, st in negocio: s.append('| `%s` | `%s` | %s | %s | %s |' % (tab, nome, (tipo or '').replace(' EACH ROW', ''), ev, 'sim' if (st or '').upper() == 'ENABLED' else '**NÃO**')) s += ['\n## Encanamento\n', 'Agrupadas de propósito: atribuem ID, gravam auditoria ou preenchem valor', 'padrão. Citá-las uma a uma afogaria as de cima.\n', '| Trigger | O que faz |', '|---|---|'] vistos = set() for tab, nome, tipo, ev, st in plumbing: motivo = _classificar(nome)[1] if motivo in vistos: continue vistos.add(motivo) s.append('| padrão `%s` | %s |' % (nome, motivo)) s.append('') s.append('São %d triggers no total nesse grupo.' % len(plumbing)) return '\n'.join(s) + '\n' def doc_fidelidade(cur, p_datasets): """ O que cada view do ERP deixa de fora, apenas onde isso e VERIFICAVEL. A tentacao aqui e publicar "a view esconde N% da tabela base". Nao publique: a diferenca entre a tabela e a view mistura duas coisas incomparaveis -- linha perdida por acidente e linha ausente de proposito. Uma view que so deve trazer UM tipo de lancamento tem dezenas de linhas contra dezenas de milhares na tabela base; chamar isso de "100% escondido" faz o agente concluir que perdeu a base inteira. Entao o documento traz o numero que se sustenta -- a linha cuja empresa nao existe mais -- e, para o resto, mostra a ESTRUTURA da view (os INNER JOIN e o WHERE) para que uma pessoa julgue. Melhor uma pergunta bem colocada que uma metrica errada com cara de certa. """ s = ['# O que cada view do ERP deixa de fora\n', '> Os datasets não vêm das tabelas: vêm de views do ERP. Uma view pode', '> **filtrar de propósito** (comissões só traz comissão) ou **perder linha', '> sem querer** (um `INNER JOIN` numa coluna vazia derruba o registro).', '> As duas coisas somem igual, sem erro — e por isso estão separadas aqui.\n'] orfas, estrutura = [], [] for ds in p_datasets: vw = ds['source_object'] texto = _linhas(cur, "SELECT text FROM user_views WHERE view_name=:v", v=vw) if not texto: continue corpo = str(texto[0][0]) achou = re.search(r'\bFROM\s+([A-Z0-9_$]+)', corpo, re.I | re.S) if not achou: continue alvo = achou.group(1).upper() inners = len(re.findall(r'\bINNER\s+JOIN\b', corpo, re.I)) lefts = len(re.findall(r'\bLEFT\s+JOIN\b', corpo, re.I)) onde = re.search(r'\bWHERE\b(.*)$', corpo, re.I | re.S) filtro = ' '.join(onde.group(1).split()) if onde else '' nv = _linhas(cur, 'SELECT COUNT(*) FROM %s' % vw)[0][0] nb = _linhas(cur, 'SELECT COUNT(*) FROM %s' % alvo)[0][0] # `x.TENANTID` SEMPRE qualificado: TENANTS tem uma coluna chamada # TENANTID, e um TENANTID solto aqui liga NELA -- a comparacao vira # t.ID = t.TENANTID e todo registro parece orfao. tem = _linhas(cur, """SELECT COUNT(*) FROM user_tab_columns WHERE table_name=:t AND column_name='TENANTID'""", t=alvo)[0][0] # So faz sentido perguntar por orfao quando a VIEW realmente junta com # TENANTS. Sem essa junção a linha nao e descartada por isso, e o numero # so confundiria. Tabela de dominio (status, tipos) cai aqui. junta = re.search(r'JOIN\s+TENANTS\b', corpo, re.I) is not None ja_vista = any(t == alvo for _, t, _, _ in orfas) if tem and junta and not ja_vista: n = _linhas(cur, """SELECT COUNT(*) FROM %s x WHERE NOT EXISTS (SELECT 1 FROM TENANTS t WHERE t.ID = x.TENANTID)""" % alvo)[0][0] # n == nb significa que a coluna nao e chave para TENANTS (a propria # TENANTS tem uma coluna TENANTID, que nao aponta para ela mesma). # Publicar "100% orfao" seria inventar um problema. if n and n < nb: orfas.append((ds['dataset'], alvo, nb, n)) estrutura.append((ds['dataset'], alvo, nb, nv, inners, lefts, filtro)) if orfas: s += ['## Linhas de empresas que não existem mais\n', 'Este número **se sustenta**: a linha tem `TENANTID`, e esse tenant não', 'está em `TENANTS`. Toda view usa `INNER JOIN TENANTS`, então nada disso', 'chega ao agente. São 913 empresas que já passaram pelo ERP — mas só duas', 'delas deixaram títulos financeiros para trás.\n', '| Tabela | Linhas na tabela | De empresa removida | |', '|---|---|---|---|'] s += ['| `%s` | %s | **%s** | %d%% |' % (t, nb, n, round(100.0 * n / nb)) for _, t, nb, n in orfas] s += ['', 'Excluí-las costuma estar certo — a empresa saiu. Vale saber que existem', 'quando alguém comparar um total do Contextia com um relatório antigo do ERP.'] s += ['\n## Como cada view é construída\n', 'Cada `INNER JOIN` é um ponto onde a linha some se a chave estiver vazia.', '**Isto não é uma medição de perda**: a diferença entre a tabela e a view', 'mistura filtro proposital com perda acidental, e separar as duas exige', 'conhecer a regra do ERP. Use a tabela para saber ONDE perguntar.\n', '| Dataset | Tabela | Tabela | View | `INNER` | `LEFT` | Filtro declarado |', '|---|---|---|---|---|---|---|'] for d, t, nb, nv, inn, lef, filtro in estrutura: s.append('| `%s` | `%s` | %s | %s | %s | %s | %s |' % (d, t, nb, nv, inn, lef, ('`%s`' % filtro[:60]) if filtro else '—')) s += ['', '> **Onde olhar primeiro:** a view com muitos `INNER JOIN` e sem filtro', '> declarado. Ali a diferença não foi pedida por ninguém — é uma chave', '> estrangeira vazia derrubando um registro legítimo.'] return '\n'.join(s) + '\n' def main(): p = argparse.ArgumentParser(description='Gera o dossie de contexto de uma app Oracle APEX.') p.add_argument('--app', required=True, type=int) p.add_argument('--packages', default='', help='packages de negocio, separados por virgula') p.add_argument('--source', default='generic', help='modulo em sources/') p.add_argument('--mapping', default=None, help='mapeamento do cliente: acrescenta os documentos de schema' ' (triggers e fidelidade das views)') p.add_argument('--out', required=True, help='diretorio de saida') args = p.parse_args() sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), 'sources')) import importlib origem = importlib.import_module(args.source) conexao = origem.conectar() _texto_longo(conexao) cur = conexao.cursor() if not os.path.isdir(args.out): os.makedirs(args.out) todas = _linhas(cur, """SELECT application_id, application_name, alias, pages, last_updated_on FROM APEX_APPLICATIONS ORDER BY application_id""") pkgs = [x.strip().upper() for x in args.packages.split(',') if x.strip()] escritos = [escrever(args.out, '00-aplicacao.md', doc_aplicacao(cur, args.app)), escrever(args.out, '10-telas.md', doc_modulos(cur, args.app))] for i, pkg in enumerate(pkgs): conteudo = doc_package(cur, args.app, pkg) if conteudo is None: print(' AVISO: package %s nao encontrado, pulado.' % pkg) continue escritos.append(escrever(args.out, '%d-regras-%s.md' % (20 + i, pkg.lower()), conteudo)) if args.mapping: import json as _json mapa = _json.load(io.open(args.mapping, encoding='utf-8')) views = [d['source_object'] for d in mapa['datasets']] tabelas = sorted(r[0] for r in _linhas( cur, """SELECT DISTINCT referenced_name FROM user_dependencies WHERE name IN (%s) AND referenced_type='TABLE'""" % ','.join(':v%d' % i for i in range(len(views))), **{('v%d' % i): v for i, v in enumerate(views)})) escritos.append(escrever(args.out, '30-regras-do-schema.md', doc_triggers(cur, tabelas))) escritos.append(escrever(args.out, '31-o-que-as-views-escondem.md', doc_fidelidade(cur, mapa['datasets']))) escritos.append(escrever(args.out, '90-lacunas.md', doc_lacunas(cur, args.app, pkgs, todas))) for c in escritos: print(' %8d bytes %s' % (os.path.getsize(c), c)) print('\n%d documentos em %s' % (len(escritos), args.out)) print('Indexe no ESCOPO DO TENANT -- e regra do ERP, nao dado de um cliente:') print(' php bin/mcp source:add --tenant= --scope=tenant \\') print(' --type=directory --name=aplicacao-apex --path=%s --recursive' % args.out) print(' php bin/mcp index --tenant= --scope=tenant') if hasattr(origem, 'fechar'): origem.fechar() return 0 if __name__ == '__main__': sys.exit(main())