From 2ce58ff1cbc1a3944ba00116bd13db44ab4c9c43 Mon Sep 17 00:00:00 2001 From: ademirrodrigo <78499173+ademirrodrigo@users.noreply.github.com> Date: Tue, 28 Oct 2025 15:19:12 -0300 Subject: [PATCH] Add Docker deployment and stabilize Streamlit imports --- .env.example | 13 + .gitignore | 17 + Dockerfile | 24 + README.md | 265 +++------- api_config.example.json | 6 - api_server.py | 370 -------------- app/__init__.py | 7 + app/collectors/__init__.py | 0 app/collectors/nfce_html.py | 68 +++ app/collectors/nfe_dfe.py | 106 ++++ app/config.py | 44 ++ app/database.py | 34 ++ app/models.py | 53 ++ app/services/__init__.py | 0 app/services/coleta.py | 34 ++ app/utils/__init__.py | 0 app/utils/certificado.py | 35 ++ app/utils/cnpj.py | 29 ++ app/utils/helpers.py | 42 ++ certs/.gitkeep | 0 data/html/.gitkeep | 0 data/xmls/.gitkeep | 0 docker-compose.yml | 14 + docker-entrypoint.sh | 14 + force_install.bat | 39 ++ install.bat | 28 ++ install.sh | 46 ++ logs/.gitkeep | 0 main.py | 936 ----------------------------------- monitor_config.example.json | 9 - nginx.conf | 34 ++ requirements.txt | 11 +- start.bat | 32 ++ static/css/app.css | 218 -------- templates/add_client.html | 113 ----- templates/base.html | 116 ----- templates/client_detail.html | 153 ------ templates/client_events.html | 111 ----- templates/edit_client.html | 111 ----- templates/index.html | 259 ---------- uninstall.sh | 23 + update.sh | 19 + verify.bat | 42 ++ web/app.py | 142 ++++++ webapp.py | 351 ------------- 45 files changed, 1021 insertions(+), 2947 deletions(-) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 Dockerfile delete mode 100644 api_config.example.json delete mode 100644 api_server.py create mode 100644 app/__init__.py create mode 100644 app/collectors/__init__.py create mode 100644 app/collectors/nfce_html.py create mode 100644 app/collectors/nfe_dfe.py create mode 100644 app/config.py create mode 100644 app/database.py create mode 100644 app/models.py create mode 100644 app/services/__init__.py create mode 100644 app/services/coleta.py create mode 100644 app/utils/__init__.py create mode 100644 app/utils/certificado.py create mode 100644 app/utils/cnpj.py create mode 100644 app/utils/helpers.py create mode 100644 certs/.gitkeep create mode 100644 data/html/.gitkeep create mode 100644 data/xmls/.gitkeep create mode 100644 docker-compose.yml create mode 100755 docker-entrypoint.sh create mode 100644 force_install.bat create mode 100644 install.bat create mode 100644 install.sh create mode 100644 logs/.gitkeep delete mode 100644 main.py delete mode 100644 monitor_config.example.json create mode 100644 nginx.conf create mode 100644 start.bat delete mode 100644 static/css/app.css delete mode 100644 templates/add_client.html delete mode 100644 templates/base.html delete mode 100644 templates/client_detail.html delete mode 100644 templates/client_events.html delete mode 100644 templates/edit_client.html delete mode 100644 templates/index.html create mode 100644 uninstall.sh create mode 100644 update.sh create mode 100644 verify.bat create mode 100644 web/app.py delete mode 100644 webapp.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..45e3b45 --- /dev/null +++ b/.env.example @@ -0,0 +1,13 @@ +# Configurações básicas do Coletor Fiscal v3.2 SaaS-Ready +BANCO=coletor.db +DATABASE_URL= +PORTA=8501 +STREAMLIT_ADDRESS=0.0.0.0 +AMBIENTE=dev +SECRET_KEY=troque-esta-chave +LOG_DIR=logs +XML_DIR=data/xmls +HTML_DIR=data/html +CERTS_DIR=certs +WEB_USER=admin +WEB_PASS=admin diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..720252d --- /dev/null +++ b/.gitignore @@ -0,0 +1,17 @@ +# Ambientes e arquivos sensíveis +.venv/ +.env +__pycache__/ +*.pyc + +# Dados coletados +certs/*.pfx +data/xmls/* +!data/xmls/.gitkeep +data/html/* +!data/html/.gitkeep +logs/* +!logs/.gitkeep + +# Artefatos diversos +.DS_Store diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..9a56f90 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,24 @@ +FROM python:3.12-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + +WORKDIR /app + +COPY requirements.txt ./ + +RUN apt-get update \ + && apt-get install -y --no-install-recommends build-essential libxml2-dev libxslt1-dev \ + && pip install --upgrade pip \ + && pip install -r requirements.txt \ + && apt-get purge -y build-essential \ + && apt-get autoremove -y \ + && rm -rf /var/lib/apt/lists/* + +COPY . . + +RUN mkdir -p data/xmls data/html logs certs + +EXPOSE 8501 + +ENTRYPOINT ["/app/docker-entrypoint.sh"] diff --git a/README.md b/README.md index 578b33f..7dc29eb 100644 --- a/README.md +++ b/README.md @@ -1,223 +1,104 @@ -# Monitoramento do eCAC +# Coletor Fiscal v3.2 SaaS-Ready -Este repositório contém uma ferramenta CLI em Python para monitorar periodicamente o eCAC para escritórios de contabilidade. Ela autentica usando o certificado digital de cada contribuinte (empresa ou pessoa física) ou, opcionalmente, apenas a procuração eletrônica do contador, consulta uma API proprietária e registra novos eventos em um banco SQLite, disparando alertas via webhook. +Plataforma para coleta automatizada de documentos fiscais eletrônicos (NF-e e NFC-e) com backend Python, banco SQLite e painel Streamlit. A versão v3.2 foi reconstruída para operar de forma consistente em contêineres Docker, mantendo scripts auxiliares para instalações tradicionais. -## Requisitos - -- Python 3.9 ou superior -- Bibliotecas [`requests`](https://pypi.org/project/requests/) e [`Flask`](https://flask.palletsprojects.com/) -- Certificados digitais (`.pem`) dos clientes que usarão o modo por certificado e procuração eletrônica ativa para o contador -- API própria do escritório capaz de autenticar e consultar notificações/obrigações do eCAC - -Instale a dependência: - -```bash -python -m pip install --upgrade pip -pip install -r requirements.txt -``` - -## Configuração da API proprietária incluída - -O repositório agora acompanha uma API real (`api_server.py`) que atende aos mesmos contratos esperados pelo monitor. Ela armazena clientes, notificações e obrigações em SQLite e valida o acesso por certificado ou pela procuração do contador. - -1. Copie `api_config.example.json` para `api_config.json` e defina seus valores reais: - - `contador_document`: informe o CPF/CNPJ do contador que assina as procurações (por exemplo, `97121215187`). - - `default_procuracao_token`: token padrão emitido no eCAC para o contador. - - `access_token_ttl`: tempo de vida (em minutos) dos tokens de acesso emitidos pela API. - - `admin_token`: segredo usado para autenticar as rotas administrativas (troque por um valor robusto). -2. Inicie a API: - - ```bash - export API_CONFIG=api_config.json - export API_DATABASE=api_data.db - python api_server.py - ``` - -3. Cadastre cada cliente autorizado (PJ ou PF). Exemplo usando apenas a procuração do contador: - - ```bash - curl -X POST http://localhost:5000/admin/clients \ - -H "Content-Type: application/json" \ - -H "X-Admin-Token: SEU_TOKEN_ADMIN" \ - -d '{ - "document": "12345678000190", - "name": "Empresa Exemplo Ltda", - "client_type": "PJ", - "procuracao_token": "token-especifico-opcional" - }' - ``` +## Visão Geral +- **Backend:** Python + SQLAlchemy + SQLite +- **Painel:** Streamlit executando na porta 8501 +- **Coletas:** + - NF-e via WebService `NFeDistribuicaoDFe` (integração real pronta e comentada) + - NFC-e via raspagem pública no portal da SEFAZ-GO (requisições comentadas por padrão) +- **Estrutura de pastas:** + - `app/`: serviços, models e utilitários + - `web/`: aplicativo Streamlit com login, cadastro de empresas e orquestração da coleta + - `certs/`: certificados digitais A1 (`CNPJ.pfx`) + - `data/xmls` e `data/html`: armazenamento dos documentos coletados + - `logs/`: registros de execução -4. Alimente a API com notificações e obrigações sempre que houver novos dados do eCAC (por integração automática ou operação manual). Exemplos: - - ```bash - curl -X POST http://localhost:5000/admin/clients/12345678000190/notifications \ - -H "Content-Type: application/json" \ - -H "X-Admin-Token: SEU_TOKEN_ADMIN" \ - -d '{ - "notification": { - "title": "Mensagem do eCAC", - "category": "Avisos", - "protocol": "2024-0001" - } - }' - - curl -X POST http://localhost:5000/admin/clients/12345678000190/obligations \ - -H "Content-Type: application/json" \ - -H "X-Admin-Token: SEU_TOKEN_ADMIN" \ - -d '{ - "obligations": [ - {"description": "Entrega DCTF", "due_date": "2024-06-30", "status": "pending"} - ] - }' - ``` - -5. Atualize `monitor_config.json` (copiado de `monitor_config.example.json`) apontando `api_base_url` para `http://localhost:5000`, definindo o mesmo `contador_document` e o token padrão desejado. Opcionalmente ajuste `poll_interval`, `verify_ssl`, `timeout` e `webhook_url`. - -6. Garanta que os certificados `.pem` dos clientes que utilizarem o modo por certificado estejam acessíveis no servidor. Para cadastros que operarão somente com a procuração do contador, basta configurar o token padrão ou individual conforme os passos anteriores. - -## Banco de dados - -O monitor cria automaticamente o arquivo SQLite definido por `--database` (padrão `monitor.db`) com as tabelas `clients` e `events`. Faça backup periódico desse arquivo se precisar de histórico. - -## Cadastro de clientes - -Utilize o comando `add-client` para registrar cada contribuinte, escolhendo o modo de autenticação mais adequado ao cenário. - -### Exemplo com certificado do contribuinte - -```bash -python main.py add-client \ - --database monitor.db \ - 12345678000190 "Empresa Exemplo Ltda" PJ \ - /caminho/certificados/empresa.pem \ - /caminho/certificados/empresa-key.pem \ - --certificate-password "senhaOpcional" \ - --procuracao-token "tokenOpcional" -``` +> ⚠️ **Integrações reais**: Toda chamada direta à SEFAZ permanece comentada. Basta remover os comentários sinalizados nos módulos de coleta para ativar os fluxos produtivos. -### Exemplo usando apenas a procuração do contador +## Requisitos +- Docker 24+ e Docker Compose Plugin (recomendado) +- Certificados A1 (arquivo `.pfx`) para cada empresa cadastrada +- Para instalação manual: Python 3.11+ com `pip` -```bash -python main.py add-client \ - --database monitor.db \ - 98765432100 "Contribuinte via Procuração" PF \ - --auth-mode procuracao \ - --procuracao-token "tokenEspecificoOpcional" -``` +## Configuração do Ambiente +1. Duplique `.env.example` para `.env` e ajuste as variáveis desejadas. +2. Garanta que as pastas `certs/`, `data/xmls`, `data/html` e `logs/` estejam criadas (o sistema fará isso automaticamente ao iniciar). -No modo `procuracao`, os campos de certificado e chave são opcionais e podem ser omitidos. Se nenhum token específico for informado, o sistema utilizará o valor padrão configurado em `monitor_config.json`. +Variáveis principais disponíveis no `.env`: -Para listar os clientes cadastrados: +| Variável | Descrição | +| --- | --- | +| `PORTA` | Porta exposta do Streamlit (default `8501`) | +| `STREAMLIT_ADDRESS` | Endereço de bind do servidor (default `0.0.0.0`) | +| `WEB_USER` / `WEB_PASS` | Credenciais básicas do painel | +| `DATABASE_URL` | URL alternativa para o banco (caso não queira o SQLite padrão) | +## Execução com Docker (recomendado) ```bash -python main.py list-clients --database monitor.db +docker compose up -d --build ``` -### Atualização, remoção e status de clientes - -- Atualize informações (nome, tipo, caminhos de arquivos ou credenciais específicas): - - ```bash - python main.py update-client \ - --database monitor.db \ - 12345678000190 \ - --name "Empresa Nova" \ - --certificate /novo/caminho/cert.pem \ - --key /novo/caminho/key.pem - ``` - - Para alternar o modo de autenticação para o uso exclusivo da procuração, execute: - - ```bash - python main.py update-client \ - --database monitor.db \ - 12345678000190 \ - --auth-mode procuracao - ``` - -- Remova um cliente (os eventos associados também são excluídos): - - ```bash - python main.py delete-client --database monitor.db 12345678000190 - ``` - -- Consulte o último status consolidado retornado pela API: - - ```bash - python main.py show-status --database monitor.db 12345678000190 - ``` - -- Liste eventos registrados, com suporte a filtros e paginação simples: - - ```bash - python main.py list-events --database monitor.db --document 12345678000190 --limit 20 - ``` - -## Execução do monitoramento - -### Execução contínua - -Para rodar continuamente (modo daemon simples), use: +O serviço ficará disponível em `http://localhost:8501` (ou na porta configurada em `PORTA`). Os volumes montados garantem persistência local dos certificados, XMLs/HTML e logs. +Para interromper: ```bash -python main.py run --database monitor.db --config monitor_config.json +docker compose down ``` -O processo fica em loop consultando a API a cada `poll_interval` segundos, atualizando o banco e enviando alertas para o webhook (quando configurado). - -### Execução de ciclo único - -Se quiser executar apenas um ciclo (por exemplo, em uma pipeline agendada ou cron job), adicione `--once`: - +### Atualização no Docker ```bash -python main.py run --database monitor.db --config monitor_config.json --once +docker compose pull +docker compose up -d --build ``` -Para executar o ciclo apenas para um cliente específico (útil em fluxos manuais ou integrações), informe `--client`: +## Instalação Manual (Linux) +Os scripts originais continuam disponíveis para quem preferir instalações tradicionais. ```bash -python main.py run --database monitor.db --config monitor_config.json --client 12345678000190 +./install.sh # cria .venv, instala dependências e configura systemd +./update.sh # atualiza o código e reinicia o serviço +./uninstall.sh # remove o serviço e opcionalmente o ambiente virtual ``` -## Interface web - -Além da CLI, o repositório inclui um painel web completo (`webapp.py`) para cadastrar clientes, visualizar métricas consolidadas e disparar ciclos manuais do monitor. - -1. Configure as variáveis de ambiente (use o mesmo banco e configuração apontados pelo monitor): - ```bash - export MONITOR_DATABASE=monitor.db - export MONITOR_CONFIG=/caminho/para/monitor_config.json - export FLASK_SECRET_KEY="uma-string-aleatoria" - ``` -2. Inicie o servidor: - ```bash - python webapp.py - ``` -3. Acesse `http://localhost:8000` para visualizar clientes, cadastrar novos, editar registros, remover cadastros, examinar o - histórico de eventos e executar ciclos sob demanda (globais ou por cliente). Caso a página inicial informe que a configuração - não foi encontrada, verifique se `MONITOR_CONFIG` aponta para o arquivo correto e que ele está acessível. +> Antes de rodar `install.sh`, certifique-se de definir as variáveis necessárias em `.env`. -### Recursos disponíveis no painel +## Instalação Manual (Windows) +Scripts `.bat` mantidos para conveniência: -- **Dashboard operacional** com contadores de clientes, eventos, últimos ciclos e destaque para clientes que precisam de nova verificação. -- **Linha do tempo dos eventos** com visualização amigável, filtros de paginação, destaque de categorias/referências e exibição do payload bruto por evento. -- **Páginas de detalhe** para cada cliente exibindo notificações mais recentes, obrigações, metadados retornados pela API e ações rápidas (rodar ciclo, editar, remover). -- **Formulários estruturados** para cadastro e edição, incluindo seleção do modo de autenticação e dicas operacionais ao lado dos campos obrigatórios. -- **Estilo responsivo** baseado em Bulma com personalizações próprias (`static/css/app.css`) para cards, timeline e blocos de informação. +- `install.bat`: configuração padrão com ambiente virtual +- `force_install.bat`: reinstalação completa do ambiente virtual +- `start.bat`: inicializa o painel Streamlit (modo manual) +- `verify.bat`: valida versões de Python, dependências e compilação do código -O painel reutiliza o mesmo banco SQLite e respeita as configurações do arquivo JSON. Certifique-se de que o processo tenha acesso -aos certificados dos clientes que utilizarem esse modo e aos tokens de procuração necessários. +## Operação do Painel +1. Acesse o painel e faça login com `WEB_USER`/`WEB_PASS`. +2. Cadastre as empresas informando **Nome**, **CNPJ**, **UF** e **Senha do certificado**. +3. Posicione o certificado em `certs/.pfx`. +4. Informe, se desejar, chaves de NFC-e (uma por linha). +5. Clique em **🔄 Coletar Agora**. +6. Os documentos ficam registrados no SQLite e armazenados nas pastas `data/xmls//` e `data/html//`. -## Logs e observabilidade +## Ativando Integrações Reais +- **NF-e (WebService):** Habilite o trecho comentado em `app/collectors/nfe_dfe.py` para usar `requests_pkcs12` + `zeep` contra o endpoint oficial da SEFAZ. Ajuste o `tpAmb`, `cUFAutor` e endpoints conforme o ambiente. +- **NFC-e (Scraping):** Ative o bloco comentado em `app/collectors/nfce_html.py` que realiza o GET na consulta pública da SEFAZ-GO e persiste o HTML retornado. -Os logs são enviados para `stdout` com nível `INFO` por padrão. Para aumentar a verbosidade, ajuste a variável de ambiente `PYTHONLOGLEVEL` ou modifique `logging.basicConfig` em `main.py`. +Certifique-se de que os certificados, permissões de rede e configurações de proxy estejam corretos antes de rodar em produção. -## Deploy sugerido +## Estrutura do Código +- `app/models.py`: modelos `Empresa` e `Documento` com SQLAlchemy +- `app/services/coleta.py`: orquestra as rotinas de coleta +- `app/utils/`: utilitários para certificados, CNPJ e logging +- `web/app.py`: painel Streamlit com autenticação, seleção de empresa e gatilho de coleta -- Configure um serviço do sistema (ex.: `systemd`) para iniciar o comando contínuo após o boot. -- Armazene os certificados (quando aplicável) em diretório protegido e mantenha os tokens de procuração em local seguro, com permissões restritas ao usuário que executa o monitor. -- Utilize `venv` dedicado para isolar as dependências Python. +## Testes Rápidos +Para garantir que o código está sintaticamente correto: +```bash +python -m compileall app web +``` -## Suporte +No Docker o comando pode ser executado via `docker compose run --rm coletor python -m compileall app web`. -Em caso de dúvidas, entre em contato com o time responsável pela API proprietária ou adapte o código para atender às particularidades do seu escritório. +--- +Sistema pronto para operação segura em VPS Ubuntu, Windows 11 ou ambientes Docker orquestrados. diff --git a/api_config.example.json b/api_config.example.json deleted file mode 100644 index c021a2b..0000000 --- a/api_config.example.json +++ /dev/null @@ -1,6 +0,0 @@ -{ - "contador_document": "97121215187", - "default_procuracao_token": "token-padrao-procuracao", - "access_token_ttl": 120, - "admin_token": "troque-este-token-admin" -} diff --git a/api_server.py b/api_server.py deleted file mode 100644 index e282e98..0000000 --- a/api_server.py +++ /dev/null @@ -1,370 +0,0 @@ -"""API proprietária para integrar o monitoramento do eCAC.""" -from __future__ import annotations - -import json -import os -import secrets -import sqlite3 -from dataclasses import dataclass -from datetime import datetime, timedelta -from pathlib import Path -from typing import Any, Dict, Optional - -from flask import Flask, abort, jsonify, request - -ISO_FORMAT = "%Y-%m-%dT%H:%M:%S" - - -@dataclass -class APIConfig: - """Configurações principais da API.""" - - contador_document: str - default_procuracao_token: str - access_token_ttl: int = 60 # minutos - admin_token: Optional[str] = None - - @classmethod - def load(cls, path: Path) -> "APIConfig": - with path.open("r", encoding="utf-8") as fp: - data = json.load(fp) - missing = [field for field in ("contador_document", "default_procuracao_token") if field not in data] - if missing: - raise ValueError( - "Configuração inválida da API, campos obrigatórios ausentes: " + ", ".join(missing) - ) - return cls( - contador_document=data["contador_document"], - default_procuracao_token=data["default_procuracao_token"], - access_token_ttl=int(data.get("access_token_ttl", 60)), - admin_token=data.get("admin_token"), - ) - - -class APIDatabase: - """Gerencia o armazenamento de clientes, notificações e tokens.""" - - def __init__(self, path: Path) -> None: - self.path = path - self._ensure_schema() - - def _connect(self) -> sqlite3.Connection: - conn = sqlite3.connect(self.path) - conn.row_factory = sqlite3.Row - conn.execute("PRAGMA foreign_keys = ON") - return conn - - def _ensure_schema(self) -> None: - with self._connect() as conn: - conn.executescript( - """ - CREATE TABLE IF NOT EXISTS clients ( - document TEXT PRIMARY KEY, - name TEXT NOT NULL, - client_type TEXT NOT NULL CHECK (client_type IN ('PJ', 'PF')), - procuracao_token TEXT, - certificate_identifier TEXT - ); - - CREATE TABLE IF NOT EXISTS notifications ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - client_document TEXT NOT NULL, - payload TEXT NOT NULL, - created_at TEXT NOT NULL, - FOREIGN KEY (client_document) REFERENCES clients(document) ON DELETE CASCADE - ); - - CREATE TABLE IF NOT EXISTS obligations ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - client_document TEXT NOT NULL, - payload TEXT NOT NULL, - status TEXT, - due_date TEXT, - created_at TEXT NOT NULL, - FOREIGN KEY (client_document) REFERENCES clients(document) ON DELETE CASCADE - ); - - CREATE TABLE IF NOT EXISTS access_tokens ( - token TEXT PRIMARY KEY, - client_document TEXT NOT NULL, - expires_at TEXT NOT NULL, - created_at TEXT NOT NULL, - FOREIGN KEY (client_document) REFERENCES clients(document) ON DELETE CASCADE - ); - """ - ) - - # ------------------------------------------------------------------ - # Clientes - # ------------------------------------------------------------------ - def upsert_client( - self, - document: str, - name: str, - client_type: str, - procuracao_token: Optional[str] = None, - certificate_identifier: Optional[str] = None, - ) -> None: - with self._connect() as conn: - conn.execute( - """ - INSERT INTO clients (document, name, client_type, procuracao_token, certificate_identifier) - VALUES (?, ?, ?, ?, ?) - ON CONFLICT(document) DO UPDATE SET - name=excluded.name, - client_type=excluded.client_type, - procuracao_token=excluded.procuracao_token, - certificate_identifier=excluded.certificate_identifier - """, - (document, name, client_type, procuracao_token, certificate_identifier), - ) - - def get_client(self, document: str) -> Optional[sqlite3.Row]: - with self._connect() as conn: - return conn.execute("SELECT * FROM clients WHERE document = ?", (document,)).fetchone() - - # ------------------------------------------------------------------ - # Tokens de acesso - # ------------------------------------------------------------------ - def create_access_token(self, document: str, ttl_minutes: int) -> str: - token = secrets.token_urlsafe(32) - expires_at = datetime.utcnow() + timedelta(minutes=ttl_minutes) - with self._connect() as conn: - conn.execute( - """ - INSERT INTO access_tokens (token, client_document, expires_at, created_at) - VALUES (?, ?, ?, ?) - """, - ( - token, - document, - expires_at.strftime(ISO_FORMAT), - datetime.utcnow().strftime(ISO_FORMAT), - ), - ) - return token - - def validate_token(self, token: str) -> Optional[str]: - with self._connect() as conn: - row = conn.execute( - "SELECT client_document, expires_at FROM access_tokens WHERE token = ?", - (token,), - ).fetchone() - if not row: - return None - expires_at = datetime.strptime(row["expires_at"], ISO_FORMAT) - if expires_at < datetime.utcnow(): - self.delete_token(token) - return None - return row["client_document"] - - def delete_token(self, token: str) -> None: - with self._connect() as conn: - conn.execute("DELETE FROM access_tokens WHERE token = ?", (token,)) - - # ------------------------------------------------------------------ - # Notificações e obrigações - # ------------------------------------------------------------------ - def list_notifications(self, document: str) -> list[Dict[str, Any]]: - with self._connect() as conn: - rows = conn.execute( - "SELECT payload, created_at FROM notifications WHERE client_document = ? ORDER BY id DESC", - (document,), - ).fetchall() - notifications = [] - for row in rows: - payload = json.loads(row["payload"]) - payload.setdefault("created_at", row["created_at"]) - notifications.append(payload) - return notifications - - def append_notification(self, document: str, payload: Dict[str, Any]) -> None: - with self._connect() as conn: - conn.execute( - """ - INSERT INTO notifications (client_document, payload, created_at) - VALUES (?, ?, ?) - """, - (document, json.dumps(payload, ensure_ascii=False), datetime.utcnow().strftime(ISO_FORMAT)), - ) - - def list_obligations(self, document: str) -> list[Dict[str, Any]]: - with self._connect() as conn: - rows = conn.execute( - """ - SELECT payload, status, due_date, created_at - FROM obligations - WHERE client_document = ? - ORDER BY id DESC - """, - (document,), - ).fetchall() - obligations: list[Dict[str, Any]] = [] - for row in rows: - payload = json.loads(row["payload"]) - if row["status"]: - payload.setdefault("status", row["status"]) - if row["due_date"]: - payload.setdefault("due_date", row["due_date"]) - payload.setdefault("updated_at", row["created_at"]) - obligations.append(payload) - return obligations - - def replace_obligations(self, document: str, obligations: list[Dict[str, Any]]) -> None: - with self._connect() as conn: - conn.execute("DELETE FROM obligations WHERE client_document = ?", (document,)) - now = datetime.utcnow().strftime(ISO_FORMAT) - for item in obligations: - conn.execute( - """ - INSERT INTO obligations (client_document, payload, status, due_date, created_at) - VALUES (?, ?, ?, ?, ?) - """, - ( - document, - json.dumps(item, ensure_ascii=False), - item.get("status"), - item.get("due_date"), - now, - ), - ) - - -def create_app(config: APIConfig, database: APIDatabase) -> Flask: - app = Flask(__name__) - - def require_admin() -> None: - if not config.admin_token: - return - provided = request.headers.get("X-Admin-Token") - if provided != config.admin_token: - abort(401, "Admin token inválido") - - def require_access_token() -> str: - header = request.headers.get("Authorization", "") - if not header.startswith("Bearer "): - abort(401, "Token ausente") - token = header.split(" ", 1)[1] - document = database.validate_token(token) - if not document: - abort(401, "Token inválido ou expirado") - return document - - @app.post("/auth/procuracao") - def auth_procuracao() -> Any: - payload = request.get_json(force=True) - for field in ("document", "client_type", "contador_document"): - if field not in payload: - abort(400, f"Campo obrigatório ausente: {field}") - if payload["contador_document"] != config.contador_document: - abort(401, "Contador não autorizado") - client = database.get_client(payload["document"]) - if not client: - abort(404, "Cliente não cadastrado") - if client["client_type"] != payload["client_type"]: - abort(400, "Tipo de cliente divergente") - provided_token = payload.get("procuracao_token") or config.default_procuracao_token - valid_token = client["procuracao_token"] or config.default_procuracao_token - if provided_token != valid_token: - abort(401, "Token de procuração inválido") - token = database.create_access_token(client["document"], config.access_token_ttl) - return jsonify({"access_token": token, "expires_in": config.access_token_ttl * 60}) - - @app.post("/auth/certificate") - def auth_certificate() -> Any: - payload = request.get_json(force=True) - for field in ("document", "client_type"): - if field not in payload: - abort(400, f"Campo obrigatório ausente: {field}") - client = database.get_client(payload["document"]) - if not client: - abort(404, "Cliente não cadastrado") - if client["client_type"] != payload["client_type"]: - abort(400, "Tipo de cliente divergente") - identifier = payload.get("certificate_identifier") - stored_identifier = client["certificate_identifier"] - if stored_identifier and identifier != stored_identifier: - abort(401, "Certificado não autorizado") - token = database.create_access_token(client["document"], config.access_token_ttl) - return jsonify({"access_token": token, "expires_in": config.access_token_ttl * 60}) - - @app.get("/ecac//notifications") - def get_notifications(document: str) -> Any: - token_document = require_access_token() - if token_document != document: - abort(403, "Token não corresponde ao cliente consultado") - data = database.list_notifications(document) - return jsonify({"notifications": data}) - - @app.get("/ecac//obligations") - def get_obligations(document: str) -> Any: - token_document = require_access_token() - if token_document != document: - abort(403, "Token não corresponde ao cliente consultado") - data = database.list_obligations(document) - return jsonify({"obligations": data}) - - # ------------------------------------------------------------------ - # Endpoints administrativos para alimentar a API - # ------------------------------------------------------------------ - @app.post("/admin/clients") - def create_or_update_client() -> Any: - require_admin() - payload = request.get_json(force=True) - for field in ("document", "name", "client_type"): - if field not in payload: - abort(400, f"Campo obrigatório ausente: {field}") - database.upsert_client( - document=payload["document"], - name=payload["name"], - client_type=payload["client_type"], - procuracao_token=payload.get("procuracao_token"), - certificate_identifier=payload.get("certificate_identifier"), - ) - return jsonify({"status": "ok"}) - - @app.post("/admin/clients//notifications") - def append_client_notification(document: str) -> Any: - require_admin() - payload = request.get_json(force=True) - if "notification" in payload: - entry = payload["notification"] - else: - entry = payload - if not isinstance(entry, dict): - abort(400, "Notification inválida") - if not database.get_client(document): - abort(404, "Cliente não encontrado") - database.append_notification(document, entry) - return jsonify({"status": "ok"}) - - @app.post("/admin/clients//obligations") - def replace_client_obligations(document: str) -> Any: - require_admin() - payload = request.get_json(force=True) - obligations = payload.get("obligations") - if not isinstance(obligations, list): - abort(400, "Campo 'obligations' deve ser uma lista") - if not database.get_client(document): - abort(404, "Cliente não encontrado") - database.replace_obligations(document, obligations) - return jsonify({"status": "ok"}) - - return app - - -def load_app() -> Flask: - config_path = Path(os.environ.get("API_CONFIG", "api_config.json")) - if not config_path.exists(): - raise FileNotFoundError( - "Arquivo de configuração da API não encontrado. Defina API_CONFIG ou crie api_config.json." - ) - config = APIConfig.load(config_path) - db_path = Path(os.environ.get("API_DATABASE", "api_data.db")) - database = APIDatabase(db_path) - return create_app(config, database) -app = load_app() - - -if __name__ == "__main__": - app.run(host="0.0.0.0", port=int(os.environ.get("API_PORT", 5000)), debug=False) diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..0b07b1b --- /dev/null +++ b/app/__init__.py @@ -0,0 +1,7 @@ +"""Inicialização do pacote principal do Coletor Fiscal v3.2 SaaS-Ready.""" + +from .config import settings +from .database import Base, engine, init_db + + +__all__ = ["settings", "init_db", "Base", "engine"] diff --git a/app/collectors/__init__.py b/app/collectors/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/collectors/nfce_html.py b/app/collectors/nfce_html.py new file mode 100644 index 0000000..6245eb5 --- /dev/null +++ b/app/collectors/nfce_html.py @@ -0,0 +1,68 @@ +"""Coletor responsável pela raspagem pública da NFC-e da SEFAZ-GO.""" + +from __future__ import annotations + +from datetime import datetime +from typing import Iterable, List + +from bs4 import BeautifulSoup +from sqlalchemy.orm import Session + +# from requests import Session as HttpSession # Ative quando habilitar a raspagem real + +from ..config import settings +from ..models import Documento, Empresa +from ..utils.helpers import setup_logging + +LOGGER = setup_logging() + + +def _salvar_html(session: Session, empresa: Empresa, chave: str, html: str) -> Documento: + """Armazena o HTML coletado e registra o documento.""" + pasta_empresa = settings.html_dir / empresa.cnpj + pasta_empresa.mkdir(parents=True, exist_ok=True) + arquivo_html = pasta_empresa / f"{chave}.html" + arquivo_html.write_text(html, encoding="utf-8") + + try: + caminho_registrado = str(arquivo_html.relative_to(settings.base_dir)) + except ValueError: + caminho_registrado = str(arquivo_html) + + documento = Documento( + empresa_id=empresa.id, + tipo="NFCE", + chave=chave, + arquivo=caminho_registrado, + resumo=f"HTML coletado em {datetime.utcnow():%Y-%m-%d %H:%M:%S}" + ) + session.add(documento) + session.flush() + LOGGER.info("NFC-e %s registrada com sucesso", chave) + return documento + + +def coletar_nfce_publica(session: Session, empresa: Empresa, chaves: Iterable[str]) -> List[Documento]: + """Efetua a raspagem pública da NFC-e a partir de chaves fornecidas.""" + documentos: List[Documento] = [] + for chave in chaves: + LOGGER.info("Iniciando raspagem pública para NFC-e %s", chave) + + # Modelo de chamada HTTP real (comentado para evitar tráfego durante testes): + # url = f"https://nfe.sefaz.go.gov.br/nfeweb/sites/nfe/consulta-publica?chave={chave}" + # response = requests.get(url, timeout=30) + # response.raise_for_status() + # html = response.text + + LOGGER.warning( + "Raspagem NFC-e preparada; habilite a chamada HTTP comentada para ativar a coleta real." + ) + html = "" # O HTML seria atribuído pela resposta real. + + if not html: + continue + + soup = BeautifulSoup(html, "html.parser") + # Aqui podem ser aplicados tratamentos adicionais ao HTML antes do armazenamento. + documentos.append(_salvar_html(session, empresa, chave, soup.prettify())) + return documentos diff --git a/app/collectors/nfe_dfe.py b/app/collectors/nfe_dfe.py new file mode 100644 index 0000000..891b65d --- /dev/null +++ b/app/collectors/nfe_dfe.py @@ -0,0 +1,106 @@ +"""Coletor responsável por interagir com o WebService NFeDistribuicaoDFe.""" + +from __future__ import annotations + +import base64 +import io +import xml.etree.ElementTree as ET +from datetime import datetime +from typing import Iterable, List +from zipfile import ZipFile + +from sqlalchemy.orm import Session + +from ..config import settings +from ..models import Documento, Empresa +from ..utils.certificado import carregar_certificado +from ..utils.helpers import setup_logging + +LOGGER = setup_logging() + + +def _salvar_documento(session: Session, empresa: Empresa, chave: str, xml_conteudo: bytes) -> Documento: + """Persiste um documento na base de dados e no diretório de XMLs.""" + destino = empresa.cnpj + pasta_empresa = settings.xml_dir / destino + pasta_empresa.mkdir(parents=True, exist_ok=True) + + arquivo_xml = pasta_empresa / f"{chave}.xml" + arquivo_xml.write_bytes(xml_conteudo) + + try: + caminho_registrado = str(arquivo_xml.relative_to(settings.base_dir)) + except ValueError: + caminho_registrado = str(arquivo_xml) + + documento = Documento( + empresa_id=empresa.id, + tipo="NFE", + chave=chave, + arquivo=caminho_registrado, + resumo=f"XML coletado em {datetime.utcnow():%Y-%m-%d %H:%M:%S}" + ) + session.add(documento) + session.flush() + LOGGER.info("Documento %s armazenado no banco", chave) + return documento + + +def _processar_distdoc(session: Session, empresa: Empresa, documentos: Iterable[ET.Element]) -> List[Documento]: + """Processa a lista de documentos retornados pelo WebService.""" + resultados: List[Documento] = [] + for doc in documentos: + namespace_uri = doc.tag.split("}")[0].strip("{") if "}" in doc.tag else "" + namespaces = {"ns": namespace_uri} if namespace_uri else {} + caminho = "ns:docZip" if namespace_uri else "docZip" + doc_zip = doc.findtext(caminho, namespaces=namespaces) + if not doc_zip: + continue + xml_bytes = base64.b64decode(doc_zip) + with ZipFile(io.BytesIO(xml_bytes)) as zipped: + for nome in zipped.namelist(): + conteudo = zipped.read(nome) + chave = nome.replace(".xml", "") + resultados.append(_salvar_documento(session, empresa, chave, conteudo)) + return resultados + + +def coletar_nfe_distribuicao(session: Session, empresa: Empresa, nsu_inicial: str | None = None) -> List[Documento]: + """Executa a consulta no serviço NFeDistribuicaoDFe.""" + LOGGER.info("Iniciando coleta NFeDistribuicaoDFe para %s", empresa.cnpj) + + # As linhas abaixo demonstram como a integração real deve ocorrer. + certificado, senha = carregar_certificado(empresa.cnpj, empresa.certificado_senha or "") + + # Exemplo comentado de chamada SOAP ao serviço: + # from zeep import Client + # from zeep.transports import Transport + # from requests import Session as HttpSession + # from requests_pkcs12 import Pkcs12Adapter + # http_session = HttpSession() + # http_session.mount( + # "https://", + # Pkcs12Adapter(pkcs12_filename=str(certificado), pkcs12_password=senha), + # ) + # transport = Transport(session=http_session, timeout=30) + # client = Client("https://www.sefazvirtual.fazenda.gov.br/NFeDistribuicaoDFe/NFeDistribuicaoDFe.asmx?wsdl", transport=transport) + # consulta = { + # "distNSU": { + # "ultNSU": nsu_inicial or "000000000000000" + # } + # } + # resposta = client.service.nfeDistDFeInteresse( + # cUFAutor=empresa.uf, + # tpAmb=1, + # CNPJ=empresa.cnpj, + # distNSU=consulta["distNSU"], + # ) + # if hasattr(resposta, "loteDistDFeInt"): + # documentos = resposta.loteDistDFeInt.docZip + # return _processar_distdoc(session, empresa, documentos) + + # Enquanto a integração real não é ativada, interrompemos aqui. + LOGGER.warning( + "Coleta NFeDistribuicaoDFe está preparada, porém a chamada ao WebService permanece comentada por segurança." + ) + return [] diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..4ac5999 --- /dev/null +++ b/app/config.py @@ -0,0 +1,44 @@ +"""Configurações principais do Coletor Fiscal.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path + +from dotenv import load_dotenv + + +BASE_DIR = Path(__file__).resolve().parent.parent +ENV_FILE = BASE_DIR / ".env" + +if ENV_FILE.exists(): + load_dotenv(ENV_FILE) +else: + # Permite carregamento alternativo via variáveis de ambiente do sistema. + load_dotenv() + + +@dataclass +class Settings: + """Agrupa as configurações carregadas do ambiente.""" + + base_dir: Path = BASE_DIR + database_url: str = os.getenv("DATABASE_URL") or f"sqlite:///{BASE_DIR / 'data' / os.getenv('BANCO', 'coletor.db')}" + porta: int = int(os.getenv("PORTA", "8501")) + server_address: str = os.getenv("STREAMLIT_ADDRESS", "0.0.0.0") + ambiente: str = os.getenv("AMBIENTE", "dev") + secret_key: str = os.getenv("SECRET_KEY", "troque-esta-chave") + log_dir: Path = Path(os.getenv("LOG_DIR", BASE_DIR / "logs")) + xml_dir: Path = Path(os.getenv("XML_DIR", BASE_DIR / "data" / "xmls")) + html_dir: Path = Path(os.getenv("HTML_DIR", BASE_DIR / "data" / "html")) + certs_dir: Path = Path(os.getenv("CERTS_DIR", BASE_DIR / "certs")) + + def ensure_directories(self) -> None: + """Garante a existência dos diretórios necessários.""" + for path in (self.log_dir, self.xml_dir, self.html_dir, self.certs_dir): + path.mkdir(parents=True, exist_ok=True) + + +settings = Settings() +settings.ensure_directories() diff --git a/app/database.py b/app/database.py new file mode 100644 index 0000000..50a43ba --- /dev/null +++ b/app/database.py @@ -0,0 +1,34 @@ +"""Configuração da camada de persistência.""" + +from __future__ import annotations + +from contextlib import contextmanager +from typing import Iterator + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session, declarative_base, sessionmaker + +from .config import settings + +engine = create_engine(settings.database_url, connect_args={"check_same_thread": False}) +SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) +Base = declarative_base() + + +@contextmanager +def session_scope() -> Iterator[Session]: + """Fornece um contexto seguro para transações com rollback automático.""" + session = SessionLocal() + try: + yield session + session.commit() + except Exception: + session.rollback() + raise + finally: + session.close() + + +def init_db() -> None: + """Cria todas as tabelas definidas nos modelos.""" + Base.metadata.create_all(bind=engine) diff --git a/app/models.py b/app/models.py new file mode 100644 index 0000000..b2289c4 --- /dev/null +++ b/app/models.py @@ -0,0 +1,53 @@ +"""Modelos de dados utilizados pelo Coletor Fiscal.""" + +from __future__ import annotations + +from datetime import datetime +from pathlib import Path +from typing import Optional + +from sqlalchemy import DateTime, ForeignKey, Integer, String, Text +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from .config import settings +from .utils.cnpj import sanitize_cnpj +from .database import Base + + +class Empresa(Base): + """Representa uma empresa cadastrada com certificado digital.""" + + __tablename__ = "empresas" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) + nome: Mapped[str] = mapped_column(String(255), nullable=False) + cnpj: Mapped[str] = mapped_column(String(14), unique=True, index=True, nullable=False) + uf: Mapped[str] = mapped_column(String(2), nullable=False, default="GO") + certificado: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) + certificado_senha: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) + criado_em: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) + atualizado_em: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) + + documentos: Mapped[list["Documento"]] = relationship("Documento", back_populates="empresa") + + def certificado_path(self) -> Path: + """Retorna o caminho absoluto do certificado A1 da empresa.""" + if not self.certificado: + return settings.certs_dir / f"{sanitize_cnpj(self.cnpj)}.pfx" + return settings.certs_dir / self.certificado + + +class Documento(Base): + """Armazena metadados dos documentos coletados.""" + + __tablename__ = "documentos" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) + empresa_id: Mapped[int] = mapped_column(ForeignKey("empresas.id"), nullable=False) + tipo: Mapped[str] = mapped_column(String(10), nullable=False) # NFE / NFCE + chave: Mapped[str] = mapped_column(String(44), unique=True, index=True, nullable=False) + arquivo: Mapped[str] = mapped_column(Text, nullable=False) + resumo: Mapped[Optional[str]] = mapped_column(Text, nullable=True) + criado_em: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) + + empresa: Mapped[Empresa] = relationship("Empresa", back_populates="documentos") diff --git a/app/services/__init__.py b/app/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/services/coleta.py b/app/services/coleta.py new file mode 100644 index 0000000..8f022e3 --- /dev/null +++ b/app/services/coleta.py @@ -0,0 +1,34 @@ +"""Serviços de orquestração das rotinas de coleta.""" + +from __future__ import annotations + +from typing import Iterable + +from sqlalchemy.orm import Session + +from ..collectors.nfce_html import coletar_nfce_publica +from ..collectors.nfe_dfe import coletar_nfe_distribuicao +from ..models import Empresa +from ..utils.helpers import setup_logging + +LOGGER = setup_logging() + + +def coletar_todos(session: Session, empresa: Empresa, chaves_nfce: Iterable[str] | None = None) -> dict[str, int]: + """Executa todas as coletas disponíveis para a empresa informada.""" + resultados = {"nfe": 0, "nfce": 0} + + documentos_nfe = coletar_nfe_distribuicao(session, empresa) + resultados["nfe"] = len(documentos_nfe) + + if chaves_nfce: + documentos_nfce = coletar_nfce_publica(session, empresa, chaves_nfce) + resultados["nfce"] = len(documentos_nfce) + + LOGGER.info( + "Coletas concluídas para %s | NFe: %s | NFC-e: %s", + empresa.cnpj, + resultados["nfe"], + resultados["nfce"], + ) + return resultados diff --git a/app/utils/__init__.py b/app/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/utils/certificado.py b/app/utils/certificado.py new file mode 100644 index 0000000..afc7233 --- /dev/null +++ b/app/utils/certificado.py @@ -0,0 +1,35 @@ +"""Rotinas relacionadas a certificados digitais A1.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Optional + +from ..config import settings +from .cnpj import sanitize_cnpj + + +def get_certificado_path(cnpj: str, arquivo_personalizado: Optional[str] = None) -> Path: + """Localiza o certificado PFX correspondente ao CNPJ informado.""" + cnpj_limpo = sanitize_cnpj(cnpj) + if arquivo_personalizado: + return settings.certs_dir / arquivo_personalizado + return settings.certs_dir / f"{cnpj_limpo}.pfx" + + +def carregar_certificado(cnpj: str, senha: str) -> tuple[Path, str]: + """Retorna metadados do certificado. A carga real via requests_pkcs12 é comentada.""" + certificado = get_certificado_path(cnpj) + if not certificado.exists(): + raise FileNotFoundError( + "Certificado não encontrado. Salve o arquivo .pfx na pasta 'certs/'." + ) + + # Exemplo de como carregar o certificado com requests_pkcs12: + # from requests_pkcs12 import Pkcs12Adapter + # session = requests.Session() + # session.mount("https://", Pkcs12Adapter(pkcs12_filename=str(certificado), pkcs12_password=senha)) + # return session + + # O retorno contém o caminho e a senha para ser utilizado quando a integração for ativada. + return certificado, senha diff --git a/app/utils/cnpj.py b/app/utils/cnpj.py new file mode 100644 index 0000000..a5c3f20 --- /dev/null +++ b/app/utils/cnpj.py @@ -0,0 +1,29 @@ +"""Utilidades para manipulação de CNPJ.""" + +from __future__ import annotations + +import re + +CNPJ_PATTERN = re.compile(r"\D") + + +def sanitize_cnpj(value: str) -> str: + """Remove qualquer caractere não numérico do CNPJ.""" + return CNPJ_PATTERN.sub("", value or "") + + +def validate_cnpj(value: str) -> bool: + """Valida um CNPJ utilizando o algoritmo oficial.""" + cnpj = sanitize_cnpj(value) + if len(cnpj) != 14 or len(set(cnpj)) == 1: + return False + + def calculate_digit(numbers: str) -> str: + weights = list(range(len(numbers) - 7, 1, -1)) + total = sum(int(digit) * weight for digit, weight in zip(numbers, weights)) + remainder = total % 11 + return "0" if remainder < 2 else str(11 - remainder) + + first_digit = calculate_digit(cnpj[:12]) + second_digit = calculate_digit(cnpj[:12] + first_digit) + return cnpj[-2:] == first_digit + second_digit diff --git a/app/utils/helpers.py b/app/utils/helpers.py new file mode 100644 index 0000000..ef273a6 --- /dev/null +++ b/app/utils/helpers.py @@ -0,0 +1,42 @@ +"""Funções auxiliares compartilhadas.""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Iterable + +from ..config import settings + + +def setup_logging() -> logging.Logger: + """Configura um logger com saída em arquivo e console.""" + logs_dir = settings.log_dir + logs_dir.mkdir(parents=True, exist_ok=True) + + logger = logging.getLogger("coletor_fiscal") + if logger.handlers: + return logger + + logger.setLevel(logging.INFO) + + formatter = logging.Formatter( + "%(asctime)s | %(levelname)s | %(name)s | %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + file_handler = logging.FileHandler(logs_dir / "coletor.log") + file_handler.setFormatter(formatter) + logger.addHandler(file_handler) + + stream_handler = logging.StreamHandler() + stream_handler.setFormatter(formatter) + logger.addHandler(stream_handler) + + return logger + + +def ensure_subdirectories(base: Path, subfolders: Iterable[str]) -> None: + """Garante que subdiretórios existam dentro de uma pasta base.""" + for folder in subfolders: + (base / folder).mkdir(parents=True, exist_ok=True) diff --git a/certs/.gitkeep b/certs/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/data/html/.gitkeep b/data/html/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/data/xmls/.gitkeep b/data/xmls/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..f834f5e --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,14 @@ +version: "3.9" + +services: + coletor: + build: . + container_name: coletor-fiscal + env_file: .env + ports: + - "${PORTA:-8501}:8501" + volumes: + - ./data:/app/data + - ./logs:/app/logs + - ./certs:/app/certs + restart: unless-stopped diff --git a/docker-entrypoint.sh b/docker-entrypoint.sh new file mode 100755 index 0000000..4638c68 --- /dev/null +++ b/docker-entrypoint.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env sh +set -e + +# Carrega variáveis do arquivo .env caso exista. +if [ -f "/app/.env" ]; then + set -a + . /app/.env + set +a +fi + +STREAMLIT_PORT="${PORTA:-8501}" +STREAMLIT_ADDR="${STREAMLIT_ADDRESS:-0.0.0.0}" + +exec python -m streamlit run web/app.py --server.port "${STREAMLIT_PORT}" --server.address "${STREAMLIT_ADDR}" diff --git a/force_install.bat b/force_install.bat new file mode 100644 index 0000000..5cb2ef6 --- /dev/null +++ b/force_install.bat @@ -0,0 +1,39 @@ +@echo off +SETLOCAL ENABLEDELAYEDEXPANSION + +REM Forca uma reinstalacao completa do ambiente Coletor Fiscal v3.2. +REM - Remove o ambiente virtual existente +REM - Cria um novo ambiente virtual +REM - Atualiza o pip e instala dependencias do requirements.txt com cache limpo + +SET APP_DIR=%~dp0 +SET VENV_DIR=%APP_DIR%\.venv + +IF EXIST "%VENV_DIR%" ( + echo Removendo ambiente virtual antigo em %VENV_DIR% ... + rmdir /s /q "%VENV_DIR%" +) + +python -m venv "%VENV_DIR%" +IF ERRORLEVEL 1 ( + echo [ERRO] Falha ao criar ambiente virtual. + EXIT /B 1 +) + +CALL "%VENV_DIR%\Scripts\activate.bat" + +python -m pip install --upgrade pip +IF ERRORLEVEL 1 ( + echo [ERRO] Falha ao atualizar o pip. + EXIT /B 1 +) + +python -m pip install --no-cache-dir -r "%APP_DIR%requirements.txt" +IF ERRORLEVEL 1 ( + echo [ERRO] Falha na instalacao das dependencias. Verifique sua conexao e tente novamente. + EXIT /B 1 +) + +echo. +echo Reinstalacao concluida com sucesso. +ENDLOCAL diff --git a/install.bat b/install.bat new file mode 100644 index 0000000..dcf17e4 --- /dev/null +++ b/install.bat @@ -0,0 +1,28 @@ +@echo off +SETLOCAL ENABLEDELAYEDEXPANSION + +REM Instalação do Coletor Fiscal em ambientes Windows 11. +REM - Cria ambiente virtual +REM - Instala dependências + +SET APP_DIR=%~dp0 +SET VENV_DIR=%APP_DIR%\.venv + +IF NOT EXIST "%VENV_DIR%" ( + python -m venv "%VENV_DIR%" +) + +CALL "%VENV_DIR%\Scripts\activate.bat" +python -m pip install --upgrade pip +python -m pip install -r "%APP_DIR%\requirements.txt" + +IF NOT DEFINED PORTA ( + SET PORTA=8501 +) + +echo. +echo Ambiente instalado com sucesso. +echo Execute: python -m streamlit run web/app.py --server.port %PORTA% --server.address 0.0.0.0 +echo (Defina a porta com "set PORTA=8501" ou "$env:PORTA=8501" antes de executar, se desejar alterar.) +echo Certifique-se de preencher o arquivo .env com as variaveis corretas. +ENDLOCAL diff --git a/install.sh b/install.sh new file mode 100644 index 0000000..98de0f0 --- /dev/null +++ b/install.sh @@ -0,0 +1,46 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Script de instalação para ambientes Linux (Ubuntu/Debian). +# - Cria ambiente virtual +# - Instala dependências +# - Configura serviço systemd para o painel Streamlit + +APP_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +VENV_DIR="$APP_DIR/.venv" +SERVICE_NAME="coletor-fiscal" +USER_SERVICE="$(whoami)" + +if [ ! -d "$VENV_DIR" ]; then + python3 -m venv "$VENV_DIR" +fi + +source "$VENV_DIR/bin/activate" + +pip install --upgrade pip +pip install -r "$APP_DIR/requirements.txt" + +cat </dev/null +[Unit] +Description=Coletor Fiscal v3.2 SaaS-Ready +After=network.target + +[Service] +Type=simple +User=${USER_SERVICE} +WorkingDirectory=${APP_DIR} +Environment="PYTHONPATH=${APP_DIR}" +EnvironmentFile=${APP_DIR}/.env +ExecStart=${VENV_DIR}/bin/streamlit run ${APP_DIR}/web/app.py --server.port \${PORTA:-8501} --server.address 0.0.0.0 +Restart=always +RestartSec=5 + +[Install] +WantedBy=multi-user.target +SERVICE + +sudo systemctl daemon-reload +sudo systemctl enable ${SERVICE_NAME} +sudo systemctl start ${SERVICE_NAME} + +echo "Instalação concluída. O serviço ${SERVICE_NAME} está em execução." diff --git a/logs/.gitkeep b/logs/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/main.py b/main.py deleted file mode 100644 index 1142c36..0000000 --- a/main.py +++ /dev/null @@ -1,936 +0,0 @@ -"""Ferramenta de monitoramento contínuo do eCAC para escritórios de contabilidade. - -Este módulo fornece uma linha de comando para registrar clientes (empresas ou pessoas -físicas) e acompanhar notificações do eCAC por meio de uma API própria do escritório. -As requisições podem ser autenticadas com o certificado digital do contribuinte ou -apenas com a procuração eletrônica do contador, permitindo flexibilidade para cada -cadastro. - -O código foi escrito para ser direto e pronto para uso em produção, sem mocks -ou simulações. Ajuste apenas as configurações de acesso (endereços da API, -caminhos dos certificados, tokens de procuração e URLs de alerta) antes de -colocá-lo em operação. -""" - -from __future__ import annotations - -import argparse -import hashlib -import json -import logging -import sqlite3 -import sys -import time -from dataclasses import dataclass -from datetime import datetime -from pathlib import Path -from typing import Any, Dict, Iterable, List, Optional, Sequence - -try: - import requests -except ImportError as exc: # pragma: no cover - dependência obrigatória em produção - raise SystemExit( - "Dependência 'requests' não encontrada. Instale-a com 'pip install requests'." - ) from exc - -LOGGER = logging.getLogger("ecac_monitor") - -AUTH_MODES = {"certificate", "procuracao"} - - -@dataclass -class MonitorConfig: - """Representa a configuração principal do monitor.""" - - api_base_url: str - contador_document: str - procuracao_token: str - poll_interval: int = 900 - verify_ssl: bool = True - timeout: int = 60 - webhook_url: Optional[str] = None - - @classmethod - def load(cls, path: Path) -> "MonitorConfig": - with path.open("r", encoding="utf-8") as fp: - data = json.load(fp) - required = {"api_base_url", "contador_document", "procuracao_token"} - missing = [field for field in required if field not in data] - if missing: - raise ValueError( - f"Configuração inválida: campos obrigatórios ausentes {', '.join(missing)}" - ) - return cls( - api_base_url=data["api_base_url"].rstrip("/"), - contador_document=data["contador_document"], - procuracao_token=data["procuracao_token"], - poll_interval=int(data.get("poll_interval", 900)), - verify_ssl=bool(data.get("verify_ssl", True)), - timeout=int(data.get("timeout", 60)), - webhook_url=data.get("webhook_url"), - ) - - -@dataclass -class Client: - document: str - name: str - client_type: str # PJ ou PF - auth_mode: str # certificate ou procuracao - certificate_path: Optional[Path] - key_path: Optional[Path] - certificate_password: Optional[str] - procuracao_token: Optional[str] - last_status: Optional[str] - last_checked: Optional[datetime] - - -@dataclass -class EventRecord: - id: int - client_document: str - payload: Dict[str, Any] - received_at: datetime - client_name: Optional[str] = None - - -@dataclass -class DashboardMetrics: - """Representa indicadores consolidados para o painel web.""" - - total_clients: int - pj_clients: int - pf_clients: int - total_events: int - clients_with_alerts: int - last_check: Optional[datetime] - last_event: Optional[datetime] - - -class DatabaseManager: - """Responsável por persistir dados locais do monitor.""" - - def __init__(self, db_path: Path) -> None: - self.db_path = db_path - self._ensure_schema() - - def _connect(self) -> sqlite3.Connection: - conn = sqlite3.connect(self.db_path) - conn.row_factory = sqlite3.Row - conn.execute("PRAGMA foreign_keys = ON") - return conn - - def _ensure_schema(self) -> None: - with self._connect() as conn: - conn.executescript( - """ - CREATE TABLE IF NOT EXISTS clients ( - document TEXT PRIMARY KEY, - name TEXT NOT NULL, - client_type TEXT NOT NULL CHECK (client_type IN ('PJ', 'PF')), - auth_mode TEXT NOT NULL DEFAULT 'certificate' - CHECK (auth_mode IN ('certificate', 'procuracao')), - certificate_path TEXT, - key_path TEXT, - certificate_password TEXT, - procuracao_token TEXT, - last_status TEXT, - last_checked TEXT - ); - - CREATE TABLE IF NOT EXISTS events ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - client_document TEXT NOT NULL, - event_hash TEXT NOT NULL, - payload TEXT NOT NULL, - received_at TEXT NOT NULL, - FOREIGN KEY (client_document) REFERENCES clients(document) - ); - - CREATE UNIQUE INDEX IF NOT EXISTS idx_events_document_hash - ON events(client_document, event_hash); - """ - ) - self._migrate_clients_table(conn) - - def _migrate_clients_table(self, conn: sqlite3.Connection) -> None: - info = conn.execute("PRAGMA table_info(clients)").fetchall() - if not info: - return - - columns = {row[1]: row for row in info} - auth_present = "auth_mode" in columns - certificate_nullable = not columns["certificate_path"][3] - key_nullable = not columns["key_path"][3] - - if auth_present and certificate_nullable and key_nullable: - return - - auth_select = "auth_mode" if auth_present else "'certificate'" - conn.executescript( - f""" - CREATE TABLE clients_new ( - document TEXT PRIMARY KEY, - name TEXT NOT NULL, - client_type TEXT NOT NULL CHECK (client_type IN ('PJ', 'PF')), - auth_mode TEXT NOT NULL DEFAULT 'certificate' - CHECK (auth_mode IN ('certificate', 'procuracao')), - certificate_path TEXT, - key_path TEXT, - certificate_password TEXT, - procuracao_token TEXT, - last_status TEXT, - last_checked TEXT - ); - - INSERT INTO clients_new ( - document, name, client_type, auth_mode, - certificate_path, key_path, certificate_password, - procuracao_token, last_status, last_checked - ) - SELECT - document, - name, - client_type, - COALESCE({auth_select}, 'certificate') AS auth_mode, - certificate_path, - key_path, - certificate_password, - procuracao_token, - last_status, - last_checked - FROM clients; - - DROP TABLE clients; - ALTER TABLE clients_new RENAME TO clients; - - CREATE UNIQUE INDEX IF NOT EXISTS idx_events_document_hash - ON events(client_document, event_hash); - """ - ) - - # Métodos públicos ------------------------------------------------- - def add_client( - self, - document: str, - name: str, - client_type: str, - auth_mode: str, - certificate_path: Optional[Path], - key_path: Optional[Path], - certificate_password: Optional[str], - procuracao_token: Optional[str], - ) -> None: - if auth_mode not in AUTH_MODES: - raise ValueError(f"Modo de autenticação inválido: {auth_mode}") - if auth_mode == "certificate" and (not certificate_path or not key_path): - raise ValueError( - "Certificado e chave são obrigatórios quando o modo é 'certificate'" - ) - if auth_mode == "procuracao": - certificate_path = None - key_path = None - certificate_password = None - with self._connect() as conn: - conn.execute( - """ - INSERT INTO clients ( - document, name, client_type, auth_mode, - certificate_path, key_path, - certificate_password, procuracao_token - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) - """, - ( - document, - name, - client_type, - auth_mode, - str(certificate_path) if certificate_path else None, - str(key_path) if key_path else None, - certificate_password, - procuracao_token, - ), - ) - - def list_clients(self) -> List[Client]: - with self._connect() as conn: - rows = conn.execute( - "SELECT document, name, client_type, auth_mode, certificate_path, key_path, " - "certificate_password, procuracao_token, last_status, last_checked FROM clients" - ).fetchall() - clients: List[Client] = [] - for row in rows: - last_checked = ( - datetime.fromisoformat(row["last_checked"]) if row["last_checked"] else None - ) - clients.append( - Client( - document=row["document"], - name=row["name"], - client_type=row["client_type"], - auth_mode=row["auth_mode"], - certificate_path=Path(row["certificate_path"]) - if row["certificate_path"] - else None, - key_path=Path(row["key_path"]) if row["key_path"] else None, - certificate_password=row["certificate_password"], - procuracao_token=row["procuracao_token"], - last_status=row["last_status"], - last_checked=last_checked, - ) - ) - return clients - - def get_client(self, document: str) -> Optional[Client]: - with self._connect() as conn: - row = conn.execute( - "SELECT document, name, client_type, auth_mode, certificate_path, key_path, " - "certificate_password, procuracao_token, last_status, last_checked " - "FROM clients WHERE document = ?", - (document,), - ).fetchone() - if row is None: - return None - last_checked = datetime.fromisoformat(row["last_checked"]) if row["last_checked"] else None - return Client( - document=row["document"], - name=row["name"], - client_type=row["client_type"], - auth_mode=row["auth_mode"], - certificate_path=Path(row["certificate_path"]) if row["certificate_path"] else None, - key_path=Path(row["key_path"]) if row["key_path"] else None, - certificate_password=row["certificate_password"], - procuracao_token=row["procuracao_token"], - last_status=row["last_status"], - last_checked=last_checked, - ) - - def update_status(self, document: str, status: str) -> None: - with self._connect() as conn: - conn.execute( - "UPDATE clients SET last_status = ?, last_checked = ? WHERE document = ?", - (status, datetime.utcnow().isoformat(), document), - ) - - def register_events(self, document: str, events: Sequence[Dict]) -> List[Dict]: - """Registra eventos inéditos e retorna apenas os novos.""" - - if not events: - return [] - - created: List[Dict] = [] - with self._connect() as conn: - for event in events: - normalized = json.dumps(event, sort_keys=True, ensure_ascii=False) - event_hash = hashlib.sha1(normalized.encode("utf-8")).hexdigest() - try: - conn.execute( - """ - INSERT INTO events (client_document, event_hash, payload, received_at) - VALUES (?, ?, ?, ?) - """, - ( - document, - event_hash, - normalized, - datetime.utcnow().isoformat(), - ), - ) - except sqlite3.IntegrityError: - continue - created.append(event) - return created - - def list_events( - self, - document: Optional[str] = None, - *, - limit: int = 50, - offset: int = 0, - ) -> List[EventRecord]: - query = ( - "SELECT e.id, e.client_document, e.payload, e.received_at, c.name as client_name " - "FROM events e " - "LEFT JOIN clients c ON c.document = e.client_document " - + ("WHERE e.client_document = ? " if document else "") - + "ORDER BY datetime(e.received_at) DESC, e.id DESC LIMIT ? OFFSET ?" - ) - params: List[Any] - if document: - params = [document, limit, offset] - else: - params = [limit, offset] - with self._connect() as conn: - rows = conn.execute(query, params).fetchall() - events: List[EventRecord] = [] - for row in rows: - try: - payload = json.loads(row["payload"]) - except json.JSONDecodeError: - payload = {"raw": row["payload"]} - events.append( - EventRecord( - id=row["id"], - client_document=row["client_document"], - payload=payload, - received_at=datetime.fromisoformat(row["received_at"]), - client_name=row["client_name"], - ) - ) - return events - - def get_dashboard_metrics(self) -> DashboardMetrics: - with self._connect() as conn: - total_clients = conn.execute("SELECT COUNT(*) FROM clients").fetchone()[0] - type_counts = { - row["client_type"]: row["count"] - for row in conn.execute( - "SELECT client_type, COUNT(*) as count FROM clients GROUP BY client_type" - ).fetchall() - } - last_check_raw = conn.execute("SELECT MAX(last_checked) FROM clients").fetchone()[0] - total_events = conn.execute("SELECT COUNT(*) FROM events").fetchone()[0] - clients_with_alerts = conn.execute( - "SELECT COUNT(DISTINCT client_document) FROM events" - ).fetchone()[0] - last_event_raw = conn.execute("SELECT MAX(received_at) FROM events").fetchone()[0] - - last_check = ( - datetime.fromisoformat(last_check_raw) if last_check_raw else None - ) - last_event = ( - datetime.fromisoformat(last_event_raw) if last_event_raw else None - ) - return DashboardMetrics( - total_clients=total_clients, - pj_clients=type_counts.get("PJ", 0), - pf_clients=type_counts.get("PF", 0), - total_events=total_events, - clients_with_alerts=clients_with_alerts, - last_check=last_check, - last_event=last_event, - ) - - def delete_client(self, document: str) -> None: - with self._connect() as conn: - conn.execute("DELETE FROM events WHERE client_document = ?", (document,)) - deleted = conn.execute("DELETE FROM clients WHERE document = ?", (document,)).rowcount - if not deleted: - raise ValueError(f"Cliente {document} não encontrado") - - def update_client(self, document: str, **fields: Any) -> None: - current = self.get_client(document) - if not current: - raise ValueError(f"Cliente {document} não encontrado") - - new_auth_mode = fields.get("auth_mode", current.auth_mode) - if new_auth_mode not in AUTH_MODES: - raise ValueError(f"Modo de autenticação inválido: {new_auth_mode}") - - new_certificate = fields.get("certificate_path", current.certificate_path) - new_key = fields.get("key_path", current.key_path) - - if new_auth_mode == "certificate" and (not new_certificate or not new_key): - raise ValueError( - "Certificado e chave são obrigatórios quando o modo é 'certificate'" - ) - if new_auth_mode == "procuracao": - fields.setdefault("certificate_path", None) - fields.setdefault("key_path", None) - fields.setdefault("certificate_password", None) - - allowed = { - "name": "name", - "client_type": "client_type", - "auth_mode": "auth_mode", - "certificate_path": "certificate_path", - "key_path": "key_path", - "certificate_password": "certificate_password", - "procuracao_token": "procuracao_token", - } - assignments: List[str] = [] - values: List[Any] = [] - for key, column in allowed.items(): - if key not in fields: - continue - assignments.append(f"{column} = ?") - value = fields[key] - if isinstance(value, Path): - value = str(value) - values.append(value) - if not assignments: - return - values.append(document) - with self._connect() as conn: - updated = conn.execute( - f"UPDATE clients SET {', '.join(assignments)} WHERE document = ?", - values, - ).rowcount - if not updated: - raise ValueError(f"Cliente {document} não encontrado") - - -class EcacAPIClient: - """Cliente HTTP que interage com a API proprietária.""" - - def __init__(self, config: MonitorConfig) -> None: - self.config = config - - def _prepare_session( - self, client: Client, verify_ssl: bool - ) -> requests.Session: - session = requests.Session() - session.verify = verify_ssl - if client.auth_mode == "certificate" and client.certificate_path and client.key_path: - session.cert = (str(client.certificate_path), str(client.key_path)) - if client.certificate_password: - session.headers["X-Certificate-Pin"] = client.certificate_password - session.headers.update({ - "User-Agent": "ecac-monitor/1.0", - "X-Contador-Documento": self.config.contador_document, - "X-Procuracao-Token": client.procuracao_token or self.config.procuracao_token, - }) - return session - - def authenticate(self, session: requests.Session, client: Client) -> str: - if client.auth_mode == "procuracao": - endpoint = f"{self.config.api_base_url}/auth/procuracao" - payload = { - "document": client.document, - "client_type": client.client_type, - "contador_document": self.config.contador_document, - } - if client.procuracao_token: - payload["procuracao_token"] = client.procuracao_token - else: - endpoint = f"{self.config.api_base_url}/auth/certificate" - payload = { - "document": client.document, - "client_type": client.client_type, - } - response = session.post( - endpoint, - timeout=self.config.timeout, - json=payload, - ) - response.raise_for_status() - data = response.json() - token = data.get("access_token") - if not token: - raise RuntimeError("Resposta da API sem access_token") - return token - - def fetch_notifications( - self, session: requests.Session, token: str, document: str - ) -> List[Dict]: - response = session.get( - f"{self.config.api_base_url}/ecac/{document}/notifications", - timeout=self.config.timeout, - headers={"Authorization": f"Bearer {token}"}, - ) - response.raise_for_status() - payload = response.json() - notifications = payload.get("notifications") - if notifications is None: - raise RuntimeError("Resposta inesperada da API: campo 'notifications' ausente") - if not isinstance(notifications, list): - raise RuntimeError("Campo 'notifications' deve ser uma lista") - return notifications - - def fetch_obligations( - self, session: requests.Session, token: str, document: str - ) -> List[Dict]: - response = session.get( - f"{self.config.api_base_url}/ecac/{document}/obligations", - timeout=self.config.timeout, - headers={"Authorization": f"Bearer {token}"}, - ) - response.raise_for_status() - payload = response.json() - obligations = payload.get("obligations", []) - if not isinstance(obligations, list): - raise RuntimeError("Campo 'obligations' deve ser uma lista") - return obligations - - -class AlertDispatcher: - def __init__(self, webhook_url: Optional[str], verify_ssl: bool, timeout: int) -> None: - self.webhook_url = webhook_url - self.verify_ssl = verify_ssl - self.timeout = timeout - - def dispatch(self, payload: Dict) -> None: - if not self.webhook_url: - return - response = requests.post( - self.webhook_url, - json=payload, - timeout=self.timeout, - verify=self.verify_ssl, - ) - try: - response.raise_for_status() - except requests.HTTPError: - LOGGER.exception("Falha ao enviar alerta para %s", self.webhook_url) - raise - - -class EcacMonitor: - def __init__( - self, - db: DatabaseManager, - api_client: EcacAPIClient, - dispatcher: AlertDispatcher, - poll_interval: int, - verify_ssl: bool, - ) -> None: - self.db = db - self.api_client = api_client - self.dispatcher = dispatcher - self.poll_interval = poll_interval - self.verify_ssl = verify_ssl - - def run_forever(self) -> None: - LOGGER.info("Monitoramento iniciado") - while True: - start = time.monotonic() - self.run_cycle() - elapsed = time.monotonic() - start - sleep_time = max(self.poll_interval - elapsed, 5) - LOGGER.debug("Aguardando %.1f segundos até o próximo ciclo", sleep_time) - time.sleep(sleep_time) - - def run_cycle(self) -> None: - for client in self.db.list_clients(): - self._process_client(client) - - def run_for_client(self, document: str) -> None: - client = self.db.get_client(document) - if not client: - raise ValueError(f"Cliente {document} não encontrado") - self._process_client(client) - - def _process_client(self, client: Client) -> None: - LOGGER.info("Verificando %s (%s)", client.name, client.document) - session = self.api_client._prepare_session(client, self.verify_ssl) - try: - token = self.api_client.authenticate(session, client) - notifications = self.api_client.fetch_notifications(session, token, client.document) - obligations = self.api_client.fetch_obligations(session, token, client.document) - except Exception as exc: # noqa: BLE001 - LOGGER.exception("Falha ao comunicar com a API do eCAC: %s", exc) - return - - new_events = self.db.register_events(client.document, notifications) - status_payload = { - "notifications": notifications, - "obligations": obligations, - "updated_at": datetime.utcnow().isoformat(), - } - self.db.update_status(client.document, json.dumps(status_payload, ensure_ascii=False)) - - if new_events: - LOGGER.info("%s novos eventos encontrados para %s", len(new_events), client.document) - self._emit_alert(client, new_events, obligations) - else: - LOGGER.info("Nenhum evento novo para %s", client.document) - - def _emit_alert( - self, - client: Client, - new_events: Iterable[Dict], - obligations: Sequence[Dict], - ) -> None: - payload = { - "client": { - "document": client.document, - "name": client.name, - "type": client.client_type, - }, - "new_notifications": list(new_events), - "open_obligations": obligations, - "generated_at": datetime.utcnow().isoformat(), - } - try: - self.dispatcher.dispatch(payload) - except Exception: # noqa: BLE001 - LOGGER.exception("Erro ao enviar alerta do cliente %s", client.document) - - -# --------------------------------------------------------------------------- -# Linha de comando -# --------------------------------------------------------------------------- - -def build_parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser(description="Monitoramento contínuo do eCAC") - parser.add_argument( - "--database", - default="monitor.db", - help="Caminho do arquivo SQLite (padrão: monitor.db)", - ) - subparsers = parser.add_subparsers(dest="command", required=True) - - add_client_cmd = subparsers.add_parser("add-client", help="Cadastra um novo cliente") - add_client_cmd.add_argument("document", help="CPF ou CNPJ do contribuinte") - add_client_cmd.add_argument("name", help="Nome ou razão social") - add_client_cmd.add_argument("client_type", choices=["PJ", "PF"], help="Tipo do cliente") - add_client_cmd.add_argument( - "certificate", - nargs="?", - default=None, - help="Arquivo PEM do certificado da empresa (obrigatório para auth_mode=certificate)", - ) - add_client_cmd.add_argument( - "key", - nargs="?", - default=None, - help="Arquivo PEM da chave privada correspondente", - ) - add_client_cmd.add_argument( - "--auth-mode", - choices=sorted(AUTH_MODES), - default="certificate", - help="Define se o acesso usa certificado ou procuração (padrão: certificate)", - ) - add_client_cmd.add_argument( - "--certificate-password", - dest="certificate_password", - help="Senha do certificado (se aplicável)", - ) - add_client_cmd.add_argument( - "--procuracao-token", - dest="procuracao_token", - help="Token de procuração específico do cliente (opcional)", - ) - - subparsers.add_parser("list-clients", help="Lista clientes cadastrados") - - update_client_cmd = subparsers.add_parser( - "update-client", help="Atualiza informações de um cliente" - ) - update_client_cmd.add_argument("document", help="CPF ou CNPJ do contribuinte") - update_client_cmd.add_argument("--name", help="Novo nome ou razão social") - update_client_cmd.add_argument( - "--client-type", - choices=["PJ", "PF"], - dest="client_type", - help="Atualiza o tipo do cliente", - ) - update_client_cmd.add_argument( - "--auth-mode", - choices=sorted(AUTH_MODES), - dest="auth_mode", - help="Altera o modo de autenticação do cliente", - ) - update_client_cmd.add_argument("--certificate", help="Novo caminho do certificado") - update_client_cmd.add_argument("--key", help="Novo caminho da chave privada") - update_client_cmd.add_argument( - "--certificate-password", - dest="certificate_password", - help="Atualiza a senha do certificado", - ) - update_client_cmd.add_argument( - "--procuracao-token", - dest="procuracao_token", - help="Atualiza o token de procuração", - ) - update_client_cmd.add_argument( - "--clear-certificate-password", - action="store_true", - help="Remove a senha do certificado armazenada", - ) - update_client_cmd.add_argument( - "--clear-procuracao-token", - action="store_true", - help="Remove o token de procuração armazenado", - ) - - delete_client_cmd = subparsers.add_parser( - "delete-client", help="Remove um cliente e seu histórico" - ) - delete_client_cmd.add_argument("document", help="CPF ou CNPJ do contribuinte") - - events_cmd = subparsers.add_parser( - "list-events", help="Lista eventos registrados no banco" - ) - events_cmd.add_argument( - "--document", - help="Filtra eventos por documento do cliente", - ) - events_cmd.add_argument( - "--limit", - type=int, - default=50, - help="Quantidade máxima de eventos (padrão: 50)", - ) - events_cmd.add_argument( - "--offset", - type=int, - default=0, - help="Deslocamento para paginação (padrão: 0)", - ) - - status_cmd = subparsers.add_parser( - "show-status", help="Mostra o último status consolidado de um cliente" - ) - status_cmd.add_argument("document", help="CPF ou CNPJ do contribuinte") - - run_cmd = subparsers.add_parser("run", help="Executa o monitoramento contínuo") - run_cmd.add_argument( - "--config", - required=True, - help="Arquivo JSON com configuração do monitor", - ) - run_cmd.add_argument( - "--once", - action="store_true", - help="Executa apenas um ciclo de consulta (útil para integrações)", - ) - run_cmd.add_argument( - "--client", - help="Documento de um cliente específico para executar o ciclo", - ) - - return parser - - -def handle_add_client(args: argparse.Namespace, db: DatabaseManager) -> None: - certificate_path = Path(args.certificate).expanduser() if args.certificate else None - key_path = Path(args.key).expanduser() if args.key else None - db.add_client( - document=args.document, - name=args.name, - client_type=args.client_type, - auth_mode=args.auth_mode, - certificate_path=certificate_path, - key_path=key_path, - certificate_password=args.certificate_password, - procuracao_token=args.procuracao_token, - ) - LOGGER.info("Cliente %s cadastrado com sucesso", args.document) - - -def handle_list_clients(db: DatabaseManager) -> None: - clients = db.list_clients() - if not clients: - print("Nenhum cliente cadastrado.") - return - for client in clients: - last_checked = client.last_checked.isoformat() if client.last_checked else "nunca" - print( - f"{client.document} | {client.name} | {client.client_type} | " - f"última verificação: {last_checked}" - ) - - -def handle_update_client(args: argparse.Namespace, db: DatabaseManager) -> None: - fields: Dict[str, Any] = {} - if args.name: - fields["name"] = args.name - if args.client_type: - fields["client_type"] = args.client_type - if args.auth_mode: - fields["auth_mode"] = args.auth_mode - if args.certificate: - fields["certificate_path"] = Path(args.certificate).expanduser() - if args.key: - fields["key_path"] = Path(args.key).expanduser() - if args.certificate_password is not None: - fields["certificate_password"] = args.certificate_password - if args.procuracao_token is not None: - fields["procuracao_token"] = args.procuracao_token - if args.clear_certificate_password: - fields["certificate_password"] = None - if args.clear_procuracao_token: - fields["procuracao_token"] = None - if not fields: - LOGGER.info("Nenhum campo informado para atualização") - return - db.update_client(args.document, **fields) - LOGGER.info("Cliente %s atualizado com sucesso", args.document) - - -def handle_delete_client(args: argparse.Namespace, db: DatabaseManager) -> None: - db.delete_client(args.document) - LOGGER.info("Cliente %s removido", args.document) - - -def handle_list_events(args: argparse.Namespace, db: DatabaseManager) -> None: - events = db.list_events(args.document, limit=args.limit, offset=args.offset) - if not events: - print("Nenhum evento encontrado.") - return - for event in events: - print( - f"[{event.received_at.isoformat()}] {event.client_document} - " - f"payload: {json.dumps(event.payload, ensure_ascii=False)}" - ) - - -def handle_show_status(args: argparse.Namespace, db: DatabaseManager) -> None: - client = db.get_client(args.document) - if not client: - print(f"Cliente {args.document} não encontrado.") - return - if not client.last_status: - print("Nenhum status registrado. Execute um ciclo de monitoramento.") - return - try: - status = json.loads(client.last_status) - except json.JSONDecodeError: - print(client.last_status) - return - print(json.dumps(status, indent=2, ensure_ascii=False)) - - -def handle_run(args: argparse.Namespace, db: DatabaseManager) -> None: - config = MonitorConfig.load(Path(args.config)) - api_client = EcacAPIClient(config) - dispatcher = AlertDispatcher(config.webhook_url, config.verify_ssl, config.timeout) - monitor = EcacMonitor(db, api_client, dispatcher, config.poll_interval, config.verify_ssl) - if args.client: - monitor.run_for_client(args.client) - return - if args.once: - monitor.run_cycle() - else: - monitor.run_forever() - - -def main(argv: Optional[Sequence[str]] = None) -> int: - logging.basicConfig( - level=logging.INFO, - format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", - ) - parser = build_parser() - args = parser.parse_args(argv) - db = DatabaseManager(Path(args.database)) - - if args.command == "add-client": - handle_add_client(args, db) - return 0 - if args.command == "list-clients": - handle_list_clients(db) - return 0 - if args.command == "update-client": - handle_update_client(args, db) - return 0 - if args.command == "delete-client": - handle_delete_client(args, db) - return 0 - if args.command == "list-events": - handle_list_events(args, db) - return 0 - if args.command == "show-status": - handle_show_status(args, db) - return 0 - if args.command == "run": - handle_run(args, db) - return 0 - - parser.print_help() - return 1 - - -if __name__ == "__main__": - sys.exit(main()) diff --git a/monitor_config.example.json b/monitor_config.example.json deleted file mode 100644 index 801a5ca..0000000 --- a/monitor_config.example.json +++ /dev/null @@ -1,9 +0,0 @@ -{ - "api_base_url": "http://localhost:5000", - "contador_document": "97121215187", - "procuracao_token": "token-padrao-procuracao", - "poll_interval": 900, - "verify_ssl": true, - "timeout": 60, - "webhook_url": "https://seu-servidor-de-alertas.exemplo.com/ecac" -} diff --git a/nginx.conf b/nginx.conf new file mode 100644 index 0000000..f978fc0 --- /dev/null +++ b/nginx.conf @@ -0,0 +1,34 @@ +server { + listen 80; + server_name fiscal.goianiacontabil.com; + + location /.well-known/acme-challenge/ { + root /var/www/letsencrypt; + } + + location / { + return 301 https://$host$request_uri; + } +} + +server { + listen 443 ssl http2; + server_name fiscal.goianiacontabil.com; + + ssl_certificate /etc/letsencrypt/live/fiscal.goianiacontabil.com/fullchain.pem; + ssl_certificate_key /etc/letsencrypt/live/fiscal.goianiacontabil.com/privkey.pem; + include /etc/letsencrypt/options-ssl-nginx.conf; + ssl_dhparam /etc/letsencrypt/ssl-dhparams.pem; + + access_log /var/log/nginx/coletor_access.log; + error_log /var/log/nginx/coletor_error.log; + + location / { + proxy_pass http://127.0.0.1:8501; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_read_timeout 300; + } +} diff --git a/requirements.txt b/requirements.txt index 5b10b32..c31d251 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,9 @@ -requests>=2.31.0 -Flask>=2.3.0 +streamlit==1.36.0 +sqlalchemy==2.0.30 +python-dotenv==1.0.1 +requests==2.31.0 +requests-pkcs12==1.15 +zeep==4.2.1 +beautifulsoup4==4.12.3 +# LXML 5.3.x oferece wheels para Python 3.13 no Windows, evitando compilações manuais +lxml==5.3.2 diff --git a/start.bat b/start.bat new file mode 100644 index 0000000..c3b8391 --- /dev/null +++ b/start.bat @@ -0,0 +1,32 @@ +@echo off +SETLOCAL ENABLEDELAYEDEXPANSION + +REM Inicializa o painel Streamlit do Coletor Fiscal v3.2. +REM - Ativa o ambiente virtual existente +REM - Define porta e endereço de escuta +REM - Executa o painel web + +SET APP_DIR=%~dp0 +SET VENV_DIR=%APP_DIR%\.venv +SET STREAMLIT_ENTRY=%APP_DIR%web\app.py + +IF NOT EXIST "%VENV_DIR%\Scripts\activate.bat" ( + echo [ERRO] Ambiente virtual nao encontrado em %VENV_DIR%. + echo Execute install.bat ou force_install.bat para preparar o ambiente. + EXIT /B 1 +) + +CALL "%VENV_DIR%\Scripts\activate.bat" + +IF NOT DEFINED PORTA ( + SET PORTA=8501 +) + +IF NOT DEFINED ENDERECO ( + SET ENDERECO=0.0.0.0 +) + +echo Iniciando Streamlit em %ENDERECO%:%PORTA% ... +python -m streamlit run "%STREAMLIT_ENTRY%" --server.port %PORTA% --server.address %ENDERECO% + +ENDLOCAL diff --git a/static/css/app.css b/static/css/app.css deleted file mode 100644 index 60219a5..0000000 --- a/static/css/app.css +++ /dev/null @@ -1,218 +0,0 @@ -:root { - --card-shadow: 0 15px 35px rgba(10, 10, 10, 0.1); -} - -.hero-compact { - padding-top: 1.5rem; - padding-bottom: 1.5rem; -} - -.hero-dashboard .hero-body { - padding-top: 3rem; - padding-bottom: 3rem; -} - -.stat-card { - background: linear-gradient(135deg, #3273dc, #23d160); - border-radius: 12px; - box-shadow: var(--card-shadow); - color: #fff; - padding: 1.75rem; - height: 100%; -} - -.stat-card.is-secondary { - background: linear-gradient(135deg, #00c4a7, #209cee); -} - -.stat-card.is-danger { - background: linear-gradient(135deg, #ff3860, #ff9f43); -} - -.stat-card .stat-title { - font-size: 0.85rem; - text-transform: uppercase; - letter-spacing: 0.08em; - margin-bottom: 0.75rem; - opacity: 0.9; -} - -.stat-card .stat-value { - font-size: 2.5rem; - font-weight: 700; -} - -.stat-card .stat-footnote { - margin-top: 1rem; - font-size: 0.9rem; - opacity: 0.85; -} - -.notifications-stack > .notification + .notification { - margin-top: 0.75rem; -} - -.dashboard-card { - border-radius: 12px; - box-shadow: var(--card-shadow); -} - -.dashboard-card .card-header { - border-bottom: none; -} - -.dashboard-card .card-content { - padding: 1.5rem; -} - -.timeline { - position: relative; - padding-left: 1.75rem; -} - -.timeline::before { - content: ""; - position: absolute; - top: 0.5rem; - bottom: 0.5rem; - left: 0.55rem; - width: 2px; - background: linear-gradient(180deg, #3273dc, rgba(50, 115, 220, 0)); -} - -.timeline-item { - position: relative; - margin-bottom: 1.75rem; -} - -.timeline-item:last-child { - margin-bottom: 0; -} - -.timeline-marker { - position: absolute; - left: -1.5rem; - width: 14px; - height: 14px; - border-radius: 50%; - background-color: #3273dc; - box-shadow: 0 0 0 4px rgba(50, 115, 220, 0.15); - top: 0.35rem; -} - -.timeline-content { - background: #fff; - border-radius: 10px; - box-shadow: var(--card-shadow); - padding: 1.25rem 1.5rem; -} - -.timeline-content .heading { - color: #7a7a7a; - font-size: 0.85rem; - margin-bottom: 0.5rem; -} - -.timeline-content .title { - margin-bottom: 0.5rem; -} - -.timeline-content .tag + .tag { - margin-left: 0.5rem; -} - -.payload-preview { - background: #0a0a0a; - border-radius: 8px; - color: #f5f5f5; - padding: 1rem; - overflow: auto; -} - -.payload-preview code, -.payload-preview pre { - background: transparent; - color: inherit; - font-family: "Fira Code", "SFMono-Regular", Consolas, "Liberation Mono", Menlo, Courier, monospace; - font-size: 0.85rem; -} - -.quick-actions .button { - margin-bottom: 0.5rem; - width: 100%; -} - -.quick-actions form { - width: 100%; -} - -.quick-actions form .button { - width: 100%; -} - -.meta-list { - list-style: none; - margin: 0; - padding: 0; -} - -.meta-list li + li { - margin-top: 0.5rem; -} - -.meta-list strong { - display: inline-block; - min-width: 120px; -} - -.empty-state { - align-items: center; - border: 2px dashed #b5b5b5; - border-radius: 10px; - display: flex; - justify-content: center; - min-height: 160px; - padding: 2rem; - text-align: center; -} - -.details-block { - border-radius: 12px; - box-shadow: var(--card-shadow); - padding: 1.5rem; -} - -.details-block + .details-block { - margin-top: 1.5rem; -} - -.details-block h3 { - margin-bottom: 1rem; -} - -.raw-status details { - margin-top: 1rem; -} - -.raw-status summary { - cursor: pointer; - font-weight: 600; -} - -@media (max-width: 768px) { - .stat-card { - margin-bottom: 1.25rem; - } - - .quick-actions .button { - margin-bottom: 0.75rem; - } - - .timeline::before { - left: 0.35rem; - } - - .timeline-marker { - left: -1.75rem; - } -} diff --git a/templates/add_client.html b/templates/add_client.html deleted file mode 100644 index d82942c..0000000 --- a/templates/add_client.html +++ /dev/null @@ -1,113 +0,0 @@ -{% extends 'base.html' %} -{% block title %}Novo cliente - Monitor eCAC{% endblock %} - -{% block hero %} -
-
-
-

