Direitos do titular: fluxo completo de atendimento para dados de CPF

import json
import hashlib
import sqlite3
from datetime import datetime, timezone, timedelta
from enum import Enum
from typing import Optional, List
import requests

class TipoDireito(Enum):
    CONFIRMACAO = "confirmacao_tratamento"
    ACESSO = "acesso_dados"
    CORRECAO = "correcao_dados"
    ELIMINACAO = "eliminacao_dados"
    PORTABILIDADE = "portabilidade"
    INFORMACAO_COMPARTILHAMENTO = "info_compartilhamento"
    REVOGACAO_CONSENTIMENTO = "revogacao_consentimento"
    ANONIMIZACAO = "anonimizacao"

class StatusSolicitacao(Enum):
    RECEBIDA = "recebida"
    EM_ANALISE = "em_analise"
    EM_PROCESSAMENTO = "em_processamento"
    CONCLUIDA = "concluida"
    INDEFERIDA = "indeferida"

class GerenciadorDireitosTitular:
    """Gerencia solicitações de direitos do titular para dados de CPF."""

    PRAZO_DIAS = 15 # Prazo legal para atendimento

    def __init__(self, db_path: str = "direitos_titular.db"):
    self.conn = sqlite3.connect(db_path, check_same_thread=False)
    self._criar_tabelas()

    def _criar_tabelas(self):
    self.conn.executescript("""
    CREATE TABLE IF NOT EXISTS solicitacoes (
    protocolo TEXT PRIMARY KEY,
    cpf_hash TEXT NOT NULL,
    tipo_direito TEXT NOT NULL,
    status TEXT NOT NULL,
    detalhes TEXT,
    criado_em TEXT NOT NULL,
    prazo_limite TEXT NOT NULL,
    concluido_em TEXT,
    resposta TEXT,
    atendente TEXT
    );

    CREATE TABLE IF NOT EXISTS historico_solicitacoes (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    protocolo TEXT NOT NULL,
    status_anterior TEXT,
    status_novo TEXT NOT NULL,
    timestamp TEXT NOT NULL,
    observacao TEXT
    );
    """)
    self.conn.commit()

    def abrir_solicitacao(
    self,
    cpf: str,
    tipo_direito: TipoDireito,
    detalhes: str = ""
    ) -> dict:
    """Abre uma nova solicitação de direito do titular."""

    cpf_hash = hashlib.sha256(cpf.encode()).hexdigest()
    protocolo = f"SOL-{datetime.now().strftime('%Y%m%d')}-{hashlib.sha256(f'{cpf}{datetime.now().isoformat()}'.encode()).hexdigest()[:8].upper()}"

    agora = datetime.now(timezone.utc)
    prazo = agora + timedelta(days=self.PRAZO_DIAS)

    self.conn.execute(
    """INSERT INTO solicitacoes
    (protocolo, cpf_hash, tipo_direito, status, detalhes, criado_em, prazo_limite)
    VALUES (?, ?, ?, ?, ?, ?, ?)""",
    (protocolo, cpf_hash, tipo_direito.value,
    StatusSolicitacao.RECEBIDA.value, detalhes,
    agora.isoformat(), prazo.isoformat())
    )

    self._registrar_historico(protocolo, None, StatusSolicitacao.RECEBIDA.value, "Solicitação recebida")
    self.conn.commit()

    return {
    "protocolo": protocolo,
    "tipo_direito": tipo_direito.value,
    "status": StatusSolicitacao.RECEBIDA.value,
    "criado_em": agora.isoformat(),
    "prazo_limite": prazo.isoformat()
    }

    def processar_solicitacao(self, protocolo: str, atendente: str):
    """Atualiza o status da solicitação para em processamento."""

    self.conn.execute(
    "UPDATE solicitacoes SET status = ?, atendente = ? WHERE protocolo = ?",
    (StatusSolicitacao.EM_PROCESSAMENTO.value, atendente, protocolo)
    )
    self._registrar_historico(
    protocolo, StatusSolicitacao.EM_ANALISE.value,
    StatusSolicitacao.EM_PROCESSAMENTO.value,
    f"Atendente: {atendente}"
    )
    self.conn.commit()

    def concluir_solicitacao(self, protocolo: str, resposta: str):
    """Conclui a solicitação com a resposta ao titular."""

    agora = datetime.now(timezone.utc).isoformat()
    self.conn.execute(
    "UPDATE solicitacoes SET status = ?, concluido_em = ?, resposta = ? WHERE protocolo = ?",
    (StatusSolicitacao.CONCLUIDA.value, agora, resposta, protocolo)
    )
    self._registrar_historico(
    protocolo, StatusSolicitacao.EM_PROCESSAMENTO.value,
    StatusSolicitacao.CONCLUIDA.value,
    "Solicitação concluída"
    )
    self.conn.commit()

    def consultar_status(self, protocolo: str) -> Optional[dict]:
    """Consulta o status de uma solicitação."""

    cursor = self.conn.execute(
    "SELECT protocolo, tipo_direito, status, criado_em, prazo_limite, concluido_em FROM solicitacoes WHERE protocolo = ?",
    (protocolo,)
    )
    row = cursor.fetchone()
    if row:
    return {
    "protocolo": row[0],
    "tipo_direito": row[1],
    "status": row[2],
    "criado_em": row[3],
    "prazo_limite": row[4],
    "concluido_em": row[5]
    }
    return None

    def solicitacoes_pendentes(self) -> List[dict]:
    """Lista solicitações pendentes próximas do prazo."""

    cursor = self.conn.execute(
    "SELECT protocolo, tipo_direito, status, prazo_limite FROM solicitacoes WHERE status != 'concluida' AND status != 'indeferida' ORDER BY prazo_limite ASC"
    )
    return [
    {"protocolo": r[0], "tipo": r[1], "status": r[2], "prazo": r[3]}
    for r in cursor.fetchall()
    ]

    def _registrar_historico(self, protocolo, status_anterior, status_novo, obs):
    self.conn.execute(
    "INSERT INTO historico_solicitacoes (protocolo, status_anterior, status_novo, timestamp, observacao) VALUES (?, ?, ?, ?, ?)",
    (protocolo, status_anterior, status_novo,
    datetime.now(timezone.utc).isoformat(), obs)
    )

def processar_acesso(cpf: str, api_key: str) -> dict:
    """Processa solicitação de acesso aos dados de CPF."""

    try:
    response = requests.get(
    f"https://api.cpfhub.io/cpf/{cpf}",
    headers={
    "x-api-key": api_key,
    "Accept": "application/json"
    },
    timeout=30
    )

    if response.status_code == 200:
    dados = response.json()["data"]
    return {
    "dados_tratados": dados,
    "base_legal": "execucao_contrato",
    "finalidade": "verificacao_cadastral",
    "data_coleta": "2025-01-15",
    "compartilhado_com": ["Provedor de pagamento XYZ"],
    "retencao_prevista": "365 dias"
    }

    except requests.exceptions.Timeout:
    return {"error": "Timeout ao recuperar dados"}

    return {"error": "Dados não encontrados"}

curl -X GET "https://api.cpfhub.io/cpf/12345678901" \
    -H "x-api-key: SUA_API_KEY" \
    -H "Accept: application/json" \
    --max-time 30

Titular -> [Solicitação] -> Recebida -> Verificação de Identidade
    -> Em Análise -> Classificação do Direito -> Em Processamento
    -> Execução da Ação -> Geração da Resposta -> Concluída
    -> Notificação ao Titular -> Registro de Auditoria