Capítulo 36, Projetos

Projeto Básico: relatório de funcionários

O projeto final do curso original, completo, testado e transformado em programa: carregar um CSV sujo, limpar, responder às perguntas de negócio e entregar um relatório. Faça depois do capítulo 15.

O problema

Todos os capítulos do nível Básico trabalharam com 15 funcionários. Aqui o mesmo roteiro roda sobre o funcionarios_100.csv, com 104 linhas e mais sujeira, e vira um programa: recebe um arquivo, limpa, calcula e imprime um relatório. O que o capítulo 14 fez em uma função solta vira um pacote com testes.

Etapa do curso	Onde está no projeto
Carregar e inspecionar (caps. 5 e 6)	`cli.py` lê o arquivo, e `relatorio.qualidade` mostra o diagnóstico
Selecionar e filtrar (caps. 7 e 8)	Usados dentro de `analise.py`
Limpar: ausentes, duplicatas, tipos, texto (caps. 9 e 10)	`limpeza.py`
Novas colunas (cap. 11)	`nivel` e `anos_de_casa`, no fim de `limpar`
Ordenar e ranquear (cap. 12)	`top_desempenho`, com desempate explícito
Agrupar (cap. 13)	`resumo_por_departamento`
Perguntas de negócio (cap. 14)	`perguntas`, e o `relatorio.py` as imprime

As decisões que eu tomei

Cada passo de limpeza é uma função pequena e pura, que recebe uma Series e devolve outra: limpar_salario, padronizar_departamento, converter_datas. Isso permite testar cada regra sozinha, com três valores escritos à mão.

A limpeza não adivinha. O departamento que não está no dicionário vira ausente (e é visível). A data em um formato fora dos dois declarados levanta um erro, em vez de ser interpretada de um jeito qualquer. O salário escrito como R$ 4650 é convertido sem perder nenhum valor.

O último cadastro de cada id vence, porque os repetidos do arquivo são atualizações (capítulo 10). Preencher não apaga a evidência: as colunas salario_informado e idade_informada guardam o que faltava antes do preenchimento.

Empates são mostrados, não escondidos. "Qual cidade tem mais funcionários?" devolve a lista das empatadas, e o top 5 tem desempate explícito (nota, depois projetos, depois o menor id), o que o torna independente da ordem das linhas.

O código

A limpeza é o coração do projeto. Repare que a função limpar devolve uma tabela nova (a cópia sob escrita, do capítulo 27, e o assign garantem que a recebida não muda) e que cada regra tem a sua função:

projetos/funcionarios/src/funcionarios/limpeza.py

import numpy as np
import pandas as pd

CANONICO = {
    "engenharia": "Engenharia",
    "marketing": "Marketing",
    "finanças": "Finanças",
    "ciência de dados": "Ciência de Dados",
    "rh": "RH",
}
REFERENCIA = pd.Timestamp("2026-01-01")


def limpar_salario(serie: pd.Series) -> pd.Series:
    """Converte para número, tirando o "R$ " dos valores escritos como texto, sem perder nenhum."""
    if pd.api.types.is_numeric_dtype(serie):
        return serie.astype(float)
    return pd.to_numeric(serie.str.replace("R$ ", "", regex=False))


def padronizar_departamento(serie: pd.Series) -> pd.Series:
    """Mapeia para os nomes corretos. O que não está em CANONICO vira ausente (e é visível)."""
    return serie.str.strip().str.lower().map(CANONICO)


def converter_datas(serie: pd.Series) -> pd.Series:
    """Aceita só os formatos declarados (ISO e dd/mm/aaaa) e recusa o resto, sem adivinhar."""
    if pd.api.types.is_datetime64_any_dtype(serie):
        return serie
    iso = pd.to_datetime(serie, format="%Y-%m-%d", errors="coerce")
    br = pd.to_datetime(serie, format="%d/%m/%Y", errors="coerce")
    resultado = iso.fillna(br)
    desconhecidas = resultado.isna() & serie.notna()
    if desconhecidas.any():
        raise ValueError(f"{int(desconhecidas.sum())} datas em formato desconhecido")
    return resultado