Cadastrar novo cliente

-

Informe os dados do certificado ou utilize apenas a procuração eletrônica do escritório para habilitar o monitoramento automático.

-
-
-
-{% endblock %} - -{% block content %} -
-
-
-
- -
- -
-
-
- -
- -
-
-
- -
-
- -
-
-
-
- -
-
- -
-
-

Escolha entre usar o certificado A1 do cliente ou apenas a procuração eletrônica habilitada na API.

-
-
- -
- -
-

Obrigatório apenas quando o modo de autenticação for certificado.

-
-
- -
- -
-

Obrigatório apenas quando o modo de autenticação for certificado.

-
-
- -
- -
-
-
- -
- -
-
-
-
- -
-
- Cancelar -
-
-
-
-
-
-

Dicas importantes

-
    -
  • Certificado: utilize arquivos no formato PEM já convertidos a partir do A1 quando optar pelo modo com certificado.
    • -
  • Procuração: ao usar apenas a procuração, confirme se o token geral ou específico está ativo.
    • -
  • Permissões: valide se o contador possui procuração eletrônica ativa no eCAC.
    • -
  • Ambiente: mantenha os arquivos em diretório seguro e com permissões restritas.
    • -
-
-
-

Próximos passos

-

Após salvar o cliente, execute um ciclo manual para validar a comunicação com a API e revisar os eventos iniciais.

