---
title: "Full Pipeline with DuckDB"
output: rmarkdown::html_vignette
vignette: >
  %\VignetteIndexEntry{Full Pipeline with DuckDB}
  %\VignetteEngine{knitr::rmarkdown}
  %\VignetteEncoding{UTF-8}
---

```{r setup, include = FALSE}
knitr::opts_chunk$set(
  collapse = TRUE,
  comment  = "#>",
  eval     = FALSE
)
```

This guide demonstrates the full `datacaged` pipeline: from checking the
HuggingFace repository to analytical queries in DuckDB with `dplyr` and SQL.

## 1. Check availability before downloading

Antes de iniciar qualquer download, check that the HuggingFace repository is accessible
and which competencies are available.

```{r status}
library(datacaged)

# Verifica se o repositório HuggingFace está online e mede a latência
caged_status()
#> OK Repositório HuggingFace online
#> ℹ Latência: 312 ms
#> ℹ Dataset: https://huggingface.co/datasets/alexsandroprado/caged
```

```{r ftp-files}
# List the 12 most recent competencies available in Novo CAGED
caged_hf_files()

# Lista todas as competências do Legacy CAGED
caged_hf_files(type = "antigo", n = Inf)

# Lista competências do CAGED Adjustments
caged_hf_files(type = "ajustes", n = 24)

# Capture result for programmatic use
disponivel <- caged_hf_files(n = 1, verbose = FALSE)
cat("Competência mais recente:", disponivel$competencia, "\n")
```

## 2. Download microdata

The `caged_download()` function manages local cache automatically: already-downloaded
files are not re-downloaded. It returns a `manifest` — a data.frame with the status of
each file.

```{r download-novo}
# Baixa Novo CAGED de 2023 inteiro (MOV + FOR + EXC por competência)
manifest <- caged_download(
  years    = 2023,
  months   = seq_len(12L),
  destdir = "~/dados/caged_cache"   # omitir para usar cache padrão
)

# Inspect o manifest
dplyr::count(manifest, type, status)
#> # A tibble: 4 × 3
#>   type  status       n
#>   <chr> <chr>    <int>
#> 1 EXC   baixado     12
#> 2 FOR   baixado     12
#> 3 MOV   baixado     12
#> 4 MOV   cache        0

# Files available for processing
manifest |>
  dplyr::filter(status %in% c("baixado", "cache")) |>
  dplyr::select(type, competencia, nome_arquivo, arquivo)
```

```{r download-antigo}
# Baixa Legacy CAGED — arquivo nacional único por competência
manifest_ant <- caged_download(years = 2018:2019, months = seq_len(12L))

# Download adjustments for the same period too
caged_adjustments_load(
  years    = 2018:2019,
  months   = seq_len(12L),
  db_path = "caged_historico.duckdb"
)
```

## 3. Parse files

`caged_parse()` extracts the `.7z`, reads the `.txt` and returns a normalised tibble.

```{r parse-individual}
# Parse a single file
df_jan <- caged_parse("~/dados/caged_cache/NOVO_CAGED/2023/202301/CAGEDMOV202301.7z")

# Inspect
dplyr::glimpse(df_jan)
#> Rows: 3,665,155
#> Columns: 29
#> $ competenciamov      <dbl> 202301, 202301, ...
#> $ uf                  <dbl> 11, 11, ...
#> $ municipio           <dbl> 1100015, ...
#> $ saldomovimentacao   <dbl> 1, -1, ...
#> $ salario             <dbl> 1412.00, 2800.50, ...

# Parse multiple files of the same type (devem ser todos MOV, ou todos FOR, etc.)
arquivos_mov <- list.files(
  "~/dados/caged_cache/NOVO_CAGED/2023",
  pattern    = "CAGEDMOV",
  recursive  = TRUE,
  full.names = TRUE
)
df_mov_2023 <- caged_parse_batch(arquivos_mov)
```

## 4. Write to DuckDB