def limpar(bruto: pd.DataFrame, referencia: pd.Timestamp = REFERENCIA) -> pd.DataFrame:
    """Devolve uma tabela nova e limpa. O último cadastro de cada id vence (é a atualização)."""
    df = (
        bruto.assign(
            salario=limpar_salario(bruto["salario"]),
            departamento=padronizar_departamento(bruto["departamento"]),
            data_admissao=converter_datas(bruto["data_admissao"]),
        )
        .drop_duplicates("id_funcionario", keep="last")
        .reset_index(drop=True)
    )
    df = df.assign(salario_informado=df["salario"].notna(), idade_informada=df["idade"].notna())
    mediana_do_depto = df.groupby("departamento")["salario"].transform("median")
    df = df.assign(
        salario=df["salario"].fillna(mediana_do_depto).fillna(df["salario"].median()),
        idade=df["idade"].fillna(df["idade"].median()),
        departamento=df["departamento"].fillna("Não informado"),
        cidade=df["cidade"].fillna("Não informada"),
    )
    return df.assign(
        nivel=pd.cut(
            df["experiencia"],
            bins=[-np.inf, 2, 5, np.inf],
            right=False,
            labels=["Iniciante", "Pleno", "Sênior"],
        ).astype(str),
        anos_de_casa=(referencia - df["data_admissao"]).dt.days // 365,
    )

A análise separa calcular de mostrar: cada função devolve dados, e o texto vem depois. O empatados_no_topo é a resposta honesta ao idxmax:

projetos/funcionarios/src/funcionarios/analise.py

import pandas as pd


def resumo_por_departamento(df: pd.DataFrame) -> pd.DataFrame:
    return (
        df.groupby("departamento")
        .agg(
            salario_medio=("salario", "mean"),
            nota_media=("nota_desempenho", "mean"),
            funcionarios=("id_funcionario", "count"),
        )
        .reset_index()
    )


def empatados_no_topo(serie: pd.Series) -> list[str]:
    """Todos os que empatam no máximo. Devolver só um esconderia o empate."""
    topo = serie.max()
    return sorted(str(rotulo) for rotulo in serie.index[serie == topo])


def top_desempenho(df: pd.DataFrame, n: int = 5) -> pd.DataFrame:
    """Top n com desempate explícito: nota, depois projetos, depois o menor id."""
    ordenado = df.sort_values(
        ["nota_desempenho", "projetos_concluidos", "id_funcionario"],
        ascending=[False, False, True],
    )
    return ordenado.head(n)


def perguntas(df: pd.DataFrame) -> dict[str, object]:
    resumo = resumo_por_departamento(df).set_index("departamento")
    return {
        "maior_salario_medio": empatados_no_topo(resumo["salario_medio"]),
        "maior_nota_media": empatados_no_topo(resumo["nota_media"]),
        "cidades_com_mais_funcionarios": empatados_no_topo(df["cidade"].value_counts()),
        "top5": top_desempenho(df)["nome"].tolist(),
        "correlacao_experiencia_projetos": round(
            float(df["experiencia"].corr(df["projetos_concluidos"])), 3
        ),
        "funcionarios_por_departamento": {k: int(v) for k, v in resumo["funcionarios"].items()},
    }

O relatório monta o texto, e a linha de comando expõe dois comandos (relatorio e limpar), recusando um arquivo inexistente com o código de saída 1:

projetos/funcionarios/src/funcionarios/relatorio.py

import pandas as pd

from funcionarios.analise import perguntas, resumo_por_departamento


def qualidade(bruto: pd.DataFrame, limpo: pd.DataFrame) -> list[str]:
    ausentes = bruto.isna().sum()
    ausentes = ausentes[ausentes > 0]
    salarios_texto = int(bruto["salario"].astype(str).str.contains("R$", regex=False).sum())
    datas_br = int(bruto["data_admissao"].astype(str).str.contains("/", regex=False).sum())
    return [
        f"linhas no arquivo: {len(bruto)}",
        f"linhas depois da limpeza: {len(limpo)}",
        f"cadastros repetidos removidos: {len(bruto) - len(limpo)}",
        f"salários escritos como texto: {salarios_texto}",
        f"datas em formato brasileiro: {datas_br}",
        "ausentes no arquivo: " + ", ".join(f"{c}={int(n)}" for c, n in ausentes.items()),
    ]


