trabalho_extra_gces

Imported from Atyrson/trabalho_extra_gces (AGENTS.md). Install upstream with npx skills add Atyrson/trabalho_extra_gces. Copyright stays with the author.

Pipeline UDA — Setor Habitacional

Pipeline de Análise de Dados Não Estruturados que coleta Prévias Operacionais em PDF dos portais de RI de incorporadoras brasileiras, extrai métricas operacionais via LLM, e as serve por uma API REST. A especificação completa do sistema está em docs/especificacao_pipeline_uda.md — leia antes de qualquer implementação.

Spec-Driven Development

Este projeto usa spec-driven development. O fluxo de trabalho obrigatório é:

Ler spec → Propor plano → Aguardar aprovação → Implementar → Verificar critérios de aceite

NUNCA escreva código sem antes:

1. Ler a seção relevante da spec em docs/especificacao_pipeline_uda.md.
2. Declarar em linguagem natural o que será implementado e por qual seção da spec.
3. Listar os critérios de aceite que a implementação deverá satisfazer.

Se houver ambiguidade entre a spec e os requisitos do usuário na sessão atual, pergunte antes de agir.

Setup

# Dependências Python
pip install -r requirements.txt

# Playwright (necessário para scraping de portais com Cloudflare)
playwright install chromium

# Variáveis de ambiente
cp .env.example .env
# Edite .env com suas chaves de API

# Banco de dados
python -m storage.migrations.run

# Verificar instalação
python -m pytest tests/ -x -q

Dependências críticas e versões fixas:

python>=3.11
pymupdf>=1.24.0          # parser de PDF — use sempre como `import fitz`
pydantic>=2.0.0          # validação do contrato semântico
fastapi>=0.111.0         # camada de API
sqlalchemy>=2.0.0        # ORM — use sempre sintaxe 2.x (não legado 1.x)
playwright>=1.44.0       # scraping anti-WAF
httpx>=0.27.0            # cliente HTTP assíncrono (não requests)
apscheduler>=3.10.0      # scheduler do crawler

Comandos de Desenvolvimento

# Rodar todos os testes
python -m pytest tests/ -v

# Rodar teste específico
python -m pytest tests/test_llm_extraction.py::test_vso_extraido -v

# Rodar apenas testes de uma camada
python -m pytest tests/test_parser.py -v        # Camada 2
python -m pytest tests/test_api.py -v           # Camada 3
python -m pytest tests/test_idempotency.py -v   # Controle de duplicidade

# Lint
ruff check .

# Typecheck
mypy . --ignore-missing-imports

# Subir a API em modo desenvolvimento
uvicorn api.main:app --reload --port 8000

# Rodar o crawler manualmente (uma varredura)
python -m crawler.scheduler --once

# Ingestão manual de um PDF para teste
python -m crawler.scraper --url "https://..." --empresa-id mrv

Arquitetura

O sistema tem três camadas obrigatórias mais um catálogo transversal:

Camada 1 — crawler/          Extração automatizada (polling + anti-WAF + hash)
Camada 2 — processing/       Motor UDA: parse PDF → chunking → LLM → Pydantic
Camada 3 — api/              API REST FastAPI
Transversal — storage/       Catálogo de linhagem + operational_reports (PostgreSQL)

Cada PDF percorre este fluxo exato:

scheduler → scraper → hash_checker
                           │ hash novo
                      pdf_parser → chunker (se >15 páginas)
                                       │
                                  llm_client → prompt_builder → schema.OperationalReport
                                                                        │ Pydantic válido
                                                                   catalog.save() + upsert

Mapa de arquivos:

pipeline-uda/
├── AGENTS.md
├── docs/
│   └── especificacao_pipeline_uda.md   ← fonte da verdade
├── sources.yaml                        ← configuração das fontes de RI
├── .env.example
├── requirements.txt
├── crawler/
│   ├── scheduler.py       # APScheduler — poll por empresa em POLL_INTERVAL_HOURS
│   ├── scraper.py         # httpx first; escalona para browser.py em 403/429
│   ├── browser.py         # wrapper Playwright para portais com WAF
│   └── hash_checker.py    # SHA-256 do binário → consulta data_catalog
├── processing/
│   ├── pdf_parser.py      # fitz (PyMuPDF) — extrai texto preservando estrutura de tabelas
│   ├── chunker.py         # divide por títulos/seções quando num_pages > 15
│   ├── llm_client.py      # wrapper da API LLM — temperatura 0, max_tokens 1500
│   ├── prompt_builder.py  # monta system prompt + user prompt com contexto do PDF
│   └── schema.py          # OperationalReport (Pydantic) — o contrato semântico
├── storage/
│   ├── models.py          # ORM SQLAlchemy 2.x (DataCatalog, OperationalReport)
│   ├── catalog.py         # operações de linhagem — save, get_by_hash, update_status
│   └── migrations/
│       └── run.py         # cria tabelas via models.Base.metadata.create_all()
├── api/
│   ├── main.py            # FastAPI app + lifespan
│   ├── routes/
│   │   ├── conjuntura.py  # GET /api/conjuntura
│   │   ├── catalogo.py    # GET /api/catalogo
│   │   └── ingestao.py    # POST /api/ingestao/manual
│   └── schemas.py         # Pydantic models de request/response da API
└── tests/
    ├── fixtures/           # PDFs reais de pelo menos 2 empresas com layouts diferentes
    ├── test_parser.py
    ├── test_llm_extraction.py
    ├── test_api.py
    └── test_idempotency.py