```{r gravar}
# Write the parsed tibble to the database
caged_to_duckdb(df_mov_2023, db_path = "caged.duckdb")

# Re-runs are safe: competencies already in the database are skipped
caged_to_duckdb(df_mov_2023, db_path = "caged.duckdb")
#> ℹ 12 competências já no banco — pulando.

# To overwrite (e.g. after correcting data):
caged_to_duckdb(df_mov_2023, db_path = "caged.duckdb",
                overwrite_competencies = TRUE)

# Write to a specific table without auto-detection
caged_to_duckdb(df_jan, db_path = "caged.duckdb", table = "caged_mov")
```

## 5. Full pipeline in a single command

For the most common case — download everything and write to the database — use `caged_load()`:

```{r pipeline-completo}
library(datacaged)

# ── Novo CAGED 2022-2023 ─────────────────────────────────────────────────────
caged_load(
  years    = 2022:2023,
  months   = seq_len(12L),
  db_path = "caged.duckdb"
)

# ── Legacy CAGED 2015-2019 ────────────────────────────────────────────────────
caged_load(
  years    = 2015:2019,
  months   = seq_len(12L),
  db_path = "caged.duckdb"
)

# ── CAGED Adjustments 2015-2019 ───────────────────────────────────────────────────
caged_adjustments_load(
  years    = 2015:2019,
  months   = seq_len(12L),
  db_path = "caged.duckdb"
)

# ── Check the resulting database ─────────────────────────────────────────────
caged_info("caged.duckdb")
#> ── caged.duckdb ──────────────────────────────────────────────────────────
#> Tamanho do arquivo: 4.2 GB
#> ── Tabelas ───────────────────────────────────────────────────────────────
#> * "caged_mov"     Registros: 43,981,860   Competências: 202201 – 202312
#> * "caged_for"     Registros:  1,093,176   Competências: 202201 – 202312
#> * "caged_exc"     Registros:     94,800   Competências: 202201 – 202312
#> * "caged_antigo"  Registros: 28,450,000   Competências: 201501 – 201912
#> * "caged_ajustes" Registros:    342,000   Competências: 201501 – 201912
```

## 6. Queries with dplyr

```{r conexao}
library(dplyr)

con <- caged_connect("caged.duckdb")
```

```{r consultas-dplyr}
# ── Available tables ─────────────────────────────────────────────────────────
DBI::dbListTables(con)

# ── Lazy reference (does not load into memory) ───────────────────────────────
mov <- tbl(con, "caged_mov")

# ── Monthly balance ───────────────────────────────────────────────────────────
mov |>
  group_by(competenciamov) |>
  summarise(
    admissoes    = sum(saldomovimentacao == 1,  na.rm = TRUE),
    desligamentos = sum(saldomovimentacao == -1, na.rm = TRUE),
    saldo        = sum(saldomovimentacao,        na.rm = TRUE)
  ) |>
  arrange(competenciamov) |>
  collect()

# ── Balance by state and sector ──────────────────────────────────────────────
mov |>
  filter(competenciamov >= 202301) |>
  group_by(uf, secao) |>
  summarise(saldo = sum(saldomovimentacao, na.rm = TRUE)) |>
  arrange(desc(saldo)) |>
  collect()

# ── Average wage by education level and gender ───────────────────────────────
mov |>
  filter(!is.na(salario), salario > 0) |>
  group_by(graudeinstrucao, sexo) |>
  summarise(
    salario_medio  = mean(salario, na.rm = TRUE),
    n              = n()
  ) |>
  collect()

# ── Top 10 municipalities by hires ───────────────────────────────────────────
mov |>
  filter(saldomovimentacao == 1) |>
  group_by(municipio) |>
  summarise(admissoes = n()) |>
  slice_max(admissoes, n = 10) |>
  collect()
```

## 7. Direct SQL queries

