Integrações do Sistema

Visão Geral das Integrações

O Acesso ao Emprego possui integrações implementadas e planejadas para conectar candidatos, empresas e oportunidades de forma eficiente.

Integrações Implementadas

1. MinIO - Armazenamento de Objetos

O sistema utiliza MinIO como solução de armazenamento de objetos compatível com S3 para gerenciar arquivos de mídia de forma escalável e segura.

Arquitetura MinIO

Configuração MinIO

# settings/base.py
from storages.backends.s3boto3 import S3Boto3Storage

# MinIO Configuration
DEFAULT_FILE_STORAGE = 'storages.backends.s3boto3.S3Boto3Storage'
STATICFILES_STORAGE = 'storages.backends.s3boto3.S3StaticStorage'

AWS_ACCESS_KEY_ID = env('MINIO_ACCESS_KEY', default='minioadmin')
AWS_SECRET_ACCESS_KEY = env('MINIO_SECRET_KEY', default='minioadmin')
AWS_STORAGE_BUCKET_NAME = env('MINIO_BUCKET_NAME', default='portal-media')
AWS_S3_ENDPOINT_URL = env('MINIO_ENDPOINT', default='http://localhost:9000')
AWS_S3_REGION_NAME = 'us-east-1'
AWS_S3_USE_SSL = env.bool('MINIO_USE_SSL', default=False)
AWS_S3_VERIFY = env.bool('MINIO_VERIFY_SSL', default=False)
AWS_DEFAULT_ACL = None
AWS_QUERYSTRING_AUTH = True
AWS_S3_FILE_OVERWRITE = False

# Estrutura de buckets
MINIO_BUCKETS = {
    'media': 'portal-media',      # Arquivos de usuários
    'static': 'portal-static',    # Assets estáticos
    'backups': 'portal-backups',  # Backups do sistema
}

# URL de acesso
MEDIA_URL = f'{AWS_S3_ENDPOINT_URL}/{AWS_STORAGE_BUCKET_NAME}/'

Docker Compose - MinIO

# docker-compose.dev.yml
services:
  minio:
    image: minio/minio:latest
    ports:
      - "9000:9000" # API
      - "9001:9001" # Console Web
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin
    volumes:
      - minio_data:/data
    command: server /data --console-address ":9001"
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
      interval: 30s
      timeout: 20s
      retries: 3

volumes:
  minio_data:

Storage Backend Customizado

# storage_backends.py
from storages.backends.s3boto3 import S3Boto3Storage

class MediaStorage(S3Boto3Storage):
    """Storage para arquivos de mídia do usuário"""
    bucket_name = 'portal-media'
    file_overwrite = False
    default_acl = 'private'

    def get_object_parameters(self, name):
        """Define metadados para objetos"""
        params = super().get_object_parameters(name)

        # Adicionar cache control para imagens
        if name.endswith(('.jpg', '.jpeg', '.png', '.gif')):
            params['CacheControl'] = 'max-age=86400'

        return params

class StaticStorage(S3Boto3Storage):
    """Storage para arquivos estáticos"""
    bucket_name = 'portal-static'
    default_acl = 'public-read'
    file_overwrite = True

    def get_object_parameters(self, name):
        params = super().get_object_parameters(name)
        params['CacheControl'] = 'max-age=31536000'  # 1 ano
        return params

Organização de Arquivos

# models.py - Upload paths
def user_avatar_path(instance, filename):
    """Caminho para avatares de usuário"""
    ext = filename.split('.')[-1]
    filename = f'profile_avatar_{timezone.now().strftime("%Y%m%d%H%M%S")}.{ext}'
    return f'user/{instance.user.id}/{filename}'

def company_image_path(instance, filename):
    """Caminho para imagens de empresa"""
    ext = filename.split('.')[-1]
    filename = f'company_image_{timezone.now().strftime("%Y%m%d%H%M%S")}.{ext}'
    return f'company/{instance.id}/{filename}'