def montar_relatorio(bruto: pd.DataFrame, limpo: pd.DataFrame) -> str:
    linhas = ["Relatório de funcionários", "", "Qualidade dos dados"]
    linhas += [f"  {item}" for item in qualidade(bruto, limpo)]
    linhas += ["", "Resumo por departamento"]
    resumo = resumo_por_departamento(limpo).round({"salario_medio": 0, "nota_media": 2})
    linhas.append(resumo.to_string(index=False))
    respostas = perguntas(limpo)
    linhas += ["", "Perguntas de negócio"]
    linhas.append(f"  maior salário médio: {', '.join(respostas['maior_salario_medio'])}")
    linhas.append(f"  maior nota média: {', '.join(respostas['maior_nota_media'])}")
    linhas.append(
        f"  cidades com mais funcionários: {', '.join(respostas['cidades_com_mais_funcionarios'])}"
    )
    linhas.append(f"  top 5 (nota, projetos, id): {', '.join(respostas['top5'])}")
    linhas.append(
        f"  correlação experiência x projetos: {respostas['correlacao_experiencia_projetos']}"
    )
    return "\n".join(linhas)

projetos/funcionarios/src/funcionarios/cli.py

import argparse
from pathlib import Path

import pandas as pd

from funcionarios.limpeza import limpar
from funcionarios.relatorio import montar_relatorio


def criar_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="funcionarios", description="Relatório de funcionários")
    parser.add_argument(
        "--arquivo",
        type=Path,
        default=Path("../../dados/funcionarios_100.csv"),
        help="CSV de entrada",
    )
    sub = parser.add_subparsers(dest="comando", required=True)
    sub.add_parser("relatorio", help="mostra o relatório completo")
    limpar_p = sub.add_parser("limpar", help="grava os dados limpos em CSV")
    limpar_p.add_argument("--saida", type=Path, required=True)
    return parser


def main(argv: list[str] | None = None) -> int:
    args = criar_parser().parse_args(argv)
    try:
        bruto = pd.read_csv(args.arquivo)
        limpo = limpar(bruto)
    except (OSError, ValueError, KeyError) as erro:
        print(f"Erro: {erro}")
        return 1
    if args.comando == "relatorio":
        print(montar_relatorio(bruto, limpo))
    else:
        args.saida.parent.mkdir(parents=True, exist_ok=True)
        limpo.to_csv(args.saida, index=False)
        print(f"{len(limpo)} funcionários gravados em {args.saida}")
    return 0

Os testes

Os testes usam uma tabela de cinco linhas escrita à mão, com cada problema do arquivo real (um salário com R$, uma data brasileira, um id repetido, um departamento com espaços). Como a resposta é conhecida, cada teste diz o que deveria acontecer, e não "o que aconteceu ontem":

projetos/funcionarios/tests/conftest.py

import pandas as pd
import pytest


@pytest.fixture
def bruto() -> pd.DataFrame:
    """Uma tabela pequena, escrita à mão, com cada problema do arquivo real."""
    return pd.DataFrame(
        {
            "id_funcionario": [1, 2, 3, 4, 1],
            "nome": ["Ana", "Bruno", "Carla", "Diego", "Ana"],
            "departamento": ["Engenharia", " marketing ", "RH", None, "RH"],
            "cidade": ["São Paulo", None, "Curitiba", "São Paulo", "Curitiba"],
            "idade": [30.0, 40.0, None, 25.0, 31.0],
            "salario": ["5000", "R$ 6000", None, "4000", "5500"],
            "experiencia": [1, 3, 6, 2, 2],
            "nota_desempenho": [8, 6, 9, 5, 9],
            "projetos_concluidos": [3, 4, 9, 2, 5],
            "data_admissao": ["2020-01-10", "15/03/2019", "2021-07-01", "2022-02-02", "2023-05-05"],
        }
    )

projetos/funcionarios/tests/test_limpeza.py

import numpy as np
import pandas as pd
import pytest
from pandas.testing import assert_frame_equal

from funcionarios.limpeza import (
    converter_datas,
    limpar,
    limpar_salario,
    padronizar_departamento,
)


def test_salario_nao_perde_o_que_estava_escrito_com_simbolo() -> None:
    entrada = pd.Series(["R$ 4650", "4800", None])
    saida = limpar_salario(entrada)
    assert saida.tolist()[:2] == [4650.0, 4800.0] and np.isnan(saida.iloc[2])


def test_salario_ja_numerico_passa_direto() -> None:
    assert limpar_salario(pd.Series([1, 2])).tolist() == [1.0, 2.0]