-
-
-
-{% endblock %} diff --git a/templates/base.html b/templates/base.html deleted file mode 100644 index 2831796..0000000 --- a/templates/base.html +++ /dev/null @@ -1,116 +0,0 @@ - - - - - {% block title %}Monitor eCAC{% endblock %} - - - - - - - - - {% block hero %} -
-
-
-

Monitoramento do eCAC

-

Controle centralizado das notificações fiscais dos clientes.

-
-
-
- {% endblock %} - -
-
- {% with messages = get_flashed_messages(with_categories=true) %} - {% if messages %} -
- {% for category, message in messages %} -
- {{ message }} -
- {% endfor %} -
- {% endif %} - {% endwith %} - {% block content %}{% endblock %} -
-
- -
-
-

- Monitor eCAC — Automação completa para escritórios de contabilidade. - Utilize este painel para acompanhar certificados, obrigações e alertas em tempo real. -

-
-
- - - diff --git a/templates/client_detail.html b/templates/client_detail.html deleted file mode 100644 index 681412b..0000000 --- a/templates/client_detail.html +++ /dev/null @@ -1,153 +0,0 @@ -{% extends 'base.html' %} -{% block title %}{{ client.name }} - Monitor eCAC{% endblock %} - -{% block hero %} -
-
-
-

