Advanced Project: Concurrent Data Processor

CSV  ->  [semáforo]  ->  ler (thread, com retentativas e timeout)
                              |
                              v
                    agregar (outro processo)  ->  Resultado
                              |
                         erro em qualquer etapa  ->  Falha

from dataclasses import dataclass, field


@dataclass(frozen=True)
class Resultado:
    arquivo: str
    linhas: int
    total_centavos: int
    por_regiao: dict[str, int]


@dataclass(frozen=True)
class Falha:
    arquivo: str
    motivo: str


@dataclass
class Relatorio:
    resultados: list[Resultado] = field(default_factory=list)
    falhas: list[Falha] = field(default_factory=list)

    @property
    def total_centavos(self) -> int:
        return sum(r.total_centavos for r in self.resultados)

    @property
    def por_regiao(self) -> dict[str, int]:
        totais: dict[str, int] = {}
        for resultado in self.resultados:
            for regiao, valor in resultado.por_regiao.items():
                totais[regiao] = totais.get(regiao, 0) + valor
        return totais

import csv
import io


def agregar(texto: str) -> tuple[int, int, dict[str, int]]:
    """Etapa de CPU: roda em outro processo. Devolve (linhas, total, total por região).

    Precisa ser uma função do nível do módulo, com argumentos e retorno serializáveis,
    para poder ser enviada a um processo filho.
    """
    leitor = csv.reader(io.StringIO(texto))
    cabecalho = next(leitor, None)
    if cabecalho != ["regiao", "valor_centavos"]:
        raise ValueError(f"cabeçalho inesperado: {cabecalho}")
    linhas = total = 0
    por_regiao: dict[str, int] = {}
    for numero, linha in enumerate(leitor, start=2):
        try:
            regiao, valor_texto = linha
            valor = int(valor_texto)
        except ValueError as erro:
            raise ValueError(f"linha {numero} inválida: {linha}") from erro
        por_regiao[regiao] = por_regiao.get(regiao, 0) + valor
        total += valor
        linhas += 1
    return linhas, total, por_regiao

import asyncio
import logging
from collections.abc import Awaitable, Callable
from pathlib import Path

log = logging.getLogger(__name__)


def ler_texto(caminho: Path) -> str:
    return caminho.read_text(encoding="utf-8")


async def ler_com_retentativas(
    caminho: Path,
    *,
    tentativas: int = 3,
    base: float = 0.1,
    dormir: Callable[[float], Awaitable[None]] = asyncio.sleep,
    ler: Callable[[Path], str] = ler_texto,
) -> str:
    """Lê o arquivo em uma thread, para não bloquear o laço de eventos, com espera crescente."""
    for numero in range(1, tentativas + 1):
        try:
            return await asyncio.to_thread(ler, caminho)
        except OSError as erro:
            if numero == tentativas:
                raise
            espera = base * 2 ** (numero - 1)
            log.warning(
                "leitura falhou (%s), tentativa %d, nova tentativa em %.2fs", erro, numero, espera
            )
            await dormir(espera)
    raise AssertionError("inalcançável")

import logging
from contextvars import ContextVar

id_arquivo: ContextVar[str] = ContextVar("id_arquivo", default="-")


class FiltroArquivo(logging.Filter):
    """Acrescenta o nome do arquivo em processamento a cada linha de log."""

    def filter(self, record: logging.LogRecord) -> bool:
        record.arquivo = id_arquivo.get()
        return True

import asyncio
import logging
import multiprocessing
from collections.abc import Awaitable, Callable
from concurrent.futures import Executor, ProcessPoolExecutor
from pathlib import Path

from processador.calculo import agregar
from processador.contexto import id_arquivo
from processador.leitura import ler_com_retentativas, ler_texto
from processador.modelos import Falha, Relatorio, Resultado

log = logging.getLogger(__name__)


def criar_pool(processos: int) -> ProcessPoolExecutor:
    """Pool de processos com o método de início 'spawn' escolhido de forma explícita.

    O fork copiaria para o filho threads em estado inconsistente (como as do asyncio.to_thread),
    e o método padrão muda conforme o sistema e a versão do Python. O spawn se comporta igual
    em todos, ao custo de iniciar cada processo importando o código de novo.
    """
    return ProcessPoolExecutor(
        max_workers=processos, mp_context=multiprocessing.get_context("spawn")
    )


async def processar_arquivo(
    caminho: Path,
    *,
    semaforo: asyncio.Semaphore,
    pool: Executor,
    timeout: float,
    dormir: Callable[[float], Awaitable[None]],
    ler: Callable[[Path], str],
) -> Resultado | Falha:
    token = id_arquivo.set(caminho.name)
    try:
        async with semaforo:
            async with asyncio.timeout(timeout):
                texto = await ler_com_retentativas(caminho, dormir=dormir, ler=ler)
        laco = asyncio.get_running_loop()
        linhas, total, por_regiao = await laco.run_in_executor(pool, agregar, texto)
        log.info("processado: %d linhas", linhas)
        return Resultado(caminho.name, linhas, total, por_regiao)
    except (OSError, TimeoutError, ValueError) as erro:
        motivo = str(erro) or type(erro).__name__
        log.error("falhou: %s", motivo)
        return Falha(caminho.name, motivo)
    finally:
        id_arquivo.reset(token)