def test_departamento_padroniza_e_marca_o_desconhecido() -> None:
    saida = padronizar_departamento(pd.Series([" engenharia ", "RH", "Tecnologia", None]))
    assert saida.tolist()[:2] == ["Engenharia", "RH"]
    assert saida.isna().tolist() == [False, False, True, True]


def test_datas_em_dois_formatos_sem_trocar_dia_e_mes() -> None:
    saida = converter_datas(pd.Series(["2019-08-07", "07/08/2019"]))
    assert saida.dt.strftime("%Y-%m-%d").tolist() == ["2019-08-07", "2019-08-07"]


def test_data_em_formato_desconhecido_e_recusada() -> None:
    with pytest.raises(ValueError, match="formato desconhecido"):
        converter_datas(pd.Series(["2019-08-07", "7 de agosto"]))


def test_limpar_remove_repetidos_mantendo_o_ultimo(bruto: pd.DataFrame) -> None:
    limpo = limpar(bruto)
    assert len(limpo) == 4
    ana = limpo.loc[limpo["id_funcionario"] == 1].iloc[0]
    assert ana["departamento"] == "RH" and ana["experiencia"] == 2


def test_limpar_nao_deixa_ausentes_nas_colunas_tratadas(bruto: pd.DataFrame) -> None:
    limpo = limpar(bruto)
    for coluna in ("salario", "idade", "departamento", "cidade", "data_admissao", "nivel"):
        assert not limpo[coluna].isna().any(), coluna


def test_limpar_guarda_a_informacao_de_que_faltava(bruto: pd.DataFrame) -> None:
    limpo = limpar(bruto).set_index("id_funcionario")
    assert not limpo.loc[3, "salario_informado"] and limpo.loc[2, "salario_informado"]
    assert not limpo.loc[3, "idade_informada"]


def test_limpar_nao_altera_a_tabela_recebida(bruto: pd.DataFrame) -> None:
    copia = bruto.copy()
    limpar(bruto)
    assert_frame_equal(bruto, copia)


def test_nivel_por_experiencia(bruto: pd.DataFrame) -> None:
    limpo = limpar(bruto).set_index("id_funcionario")
    assert limpo.loc[3, "nivel"] == "Sênior"
    assert limpo.loc[4, "nivel"] == "Pleno"
    assert limpo.loc[1, "nivel"] == "Pleno"

projetos/funcionarios/tests/test_analise.py

import pandas as pd

from funcionarios.analise import (
    empatados_no_topo,
    perguntas,
    resumo_por_departamento,
    top_desempenho,
)
from funcionarios.limpeza import limpar


def test_o_resumo_nao_perde_funcionarios(bruto: pd.DataFrame) -> None:
    limpo = limpar(bruto)
    resumo = resumo_por_departamento(limpo)
    assert resumo["funcionarios"].sum() == len(limpo)
    assert "Não informado" in resumo["departamento"].tolist()


def test_empate_no_topo_mostra_todos() -> None:
    serie = pd.Series({"São Paulo": 3, "Rio": 3, "Curitiba": 1})
    assert empatados_no_topo(serie) == ["Rio", "São Paulo"]


def test_top_nao_depende_da_ordem_das_linhas(bruto: pd.DataFrame) -> None:
    limpo = limpar(bruto)
    embaralhado = limpo.sample(frac=1, random_state=7)
    assert (
        top_desempenho(limpo, 3)["id_funcionario"].tolist()
        == top_desempenho(embaralhado, 3)["id_funcionario"].tolist()
    )


def test_perguntas_tem_todas_as_respostas(bruto: pd.DataFrame) -> None:
    respostas = perguntas(limpar(bruto))
    assert set(respostas) == {
        "maior_salario_medio",
        "maior_nota_media",
        "cidades_com_mais_funcionarios",
        "top5",
        "correlacao_experiencia_projetos",
        "funcionarios_por_departamento",
    }
    assert sum(respostas["funcionarios_por_departamento"].values()) == 4

projetos/funcionarios/tests/test_cli.py

from pathlib import Path

import pandas as pd
import pytest

from funcionarios.cli import main