{{ client.name }}

-

{{ client.document }} · {{ 'Empresa' if client.client_type == 'PJ' else 'Pessoa Física' }}

-
- Última verificação: - {% if client.last_checked %} - {{ client.last_checked.strftime('%d/%m/%Y %H:%M') }} - {% else %} - nunca - {% endif %} - -
-
-
-
-{% endblock %} - -{% block content %} -
-
-
-

Dados do cadastro

-

Documento: {{ client.document }}

-

Tipo: {{ 'Pessoa Jurídica' if client.client_type == 'PJ' else 'Pessoa Física' }}

-

Modo de autenticação: - {% if client.auth_mode == 'certificate' %} - Certificado do contribuinte - {% else %} - Procuração do contador - {% endif %} -

- {% if client.certificate_path %} -

Caminho do certificado: {{ client.certificate_path }}

- {% else %} -

Caminho do certificado: não aplicável

- {% endif %} - {% if client.key_path %} -

Caminho da chave privada: {{ client.key_path }}

- {% else %} -

Caminho da chave privada: não aplicável

- {% endif %} -

Token específico de procuração: {{ client.procuracao_token or 'não informado' }}

-
- -
-

Status do monitoramento

- {% if status %} -
-
-

Notificações

- {% if notifications %} -
    - {% for item in notifications %} -
  • - #{{ loop.index }} -
    {{ (item.title or item.titulo or item.assunto or item.subject) if item is mapping else item }}
    -
    • - {% endfor %} -
- {% else %} -

Nenhuma notificação registrada.

- {% endif %} -
-
-

Obrigações pendentes

- {% if obligations %} -
    - {% for obligation in obligations %} -
  • - #{{ loop.index }} -
    - {{ (obligation.title or obligation.titulo or obligation.descricao) if obligation is mapping else obligation }} -
    -
    • - {% endfor %} -
- {% else %} -

Nenhuma obrigação pendente.

- {% endif %} -
-
- {% if metadata %} -
-

Metadados

-
    - {% for key, value in metadata.items() %} -
  • {{ key }}: {{ value }}
    • - {% endfor %} -
-
- {% endif %} -
-
- Exibir payload bruto retornado pela API -
- 
{{ status | tojson(indent=2, ensure_ascii=False) if status is mapping else status }}
-
-
-
- {% else %} -

Nenhum status registrado. Execute um ciclo para obter informações atualizadas.

- {% endif %} -
-
- -
-
-

Ações rápidas

- ← Voltar para o painel - Histórico de eventos - Editar cadastro -
- -
-
- -
-
- -
-

Últimos eventos

- {% if recent_events %} -
    - {% for event in recent_events %} -
  • - {{ event.summary }} -
    {{ event.received_at.strftime('%d/%m/%Y %H:%M') }}
    - {% if event.description %} -
    {{ event.description }}
    - {% endif %} -
    • - {% endfor %} -