class CustomUser(AbstractUser):
    profile_avatar = models.ImageField(
        upload_to=user_avatar_path,
        validators=[validate_image_size, validate_image_extension],
        blank=True,
        null=True
    )

class Company(models.Model):
    logo = models.ImageField(
        upload_to=company_image_path,
        validators=[validate_image_size, validate_image_extension],
        blank=True,
        null=True
    )

Validação de Upload

# domain_services/validators/
def validate_image_size(image):
    """Valida tamanho máximo de 5MB"""
    max_size = 5 * 1024 * 1024
    if image.size > max_size:
        raise ValidationError('Imagem maior que 5MB')

def validate_image_extension(value):
    """Aceita apenas jpg, jpeg, png"""
    valid_extensions = ['.jpg', '.jpeg', '.png']
    ext = os.path.splitext(value.name)[1]
    if ext.lower() not in valid_extensions:
        raise ValidationError('Formato inválido')

Comandos de Gerenciamento MinIO

# management/commands/setup_minio.py
from django.core.management.base import BaseCommand
import boto3
from django.conf import settings

class Command(BaseCommand):
    help = 'Configura buckets do MinIO'

    def handle(self, *args, **options):
        # Conectar ao MinIO
        s3_client = boto3.client(
            's3',
            endpoint_url=settings.AWS_S3_ENDPOINT_URL,
            aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
            aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
            region_name=settings.AWS_S3_REGION_NAME,
            use_ssl=settings.AWS_S3_USE_SSL,
            verify=settings.AWS_S3_VERIFY
        )

        # Criar buckets
        buckets = ['portal-media', 'portal-static', 'portal-backups']

        for bucket in buckets:
            try:
                s3_client.create_bucket(Bucket=bucket)
                self.stdout.write(
                    self.style.SUCCESS(f'Bucket {bucket} criado')
                )

                # Configurar política de retenção
                if bucket == 'portal-backups':
                    s3_client.put_bucket_lifecycle_configuration(
                        Bucket=bucket,
                        LifecycleConfiguration={
                            'Rules': [{
                                'ID': 'delete-old-backups',
                                'Status': 'Enabled',
                                'Expiration': {'Days': 30}
                            }]
                        }
                    )
            except Exception as e:
                if 'BucketAlreadyExists' not in str(e):
                    self.stdout.write(
                        self.style.ERROR(f'Erro ao criar {bucket}: {e}')
                    )

2. PostgreSQL - Banco de Dados Principal

Sistema de gerenciamento de banco de dados relacional para armazenamento de dados estruturados.

Configuração

# settings/base.py
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': env('DB_NAME', default='portal_talentos'),
        'USER': env('DB_USER', default='postgres'),
        'PASSWORD': env('DB_PASSWORD', default='postgres'),
        'HOST': env('DB_HOST', default='localhost'),
        'PORT': env('DB_PORT', default='5432'),
        'CONN_MAX_AGE': 600,
        'OPTIONS': {
            'connect_timeout': 10,
        }
    }
}

3. OpenRouter - API de IA Generativa

Integração com OpenRouter para funcionalidades de processamento de linguagem natural e melhoria de textos usando modelos de IA generativa.

Arquitetura OpenRouter

Funcionalidades

• Melhoria de Texto: Aprimoramento de descrições profissionais
• Resumo de Conteúdo: Condensação de textos longos
• Correção Gramatical: Correção automática de erros
• Análise Contextual: Entendimento de contexto para melhores sugestões

Configuração

# settings/base.py
OPENROUTER_API_KEY = env('OPENROUTER_API_KEY', default=None)
OPENROUTER_SITE_URL = env('OPENROUTER_SITE_URL', default='')
OPENROUTER_SITE_NAME = env('OPENROUTER_SITE_NAME', default='')
OPENROUTER_MODEL = env('OPENROUTER_MODEL', default='openai/gpt-4o')

Implementação

# domain/services/ai_text_improvement_service.py
from openai import OpenAI

class AITextImprovementService:
    def __init__(self):
        self.client = OpenAI(
            base_url="https://openrouter.ai/api/v1",
            api_key=settings.OPENROUTER_API_KEY
        )
        self.model = settings.OPENROUTER_MODEL

    def improve_candidate_description(
        self,
        original_text: str,
        operation_type: str,
        mode: str = 'suggestion',
        char_limit: Optional[int] = None,
        **context
    ) -> dict:
        """
        Melhora texto do candidato usando IA.

        Modos:
        - suggestion: Melhora mantendo essência
        - summarize: Resume o texto
        - correction: Corrige apenas erros
        """
        prompt = self._build_prompt(original_text, operation_type, mode, **context)

        completion = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=int(char_limit * 0.35) if char_limit else None
        )

        improved_text = completion.choices[0].message.content.strip()

        # Validação de limite de caracteres
        if char_limit and len(improved_text) > char_limit:
            improved_text = self._truncate_intelligently(improved_text, char_limit)

        return {
            'improved_text': improved_text,
            'success': True
        }

Tipos de Operação Suportados

Tipo	Código	Limite de Caracteres
Experiência Profissional	professional_experience	2000
Certificados	certificates	2000
Formação Acadêmica	education	2000
Candidatura	application	500

Endpoint da API

POST /api/ai/improve-description/
Authorization: Bearer <jwt_token>
Content-Type: application/json

{
  "text": "Texto original",
  "mode": "suggestion",
  "operation_type": "professional_experience",
  "position": "Desenvolvedor",
  "company": "Empresa XYZ"
}

Response:

{
  "improved_text": "Texto melhorado pela IA",
  "success": true
}

Custos e Limites

• Custo: Baseado no modelo escolhido (GPT-4o: ~$0.005/1K tokens)
• Rate Limit: Conforme plano OpenRouter
• Timeout: 30 segundos por requisição
• Retry: Não implementado (falha retorna texto original)

Monitoramento

# Logs automáticos
logger.info(f"AI improvement requested - mode: {mode}, operation: {operation_type}")
logger.info(f"AI improvement completed - chars: {len(improved_text)}")
logger.error(f"AI improvement failed: {error}")

4. Motor Fuzzy/PLN Integrado

Sistema de inteligência artificial para matching entre candidatos e vagas.

Configuração

# settings/base.py
FUZZY_FILES_DOWNLOAD_URL = "https://sistema.carreirarh.com/download/skip_s300.zip"
FUZZY_FILES_ROOT = Path.joinpath(BASE_DIR, "fuzzy_files")

Implementação

# fuzzy/pln_processing.py
class FuzzyProcessor:
    def __init__(self):
        self.rules_count = 125  # Regras de inferência
        self.word2vec_model = self.load_word2vec()

    def calculate_similarity(self, text1, text2):
        """
        Calcula similaridade entre textos usando:
        - Word2Vec para semântica
        - Fuzzy logic para pontuação
        """
        tokens1 = self.tokenize(text1)
        tokens2 = self.tokenize(text2)

        semantic_score = self.word2vec_similarity(tokens1, tokens2)
        fuzzy_score = self.apply_fuzzy_rules(semantic_score)

        return fuzzy_score

Integrações Planejadas (Futuro Próximo)

1. Integração com CadÚnico 🔜

Integração com o Cadastro Único para validação e importação de dados de cidadãos.

Arquitetura Proposta

Implementação Proposta

# Futuro: services/cadunico_service.py
class CadUnicoService:
    def __init__(self):
        self.base_url = settings.CADUNICO_API_URL
        self.api_key = settings.CADUNICO_API_KEY
        self.cache_ttl = 3600  # 1 hora

    async def get_citizen_data(self, cpf: str) -> dict:
        """
        Busca dados do cidadão no CadÚnico
        """
        # Verificar cache primeiro
        cached = cache.get(f'cadunico:{cpf}')
        if cached:
            return cached

        # Buscar na API
        headers = {
            'X-API-Key': self.api_key,
            'Accept': 'application/json'
        }

        response = await self.http_client.get(
            f'{self.base_url}/api/v2/cidadao/{cpf}',
            headers=headers
        )

        if response.status_code == 200:
            data = response.json()
            cache.set(f'cadunico:{cpf}', data, self.cache_ttl)
            return self.map_to_candidate(data)

        raise CadUnicoAPIException('Erro ao buscar dados')

    def map_to_candidate(self, cadunico_data: dict) -> dict:
        """
        Mapeia dados do CadÚnico para modelo Candidate
        """
        return {
            'cpf': cadunico_data['nu_cpf_pessoa'],
            'full_name': cadunico_data['no_pessoa'],
            'birth_date': cadunico_data['dt_nasc_pessoa'],
            'mother_name': cadunico_data['no_mae_pessoa'],
            'address': {
                'street': cadunico_data['no_logradouro_fam'],
                'number': cadunico_data['nu_logradouro_fam'],
                'neighborhood': cadunico_data['no_bairro_fam'],
                'city': cadunico_data['no_municipio_fam'],
                'state': cadunico_data['sg_uf_munic_fam'],
                'zip_code': cadunico_data['nu_cep_logradouro_fam']
            },
            'phone': cadunico_data.get('nu_telefone_pessoa'),
            'email': cadunico_data.get('no_email_pessoa')
        }

Benefícios da Integração

• ✅ Validação automática de CPF
• ✅ Importação de dados básicos do cidadão
• ✅ Verificação de elegibilidade para programas sociais
• ✅ Atualização periódica de informações
• ✅ Redução de fraudes no cadastro

2. Serviço de Email para Notificações 🔜

Sistema de envio de emails para notificações e comunicação com usuários.

Arquitetura Proposta

Implementação Proposta

# Futuro: services/email_service.py
from django.core.mail import send_mail
from django.template.loader import render_to_string
from celery import shared_task

class EmailService:
    def __init__(self):
        self.from_email = settings.DEFAULT_FROM_EMAIL
        self.templates_path = 'emails/'

    def send_notification(self, user, template, context):
        """
        Envia email de notificação para usuário
        """
        # Adicionar à fila
        send_email_task.delay(
            user.email,
            template,
            context
        )

    def render_template(self, template_name, context):
        """
        Renderiza template de email com contexto
        """
        html_content = render_to_string(
            f'{self.templates_path}{template_name}.html',
            context
        )
        return html_content

@shared_task
def send_email_task(to_email, template, context):
    """
    Task Celery para envio assíncrono de email
    """
    service = EmailService()
    html_content = service.render_template(template, context)

    send_mail(
        subject=context.get('subject', 'Acesso ao Emprego'),
        message='',  # Texto plano
        from_email=service.from_email,
        recipient_list=[to_email],
        html_message=html_content,
        fail_silently=False
    )

Templates de Email Propostos

# Templates planejados
EMAIL_TEMPLATES = {
    'welcome': {
        'subject': 'Bem-vindo ao Acesso ao Emprego',
        'template': 'welcome.html'
    },
    'job_match': {
        'subject': 'Nova vaga compatível com seu perfil!',
        'template': 'job_match.html'
    },
    'application_status': {
        'subject': 'Atualização sobre sua candidatura',
        'template': 'application_status.html'
    },
    'company_approved': {
        'subject': 'Sua empresa foi aprovada!',
        'template': 'company_approved.html'
    },
    'interview_scheduled': {
        'subject': 'Entrevista agendada',
        'template': 'interview_scheduled.html'
    }
}

Configuração SMTP Proposta

# settings/production.py (Futuro)
EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_HOST = env('EMAIL_HOST', default='smtp.gmail.com')
EMAIL_PORT = env('EMAIL_PORT', default=587)
EMAIL_USE_TLS = True
EMAIL_HOST_USER = env('EMAIL_HOST_USER')
EMAIL_HOST_PASSWORD = env('EMAIL_HOST_PASSWORD')
DEFAULT_FROM_EMAIL = 'noreply@portal-talentos.gov.br'

# Alternativamente, usar serviço dedicado
# EMAIL_BACKEND = 'sendgrid_backend.SendgridBackend'
# SENDGRID_API_KEY = env('SENDGRID_API_KEY')

Segurança das Integrações

Práticas de Segurança Implementadas

1. Validação de Dados: Toda entrada é validada antes do processamento
2. Sanitização: Remoção de caracteres perigosos em uploads
3. Limites de Taxa: Proteção contra abuso (implementar com django-ratelimit)
4. Logs de Auditoria: Registro de todas as operações sensíveis
5. MinIO Security:
   • Acesso via HTTPS em produção
   • Políticas de bucket restritivas
   • URLs assinadas com expiração
   • Criptografia de objetos em repouso

Práticas Futuras

# Futuro: middleware/security.py
class IntegrationSecurityMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # Verificar assinatura de webhooks
        if request.path.startswith('/webhooks/'):
            self.verify_webhook_signature(request)

        # Rate limiting por IP
        self.check_rate_limit(request)

        response = self.get_response(request)
        return response

    def verify_webhook_signature(self, request):
        """
        Verifica assinatura HMAC de webhooks
        """
        signature = request.headers.get('X-Signature')
        if not signature:
            raise PermissionDenied('Missing signature')

        expected = hmac.new(
            settings.WEBHOOK_SECRET.encode(),
            request.body,
            hashlib.sha256
        ).hexdigest()

        if not hmac.compare_digest(signature, expected):
            raise PermissionDenied('Invalid signature')

Monitoramento de Integrações

Sistema de Health Check

# monitoring/health_checks.py
class IntegrationHealthCheck:
    def check_all(self):
        results = {
            'database': self.check_database(),
            'minio_storage': self.check_minio(),
            'fuzzy_engine': self.check_fuzzy_engine(),
            'cadunico_api': self.check_cadunico_api(),  # Futuro
            'email_service': self.check_email_service()  # Futuro
        }
        return results

    def check_database(self):
        """Verifica conexão com PostgreSQL"""
        from django.db import connection
        try:
            with connection.cursor() as cursor:
                cursor.execute("SELECT 1")
            return {'status': 'healthy', 'message': 'Database conectado'}
        except Exception as e:
            return {'status': 'unhealthy', 'message': str(e)}

    def check_minio(self):
        """Verifica se o MinIO está acessível"""
        import boto3
        try:
            s3_client = boto3.client(
                's3',
                endpoint_url=settings.AWS_S3_ENDPOINT_URL,
                aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
                aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY
            )

            # Listar buckets para verificar conexão
            buckets = s3_client.list_buckets()

            # Verificar bucket principal
            s3_client.head_bucket(Bucket=settings.AWS_STORAGE_BUCKET_NAME)

            return {
                'status': 'healthy',
                'message': 'MinIO acessível',
                'buckets': len(buckets.get('Buckets', []))
            }
        except Exception as e:
            return {'status': 'unhealthy', 'message': str(e)}

    def check_fuzzy_engine(self):
        """Verifica se o motor fuzzy está operacional"""
        try:
            from fuzzy.pln_processing import FuzzyProcessor
            processor = FuzzyProcessor()
            score = processor.calculate_similarity("test", "teste")
            return {'status': 'healthy', 'score': score}
        except Exception as e:
            return {'status': 'unhealthy', 'message': str(e)}

---

Documentação Atualizada

Esta documentação reflete o estado atual das integrações e o planejamento para implementações futuras, focando especialmente na integração com CadÚnico e serviços de email.