Convenções de Código

Geral

• Python 3.11+. Type hints em todas as funções públicas. Sem Any desnecessário.
• async/await em toda a Camada 1 (crawler) e chamadas ao LLM. A API FastAPI é assíncrona.
• Use httpx.AsyncClient para requisições HTTP — nunca requests (síncrono).
• Logs estruturados em JSON via structlog. Nunca use print() fora de scripts utilitários.
• Variáveis de ambiente via pydantic_settings.BaseSettings — nunca os.getenv() direto.

Banco de Dados

• SQLAlchemy 2.x com sintaxe select() — nunca a sintaxe legada session.query().
• Toda escrita em operational_reports deve usar Upsert (ON CONFLICT DO UPDATE). O INSERT simples está proibido nessa tabela por risco de violação de constraint UNIQUE (empresa, ano, trimestre).
• Migrations via models.Base.metadata.create_all() durante o desenvolvimento. Em produção, usar Alembic.

Contrato Semântico (Camada 2)

• O schema Pydantic OperationalReport em processing/schema.py é imutável durante a sessão — não altere campos sem antes atualizar a spec e os testes.
• Campos Optional devem ser None quando ausentes. Nunca preencher com 0, "" ou strings placeholder.
• O campo vso é o único campo do schema que aceita um valor percentual — todos os outros campos numéricos são valores absolutos.
• Respostas inválidas do LLM disparam retry (máximo LLM_MAX_RETRIES vezes). Após esgotar retries, registrar status = 'extraction_failed' no catálogo com o texto bruto da resposta em erro_detalhe.

Scraper

• Toda fonte começa com estratégia requests via scraper.py.
• Se a resposta for 403 ou 429, escalonar automaticamente para browser.py (Playwright).
• O campo scraper_mode em sources.yaml pode forçar Playwright direto para uma fonte específica.
• Rotacionar User-Agent a cada requisição usando a lista em crawler/scraper.py.
• Nunca hardcodar lógica por empresa (if empresa_id == 'mrv': ...). Toda variação deve vir de sources.yaml.

API

• Filtros de query são case-insensitive. Use ilike no SQLAlchemy para buscas por string.
• GET /api/conjuntura sem resultados retorna 200 OK com {"total": 0, "data": []} — nunca 404.
• Erros de validação de parâmetros retornam 400 com o formato padrão {"error": ..., "message": ..., "field": ...}.
• Nunca expor DATABASE_URL ou stack traces completos em respostas de erro.

Critérios de Aceite Obrigatórios por Camada

Antes de declarar qualquer implementação concluída, verifique os itens abaixo. São extraídos diretamente da spec e representam os testes que o avaliador vai rodar.

Camada 1 — Crawler

• Polling executa automaticamente sem intervenção manual.
• Frequência configurável via POLL_INTERVAL_HOURS; não excede 1 req/6h por empresa.
• Falha em uma fonte não interrompe as demais.
• SHA-256 do binário do PDF é calculado antes de qualquer chamada ao LLM.
• PDF com hash já existente retorna status = 'duplicate_skipped' sem chamar o LLM.
• PDF com mesma URL mas conteúdo diferente (hash diferente) é processado normalmente.
• HTTP 403 em qualquer fonte escalona automaticamente para Playwright.
• User-Agent é rotacionado a cada requisição.

Camada 2 — Motor UDA

• PDFs com até 15 páginas usam estratégia Full-Scan.
• PDFs com mais de 15 páginas usam Chunking Semântico.
• A escolha de estratégia é registrada no log (estrategia_parse).
• Tabelas do PDF são preservadas de forma legível ao LLM.
• Nenhum campo Optional é preenchido com 0 quando o dado está ausente no PDF.
• O campo vso é extraído corretamente como percentual (ex: 18.5) quando presente.
• Valores monetários em R$ milhões são convertidos para R$ mil antes de salvar.
• O campo confianca_extracao está presente em 100% das respostas do LLM.
• O pipeline extrai dados corretamente de pelo menos 2 layouts distintos (tabela e slides).

Camada 3 — API

