---
title: "Data Wrangling"
subtitle: "Parsing, classificação, mídia e enriquecimento"
author: "Marlon L."
date: today
format:
html:
toc: true
toc-depth: 3
code-fold: true
theme: cosmo
execute:
warning: false
echo: false
---
```{python}
#| label: setup
#| code-fold: false
import os
from pathlib import Path
import pandas as pd
from whatsapp.pipeline.config import PROJECT_ROOT, PATHS
from whatsapp.pipeline.wrangling import run_wrangling_pipeline, WRANGLING_STEPS
```
📁 Projeto: `{python} PROJECT_ROOT.name`
# Data Wrangling
Com o arquivo limpo gerado no [Data Cleaning](01-data-cleaning.qmd), esta etapa transforma o TXT em um DataFrame estruturado, vincula arquivos de mídia, integra transcrições e exporta os dados enriquecidos.
## Objetivo
- ✅ Parsing: TXT → DataFrame
- ✅ Classificação: Identificar tipos de mensagem
- ✅ Mídia: Vincular arquivos físicos
- ✅ Transcrição: Integrar transcrições existentes
- ✅ Enriquecimento: Substituir mídias por transcrições
- ✅ Exportação: CSV + TXTs de corpus
------------------------------------------------------------------------
## Configuração do Pipeline
```{python}
#| code-fold: false
PIPELINE_ORDER = ['parse', 'classify', 'media', 'transcriptions', 'enrich', 'export']
INPUT_FILE = PATHS['interim'] / 'raw-data_cln7.txt'
OUTPUT_DIR = PATHS['processed']
MEDIA_DIR = PATHS['media']
TRANSCRIPTION_FILE = OUTPUT_DIR / 'transcriptions.csv'
# Helper: converte path absoluto em relativo ao projeto
def format_relative_path(full_path):
full_str = str(full_path)
if 'whatsapp-interaction-analysis' in full_str:
idx = full_str.find('whatsapp-interaction-analysis')
return '~/' + full_str[idx:]
return full_str
# Exibe configuração em tabela
config_data = [
{'Parâmetro': 'Entrada', 'Valor': f'<code>{INPUT_FILE.name}</code>', 'Path': format_relative_path(INPUT_FILE.parent)},
{'Parâmetro': 'Saída', 'Valor': f'<code>{OUTPUT_DIR.name}</code>', 'Path': format_relative_path(OUTPUT_DIR)},
{'Parâmetro': 'Mídia', 'Valor': f'<code>{MEDIA_DIR.name}</code>', 'Path': format_relative_path(MEDIA_DIR)},
{'Parâmetro': 'Transcrições', 'Valor': f'<code>{TRANSCRIPTION_FILE.name}</code>', 'Path': format_relative_path(TRANSCRIPTION_FILE.parent)},
]
df_config = pd.DataFrame(config_data)
display(df_config.style.hide(axis='index'))
```
------------------------------------------------------------------------
## Execução do Pipeline
```{python}
#| output: false
result = run_wrangling_pipeline(
order=PIPELINE_ORDER,
input_file=INPUT_FILE,
output_dir=OUTPUT_DIR,
media_dir=MEDIA_DIR,
transcription_file=TRANSCRIPTION_FILE,
show_progress=True
)
df = result['df']
outputs = result['outputs']
stats = result['stats']
```
```{python}
# Tabela de execução do pipeline
n_msgs = stats.get('total_mensagens', len(df))
n_tipos = len(stats.get('tipos_mensagem', {}))
n_midias = stats.get('midias_anexadas', 0)
n_trans = stats.get('com_transcricao', 0)
n_arquivos = len([k for k in stats.get('export_summary', {}).keys()])
etapas = [
{'#': '1', 'Etapa': 'Parse', 'Resultado': f'{n_msgs:,} mensagens'},
{'#': '2', 'Etapa': 'Classificação', 'Resultado': f'{n_tipos} tipos identificados'},
{'#': '3', 'Etapa': 'Mídia', 'Resultado': f'{n_midias:,} arquivos vinculados'},
{'#': '4', 'Etapa': 'Transcrições', 'Resultado': f'{n_trans:,} transcrições integradas'},
{'#': '4.1', 'Etapa': 'Enriquecimento', 'Resultado': f'{n_trans:,} mensagens enriquecidas'},
{'#': '5', 'Etapa': 'Exportação', 'Resultado': f'{n_arquivos} arquivos gerados'},
]
display(pd.DataFrame(etapas).style.hide(axis='index'))
print()
print("✅ Pipeline concluído com sucesso!")
```
------------------------------------------------------------------------
## Pipeline de Transformação
### Etapa 1: Parsing
Converte TXT em DataFrame, agregando mensagens multilinha.
```{python}
if 'total_mensagens' in stats:
dados = [
{'Métrica': 'Mensagens parseadas', 'Valor': f"{stats['total_mensagens']:,}"},
{'Métrica': 'Linhas no TXT', 'Valor': f"{stats.get('total_linhas_txt', 0):,}"},
]
if stats.get('total_linhas_txt', 0) > 0:
ratio = stats['total_linhas_txt'] / stats['total_mensagens']
dados.append({'Métrica': 'Razão linhas/mensagem', 'Valor': f"{ratio:.2f}"})
display(pd.DataFrame(dados).style.hide(axis='index'))
```
### Etapa 2: Classificação
Classifica cada mensagem por tipo (21 tipos possíveis) e agrupa em categorias.
::: panel-tabset
#### Por Tipo
```{python}
if 'tipos_mensagem' in stats and 'grupo_mensagem' in df.columns:
# Mapeia cada tipo ao seu grupo
tipo_to_grupo = df.groupby('tipo_mensagem')['grupo_mensagem'].first().to_dict()
tipos = stats['tipos_mensagem']
total = sum(tipos.values())
df_tipos = pd.DataFrame([
{
'Tipo': f'<code>{t}</code>',
'Grupo': f'<code>{tipo_to_grupo.get(t, "?")}</code>',
'Contagem': f'{c:,}',
'%': f'{(c / total * 100):.1f}%'
}
for t, c in sorted(tipos.items(), key=lambda x: x[1], reverse=True)
])
display(df_tipos.style.hide(axis='index'))
```
#### Por Grupo
```{python}
if 'grupo_mensagem' in df.columns:
grupos = df['grupo_mensagem'].value_counts()
total_msgs = len(df)
# Para cada grupo, lista os tipos que pertencem a ele
dados_grupos = []
for grupo in grupos.index:
tipos_do_grupo = df[df['grupo_mensagem'] == grupo]['tipo_mensagem'].unique()
tipos_str = ', '.join(sorted(tipos_do_grupo))
dados_grupos.append({
'Grupo': f'<code>{grupo}</code>',
'Contagem': f'{grupos[grupo]:,}',
'%': f'{(grupos[grupo] / total_msgs * 100):.1f}%',
'Tipos Incluídos': tipos_str
})
df_grupos = pd.DataFrame(dados_grupos)
display(df_grupos.style.hide(axis='index'))
```
:::
### Etapa 3: Mídia
Vincula arquivos físicos de mídia ao DataFrame, criando metadados (`arquivo_existe`, `arquivo_path`) que serão usados nas etapas seguintes para merge com transcrições.
```{python}
# Lista TODOS os 11 grupos de mensagem
# Grupos que podem ter arquivo anexado vs grupos sem arquivo
GRUPOS_COM_ARQUIVO = ['AUDIO', 'VID', 'IMG', 'STICKER', 'GIF', 'DOC', 'FILE', 'CONTACT']
GRUPOS_SEM_ARQUIVO = ['TEXT', 'SYSTEM', 'CALL']
dados_midia = []
if 'arquivo' in df.columns and 'grupo_mensagem' in df.columns:
df_com_arquivo = df[df['arquivo'].notna()]
# Grupos que podem ter arquivo
for grupo in GRUPOS_COM_ARQUIVO:
df_grupo = df_com_arquivo[df_com_arquivo['grupo_mensagem'] == grupo]
anexadas = len(df_grupo)
if 'arquivo_existe' in df.columns and anexadas > 0:
encontradas = int(df_grupo['arquivo_existe'].sum())
else:
encontradas = anexadas
dados_midia.append({
'Grupo': f'<code>{grupo}</code>',
'Anexadas': f'{anexadas:,}',
'Encontradas': f'{encontradas:,}'
})
# Grupos sem arquivo (TEXT, SYSTEM, CALL)
for grupo in GRUPOS_SEM_ARQUIVO:
dados_midia.append({
'Grupo': f'<code>{grupo}</code>',
'Anexadas': '—',
'Encontradas': '—'
})
if dados_midia:
display(pd.DataFrame(dados_midia).style.hide(axis='index'))
```
*Esta etapa prepara o DataFrame para receber transcrições (Etapa 4) que serão vinculadas pelo nome do arquivo.*
### Etapa 4: Transcrições
Integra transcrições de áudio/vídeo (geradas pelo script `transcribe_media.py`) ao DataFrame através de merge pela coluna `arquivo`.
```{python}
# Verifica se arquivo de transcrições existe
transcription_exists = TRANSCRIPTION_FILE.exists() if 'TRANSCRIPTION_FILE' in locals() else False
if not transcription_exists:
print("⚠️ TRANSCRIÇÕES NÃO ENCONTRADAS")
print()
print("Para processar áudios e vídeos, execute:")
print(" python scripts/transcribe_media.py")
print()
print("Tempo estimado: ~40 min para ~700 arquivos")
print("Após concluir, re-execute este notebook.")
```
```{python}
if transcription_exists:
# Carrega transcriptions.csv diretamente (fonte verdadeira)
df_csv_trans = pd.read_csv(TRANSCRIPTION_FILE)
# Conta por status e tipo
total_csv = len(df_csv_trans)
completed = len(df_csv_trans[df_csv_trans['transcription_status'] == 'completed'])
df_completed = df_csv_trans[df_csv_trans['transcription_status'] == 'completed']
audios_csv = len(df_completed[df_completed['media_type'] == 'audio'])
videos_csv = len(df_completed[df_completed['media_type'] == 'video'])
dados = [
{'Métrica': 'Arquivos no CSV', 'Valor': f'{total_csv:,}'},
{'Métrica': 'Transcrições OK', 'Valor': f'{completed:,}'},
{'Métrica': '↳ Áudios', 'Valor': f'{audios_csv:,}'},
{'Métrica': '↳ Vídeos', 'Valor': f'{videos_csv:,}'},
]
display(pd.DataFrame(dados).style.hide(axis='index'))
print()
print(f"✅ Carregado de: {TRANSCRIPTION_FILE.name}")
```
*Transcrições são vinculadas às mensagens pelo nome do arquivo (`arquivo` column).*
### Etapa 4.1: Enriquecimento de Conteúdo
Substitui tags de mídia (`<attached: arquivo>`) pelo texto da transcrição na coluna `conteudo`.
```{python}
#| output: false
# Carrega transcriptions.csv para exemplos
tem_transcricoes = False
df_transcriptions_ok = None
if TRANSCRIPTION_FILE.exists():
df_transcriptions = pd.read_csv(TRANSCRIPTION_FILE)
df_transcriptions_ok = df_transcriptions[df_transcriptions['transcription_status'] == 'completed']
tem_transcricoes = len(df_transcriptions_ok) > 0
```
```{python}
if not tem_transcricoes:
print("⚠️ SEM TRANSCRIÇÕES PARA ENRIQUECER")
print()
print("Esta etapa depende da Etapa 4.")
print("Execute: python scripts/transcribe_media.py")
```
```{python}
if tem_transcricoes:
# Métricas de impacto do enriquecimento
n_transcritas = stats.get('com_transcricao', 0)
n_texto_original = len(df[df['grupo_mensagem'] == 'TEXT']) if 'grupo_mensagem' in df.columns else 0
total_textual = n_texto_original + n_transcritas
# Calcula aumento percentual
aumento_pct = (n_transcritas / n_texto_original * 100) if n_texto_original > 0 else 0
# Calcula caracteres (impacto em volume de texto)
chars_texto = df[df['grupo_mensagem'] == 'TEXT']['conteudo'].astype(str).str.len().sum() if 'grupo_mensagem' in df.columns else 0
chars_transcritas = df_transcriptions_ok['transcription'].astype(str).str.len().sum() if df_transcriptions_ok is not None else 0
total_chars = chars_texto + chars_transcritas
dados = [
{'Métrica': 'Mensagens de texto originais', 'Valor': f'{n_texto_original:,}'},
{'Métrica': 'Áudios/vídeos transcritos', 'Valor': f'{n_transcritas:,}'},
{'Métrica': '**Total de mensagens textuais**', 'Valor': f'**{total_textual:,}**'},
{'Métrica': '**Aumento de conteúdo**', 'Valor': f'**+{aumento_pct:.1f}%**'},
]
display(pd.DataFrame(dados).style.hide(axis='index'))
# Tabela secundária: impacto em caracteres
print()
dados_chars = [
{'Métrica': 'Caracteres em texto original', 'Valor': f'{chars_texto:,}'},
{'Métrica': 'Caracteres em transcrições', 'Valor': f'{chars_transcritas:,}'},
{'Métrica': '**Total de caracteres**', 'Valor': f'**{total_chars:,}**'},
]
display(pd.DataFrame(dados_chars).style.hide(axis='index'))
```
**Exemplo de transformação:**
```{python}
#| echo: false
#| output: false
# Pega 3 exemplos diretamente do transcriptions.csv
tem_exemplos = False
df_exemplos = None
if tem_transcricoes and df_transcriptions_ok is not None:
df_exemplos = df_transcriptions_ok[['file_path', 'transcription']].head(3)
tem_exemplos = len(df_exemplos) > 0
```
::: panel-tabset
#### Antes
```{python}
#| echo: false
if tem_exemplos:
# Reconstrói como seria ANTES do enriquecimento
df_antes = df_exemplos[['file_path']].copy()
df_antes['Conteúdo Original'] = df_antes['file_path'].apply(lambda x: f'<attached: {x}>')
df_antes = df_antes[['Conteúdo Original']]
display(df_antes.style.hide(axis='index'))
else:
print("Nenhum exemplo disponível.")
```
*Formato original no arquivo TXT exportado do WhatsApp (antes do wrangling).*
#### Depois
```{python}
#| echo: false
if tem_exemplos:
# Mostra transcrição (truncada)
df_depois = df_exemplos[['transcription']].copy()
df_depois['Conteúdo Enriquecido'] = df_depois['transcription'].apply(
lambda x: (str(x)[:120] + '...') if len(str(x)) > 120 else str(x)
)
df_depois = df_depois[['Conteúdo Enriquecido']]
display(df_depois.style.hide(axis='index'))
else:
print("Nenhum exemplo disponível.")
```
*Conteúdo após substituição pela transcrição via API Groq Whisper.*
:::
*Mensagens sem transcrição (ex: `image omitted`) permanecem inalteradas.*
### Etapa 5: Exportação
Exporta datasets em múltiplos formatos (CSV, Parquet, TXT).
#### 📦 Arquivos Gerados
::: panel-tabset
### 🎯 Principais
**Datasets estruturados:**
```{python}
if 'export_summary' in stats:
dados = []
n_linhas = stats.get('total_mensagens', len(df))
n_colunas = 8 # colunas do dataset principal
for key in ['csv_core', 'parquet']:
if key in stats['export_summary']:
info = stats['export_summary'][key]
uso_map = {
'csv_core': 'Alternativa ao Parquet',
'parquet': 'Recomendado (tipos otimizados)'
}
dados.append({
'Arquivo': f"<code>{info.get('arquivo', 'messages')}</code>",
'Formato': info.get('formato', key).upper(),
'Tamanho': f"{info.get('size_mb', 0):.2f} MB",
'Linhas': f'{n_linhas:,}',
'Colunas': n_colunas,
'Uso': uso_map.get(key, '')
})
if dados:
display(pd.DataFrame(dados).style.hide(axis='index'))
```
**Colunas do dataset (8):**
```{python}
colunas_info = [
('<code>timestamp</code>', 'datetime', 'Data e hora da mensagem'),
('<code>remetente</code>', 'category', 'Participante (P1 ou P2)'),
('<code>tipo_mensagem</code>', 'category', 'Tipo específico - 21 possíveis (text_pure, audio_omitted, video_attached, etc.)'),
('<code>grupo_mensagem</code>', 'category', 'Categoria agrupada: TEXT, AUDIO, VID, IMG, STICKER, GIF, DOC, CONTACT, FILE, SYSTEM, CALL'),
('<code>conteudo</code>', 'text', 'Conteúdo enriquecido: transcrição (se mídia transcrita) ou texto original. Renomeado de conteudo_enriquecido'),
('<code>arquivo</code>', 'text', 'Nome do arquivo de mídia anexado (ex: 00001-AUDIO-2024-10-21.opus), se aplicável'),
('<code>transcricao</code>', 'bool', 'Indica se a mensagem possui transcrição de áudio/vídeo. O texto da transcrição fica em conteudo. Renomeado de tem_transcricao'),
('<code>date_match</code>', 'bool', 'Indica se é mensagem órfã/sintética (transcrição sem match de timestamp). Renomeado de is_synthetic'),
]
df_colunas = pd.DataFrame(colunas_info, columns=['Coluna', 'Tipo', 'Descrição'])
display(df_colunas.style.hide(axis='index'))
```
**Como usar:**
``` python
# Recomendado: Parquet (rápido, tipos corretos)
df = pd.read_parquet('data/processed/messages.parquet')
# Alternativa: CSV
df = pd.read_csv('data/processed/messages.csv')
```
**Exemplos de filtros:**
``` python
# Apenas mensagens de texto
df_text = df[df['grupo_mensagem'] == 'TEXT']
# Mensagens com transcrição
df_transcribed = df[df['transcricao'] == True]
# Mensagens de P1 no mês de outubro
df_p1_oct = df[(df['remetente'] == 'P1') & (df['timestamp'].dt.month == 10)]
```
### 📊 Corpus
**Arquivos de texto puro para análises de linguagem natural:**
```{python}
corpus_files = []
for key in ['chat_complete', 'corpus_full', 'chat_p1', 'chat_p2', 'corpus_p1', 'corpus_p2']:
if key in outputs and outputs[key].exists():
file_path = outputs[key]
size_mb = os.path.getsize(file_path) / (1024 * 1024)
try:
num_lines = sum(1 for _ in open(file_path, 'r', encoding='utf-8'))
except:
num_lines = 0
# Descrição por arquivo
desc_map = {
'chat_complete': 'Chat completo. Formato: DD/MM/YY HH:MM:SS Remetente: Conteúdo',
'corpus_full': 'Corpus completo (apenas texto). Word clouds, vocabulário.',
'chat_p1': 'Chat de P1 com timestamps. Análise temporal.',
'chat_p2': 'Chat de P2 com timestamps. Análise temporal.',
'corpus_p1': 'Corpus de P1 (apenas texto). Comparação de estilos.',
'corpus_p2': 'Corpus de P2 (apenas texto). Comparação de estilos.',
}
corpus_files.append({
'Arquivo': f'<code>{file_path.name}</code>',
'Tamanho': f'{size_mb:.2f} MB',
'Linhas': f'{num_lines:,}',
'Descrição': desc_map.get(key, '')
})
if corpus_files:
display(pd.DataFrame(corpus_files).style.hide(axis='index'))
```
**Como usar:**
``` python
# Corpus completo (apenas texto, sem metadados)
with open('data/processed/corpus_full.txt', 'r', encoding='utf-8') as f:
texto = f.read()
# Análise de vocabulário
palavras = texto.split()
vocab_size = len(set(palavras))
```
**Por participante:**
``` python
# Corpus de P1 (apenas texto)
with open('data/processed/corpus_p1.txt', 'r', encoding='utf-8') as f:
texto_p1 = f.read()
# Corpus de P2 (apenas texto)
with open('data/processed/corpus_p2.txt', 'r', encoding='utf-8') as f:
texto_p2 = f.read()
```
**Chat com timestamps:**
``` python
# Lê chat completo
with open('data/processed/chat_complete.txt', 'r', encoding='utf-8') as f:
chat_lines = f.readlines()
# Parsing manual (se necessário)
import re
pattern = r'^(\d{2}/\d{2}/\d{2}) (\d{2}:\d{2}:\d{2}) (.+?): (.*)$'
for line in chat_lines:
match = re.match(pattern, line)
if match:
data, hora, remetente, conteudo = match.groups()
```
### 🔍 Debug
**Arquivo completo:**
```{python}
if 'csv_full' in stats.get('export_summary', {}):
info = stats['export_summary']['csv_full']
n_linhas = stats.get('total_mensagens', len(df))
n_colunas_full = info.get('colunas', 17) # dataset full tem mais colunas
df_debug = pd.DataFrame([{
'Arquivo': f"<code>{info.get('arquivo', 'messages_full.csv')}</code>",
'Tamanho': f"{info.get('size_mb', 0):.2f} MB",
'Linhas': f'{n_linhas:,}',
'Colunas': n_colunas_full,
'Descrição': 'Todas as colunas (debug)'
}])
display(df_debug.style.hide(axis='index'))
```
**Colunas extras (debug):**
```{python}
colunas_extras = [
('<code>linha_original</code>', 'Referência ao TXT original'),
('<code>data</code>', 'Redundante (já tem timestamp)'),
('<code>hora</code>', 'Redundante (já tem timestamp)'),
('<code>conteudo</code>', 'Conteúdo original (antes do enriquecimento)'),
('<code>arquivo_existe</code>', 'Bool de verificação'),
('<code>extensao</code>', "Derivável de 'arquivo'"),
('<code>tipo_arquivo</code>', "Derivável de 'arquivo'"),
('<code>arquivo_path</code>', 'Path completo do arquivo'),
('<code>transcricao</code>', 'Redundante - já tem coluna transcricao (bool) no dataset principal'),
('<code>transcription_status</code>', 'Status da API de transcrição (completed/error)'),
]
df_extras = pd.DataFrame(colunas_extras, columns=['Coluna', 'Descrição'])
display(df_extras.style.hide(axis='index'))
```
**Exportação manual (opcional):**
``` python
# Re-exportar só texto (sem tags de mídia)
from wrangling import export_corpus_files
df_text = df[df['grupo_mensagem'] == 'TEXT']
export_corpus_files(df_text, OUTPUT_DIR / 'text_only', use_enriched=False)
```
``` python
# Exportar CSV com colunas específicas
from wrangling import export_to_csv
cols = ['timestamp', 'remetente', 'conteudo', 'tipo_mensagem']
export_to_csv(df, OUTPUT_DIR / 'messages_minimal.csv', columns=cols)
```
**Pode deletar após verificação:** - `messages_full.csv` (só precisa se for debugar pipeline)
### 📥 Entrada
**Arquivo de input:**
```{python}
if TRANSCRIPTION_FILE.exists():
size_mb = os.path.getsize(TRANSCRIPTION_FILE) / (1024 * 1024)
try:
num_lines = sum(1 for _ in open(TRANSCRIPTION_FILE, 'r', encoding='utf-8')) - 1
except:
num_lines = 0
df_entrada = pd.DataFrame([{
'Arquivo': f'<code>{TRANSCRIPTION_FILE.name}</code>',
'Formato': 'CSV',
'Tamanho': f'{size_mb:.2f} MB',
'Linhas': f'{num_lines:,}',
'Descrição': 'Transcrições de áudio/vídeo (input da Etapa 4)'
}])
display(df_entrada.style.hide(axis='index'))
```
**Sobre este arquivo:**
Gerado por `scripts/transcribe_media.py`, contém:
- Nome do arquivo de áudio/vídeo
- Transcrição do conteúdo
- Status da API (success/error)
- Timestamp de processamento
**Como regenerar:**
``` bash
python scripts/transcribe_media.py
```
*Nota: Requer configuração da API Groq no arquivo `.env`*
:::
------------------------------------------------------------------------
# Próximos Passos
1. [**Feature Engineering**](04-feature-engineering.qmd) — Criação de variáveis derivadas