-
- Ver histórico completo -
- {% else %} -

Nenhum evento registrado para este cliente.

- {% endif %} -
-
-
-{% endblock %} diff --git a/templates/client_events.html b/templates/client_events.html deleted file mode 100644 index a443c89..0000000 --- a/templates/client_events.html +++ /dev/null @@ -1,111 +0,0 @@ -{% extends 'base.html' %} -{% block title %}Eventos de {{ client.name }} - Monitor eCAC{% endblock %} - -{% block hero %} -
-
-
-

Histórico de eventos

-

{{ client.name }} · {{ client.document }}

-
-
-
-{% endblock %} - -{% block content %} -
-
-
-
-
-
-

Linha do tempo

-
-
-
-
- ← Voltar - Atualizar -
-
-
- {% if events %} -
- {% for event in events %} -
-
-
-

{{ event.received_at.strftime('%d/%m/%Y %H:%M:%S') }}

-

{{ event.summary }}

-
- {% if event.category %} - {{ event.category }} - {% endif %} - {% if event.reference %} - Ref: {{ event.reference }} - {% endif %} - ID {{ event.id }} -
- {% if event.description %} -

{{ event.description }}

- {% endif %} -
- Ver payload bruto -
- 
{{ event.payload | tojson(indent=2, ensure_ascii=False) }}
-
-
-
-
- {% endfor %} -
-
- {% if prev_offset is not none %} - ← Anteriores - {% endif %} - {% if next_offset is not none %} - Próximos → - {% endif %} -
- {% else %} -
-
-

