/
Chapter 36, Projects
Basic Project: employee report
The final project of the original course, complete, tested, and transformed into a program: load a dirty CSV, clean it, answer business questions, and deliver a report. Do this after chapter 15.
The problem
All chapters of the Basic level worked with 15 employees. Here the same script runs on the funcionarios_100.csv, with 104 lines and more dirt, and turns into a program: it receives a file, cleans it, calculates, and prints a report. What chapter 14 did in a loose function becomes a package with tests.
| Course stage | Where it is in the project |
|---|---|
| Load and inspect (caps. 5 and 6) | cli.py reads the file, and relatorio.qualidade shows the diagnosis |
| Select and filter (caps. 7 and 8) | Used within analise.py |
| Clean: missing, duplicates, types, text (caps. 9 and 10) | limpeza.py |
| New columns (cap. 11) | nivel and anos_de_casa, at the end of limpar |
| Sort and rank (cap. 12) | top_desempenho, with explicit tie-breaking |
| Group (cap. 13) | resumo_por_departamento |
| Business questions (cap. 14) | perguntas, and the relatorio.py it prints |
The decisions I made
Each cleaning step is a small, pure function, which receives a Series and returns another: limpar_salario, padronizar_departamento, converter_datas. This allows testing each rule alone, with three values written by hand.
Cleaning does not guess. The department that is not in the dictionary becomes missing (and is visible). The date in a format outside the two declared raises an error, instead of being interpreted in any way. The salary written as R$ 4650 is converted without losing any value.
The last registration of each id wins, because the duplicates in the file are updates (chapter 10). Filling in does not erase the evidence: the columns salario_informado and idade_informada keep what was missing before the filling.
Ties are shown, not hidden. "Which city has the most employees?" returns the list of the tied ones, and the top 5 has explicit ties (score, then projects, then the lowest id), which makes it independent of the order of the rows.
The code
The cleaning is the heart of the project. Note that the function limpar returns a table new (the copy under writing, from chapter 27, and the assign ensure that the received one does not change) and that each rule has its function:
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,
)
The analysis separates calculating from showing: each function returns data, and the text comes afterwards. The empatados_no_topo is the honest answer to the idxmax:
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()},
}
The report assembles the text, and the command line exposes two commands (relatorio and limpar), refusing a nonexistent file with exit code 1:
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)
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
The tests
The tests use a table of five handwritten lines, with each problem from the real file (a salary with R$, a Brazilian date, a repeated id, a department with spaces). Since the answer is known, each test states what should happen, and not "what happened yesterday":
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"],
}
)
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"
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
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
The last test runs the program on the real course data and checks that 104 lines become 100, and it is ignored (skip) when the file is not next to it, for the project to remain testable on its own.
Run
cd projetos/funcionarios
uv sync
uv run pytest
uv run funcionarios --arquivo ../../dados/funcionarios_100.csv relatorio
Output
.................. [100%]
18 passed
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
Reading the result
- The sum closes. The departments sum 20 + 23 + 27 + 16 + 3 + 11 = 100 employees, the total after cleaning. The 3 without a department appear as
Não informado, instead of disappearing from the summary. - The 4 duplicates became 0 duplicates, and the cleaning kept the most recent registration of each.
- The group "Not informed" has the lowest average salary (5,560). It's a signal, not a conclusion: there are only 3 people, and the missing salary was filled by the median (of the group, or the overall, when the group is also missing). I wouldn't make a decision based on this number.
- The correlation between experience and projects is 0.92, much stronger than that of the set of 15 (0.22). The 100 employees were generated with this embedded dependency, and that's why it appears. In real data, such a high correlation would deserve suspicion.
Challenges
- Percentiles. Add to the report the median salary and the 90th percentile by department, and check with a known response test.
- Graph. With
matplotlib, draw the average salary by department and save in PNG (the program should not require thematplotlibto generate the report). - Another date format. Also accept
aaaa/mm/ddinconverter_datas, with a test, and check that an undeclared format continues to raise an error. - Parquet. Change the command
limparto also save in Parquet, and check that the types survive the back and forth (chapter 26).