async def processar(
    pasta: Path,
    *,
    pool: Executor,
    leituras: int = 4,
    timeout: float = 5.0,
    dormir: Callable[[float], Awaitable[None]] = asyncio.sleep,
    ler: Callable[[Path], str] = ler_texto,
) -> Relatorio:
    """Lê vários arquivos com concorrência limitada e calcula em processos separados.

    Uma falha em um arquivo vira um registro de Falha e não derruba os demais.
    """
    semaforo = asyncio.Semaphore(leituras)
    saidas = await asyncio.gather(
        *(
            processar_arquivo(
                caminho, semaforo=semaforo, pool=pool, timeout=timeout, dormir=dormir, ler=ler
            )
            for caminho in sorted(pasta.glob("*.csv"))
        )
    )
    relatorio = Relatorio()
    for saida in saidas:
        if isinstance(saida, Resultado):
            relatorio.resultados.append(saida)
        else:
            relatorio.falhas.append(saida)
    return relatorio

import argparse
import asyncio
import csv
import logging
import sys
from pathlib import Path

from processador.contexto import FiltroArquivo
from processador.modelos import Relatorio
from processador.pipeline import criar_pool, processar

REGIOES = ["norte", "nordeste", "centro-oeste", "sudeste", "sul"]


def gerar_dados(pasta: Path, arquivos: int, linhas: int) -> None:
    """Gera arquivos CSV determinísticos (sem aleatoriedade, para o resultado ser repetível)."""
    pasta.mkdir(parents=True, exist_ok=True)
    for i in range(arquivos):
        with (pasta / f"vendas_{i:02d}.csv").open("w", encoding="utf-8", newline="") as arquivo:
            escritor = csv.writer(arquivo)
            escritor.writerow(["regiao", "valor_centavos"])
            for j in range(linhas):
                escritor.writerow([REGIOES[(i + j) % 5], (i * 7919 + j * 104729) % 50_000 + 100])


def formatar_reais(centavos: int) -> str:
    reais, resto = divmod(centavos, 100)
    return f"R$ {reais:,}".replace(",", ".") + f",{resto:02d}"


def imprimir(relatorio: Relatorio) -> None:
    print(f"arquivos processados: {len(relatorio.resultados)}  falhas: {len(relatorio.falhas)}")
    for regiao, total in sorted(relatorio.por_regiao.items()):
        print(f"  {regiao:<13}{formatar_reais(total):>16}")
    print(f"  {'total':<13}{formatar_reais(relatorio.total_centavos):>16}")
    for falha in relatorio.falhas:
        print(f"  FALHA {falha.arquivo}: {falha.motivo}")


def configurar_logs(verboso: bool) -> None:
    manipulador = logging.StreamHandler(sys.stderr)
    manipulador.addFilter(FiltroArquivo())
    manipulador.setFormatter(logging.Formatter("%(levelname)s [%(arquivo)s] %(message)s"))
    logging.basicConfig(
        level=logging.INFO if verboso else logging.WARNING, handlers=[manipulador], force=True
    )


def criar_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="processador")
    parser.add_argument("-v", "--verboso", action="store_true")
    sub = parser.add_subparsers(dest="comando", required=True)

    gerar = sub.add_parser("gerar", help="gera arquivos CSV de exemplo")
    gerar.add_argument("pasta", type=Path)
    gerar.add_argument("--arquivos", type=int, default=6)
    gerar.add_argument("--linhas", type=int, default=50_000)

    proc = sub.add_parser("processar", help="processa todos os CSV de uma pasta")
    proc.add_argument("pasta", type=Path)
    proc.add_argument("--leituras", type=int, default=4, help="leituras simultâneas")
    proc.add_argument("--processos", type=int, default=2, help="processos de cálculo")
    proc.add_argument("--timeout", type=float, default=5.0, help="segundos por arquivo")
    return parser


def main(argv: list[str] | None = None) -> int:
    args = criar_parser().parse_args(argv)
    configurar_logs(args.verboso)
    if args.comando == "gerar":
        gerar_dados(args.pasta, args.arquivos, args.linhas)
        print(f"{args.arquivos} arquivos gerados em {args.pasta}")
        return 0
    with criar_pool(args.processos) as pool:
        relatorio = asyncio.run(
            processar(args.pasta, pool=pool, leituras=args.leituras, timeout=args.timeout)
        )
    imprimir(relatorio)
    return 1 if relatorio.falhas else 0

from processador.cli import main

if __name__ == "__main__":
    raise SystemExit(main())

import pytest

from processador.calculo import agregar