Nenhum evento registrado até o momento.

-

Execute um ciclo de monitoramento para captar novas notificações.

-
-
- {% endif %} -
-
-
-
-

Configurações de exibição

-
-
- -
-
- -
-
-
-
- -
- -
-
-
-
- -
-
-
-
-
-
-{% endblock %} diff --git a/templates/edit_client.html b/templates/edit_client.html deleted file mode 100644 index 07b73c2..0000000 --- a/templates/edit_client.html +++ /dev/null @@ -1,111 +0,0 @@ -{% extends 'base.html' %} -{% block title %}Editar {{ client.name }} - Monitor eCAC{% endblock %} - -{% block hero %} -
-
-
-

Atualizar cadastro

-

Faça ajustes de certificados, senhas, tokens ou altere o modo de autenticação sempre que houver troca de representantes ou renovação.

-
-
-
-{% endblock %} - -{% block content %} -
-
-
-
- -

-
-
- -
- -
-
-
- -
-
- -
-
-
-
- -
-
- -
-
-

Selecione como o monitor irá autenticar no eCAC para este cliente.

-
-
- -
- -
-

Informe apenas quando o modo selecionado utilizar certificado.

-
-
- -
- -
-

Obrigatória somente quando o modo for certificado.

-
-
- -
- -
-
-
- -
- -
-
-
-
- -
-
- Cancelar -
-
-
-
-
-
-

Boas práticas

-
    -
  • Atualize o certificado sempre que o arquivo for substituído ou expirar.
    • -
  • Ao migrar para o modo por procuração, deixe os campos de certificado e chave em branco.
    • -
  • Ao remover a senha, deixe o campo em branco e salve.
    • -
  • O token de procuração pode ser individual para o cliente ou o padrão definido no arquivo de configuração.
    • -
-
-
-

Monitoramento

-

Após salvar, execute um ciclo para validar o novo certificado e garantir que o escritório continue recebendo notificações.

-
-
-
-{% endblock %} diff --git a/templates/index.html b/templates/index.html deleted file mode 100644 index ebadbf1..0000000 --- a/templates/index.html +++ /dev/null @@ -1,259 +0,0 @@ -{% extends 'base.html' %} -{% block title %}Painel - Monitor eCAC{% endblock %} - -{% block hero %} -
-
-
-
-
-

Painel operacional do Monitor eCAC

-

- Visualize rapidamente o status dos clientes, execute ciclos sob demanda e acompanhe - as últimas notificações recebidas pela integração com a API proprietária. -

-
- - - Adicionar cliente - -
- -
-
-
-
-
-
-
-

Clientes monitorados

-

{{ metrics.total_clients }}

-

{{ metrics.pj_clients }} PJ · {{ metrics.pf_clients }} PF

-
-
-
-
-

Eventos registrados

-

{{ metrics.total_events }}

-

{{ metrics.clients_with_alerts }} clientes com alertas

-
-
-
-
-

Último ciclo

-

- {% if metrics.last_check %} - {{ metrics.last_check.strftime('%d/%m %H:%M') }} - {% else %} - -- - {% endif %} -

-

- {% if metrics.last_event %} - Último evento {{ metrics.last_event.strftime('%d/%m %H:%M') }} - {% else %} - Nenhum evento registrado - {% endif %} -

-
-
-
-
-
-
-
-
-{% endblock %} - -{% block content %} -
-
- {% if not config_available %} -
-
Configuração ausente
-
- Defina a variável de ambiente MONITOR_CONFIG apontando para o arquivo JSON - de configuração antes de executar um ciclo. A interface continua disponível para gestão de clientes. -
-
- {% endif %} - -
-
-

Clientes cadastrados

-
- {{ clients|length }} registros -
-
-
-
- - - - - - - - - - - - - {% for entry in clients %} - {% set client = entry.client %} - - - - - - - - - {% else %} - - - - {% endfor %} - -
ClienteDocumentoTipoÚltima verificaçãoAlertasAções
- {{ client.name }} -
- {{ client.document }} - 		{{ client.document }} - - {{ 'Empresa' if client.client_type == 'PJ' else 'Pessoa Física' }} - - - {% if client.auth_mode == 'certificate' %} - Certificado - {% else %} - Procuração - {% endif %} - - - {% if client.last_checked %} - {{ client.last_checked.strftime('%d/%m/%Y %H:%M') }} - {% else %} - Nunca verificado - {% endif %} - - {% if entry.notifications > 0 %} - {{ entry.notifications }} notificações - {% else %} - Sem alertas - {% endif %} - {% if entry.obligations > 0 %} - {{ entry.obligations }} obrigações - {% endif %} - -
- Detalhes - Eventos - Editar -
- -
-
-
-
-
-

Nenhum cliente cadastrado ainda

-

Clique em “Adicionar cliente” para iniciar o monitoramento.

-
-
-
-
-
-
-
- -
-
-
-

Últimos eventos

-
-
- {% if recent_events %} -
    - {% for event in recent_events %} -
  • -

    {{ event.summary }}

    -

    - {{ event.received_at.strftime('%d/%m/%Y %H:%M') }} · {{ event.client_name or event.client_document }} -

    - {% if event.description %} -

    {{ event.description }}

    - {% endif %} -
    • - {% endfor %} -
-
- - Ver histórico completo - -
- {% else %} -

Nenhum evento recebido até o momento.

- {% endif %} -
-
- -
-
-

Clientes a verificar

-
-
- {% if stale_clients %} -
    - {% for entry in stale_clients %} -
  • - {{ entry.client.name }} -
    - Última verificação: - {% if entry.client.last_checked %} - {{ entry.client.last_checked.strftime('%d/%m/%Y %H:%M') }} - {% else %} - nunca - {% endif %} -
    -
    - -
    -
    • - {% endfor %} -
- {% else %} -

Todos os clientes foram verificados nas últimas 24 horas.

- {% endif %} -
-
- -
-
-

Guia rápido de operação

-
-
-

- • Atualize certificados e tokens nas ações de cada cliente.
- • Utilize o ciclo global diário para consolidar notificações.
- • Configure o MONITOR_CONFIG e o MONITOR_DATABASE em produção. -

-

- A documentação completa está no arquivo README.md do projeto. -