def test_relatorio_de_ponta_a_ponta(
    tmp_path: Path, bruto: pd.DataFrame, capsys: pytest.CaptureFixture[str]
) -> None:
    arquivo = tmp_path / "f.csv"
    bruto.to_csv(arquivo, index=False)
    assert main(["--arquivo", str(arquivo), "relatorio"]) == 0
    saida = capsys.readouterr().out
    assert "Relatório de funcionários" in saida
    assert "cadastros repetidos removidos: 1" in saida
    assert "salários escritos como texto: 1" in saida


def test_limpar_grava_o_csv(tmp_path: Path, bruto: pd.DataFrame) -> None:
    arquivo = tmp_path / "f.csv"
    bruto.to_csv(arquivo, index=False)
    destino = tmp_path / "saida" / "limpo.csv"
    assert main(["--arquivo", str(arquivo), "limpar", "--saida", str(destino)]) == 0
    assert len(pd.read_csv(destino)) == 4


def test_arquivo_inexistente_devolve_codigo_1(
    tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
    assert main(["--arquivo", str(tmp_path / "nao_existe.csv"), "relatorio"]) == 1
    assert "Erro" in capsys.readouterr().out


def test_com_os_dados_reais_do_livro(capsys: pytest.CaptureFixture[str]) -> None:
    caminho = Path(__file__).resolve().parents[3] / "dados" / "funcionarios_100.csv"
    if not caminho.exists():
        pytest.skip("os dados do livro não estão ao lado do projeto")
    assert main(["--arquivo", str(caminho), "relatorio"]) == 0
    assert "linhas depois da limpeza: 100" in capsys.readouterr().out

O último teste roda o programa sobre os dados reais do curso e confere que 104 linhas viram 100, e ele é ignorado (skip) quando o arquivo não está ao lado, para o projeto continuar testável sozinho.

Rodar

Terminal

cd projetos/funcionarios
uv sync
uv run pytest
uv run funcionarios --arquivo ../../dados/funcionarios_100.csv relatorio

Saída

..................                                                       [100%]
18 passed

O relatório (execução real)

Relatório de funcionários

Qualidade dos dados
  linhas no arquivo: 104
  linhas depois da limpeza: 100
  cadastros repetidos removidos: 4
  salários escritos como texto: 6
  datas em formato brasileiro: 3
  ausentes no arquivo: departamento=3, cidade=5, idade=4, salario=7

Resumo por departamento
    departamento  salario_medio  nota_media  funcionarios
Ciência de Dados        10152.0        6.46            20
      Engenharia         9012.0        6.85            23
        Finanças         9346.0        7.01            27
       Marketing         8460.0        6.55            16
   Não informado         5560.0        5.83             3
              RH         8016.0        6.68            11

Perguntas de negócio
  maior salário médio: Ciência de Dados
  maior nota média: Finanças
  cidades com mais funcionários: Porto Alegre
  top 5 (nota, projetos, id): Lucas Teixeira, Elisa Gomes, Sérgio Teixeira, Karina Martins, Lucas Souza
  correlação experiência x projetos: 0.923

Lendo o resultado

A soma fecha. Os departamentos somam 20 + 23 + 27 + 16 + 3 + 11 = 100 funcionários, o total depois da limpeza. Os 3 sem departamento aparecem como Não informado, em vez de sumirem do resumo.
Os 4 repetidos viraram 0 duplicatas, e a limpeza ficou com o cadastro mais recente de cada um.
O grupo "Não informado" tem o salário médio mais baixo (5.560). É um sinal, e não uma conclusão: são só 3 pessoas, e o salário que falta foi preenchido pela mediana (do grupo, ou a geral, quando o grupo também falta). Eu não tomaria uma decisão com base nesse número.
A correlação entre experiência e projetos é 0,92, bem mais forte que a do conjunto de 15 (0,22). Os 100 funcionários foram gerados com essa dependência embutida, e é por isso que ela aparece. Em dados reais, uma correlação tão alta mereceria desconfiança.

Desafios

Percentis. Acrescente ao relatório o salário mediano e o percentil 90 por departamento, e confira com um teste de resposta conhecida.
Gráfico. Com matplotlib, desenhe o salário médio por departamento e grave em PNG (o programa não deve exigir o matplotlib para gerar o relatório).
Mais um formato de data. Aceite também aaaa/mm/dd em converter_datas, com um teste, e confira que um formato não declarado continua levantando erro.
Parquet. Troque o comando limpar para gravar também em Parquet, e confira que os tipos sobrevivem à ida e volta (capítulo 26).