def test_agrega_por_regiao() -> None:
    texto = "regiao,valor_centavos\nsul,100\nnorte,50\nsul,25\n"
    assert agregar(texto) == (3, 175, {"sul": 125, "norte": 50})


def test_cabecalho_errado_e_recusado() -> None:
    with pytest.raises(ValueError, match="cabeçalho"):
        agregar("a,b\n1,2\n")


def test_linha_invalida_informa_o_numero_da_linha() -> None:
    with pytest.raises(ValueError, match="linha 3"):
        agregar("regiao,valor_centavos\nsul,100\nsul,abc\n")

import asyncio
import threading
import time
from pathlib import Path

from processador.cli import gerar_dados
from processador.pipeline import criar_pool, processar


def rodar(pasta: Path, **opcoes):  # type: ignore[no-untyped-def]
    async def principal():  # type: ignore[no-untyped-def]
        with criar_pool(2) as pool:
            return await processar(pasta, pool=pool, **opcoes)

    return asyncio.run(principal())


def test_processa_todos_e_soma(tmp_path: Path) -> None:
    gerar_dados(tmp_path, arquivos=3, linhas=100)
    relatorio = rodar(tmp_path)
    assert len(relatorio.resultados) == 3
    assert relatorio.falhas == []
    assert sum(r.linhas for r in relatorio.resultados) == 300


def test_arquivo_corrompido_nao_derruba_os_outros(tmp_path: Path) -> None:
    gerar_dados(tmp_path, arquivos=2, linhas=10)
    (tmp_path / "quebrado.csv").write_text("regiao,valor_centavos\nsul,abc\n", encoding="utf-8")
    relatorio = rodar(tmp_path)
    assert len(relatorio.resultados) == 2
    assert [f.arquivo for f in relatorio.falhas] == ["quebrado.csv"]
    assert "linha 2" in relatorio.falhas[0].motivo


def test_retentativas_com_espera_crescente(tmp_path: Path) -> None:
    gerar_dados(tmp_path, arquivos=1, linhas=5)
    esperas: list[float] = []
    falhas_restantes = {"n": 2}

    async def dormir_falso(segundos: float) -> None:
        esperas.append(segundos)

    def ler_instavel(caminho: Path) -> str:
        if falhas_restantes["n"] > 0:
            falhas_restantes["n"] -= 1
            raise OSError("disco ocupado")
        return caminho.read_text(encoding="utf-8")

    relatorio = rodar(tmp_path, dormir=dormir_falso, ler=ler_instavel)
    assert len(relatorio.resultados) == 1
    assert esperas == [0.1, 0.2]


def test_timeout_vira_falha(tmp_path: Path) -> None:
    gerar_dados(tmp_path, arquivos=1, linhas=5)

    def ler_lento(caminho: Path) -> str:
        time.sleep(0.3)
        return caminho.read_text(encoding="utf-8")

    relatorio = rodar(tmp_path, timeout=0.05, ler=ler_lento)
    assert relatorio.resultados == []
    assert relatorio.falhas[0].motivo == "TimeoutError"


def test_limite_de_leituras_simultaneas(tmp_path: Path) -> None:
    gerar_dados(tmp_path, arquivos=6, linhas=5)
    trava = threading.Lock()
    ativas = {"agora": 0, "maximo": 0}

    def ler_medindo(caminho: Path) -> str:
        with trava:
            ativas["agora"] += 1
            ativas["maximo"] = max(ativas["maximo"], ativas["agora"])
        time.sleep(0.05)
        with trava:
            ativas["agora"] -= 1
        return caminho.read_text(encoding="utf-8")

    rodar(tmp_path, leituras=2, ler=ler_medindo)
    assert ativas["maximo"] <= 2

from pathlib import Path

import pytest

from processador.cli import main


def test_gerar_e_processar(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
    pasta = str(tmp_path / "dados")
    assert main(["gerar", pasta, "--arquivos", "3", "--linhas", "50"]) == 0
    assert main(["processar", pasta, "--processos", "2"]) == 0
    saida = capsys.readouterr().out
    assert "arquivos processados: 3  falhas: 0" in saida
    assert "total" in saida


def test_codigo_de_saida_1_quando_ha_falha(tmp_path: Path) -> None:
    (tmp_path / "ruim.csv").write_text("x,y\n", encoding="utf-8")
    assert main(["processar", str(tmp_path), "--processos", "1"]) == 1

cd projetos/processador
uv sync
uv run pytest
uv run mypy
uv run ruff check .

uv run processador gerar dados --arquivos 6 --linhas 50000
uv run processador processar dados

arquivos processados: 6  falhas: 1
  (as seis regiões, somando só os arquivos válidos)
  FALHA quebrado.csv: linha 2 inválida: ['sul', 'abc']

INFO [vendas_00.csv] processado: 50000 linhas
INFO [vendas_01.csv] processado: 50000 linhas
INFO [vendas_02.csv] processado: 50000 linhas