-
-
-
-
-{% endblock %} diff --git a/uninstall.sh b/uninstall.sh new file mode 100644 index 0000000..2a80760 --- /dev/null +++ b/uninstall.sh @@ -0,0 +1,23 @@ +#!/usr/bin/env bash +set -euo pipefail + +SERVICE_NAME="coletor-fiscal" +APP_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +VENV_DIR="$APP_DIR/.venv" + +sudo systemctl stop ${SERVICE_NAME} || true +sudo systemctl disable ${SERVICE_NAME} || true +sudo rm -f /etc/systemd/system/${SERVICE_NAME}.service +sudo systemctl daemon-reload + +read -rp "Deseja remover o ambiente virtual (.venv)? [s/N] " resposta +case "$resposta" in + s|S|sim|SIM) + rm -rf "$VENV_DIR" + ;; + *) + echo "Ambiente virtual preservado." + ;; +esac + +echo "Remoção concluída." diff --git a/update.sh b/update.sh new file mode 100644 index 0000000..1bfdb0f --- /dev/null +++ b/update.sh @@ -0,0 +1,19 @@ +#!/usr/bin/env bash +set -euo pipefail + +SERVICE_NAME="coletor-fiscal" +APP_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +VENV_DIR="$APP_DIR/.venv" + +cd "$APP_DIR" + +git pull --ff-only + +if [ -d "$VENV_DIR" ]; then + source "$VENV_DIR/bin/activate" + pip install -r requirements.txt +fi + +sudo systemctl restart ${SERVICE_NAME} + +echo "Atualização concluída. Serviço reiniciado." diff --git a/verify.bat b/verify.bat new file mode 100644 index 0000000..ca035ca --- /dev/null +++ b/verify.bat @@ -0,0 +1,42 @@ +@echo off +SETLOCAL ENABLEDELAYEDEXPANSION + +REM Executa verificacoes rapidas do ambiente Coletor Fiscal v3.2. +REM - Valida existencia do ambiente virtual +REM - Checa conflitos de dependencias com "pip check" +REM - Compila o codigo fonte para garantir ausencia de erros de sintaxe + +SET APP_DIR=%~dp0 +SET VENV_DIR=%APP_DIR%\.venv + +IF NOT EXIST "%VENV_DIR%\Scripts\activate.bat" ( + echo [ERRO] Ambiente virtual nao encontrado em %VENV_DIR%. + echo Execute install.bat ou force_install.bat para preparar o ambiente. + EXIT /B 1 +) + +CALL "%VENV_DIR%\Scripts\activate.bat" + +echo [1/3] Verificando versao do Python... +python --version || EXIT /B 1 + +echo [2/3] Conferindo dependencias com pip check... +python -m pip check +IF ERRORLEVEL 1 ( + echo. + echo [AVISO] Conflitos detectados. Execute "force_install.bat" ou reinstale dependencias manualmente. + EXIT /B 1 +) + +echo [3/3] Compilando fontes (app/ e web/)... +python -m compileall "%APP_DIR%app" "%APP_DIR%web" +IF ERRORLEVEL 1 ( + echo. + echo [ERRO] Falha na compilacao. Analise as mensagens acima. + EXIT /B 1 +) + +echo. +echo Ambiente validado com sucesso. + +ENDLOCAL diff --git a/web/app.py b/web/app.py new file mode 100644 index 0000000..d417c44 --- /dev/null +++ b/web/app.py @@ -0,0 +1,142 @@ +"""Aplicativo Streamlit para o Coletor Fiscal v3.2 SaaS-Ready.""" + +from __future__ import annotations + +import os +import sys +from pathlib import Path +from typing import List + +import streamlit as st + +# Garante que o pacote backend ``app`` seja resolvido corretamente quando este +# arquivo for executado pelo Streamlit como ``app`` (nome do arquivo). +FRONTEND_FILE = Path(__file__).resolve() +PROJECT_ROOT = FRONTEND_FILE.parents[1] + +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +existing_app_module = sys.modules.get("app") +if existing_app_module and getattr(existing_app_module, "__file__", None): + if Path(existing_app_module.__file__).resolve() == FRONTEND_FILE: + del sys.modules["app"] + +# Após normalizar o caminho garantimos que as importações usem o backend real. +from app.database import init_db, session_scope # type: ignore # noqa: E402 +from app.config import settings # type: ignore # noqa: E402 +from app.models import Empresa # type: ignore # noqa: E402 +from app.services.coleta import coletar_todos # type: ignore # noqa: E402 +from app.utils.cnpj import sanitize_cnpj, validate_cnpj # type: ignore # noqa: E402 +from app.utils.helpers import setup_logging # type: ignore # noqa: E402 + +LOGGER = setup_logging() + +USUARIO_PADRAO = os.getenv("WEB_USER", "admin") +SENHA_PADRAO = os.getenv("WEB_PASS", "admin") + + +def autenticar(usuario: str, senha: str) -> bool: + """Autentica o usuário utilizando credenciais básicas do ambiente.""" + return usuario == USUARIO_PADRAO and senha == SENHA_PADRAO + + +def carregar_empresas() -> List[Empresa]: + with session_scope() as session: + return session.query(Empresa).order_by(Empresa.nome).all() + + +def salvar_empresa(nome: str, cnpj: str, uf: str, senha_cert: str | None = None) -> None: + with session_scope() as session: + empresa = Empresa(nome=nome, cnpj=cnpj, uf=uf, certificado_senha=senha_cert) + session.add(empresa) + session.commit() + LOGGER.info("Empresa %s cadastrada", nome) + + +def executar_coletas(empresa: Empresa, chaves_nfce: list[str]) -> dict[str, int]: + with session_scope() as session: + empresa_refrescada = session.query(Empresa).filter_by(id=empresa.id).one() + resultado = coletar_todos(session, empresa_refrescada, chaves_nfce) + session.commit() + return resultado + + +def pagina_login() -> None: + st.set_page_config(page_title="Coletor Fiscal v3.2", page_icon="🧾", layout="centered") + st.title("Coletor Fiscal v3.2 - Login") + usuario = st.text_input("Usuário") + senha = st.text_input("Senha", type="password") + if st.button("Entrar"): + if autenticar(usuario, senha): + st.session_state["autenticado"] = True + st.experimental_rerun() + else: + st.error("Credenciais inválidas. Verifique usuário e senha.") + + +def pagina_dashboard() -> None: + st.set_page_config(page_title="Coletor Fiscal v3.2", page_icon="🧾", layout="wide") + st.sidebar.title("Coletor Fiscal v3.2") + st.sidebar.write(f"Ambiente: {settings.ambiente.upper()}") + st.sidebar.write(f"Banco de dados: {settings.database_url}") + if st.sidebar.button("Sair"): + st.session_state.clear() + st.experimental_rerun() + + st.header("Selecione a empresa para coletar documentos") + empresas = carregar_empresas() + if not empresas: + st.warning("Nenhuma empresa cadastrada. Cadastre abaixo antes de coletar.") + + empresa_nomes = {f"{emp.nome} ({emp.cnpj})": emp for emp in empresas} + selecionado = st.selectbox("Empresa", options=list(empresa_nomes.keys())) if empresas else None + empresa_escolhida = empresa_nomes.get(selecionado) if selecionado else None + + with st.expander("Cadastrar nova empresa"): + with st.form("form_empresa"): + nome = st.text_input("Nome Fantasia") + cnpj = st.text_input("CNPJ") + uf = st.text_input("UF", value="GO", max_chars=2) + senha_cert = st.text_input("Senha do Certificado", type="password") + enviar = st.form_submit_button("Salvar Empresa") + if enviar: + cnpj_limpo = sanitize_cnpj(cnpj) + if not validate_cnpj(cnpj_limpo): + st.error("CNPJ inválido.") + else: + try: + salvar_empresa(nome, cnpj_limpo, uf.upper(), senha_cert) + st.success("Empresa cadastrada com sucesso!") + st.experimental_rerun() + except Exception as exc: # pragma: no cover - exibido apenas em runtime + st.error(f"Erro ao salvar empresa: {exc}") + + chaves_nfce = st.text_area( + "Informe as chaves NFC-e (uma por linha)", + help="As chaves serão utilizadas na raspagem pública da SEFAZ-GO.", + ) + lista_chaves = [linha.strip() for linha in chaves_nfce.splitlines() if linha.strip()] + + if st.button("🔄 Coletar Agora", disabled=empresa_escolhida is None): + if not empresa_escolhida: + st.error("Selecione uma empresa antes de iniciar a coleta.") + else: + with st.spinner("Executando coletas seguras..."): + resultado = executar_coletas(empresa_escolhida, lista_chaves) + st.success( + f"Coletas finalizadas! NFe: {resultado['nfe']} | NFC-e: {resultado['nfce']}" + ) + st.info( + "As integrações com SEFAZ estão prontas e basta remover os comentários indicados no código para ativá-las." + ) + + +init_db() +if "autenticado" not in st.session_state: + st.session_state["autenticado"] = False + +if not st.session_state["autenticado"]: + pagina_login() +else: + pagina_dashboard() diff --git a/webapp.py b/webapp.py deleted file mode 100644 index a0a25aa..0000000 --- a/webapp.py +++ /dev/null @@ -1,351 +0,0 @@ -"""Aplicação web para operar o monitoramento do eCAC.""" -from __future__ import annotations - -import json -import os -from datetime import datetime, timedelta -from pathlib import Path -from typing import Any, Dict, Optional - -from flask import ( - Flask, - Response, - flash, - redirect, - render_template, - request, - url_for, -) - -from main import ( - AUTH_MODES, - AlertDispatcher, - Client, - DashboardMetrics, - DatabaseManager, - EcacAPIClient, - EcacMonitor, - MonitorConfig, -) - - -def _load_config() -> MonitorConfig: - config_path = Path(os.environ.get("MONITOR_CONFIG", "monitor_config.json")) - if not config_path.exists(): - raise FileNotFoundError( - "Arquivo de configuração do monitor não encontrado. " - "Defina MONITOR_CONFIG com o caminho correto." - ) - return MonitorConfig.load(config_path) - - -def _load_database() -> DatabaseManager: - db_path = Path(os.environ.get("MONITOR_DATABASE", "monitor.db")) - return DatabaseManager(db_path) - - -def create_app() -> Flask: - app = Flask(__name__) - app.config["SECRET_KEY"] = os.environ.get("FLASK_SECRET_KEY", "change-me") - - db = _load_database() - - def _build_monitor() -> EcacMonitor: - config = _load_config() - api_client = EcacAPIClient(config) - dispatcher = AlertDispatcher(config.webhook_url, config.verify_ssl, config.timeout) - return EcacMonitor(db, api_client, dispatcher, config.poll_interval, config.verify_ssl) - - @app.context_processor - def inject_globals() -> Dict[str, Any]: - config_env = os.environ.get("MONITOR_CONFIG") - available = bool(config_env and Path(config_env).exists()) - return {"config_available": available} - - def _prepare_status(client: Client) -> Optional[Dict[str, Any]]: - if not client.last_status: - return None - try: - return json.loads(client.last_status) - except json.JSONDecodeError: - return {"raw": client.last_status} - - def _format_event(event) -> Dict[str, Any]: - payload = event.payload - summary = None - description = None - category = None - reference = None - if isinstance(payload, dict): - for key in ("title", "titulo", "subject", "assunto"): - if payload.get(key): - summary = str(payload[key]) - break - description = ( - payload.get("description") - or payload.get("mensagem") - or payload.get("message") - ) - category = payload.get("category") or payload.get("categoria") - reference = ( - payload.get("reference") - or payload.get("protocolo") - or payload.get("id") - ) - if summary is None: - if isinstance(payload, (str, int, float)): - summary = str(payload) - else: - summary = json.dumps(payload, ensure_ascii=False) - return { - "id": event.id, - "client_document": event.client_document, - "client_name": getattr(event, "client_name", None), - "received_at": event.received_at, - "summary": summary, - "description": description, - "category": category, - "reference": reference, - "payload": payload, - } - - @app.get("/") - def index() -> str: - clients = db.list_clients() - metrics: DashboardMetrics = db.get_dashboard_metrics() - enriched = [] - for client in clients: - status = _prepare_status(client) - notifications = 0 - obligations = 0 - if isinstance(status, dict): - notifications_field = status.get("notifications") or status.get("avisos") - if isinstance(notifications_field, list): - notifications = len(notifications_field) - obligations_field = ( - status.get("obligations") - or status.get("obrigacoes") - or status.get("obrigações") - ) - if isinstance(obligations_field, list): - obligations = len(obligations_field) - enriched.append( - { - "client": client, - "status": status, - "notifications": notifications, - "obligations": obligations, - } - ) - - stale_threshold = datetime.utcnow() - timedelta(hours=24) - stale_clients = [ - entry - for entry in enriched - if not entry["client"].last_checked - or entry["client"].last_checked < stale_threshold - ] - recent_events = [_format_event(event) for event in db.list_events(limit=6)] - - return render_template( - "index.html", - clients=enriched, - metrics=metrics, - stale_clients=stale_clients, - recent_events=recent_events, - ) - - @app.get("/clients/new") - def new_client() -> str: - return render_template("add_client.html", auth_modes=sorted(AUTH_MODES)) - - @app.post("/clients") - def create_client() -> Response: - form = request.form - auth_mode = form.get("auth_mode", "certificate") - if auth_mode not in AUTH_MODES: - flash("Modo de autenticação inválido", "error") - return redirect(url_for("new_client")) - - required = ["document", "name", "client_type"] - if auth_mode == "certificate": - required.extend(["certificate", "key"]) - missing = [field for field in required if not form.get(field)] - if missing: - flash(f"Campos obrigatórios ausentes: {', '.join(missing)}", "error") - return redirect(url_for("new_client")) - - try: - db.add_client( - document=form["document"].strip(), - name=form["name"].strip(), - client_type=form["client_type"], - auth_mode=auth_mode, - certificate_path=Path(form["certificate"]).expanduser() - if form.get("certificate") - else None, - key_path=Path(form["key"]).expanduser() if form.get("key") else None, - certificate_password=form.get("certificate_password") or None, - procuracao_token=form.get("procuracao_token") or None, - ) - except Exception as exc: # noqa: BLE001 - flash(f"Erro ao cadastrar cliente: {exc}", "error") - return redirect(url_for("new_client")) - - flash("Cliente cadastrado com sucesso", "success") - return redirect(url_for("index")) - - @app.get("/clients/") - def client_detail(document: str) -> str: - client = db.get_client(document) - if not client: - flash("Cliente não encontrado", "error") - return redirect(url_for("index")) - status = _prepare_status(client) - notifications = [] - obligations = [] - metadata: Dict[str, Any] = {} - if isinstance(status, dict): - raw_notifications = status.get("notifications") or status.get("avisos") - if isinstance(raw_notifications, list): - notifications = raw_notifications - raw_obligations = ( - status.get("obligations") - or status.get("obrigacoes") - or status.get("obrigações") - ) - if isinstance(raw_obligations, list): - obligations = raw_obligations - meta_candidate = status.get("metadata") or status.get("metadados") - if isinstance(meta_candidate, dict): - metadata = meta_candidate - recent_events = [ - _format_event(event) for event in db.list_events(document, limit=5) - ] - return render_template( - "client_detail.html", - client=client, - status=status, - notifications=notifications, - obligations=obligations, - metadata=metadata, - recent_events=recent_events, - ) - - @app.post("/run-cycle") - def run_cycle() -> Response: - try: - monitor = _build_monitor() - monitor.run_cycle() - except FileNotFoundError as exc: - flash(str(exc), "error") - return redirect(url_for("index")) - except Exception as exc: # noqa: BLE001 - flash(f"Erro ao executar ciclo: {exc}", "error") - return redirect(url_for("index")) - - flash("Ciclo executado com sucesso", "success") - return redirect(url_for("index")) - - @app.post("/clients//run") - def run_cycle_for_client(document: str) -> Response: - try: - monitor = _build_monitor() - monitor.run_for_client(document) - except ValueError as exc: - flash(str(exc), "error") - return redirect(url_for("index")) - except FileNotFoundError as exc: - flash(str(exc), "error") - return redirect(url_for("client_detail", document=document)) - except Exception as exc: # noqa: BLE001 - flash(f"Erro ao executar ciclo: {exc}", "error") - return redirect(url_for("client_detail", document=document)) - flash("Ciclo executado com sucesso", "success") - return redirect(url_for("client_detail", document=document)) - - @app.get("/clients//events") - def client_events(document: str) -> str: - client = db.get_client(document) - if not client: - flash("Cliente não encontrado", "error") - return redirect(url_for("index")) - try: - limit = max(1, min(500, int(request.args.get("limit", 50)))) - offset = max(0, int(request.args.get("offset", 0))) - except ValueError: - flash("Parâmetros de paginação inválidos", "error") - return redirect(url_for("client_events", document=document)) - events = db.list_events(document, limit=limit, offset=offset) - formatted_events = [_format_event(event) for event in events] - return render_template( - "client_events.html", - client=client, - events=formatted_events, - limit=limit, - offset=offset, - next_offset=offset + limit if len(events) == limit else None, - prev_offset=offset - limit if offset - limit >= 0 else None, - ) - - @app.get("/clients//edit") - def edit_client(document: str) -> str: - client = db.get_client(document) - if not client: - flash("Cliente não encontrado", "error") - return redirect(url_for("index")) - return render_template( - "edit_client.html", client=client, auth_modes=sorted(AUTH_MODES) - ) - - @app.post("/clients/") - def update_client(document: str) -> Response: - client = db.get_client(document) - if not client: - flash("Cliente não encontrado", "error") - return redirect(url_for("index")) - form = request.form - fields: Dict[str, Any] = {} - if form.get("name"): - fields["name"] = form["name"].strip() - if form.get("client_type"): - fields["client_type"] = form["client_type"] - if form.get("auth_mode"): - fields["auth_mode"] = form.get("auth_mode") - if form.get("certificate"): - fields["certificate_path"] = Path(form["certificate"]).expanduser() - if form.get("key"): - fields["key_path"] = Path(form["key"]).expanduser() - if form.get("certificate_password") is not None: - fields["certificate_password"] = form.get("certificate_password") or None - if form.get("procuracao_token") is not None: - fields["procuracao_token"] = form.get("procuracao_token") or None - if not fields: - flash("Nenhuma alteração informada", "warning") - return redirect(url_for("edit_client", document=document)) - try: - db.update_client(document, **fields) - except Exception as exc: # noqa: BLE001 - flash(f"Erro ao atualizar cliente: {exc}", "error") - return redirect(url_for("edit_client", document=document)) - flash("Cliente atualizado com sucesso", "success") - return redirect(url_for("client_detail", document=document)) - - @app.post("/clients//delete") - def delete_client(document: str) -> Response: - try: - db.delete_client(document) - except ValueError as exc: - flash(str(exc), "error") - return redirect(url_for("index")) - flash("Cliente removido", "success") - return redirect(url_for("index")) - - return app - - -app = create_app() - - -if __name__ == "__main__": - app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 8000)), debug=False)