```{r consultas-sql}
# Net balance by CNAE section and month
DBI::dbGetQuery(con, "
  SELECT
    competenciamov                    AS competencia,
    secao,
    SUM(CASE WHEN saldomovimentacao =  1 THEN 1 ELSE 0 END) AS admissoes,
    SUM(CASE WHEN saldomovimentacao = -1 THEN 1 ELSE 0 END) AS desligamentos,
    SUM(saldomovimentacao)                                   AS saldo,
    ROUND(AVG(salario), 2)                                  AS salario_medio
  FROM caged_mov
  WHERE competenciamov >= 202301
  GROUP BY competenciamov, secao
  ORDER BY competenciamov, saldo DESC
")

# Age distribution by range and movement type
DBI::dbGetQuery(con, "
  SELECT
    CASE
      WHEN idade < 25              THEN 'Até 24 anos'
      WHEN idade BETWEEN 25 AND 34 THEN '25-34 anos'
      WHEN idade BETWEEN 35 AND 44 THEN '35-44 anos'
      WHEN idade BETWEEN 45 AND 54 THEN '45-54 anos'
      ELSE '55 anos ou mais'
    END AS faixa_etaria,
    SUM(CASE WHEN saldomovimentacao =  1 THEN 1 ELSE 0 END) AS admissoes,
    SUM(CASE WHEN saldomovimentacao = -1 THEN 1 ELSE 0 END) AS desligamentos
  FROM caged_mov
  GROUP BY faixa_etaria
  ORDER BY faixa_etaria
")
```

## 8. Historical series: joining Novo CAGED and legacy CAGED

To build long series that cross the 2020 break point, normalise
columns before stacking.

```{r serie-historica}
# Novo CAGED — colunas a normalizar
novo <- tbl(con, "caged_mov") |>
  select(
    competencia    = competenciamov,
    uf,
    saldo          = saldomovimentacao,
    salario,
    sexo,
    idade,
    escolaridade   = graudeinstrucao
  ) |>
  mutate(serie = "novo")

# Legacy CAGED — já tem coluna competencia e saldomovimentacao
antigo <- tbl(con, "caged_antigo") |>
  select(
    competencia,
    uf,
    saldo          = saldomovimentacao,
    salario,
    sexo,
    idade,
    escolaridade
  ) |>
  mutate(serie = "antigo")

# Empilha e agrega
serie_hist <- bind_rows(
  novo   |> group_by(competencia, serie) |> summarise(saldo = sum(saldo, na.rm = TRUE)),
  antigo |> group_by(competencia, serie) |> summarise(saldo = sum(saldo, na.rm = TRUE))
) |>
  collect() |>
  arrange(competencia)
```

## 9. Export results

```{r exportar}
library(dplyr)

# Export to CSV
resultado <- tbl(con, "caged_mov") |>
  group_by(competenciamov, uf) |>
  summarise(saldo = sum(saldomovimentacao, na.rm = TRUE)) |>
  collect()

readr::write_csv(resultado, "saldo_uf_mensal.csv")

# Export all database tables to Parquet (native DuckDB, much faster)
caged_to_parquet("caged.duckdb", output_dir = "~/exports")

# Export only caged_mov partitioned by state
caged_to_parquet(
  "caged.duckdb",
  output_dir   = "~/exports",
  tables       = "caged_mov",
  partition_by = "uf"
)

# Close connection
DBI::dbDisconnect(con, shutdown = TRUE)
```

## 10. Incremental update

Use `caged_update()` to keep the database up to date without re-downloading everything.
The function queries the database to find the maximum competency and downloads only
what is more recent on HuggingFace.

```{r update}
# Update Novo CAGED with months not yet in the database
caged_update(db_path = "caged.duckdb")

# Update all series
caged_update(
  db_path = "caged.duckdb",
  series  = c("novo", "antigo", "ajustes")
)
```

## 11. Best practices

- **Always close the connection** with `DBI::dbDisconnect(con, shutdown = TRUE)` when done.
- **Use `tbl()` before `collect()`**: DuckDB executes as much as possible before bringing data into R.
- **`select()` early**: bring only the necessary columns to save memory.
- **Safe re-runs**: `caged_load()` detects already-written competencies and skips them automatically.
- **Local cache**: `.7z` files are stored in `tools::R_user_dir("datacaged", "cache")` and reused between sessions.
- **Large volumes**: the pipeline processes one competency at a time to avoid running out of RAM.