• GET /api/conjuntura com filtros válidos retorna 200 com dados corretos.
• GET /api/conjuntura sem resultados retorna 200 com data: [].
• Filtro empresa é case-insensitive.
• GET /api/conjuntura?trimestre=5T retorna 400 com mensagem clara.
• fonte_url está presente e não-nula em todos os registros retornados.
• Reprocessamento com forcar_reprocessamento: true executa Upsert sem erro.
• Respostas da API em menos de 500ms para consultas com filtros indexados.

Catálogo e Linhagem

• 100% dos registros em operational_reports têm catalog_id válido.
• Query de rastreabilidade funciona: SELECT fonte_url FROM data_catalog WHERE id = (SELECT catalog_id FROM operational_reports WHERE empresa = 'MRV' AND ano = 2025 AND trimestre = '3T').

Comportamento Esperado do Agente por Fase

Fase 1 — Scaffolding

Ao criar a estrutura inicial do projeto:

1. Criar todos os diretórios e arquivos __init__.py conforme o mapa de arquivos acima.
2. Criar requirements.txt com versões fixas.
3. Criar .env.example com todas as variáveis documentadas na spec (seção 7.3).
4. Criar sources.yaml com as 5 incorporadoras do escopo mínimo, incluindo o campo scraper_mode.
5. Verificação: python -c "import fitz, pydantic, fastapi, sqlalchemy, playwright" deve passar sem erros.

Fase 2 — Storage e Migrations

Ao implementar storage/:

1. Ler a spec seção 5 (schema SQL de data_catalog e operational_reports).
2. Implementar os modelos SQLAlchemy 2.x correspondentes em storage/models.py.
3. Lembrar de adicionar coluna updated_at em operational_reports (necessária para Upsert).
4. Implementar catalog.py com as funções: get_by_hash, save, update_status, upsert_report.
5. Verificação: python -m storage.migrations.run deve criar as tabelas sem erros.

Fase 3 — Contrato Semântico

Ao implementar processing/schema.py:

1. Implementar o OperationalReport exatamente conforme a spec seção 4.3.
2. Implementar processing/prompt_builder.py com o system prompt da spec seção 4.4 — atenção especial à regra do VSO (única exceção de percentual permitida).
3. Verificação: python -c "from processing.schema import OperationalReport; print(OperationalReport.model_json_schema())" deve imprimir o schema completo.

Fase 4 — Motor de Parsing e LLM

Ao implementar processing/pdf_parser.py, chunker.py e llm_client.py:

1. pdf_parser.py: usar fitz.open() — preservar estrutura de tabelas extraindo por blocos de texto.
2. chunker.py: lógica de corte por títulos/seções; retornar lista de strings quando num_pages > 15.
3. llm_client.py: temperatura 0, max_tokens=1500, retry até LLM_MAX_RETRIES.
4. Verificação: python -m pytest tests/test_parser.py tests/test_llm_extraction.py -v

Fase 5 — Crawler

Ao implementar crawler/:

1. hash_checker.py primeiro: testar isoladamente antes de integrar ao scraper.
2. scraper.py: estratégia httpx com fallback para browser.py em 403/429.
3. scheduler.py: APScheduler com BlockingScheduler; intervalo lido de POLL_INTERVAL_HOURS.
4. Verificação: python -m pytest tests/test_idempotency.py -v

Fase 6 — API REST

Ao implementar api/:

1. Implementar os 4 endpoints da spec seção 6.1 em ordem: conjuntura, serie-historica, empresas, catalogo, ingestao/manual.
2. Usar ilike para filtros de empresa (case-insensitive).
3. Implementar o handler de erro 400 para parâmetros inválidos com o formato padrão da spec.
4. Verificação: python -m pytest tests/test_api.py -v

Restrições Absolutas

Estas regras nunca devem ser violadas, independentemente de como o usuário formular o pedido:

• Sem lógica por empresa: nenhum if empresa_id == 'X' ou if empresa == 'MRV' em qualquer módulo. Toda variação vem de sources.yaml.
• Sem coordenadas de PDF: nenhuma extração baseada em coordenadas de pixel, bounding boxes ou posição física no documento.
• Sem regex para extração de dados: expressões regulares são permitidas apenas para matching de URLs e nomes de arquivo — nunca para extrair valores numéricos de texto de PDF.
• Sem INSERT simples em operational_reports: sempre Upsert (ON CONFLICT DO UPDATE).
• Sem credenciais hardcoded: toda chave de API, URL de banco e parâmetro operacional vem de .env via pydantic_settings.
• Sem requests síncrono: use httpx.AsyncClient para todo I/O de rede.
• Sem session.query() legado do SQLAlchemy: use sempre a sintaxe 2.x com select().

Variáveis de Ambiente

Todas as variáveis obrigatórias. A aplicação deve falhar na inicialização com mensagem clara se alguma estiver ausente.

Referência da Spec

Seções da docs/especificacao_pipeline_uda.md mapeadas por módulo: