Projeto 3: Bot de Automação com Scraping e Notificações

O Projeto 3 é um bot de automação completo — monitora fontes de dados na web, processa as informações coletadas e envia notificações inteligentes por múltiplos canais. Ele junta Web Scraping com BeautifulSoup e Selenium, Automação com Python e Envio de E-mails e Integração com APIs de Comunicação. O sistema monitora vagas de emprego em tecnologia, filtra por critérios do usuário e notifica em tempo real via Telegram e e-mail.

Visão Geral do Projeto

job-bot/
├── src/
│   └── jobbot/
│       ├── __init__.py
│       ├── config.py              ← configurações e critérios
│       ├── modelos.py             ← a Vaga
│       ├── agendador.py           ← schedule + execução
│       ├── coletores/
│       │   ├── __init__.py
│       │   ├── base.py            ← classe base de coleta
│       │   ├── remoteok.py        ← API RemoteOK
│       │   └── linkedin.py        ← scraping (ver aviso adiante)
│       ├── processadores/
│       │   ├── __init__.py
│       │   ├── filtro.py          ← filtragem e pontuação
│       │   └── deduplicador.py    ← remoção de duplicatas
│       ├── armazenamento/
│       │   ├── __init__.py
│       │   └── repositorio.py     ← SQLite para vagas
│       └── notificadores/
│           ├── __init__.py
│           ├── telegram.py        ← bot Telegram
│           └── email.py           ← notificações por e-mail
├── tests/
│   ├── conftest.py
│   ├── test_filtro.py
│   └── test_deduplicador.py
├── data/
├── logs/
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
└── .env.example

Configuração

# src/jobbot/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import Field
from typing import Optional
from functools import lru_cache


class Configuracoes(BaseSettings):
    # Bot Telegram
    telegram_token:  str
    telegram_chat_id: str

    # E-mail
    email_usuario:   str
    email_senha:     str
    email_destino:   str
    smtp_host:       str  = "smtp.gmail.com"
    smtp_porta:      int  = 587

    # Critérios de busca
    palavras_chave:  list[str] = ["Python", "FastAPI", "Django", "Data Science"]
    tecnologias:     list[str] = ["Python", "FastAPI", "Django", "pandas", "scikit-learn"]
    # na unidade do RemoteOK: dólares por ano (US$ 60 mil ≈ US$ 5 mil/mês)
    salario_minimo:  float     = 60000.0
    remoto_apenas:   bool      = True
    senioridade:     list[str] = ["junior", "pleno", "senior", ""]

    # Agendamento
    intervalo_minutos: int = 60
    horario_resumo:    str = "08:00"   # resumo diário

    # Armazenamento
    db_path:    str = "data/vagas.db"
    max_vagas_notificadas: int = 5   # por execução

    model_config = SettingsConfigDict(env_file=".env")


@lru_cache
def get_config() -> Configuracoes:
    return Configuracoes()

O salário mínimo está em dólares por ano porque é essa a unidade que o RemoteOK devolve: consultada em setembro de 2026, a API trazia, por exemplo, salary_min: 70000 e salary_max: 80000. A primeira versão comparava esses valores com 5000, pensados como reais por mês, e qualquer vaga com salário informado passava no critério.

Domínio: Vaga de Emprego

# src/jobbot/modelos.py
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Optional
import hashlib


@dataclass
class Vaga:
    titulo:        str
    empresa:       str
    descricao:     str
    url:           str
    fonte:         str
    localidade:    str          = "Remoto"
    salario_min:   Optional[float] = None
    salario_max:   Optional[float] = None
    tecnologias:   list[str]    = field(default_factory=list)
    senioridade:   str          = ""
    remoto:        bool         = True
    coletado_em:   datetime     = field(
        default_factory=lambda: datetime.now(timezone.utc)
    )
    pontuacao:     float        = 0.0
    notificada:    bool         = False

    @property
    def id(self) -> str:
        """ID único baseado no conteúdo — detecta duplicatas."""
        conteudo = f"{self.titulo}{self.empresa}{self.url}"
        return hashlib.md5(conteudo.encode()).hexdigest()[:12]

    @property
    def salario_display(self) -> str:
        if self.salario_min and self.salario_max:
            return f"US$ {self.salario_min:,.0f} – {self.salario_max:,.0f}/ano"
        if self.salario_min:
            return f"A partir de US$ {self.salario_min:,.0f}/ano"
        return "Salário não informado"

    def __str__(self):
        return (
            f"[{self.fonte}] {self.titulo} @ {self.empresa}\n"
            f"  {self.localidade} | {self.salario_display}\n"
            f"  {self.url}"
        )

Coletores

# src/jobbot/coletores/base.py
from abc import ABC, abstractmethod
from typing import Iterator
from jobbot.modelos import Vaga
import logging
import time

logger = logging.getLogger("jobbot.coletor")


class ColetorBase(ABC):
    """Classe base para todos os coletores."""

    nome: str = "base"
    intervalo_requisicoes: float = 2.0   # segundos entre requisições

    def __init__(self):
        self._session = None

    @abstractmethod
    def coletar(self, termos: list[str]) -> Iterator[Vaga]:
        """Coleta vagas para os termos fornecidos."""
        ...

    def _aguardar(self):
        time.sleep(self.intervalo_requisicoes)

    def _criar_session(self):
        import requests
        from requests.adapters import HTTPAdapter
        from urllib3.util.retry import Retry

        session = requests.Session()
        session.headers.update({
            "User-Agent": (
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36"
            ),
            "Accept-Language": "pt-BR,pt;q=0.9,en;q=0.8"
        })

        retry = Retry(total=3, backoff_factor=0.5, status_forcelist=[429, 500, 503])
        session.mount("https://", HTTPAdapter(max_retries=retry))
        return session

# src/jobbot/coletores/remoteok.py
from jobbot.coletores.base import ColetorBase
from jobbot.modelos import Vaga
from typing import Iterator
import logging

logger = logging.getLogger("jobbot.remoteok")


class ColetorRemoteOK(ColetorBase):
    """
    Coleta vagas da API pública do RemoteOK.
    Documentação: https://remoteok.com/api
    """

    nome    = "RemoteOK"
    API_URL = "https://remoteok.com/api"

    def coletar(self, termos: list[str]) -> Iterator[Vaga]:
        session = self._criar_session()

        # Falha de rede ou HTTP sobe para o agendador, que avisa no Telegram;
        # engolida aqui, o bot passaria a "encontrar 0 vagas" em silêncio
        logger.info("Coletando vagas do RemoteOK...")
        resposta = session.get(self.API_URL, timeout=15)
        resposta.raise_for_status()

        dados = resposta.json()
        # o primeiro item é o aviso legal (termos de uso), sem "id"
        vagas_raw = [d for d in dados if isinstance(d, dict) and "id" in d]

        logger.info(f"RemoteOK: {len(vagas_raw)} vagas brutas")

        for raw in vagas_raw:
            try:
                vaga = self._parsear(raw, termos)
                if vaga:
                    yield vaga
            except Exception as e:
                logger.debug(f"Erro ao parsear vaga: {e}")

    def _parsear(self, raw: dict, termos: list[str]) -> Vaga | None:
        titulo    = raw.get("position", "")
        empresa   = raw.get("company",  "")
        descricao = raw.get("description", "")
        url       = raw.get("url", "")
        tags      = raw.get("tags", []) or []

        # Filtra por termos relevantes
        texto_busca = f"{titulo} {descricao} {' '.join(tags)}".lower()
        termo_encontrado = any(t.lower() in texto_busca for t in termos)
        if not termo_encontrado:
            return None

        # Salário: a API traz salary_min e salary_max numéricos, em dólares
        # por ano, e 0 quando a vaga não informa
        salario_min = float(raw.get("salary_min") or 0) or None
        salario_max = float(raw.get("salary_max") or 0) or None

        # Tecnologias detectadas
        techs_detectadas = [
            t for t in [
                "Python", "FastAPI", "Django", "Flask",
                "Docker", "Kubernetes", "PostgreSQL", "Redis",
                "React", "TypeScript", "Go", "Rust"
            ]
            if t.lower() in texto_busca
        ]

        return Vaga(
            titulo=      titulo,
            empresa=     empresa,
            descricao=   descricao[:500],
            url=         url,
            fonte=       self.nome,
            salario_min= salario_min,
            salario_max= salario_max,
            tecnologias= techs_detectadas,
            remoto=      True,
            localidade=  "Remoto"
        )

Dois ajustes vieram de consultar a API de verdade. Ela não tem campo salary em texto, como a primeira versão supunha — tem salary_min e salary_max numéricos —, e a expressão regular nunca encontrava nada: todo salário saía vazio. Com os campos certos, 14 das 79 vagas coletadas num teste vieram com salário. O outro ajuste é deixar a falha de rede subir: antes, o try em volta de tudo registrava o erro e devolvia zero vagas, e o aviso de erro do agendador, que manda uma mensagem ao Telegram, nunca era disparado — medido com a rede derrubada de propósito. O primeiro item da resposta é o aviso de termos de uso: o RemoteOK pede que quem usa a API cite a fonte e aponte um link para a vaga no site deles, sob pena de suspender o acesso. O link "Ver vaga" das notificações cumpre a segunda parte, e o e-mail passou a mostrar a fonte.

# src/jobbot/coletores/linkedin.py
from jobbot.coletores.base import ColetorBase
from jobbot.modelos import Vaga
from typing import Iterator
from bs4 import BeautifulSoup
import logging

logger = logging.getLogger("jobbot.linkedin")


class ColetorLinkedIn(ColetorBase):
    """
    Coleta vagas públicas do LinkedIn sem autenticação.
    Usa a URL de busca pública de empregos.
    """

    nome    = "LinkedIn"
    BASE_URL = "https://www.linkedin.com/jobs/search"

    def coletar(self, termos: list[str]) -> Iterator[Vaga]:
        session = self._criar_session()

        for termo in termos:
            try:
                yield from self._coletar_termo(session, termo)
                self._aguardar()
            except Exception as e:
                logger.error(f"Erro LinkedIn para '{termo}': {e}")

    def _coletar_termo(self, session, termo: str) -> Iterator[Vaga]:
        params = {
            "keywords":    termo,
            "location":    "Brazil",
            "f_WT":        "2",      # remote
            "f_TPR":       "r86400", # últimas 24h
            "start":       "0"
        }

        logger.info(f"LinkedIn: buscando '{termo}'...")
        resposta = session.get(self.BASE_URL, params=params, timeout=15)

        if resposta.status_code != 200:
            logger.warning(f"LinkedIn retornou {resposta.status_code}")
            return

        soup  = BeautifulSoup(resposta.text, "lxml")
        cards = soup.select("div.base-card")

        logger.info(f"LinkedIn: {len(cards)} vagas para '{termo}'")

        for card in cards[:10]:   # limita para não sobrecarregar
            try:
                vaga = self._parsear_card(card)
                if vaga:
                    yield vaga
            except Exception as e:
                logger.debug(f"Erro ao parsear card: {e}")

    def _parsear_card(self, card) -> Vaga | None:
        titulo_el  = card.select_one("h3.base-search-card__title")
        empresa_el = card.select_one("h4.base-search-card__subtitle")
        local_el   = card.select_one("span.job-search-card__location")
        link_el    = card.select_one("a.base-card__full-link")

        if not all([titulo_el, empresa_el, link_el]):
            return None

        titulo  = titulo_el.get_text(strip=True)
        empresa = empresa_el.get_text(strip=True)
        local   = local_el.get_text(strip=True) if local_el else "N/I"
        url     = link_el.get("href", "").split("?")[0]

        if not url:
            return None

        return Vaga(
            titulo=    titulo,
            empresa=   empresa,
            descricao= f"{titulo} em {empresa}",
            url=       url,
            fonte=     self.nome,
            localidade= local,
            remoto=    "remoto" in local.lower() or "remote" in local.lower()
        )

Um aviso sobre esse coletor: o Contrato do Usuário do LinkedIn proíbe coleta automatizada do site, e a empresa bloqueia ativamente esse tipo de acesso — a estrutura de classes usada aqui também muda sem aviso. O código fica como exemplo de técnica de scraping com BeautifulSoup; para um bot que vai rodar de verdade, prefira fontes que oferecem API ou feed, como o próprio RemoteOK, e confira os termos de cada site antes de coletar. Outro detalhe: numa vaga remota do LinkedIn, a localidade costuma ser a cidade da empresa, e com remoto_apenas ligado, o filtro por texto descartaria a vaga.

Processadores

# src/jobbot/processadores/filtro.py
from jobbot.modelos import Vaga
from jobbot.config import get_config
import logging

logger = logging.getLogger("jobbot.filtro")
config = get_config()


def pontuar_vaga(vaga: Vaga) -> float:
    """Calcula pontuação de relevância de uma vaga."""
    pontos = 0.0

    # Tecnologias relevantes
    techs_match = sum(
        1 for t in config.tecnologias
        if t.lower() in vaga.titulo.lower() or
           t.lower() in vaga.descricao.lower() or
           t in vaga.tecnologias
    )
    pontos += techs_match * 10

    # Remoto
    if vaga.remoto:
        pontos += 20

    # Salário
    if vaga.salario_min and vaga.salario_min >= config.salario_minimo:
        pontos += 15
    elif vaga.salario_min and vaga.salario_min >= config.salario_minimo * 0.8:
        pontos += 8

    # Senioridade
    titulo_lower = vaga.titulo.lower()
    if any(s in titulo_lower for s in ["senior", "sênior", "sr."]):
        pontos += 10
    elif any(s in titulo_lower for s in ["pleno", "mid", "middle"]):
        pontos += 8
    elif any(s in titulo_lower for s in ["junior", "júnior", "jr."]):
        pontos += 5

    # Palavras negativas
    palavras_negativas = ["estágio", "estagiário", "intern", "unpaid"]
    if any(p in titulo_lower for p in palavras_negativas):
        pontos -= 30

    return max(0.0, pontos)


def filtrar_vagas(vagas: list[Vaga]) -> list[Vaga]:
    """Filtra e pontua lista de vagas."""
    resultado = []

    for vaga in vagas:
        # Filtro obrigatório: remoto
        if config.remoto_apenas and not vaga.remoto:
            continue

        # Filtro obrigatório: palavras-chave
        texto = f"{vaga.titulo} {vaga.descricao}".lower()
        tem_keyword = any(
            kw.lower() in texto for kw in config.palavras_chave
        )
        if not tem_keyword:
            continue

        # Pontua
        vaga.pontuacao = pontuar_vaga(vaga)
        if vaga.pontuacao > 0:
            resultado.append(vaga)

    # Ordena por pontuação
    resultado.sort(key=lambda v: v.pontuacao, reverse=True)
    logger.info(f"Filtro: {len(vagas)} → {len(resultado)} vagas relevantes")
    return resultado

# src/jobbot/processadores/deduplicador.py
from jobbot.modelos import Vaga
from jobbot.armazenamento.repositorio import RepositorioVagas
import logging

logger = logging.getLogger("jobbot.deduplicador")


def deduplicar(vagas: list[Vaga], repo: RepositorioVagas) -> list[Vaga]:
    """Remove vagas já vistas no banco de dados."""
    novas = []
    for vaga in vagas:
        if not repo.existe(vaga.id):
            novas.append(vaga)

    logger.info(f"Deduplicação: {len(vagas)} → {len(novas)} vagas novas")
    return novas

Armazenamento

# src/jobbot/armazenamento/repositorio.py
import sqlite3
from pathlib import Path
from datetime import datetime
from jobbot.modelos import Vaga
import json
import logging

logger = logging.getLogger("jobbot.repo")


class RepositorioVagas:

    def __init__(self, db_path: str = "data/vagas.db"):
        Path(db_path).parent.mkdir(parents=True, exist_ok=True)
        self.db_path = db_path
        self._criar_tabelas()

    def _conn(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        return conn

    def _criar_tabelas(self):
        with self._conn() as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS vagas (
                    id            TEXT PRIMARY KEY,
                    titulo        TEXT NOT NULL,
                    empresa       TEXT NOT NULL,
                    descricao     TEXT,
                    url           TEXT UNIQUE,
                    fonte         TEXT,
                    localidade    TEXT,
                    salario_min   REAL,
                    salario_max   REAL,
                    tecnologias   TEXT,
                    remoto        INTEGER,
                    pontuacao     REAL DEFAULT 0,
                    notificada    INTEGER DEFAULT 0,
                    coletado_em   TEXT
                )
            """)
            conn.execute("""
                CREATE TABLE IF NOT EXISTS execucoes (
                    id          INTEGER PRIMARY KEY AUTOINCREMENT,
                    inicio      TEXT,
                    fim         TEXT,
                    n_coletadas INTEGER,
                    n_novas     INTEGER,
                    n_notificadas INTEGER
                )
            """)

    def salvar(self, vaga: Vaga) -> bool:
        try:
            with self._conn() as conn:
                conn.execute("""
                    INSERT OR IGNORE INTO vagas
                    (id, titulo, empresa, descricao, url, fonte,
                     localidade, salario_min, salario_max, tecnologias,
                     remoto, pontuacao, notificada, coletado_em)
                    VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)
                """, (
                    vaga.id, vaga.titulo, vaga.empresa,
                    vaga.descricao, vaga.url, vaga.fonte,
                    vaga.localidade, vaga.salario_min, vaga.salario_max,
                    json.dumps(vaga.tecnologias),
                    int(vaga.remoto), vaga.pontuacao,
                    int(vaga.notificada),
                    vaga.coletado_em.isoformat()
                ))
            return True
        except Exception as e:
            logger.error(f"Erro ao salvar vaga {vaga.id}: {e}")
            return False

    def existe(self, vaga_id: str) -> bool:
        with self._conn() as conn:
            row = conn.execute(
                "SELECT 1 FROM vagas WHERE id = ?", (vaga_id,)
            ).fetchone()
        return row is not None

    def marcar_notificada(self, vaga_id: str):
        with self._conn() as conn:
            conn.execute(
                "UPDATE vagas SET notificada = 1 WHERE id = ?",
                (vaga_id,)
            )

    def nao_notificadas(self, limite: int = 5) -> list[dict]:
        with self._conn() as conn:
            rows = conn.execute("""
                SELECT * FROM vagas
                WHERE notificada = 0
                ORDER BY pontuacao DESC
                LIMIT ?
            """, (limite,)).fetchall()
        return [dict(r) for r in rows]

    def estatisticas(self) -> dict:
        with self._conn() as conn:
            total      = conn.execute("SELECT COUNT(*) FROM vagas").fetchone()[0]
            notificadas = conn.execute(
                "SELECT COUNT(*) FROM vagas WHERE notificada = 1"
            ).fetchone()[0]
            por_fonte  = conn.execute(
                "SELECT fonte, COUNT(*) as n FROM vagas GROUP BY fonte"
            ).fetchall()
        return {
            "total":       total,
            "notificadas": notificadas,
            "pendentes":   total - notificadas,
            "por_fonte":   {r["fonte"]: r["n"] for r in por_fonte}
        }

    def registrar_execucao(self, inicio, fim, coletadas, novas, notificadas):
        with self._conn() as conn:
            conn.execute("""
                INSERT INTO execucoes (inicio, fim, n_coletadas, n_novas, n_notificadas)
                VALUES (?,?,?,?,?)
            """, (inicio.isoformat(), fim.isoformat(), coletadas, novas, notificadas))

Notificadores

# src/jobbot/notificadores/telegram.py
import html
import httpx
import asyncio
from jobbot.modelos import Vaga
from jobbot.config import get_config
import logging

logger = logging.getLogger("jobbot.telegram")
config = get_config()


class NotificadorTelegram:

    BASE_URL = "https://api.telegram.org/bot{token}"

    def __init__(self):
        self._url = f"https://api.telegram.org/bot{config.telegram_token}"

    async def enviar_mensagem(self, texto: str, parse_mode: str = "HTML") -> bool:
        async with httpx.AsyncClient(timeout=10.0) as client:
            try:
                resp = await client.post(
                    f"{self._url}/sendMessage",
                    json={
                        "chat_id":    config.telegram_chat_id,
                        "text":       texto,
                        "parse_mode": parse_mode,
                        "disable_web_page_preview": False
                    }
                )
                return resp.status_code == 200
            except Exception as e:
                logger.error(f"Erro Telegram: {e}")
                return False

    def formatar_vaga(self, vaga: Vaga) -> str:
        # parse_mode HTML: <, > e & no texto precisam ser escapados, senão o
        # Telegram recusa a mensagem inteira ("can't parse entities")
        e = html.escape
        techs = ", ".join(vaga.tecnologias[:5]) if vaga.tecnologias else "N/I"
        return (
            f"🎯 <b>{e(vaga.titulo)}</b>\n"
            f"🏢 {e(vaga.empresa)}\n"
            f"📍 {e(vaga.localidade)}\n"
            f"💰 {vaga.salario_display}\n"
            f"🛠️ {e(techs)}\n"
            f"⭐ Score: {vaga.pontuacao:.0f}\n"
            f"🔗 <a href=\"{e(vaga.url)}\">Ver vaga</a>"
        )

    async def notificar_vaga(self, vaga: Vaga) -> bool:
        texto = self.formatar_vaga(vaga)
        return await self.enviar_mensagem(texto)

    async def notificar_resumo(self, stats: dict) -> bool:
        texto = (
            f"📊 <b>Resumo Diário — JobBot</b>\n\n"
            f"Total coletadas: {stats['total']}\n"
            f"Notificadas hoje: {stats['notificadas']}\n"
            f"Pendentes: {stats['pendentes']}\n\n"
            f"<b>Por fonte:</b>\n"
        )
        for fonte, n in stats.get("por_fonte", {}).items():
            texto += f"  • {fonte}: {n}\n"
        return await self.enviar_mensagem(texto)

    async def notificar_erro(self, mensagem: str) -> bool:
        texto = f"⚠️ <b>Erro no JobBot</b>\n\n{mensagem}"
        return await self.enviar_mensagem(texto)

# src/jobbot/notificadores/email.py
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from jinja2 import Template
from jobbot.modelos import Vaga
from jobbot.config import get_config
import logging

logger = logging.getLogger("jobbot.email")
config = get_config()

TEMPLATE_EMAIL = """
<!DOCTYPE html>
<html lang="pt-BR">
<head>
  <meta charset="UTF-8">
  <style>
    body   { font-family: Arial, sans-serif; background: #f4f4f4; }
    .card  { background: white; border-radius: 8px; padding: 20px;
             margin: 10px 0; border-left: 4px solid #2E75B6; }
    .titulo { font-size: 1.1rem; font-weight: bold; color: #1F4E79; }
    .info   { color: #666; font-size: 0.9rem; margin: 4px 0; }
    .badge  { display: inline-block; background: #E8F4FD; color: #2E75B6;
              padding: 2px 8px; border-radius: 12px; font-size: 0.8rem;
              margin: 2px; }
    .btn    { display: inline-block; background: #2E75B6; color: white;
              padding: 8px 16px; border-radius: 4px; text-decoration: none; }
  </style>
</head>
<body>
  <h2>🎯 {{ n_vagas }} nova(s) vaga(s) encontrada(s)</h2>

  {% for vaga in vagas %}
  <div class="card">
    <div class="titulo">{{ vaga.titulo }}</div>
    <div class="info">🏢 {{ vaga.empresa }}</div>
    <div class="info">📍 {{ vaga.localidade }}</div>
    <div class="info">💰 {{ vaga.salario_display }}</div>
    <div class="info">⭐ Score: {{ "%.0f"|format(vaga.pontuacao) }}</div>
    <div class="info">Fonte: {{ vaga.fonte }}</div>
    <div style="margin: 8px 0">
      {% for tech in vaga.tecnologias[:6] %}
        <span class="badge">{{ tech }}</span>
      {% endfor %}
    </div>
    <a href="{{ vaga.url }}" class="btn">Ver Vaga →</a>
  </div>
  {% endfor %}

  <p style="color:#888;font-size:0.8rem">
    JobBot — enviado automaticamente em {{ agora }}
  </p>
</body>
</html>
"""


class NotificadorEmail:

    def enviar_vagas(self, vagas: list[Vaga]) -> bool:
        from datetime import datetime

        # autoescape: título e empresa vêm de sites de terceiros
        template = Template(TEMPLATE_EMAIL, autoescape=True)
        html = template.render(
            vagas=vagas,
            n_vagas=len(vagas),
            agora=datetime.now().strftime("%d/%m/%Y %H:%M")
        )

        msg = MIMEMultipart("alternative")
        msg["Subject"] = f"🎯 {len(vagas)} nova(s) vaga(s) Python — JobBot"
        msg["From"]    = config.email_usuario
        msg["To"]      = config.email_destino
        msg.attach(MIMEText(html, "html", "utf-8"))

        try:
            with smtplib.SMTP(config.smtp_host, config.smtp_porta) as smtp:
                smtp.ehlo()
                smtp.starttls()
                smtp.login(config.email_usuario, config.email_senha)
                smtp.sendmail(
                    config.email_usuario,
                    config.email_destino,
                    msg.as_string()
                )
            logger.info(f"E-mail enviado: {len(vagas)} vagas")
            return True
        except Exception as e:
            logger.error(f"Erro ao enviar e-mail: {e}")
            return False

Título, empresa e localidade vêm de sites de terceiros, e os dois notificadores precisam tratá-los como texto. No Telegram, com parse_mode="HTML", um título como "Dev C++ & Python <Sr>" sem escape faz a API recusar a mensagem inteira, e a vaga nunca é marcada como notificada. No e-mail, o Template do Jinja não escapa nada por padrão: medido, um título com <b> entrava como HTML de verdade no corpo da mensagem. Com html.escape no Telegram e autoescape=True no Jinja, o mesmo título sai como texto nos dois canais.

Agendador: Orquestrando Tudo

# src/jobbot/agendador.py
import asyncio
import schedule
import time
import json
import logging
from datetime import datetime, timezone
from jobbot.config import get_config
from jobbot.coletores.remoteok import ColetorRemoteOK
from jobbot.coletores.linkedin import ColetorLinkedIn
from jobbot.processadores.filtro import filtrar_vagas
from jobbot.processadores.deduplicador import deduplicar
from jobbot.armazenamento.repositorio import RepositorioVagas
from jobbot.notificadores.telegram import NotificadorTelegram
from jobbot.notificadores.email import NotificadorEmail

logger = logging.getLogger("jobbot.agendador")
config = get_config()


class JobBot:
    """Orquestrador principal do bot de vagas."""

    def __init__(self):
        self.repo         = RepositorioVagas(config.db_path)
        self.coletores    = [ColetorRemoteOK(), ColetorLinkedIn()]
        self.telegram     = NotificadorTelegram()
        self.email        = NotificadorEmail()

    async def executar_ciclo(self):
        """Um ciclo completo: coletar → filtrar → deduplicar → notificar."""
        inicio = datetime.now(timezone.utc)
        logger.info(f"=== Ciclo iniciado: {inicio.strftime('%H:%M')} ===")

        # 1. Coleta
        todas_vagas = []
        for coletor in self.coletores:
            try:
                vagas = list(coletor.coletar(config.palavras_chave))
                todas_vagas.extend(vagas)
                logger.info(f"{coletor.nome}: {len(vagas)} vagas coletadas")
            except Exception as e:
                logger.error(f"Erro em {coletor.nome}: {e}")
                await self.telegram.notificar_erro(
                    f"Falha no coletor {coletor.nome}: {e}"
                )

        # 2. Filtro
        vagas_filtradas = filtrar_vagas(todas_vagas)

        # 3. Deduplicação
        vagas_novas = deduplicar(vagas_filtradas, self.repo)

        # 4. Persistência
        for vaga in vagas_novas:
            self.repo.salvar(vaga)

        # 5. Notificação — top vagas não notificadas
        nao_notificadas = self.repo.nao_notificadas(
            limite=config.max_vagas_notificadas
        )

        notificadas = 0
        if nao_notificadas:
            from jobbot.modelos import Vaga

            vagas_obj = [
                Vaga(
                    titulo=    r["titulo"],
                    empresa=   r["empresa"],
                    descricao= r["descricao"] or "",
                    url=       r["url"],
                    fonte=     r["fonte"],
                    localidade= r["localidade"],
                    salario_min= r["salario_min"],
                    salario_max= r["salario_max"],
                    pontuacao=  r["pontuacao"],
                    # o banco guarda tudo; sem estas linhas, a mensagem saía
                    # sem tecnologias e toda vaga era tratada como remota
                    tecnologias= json.loads(r["tecnologias"] or "[]"),
                    remoto=      bool(r["remoto"]),
                )
                for r in nao_notificadas
            ]

            # Telegram — uma por uma
            for vaga in vagas_obj:
                sucesso = await self.telegram.notificar_vaga(vaga)
                if sucesso:
                    self.repo.marcar_notificada(
                        next(r["id"] for r in nao_notificadas
                             if r["url"] == vaga.url)
                    )
                    notificadas += 1
                await asyncio.sleep(1)   # evita flood

            # E-mail — lote
            self.email.enviar_vagas(vagas_obj)

        # 6. Registro da execução
        fim = datetime.now(timezone.utc)
        duracao = (fim - inicio).seconds
        self.repo.registrar_execucao(
            inicio, fim,
            len(todas_vagas), len(vagas_novas), notificadas
        )

        logger.info(
            f"Ciclo concluído em {duracao}s — "
            f"coletadas={len(todas_vagas)} "
            f"novas={len(vagas_novas)} "
            f"notificadas={notificadas}"
        )

    async def resumo_diario(self):
        """Envia resumo das estatísticas do dia."""
        stats = self.repo.estatisticas()
        await self.telegram.notificar_resumo(stats)

    def iniciar(self):
        """Inicia o bot com agendamento."""
        logger.info("JobBot iniciado.")

        # Agendamento
        schedule.every(config.intervalo_minutos).minutes.do(
            lambda: asyncio.run(self.executar_ciclo())
        )
        schedule.every().day.at(config.horario_resumo).do(
            lambda: asyncio.run(self.resumo_diario())
        )

        # Executa imediatamente na primeira vez
        asyncio.run(self.executar_ciclo())

        while True:
            schedule.run_pending()
            time.sleep(30)


if __name__ == "__main__":
    import logging
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
    )
    bot = JobBot()
    bot.iniciar()

Na primeira versão, as vagas relidas do banco para notificação eram reconstruídas sem tecnologias e sem remoto, embora os dois estivessem gravados. Medido: a mensagem do Telegram saía com "N/I" no lugar das tecnologias de toda vaga. O json.loads desfaz o json.dumps do repositório.

Testes

# tests/conftest.py
import os

# O config exige credenciais; nos testes, valores falsos bastam.
# Precisa rodar antes de qualquer import de jobbot.
for var, valor in {
    "TELEGRAM_TOKEN": "teste", "TELEGRAM_CHAT_ID": "0",
    "EMAIL_USUARIO": "bot@exemplo.com", "EMAIL_SENHA": "teste",
    "EMAIL_DESTINO": "voce@exemplo.com",
}.items():
    os.environ.setdefault(var, valor)


# tests/test_filtro.py
import pytest
from jobbot.modelos import Vaga
from jobbot.processadores.filtro import pontuar_vaga, filtrar_vagas


@pytest.fixture
def vaga_python_remoto():
    return Vaga(
        titulo="Python Developer Senior",
        empresa="TechCorp",
        descricao="Vaga para desenvolvedor Python com FastAPI",
        url="https://exemplo.com/vaga/1",
        fonte="RemoteOK",
        remoto=True,
        salario_min=80000,
        tecnologias=["Python", "FastAPI", "Docker"]
    )


@pytest.fixture
def vaga_java_presencial():
    return Vaga(
        titulo="Java Developer",
        empresa="OldCorp",
        descricao="Vaga para desenvolvedor Java",
        url="https://exemplo.com/vaga/2",
        fonte="LinkedIn",
        remoto=False,
        tecnologias=["Java", "Spring"]
    )


class TestPontuacao:

    def test_vaga_python_tem_pontuacao_positiva(self, vaga_python_remoto):
        pontos = pontuar_vaga(vaga_python_remoto)
        assert pontos > 0

    def test_remoto_adiciona_pontos(self, vaga_python_remoto):
        vaga_python_remoto.remoto = False
        pontos_sem_remoto = pontuar_vaga(vaga_python_remoto)
        vaga_python_remoto.remoto = True
        pontos_com_remoto = pontuar_vaga(vaga_python_remoto)
        assert pontos_com_remoto > pontos_sem_remoto

    def test_estagio_penalizado(self):
        vaga = Vaga(
            titulo="Estágio Python",
            empresa="Empresa",
            descricao="Vaga de estágio",
            url="https://ex.com/3",
            fonte="test",
            tecnologias=["Python"]
        )
        pontos = pontuar_vaga(vaga)
        assert pontos == 0.0


class TestFiltro:

    def test_filtra_presencial_quando_remoto_obrigatorio(
        self, vaga_java_presencial
    ):
        resultado = filtrar_vagas([vaga_java_presencial])
        assert len(resultado) == 0

    def test_mantem_vaga_relevante(self, vaga_python_remoto):
        resultado = filtrar_vagas([vaga_python_remoto])
        assert len(resultado) == 1

    def test_ordena_por_pontuacao(self):
        vagas = [
            Vaga("Junior Python Dev", "A", "Python", "https://a.com/1",
                 "t", remoto=True, tecnologias=["Python"]),
            Vaga("Senior Python FastAPI Docker", "B", "Python FastAPI Docker",
                 "https://b.com/2", "t", remoto=True,
                 tecnologias=["Python", "FastAPI", "Docker"],
                 salario_min=90000),
        ]
        resultado = filtrar_vagas(vagas)
        assert len(resultado) == 2
        assert resultado[0].titulo.startswith("Senior")



# tests/test_deduplicador.py
import pytest
from unittest.mock import MagicMock
from jobbot.processadores.deduplicador import deduplicar
from jobbot.modelos import Vaga


def test_remove_vagas_ja_vistas():
    vaga1 = Vaga("Vaga 1", "Empresa", "desc", "https://ex.com/1", "t")
    vaga2 = Vaga("Vaga 2", "Empresa", "desc", "https://ex.com/2", "t")

    repo = MagicMock()
    repo.existe.side_effect = lambda id: id == vaga1.id

    resultado = deduplicar([vaga1, vaga2], repo)
    assert len(resultado) == 1
    assert resultado[0].url == vaga2.url


def test_retorna_tudo_quando_nada_visto():
    vagas = [
        Vaga("V1", "E", "d", "https://ex.com/1", "t"),
        Vaga("V2", "E", "d", "https://ex.com/2", "t"),
    ]
    repo = MagicMock()
    repo.existe.return_value = False

    resultado = deduplicar(vagas, repo)
    assert len(resultado) == 2

O conftest.py entrou porque, sem ele, nenhum teste rodava: o módulo de filtro importa o config, que exige o token do Telegram e as credenciais de e-mail, e o pytest parava na coleta com cinco erros de validação do Pydantic. Valores falsos no ambiente bastam para testar a lógica. O teste de ordenação também mudou: a primeira versão só conferia a ordem if len(resultado) >= 2, e passaria sem verificar nada se o filtro descartasse uma das vagas.

Docker

# Dockerfile
FROM python:3.12-slim
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 \
    PYTHONPATH=/app/src
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY src/ ./src/
# O .env NÃO entra na imagem: as credenciais chegam pelo compose
RUN adduser --disabled-password --gecos "" appuser \
    && mkdir -p /app/data /app/logs \
    && chown appuser /app/data /app/logs
USER appuser
CMD ["python", "-m", "jobbot.agendador"]

# docker-compose.yml
services:
  jobbot:
    build: .
    container_name: job-bot
    volumes:
      - ./data:/app/data
      - ./logs:/app/logs
    environment:
      - TELEGRAM_TOKEN=${TELEGRAM_TOKEN}
      - TELEGRAM_CHAT_ID=${TELEGRAM_CHAT_ID}
      - EMAIL_USUARIO=${EMAIL_USUARIO}
      - EMAIL_SENHA=${EMAIL_SENHA}
      - EMAIL_DESTINO=${EMAIL_DESTINO}
    restart: unless-stopped

A primeira versão do Dockerfile fazia COPY .env ., e isso grava o token do Telegram e a senha do e-mail dentro da imagem, em texto, visíveis para quem tiver acesso a ela ou ao registro onde for publicada. As credenciais já chegam pelo environment do compose, então a linha saiu. O PYTHONPATH=/app/src faz o python -m jobbot.agendador encontrar o pacote, e as pastas de dados e logs passam a pertencer ao usuário que roda o bot.

O bot junta num ciclo só as peças de automação da série: coletores que buscam vagas numa API e numa página, um filtro que pontua cada uma pelos critérios do usuário, um repositório SQLite que lembra o que já foi visto, notificadores para Telegram e e-mail e um agendador que repete tudo a cada hora. É o mesmo desenho em camadas dos projetos anteriores, como Projeto 2: Pipeline de Data Science do Zero ao Deploy, aplicado a um processo que roda sozinho.

Um bot que roda sozinho tem uma exigência a mais: quando algo falha, ninguém está olhando. Por isso os defeitos desta página pesam mais do que pareceriam num script. Um coletor que engole a falha de rede faz o bot "não encontrar vagas" por dias; um salário comparado na unidade errada deixa passar tudo; um título com & faz o Telegram recusar a mensagem e a vaga voltar a cada ciclo; um .env copiado para a imagem espalha a senha junto com ela. A regra que resume todos: falhas precisam chegar a alguém, dados de terceiros entram como texto, e credenciais nunca viajam dentro do artefato.

Fontes e leituras recomendadas

• python-telegram-bot — https://python-telegram-bot.org/
• schedule — agendamento simples — https://schedule.readthedocs.io/
• BeautifulSoup4 — https://www.crummy.com/software/BeautifulSoup/bs4/doc/
• RemoteOK API — https://remoteok.com/api
• SWEIGART, Al. Automate the Boring Stuff with Python. 2. ed. No Starch Press, 2019. — automação de tarefas do dia a dia com Python.

Exercícios

Exercício 1

O bot passou uma semana sem notificar nenhuma vaga, e ninguém percebeu: nenhuma mensagem de erro chegou ao Telegram. O log mostra, a cada hora, "Erro ao coletar RemoteOK: ..." seguido de "RemoteOK: 0 vagas coletadas". O agendador tem um try/except que avisa no Telegram quando um coletor falha. Por que o aviso nunca saiu?

Exercício 2

O usuário configurou salario_minimo = 5000, pensando em reais por mês, e reclama que o bot só manda vagas "sem salário informado". Consultada, a API do RemoteOK devolve, numa vaga, "salary_min": 70000, "salary_max": 80000. O coletor da primeira versão lia raw.get("salary", "") e procurava números no texto. Quantos problemas há aqui?

Exercício 3

Uma vaga com o título "Dev C++ & Python <Sênior>" é coletada, filtrada e salva, mas nunca chega ao Telegram. Pior: a cada hora o bot tenta de novo. O notificador envia com parse_mode="HTML" e monta o texto com f"🎯 <b>{vaga.titulo}</b>". Explique o ciclo e a correção.

Exercício 4

Na mensagem do Telegram, toda vaga aparece com "🛠️ N/I" no lugar das tecnologias, embora o coletor detecte "Python", "Docker" e outras, e o repositório as grave numa coluna tecnologias. Onde elas se perdem?

Exercício 5

Para facilitar o deploy, o Dockerfile da primeira versão tinha COPY .env .. A imagem foi publicada num registro para ser baixada no servidor. Que risco isso cria, se o compose já passa as mesmas variáveis em environment?