Artigo 50 — Projeto 3: Bot de Automação com Scraping e Notificações
Prof. Ricardo Matos Módulo 9 · Projetos Reais e Carreira · Artigo 50 de 52
Introdução
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. Aplicamos aqui os Módulos 6 (automação, scraping), 4 (agendamento, arquivos) e 5 (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
│ ├── coletores/
│ │ ├── __init__.py
│ │ ├── base.py ← classe base de coleta
│ │ ├── linkedin.py ← scraping LinkedIn
│ │ └── remoteok.py ← API RemoteOK
│ ├── processadores/
│ │ ├── __init__.py
│ │ ├── filtro.py ← filtragem e pontuação
│ │ └── deduplicador.py ← remoção de duplicatas
│ ├── notificadores/
│ │ ├── __init__.py
│ │ ├── telegram.py ← bot Telegram
│ │ ├── email.py ← notificações por e-mail
│ │ └── base.py ← interface comum
│ ├── armazenamento/
│ │ ├── __init__.py
│ │ └── repositorio.py ← SQLite para vagas
│ └── agendador.py ← schedule + execução
├── tests/
│ ├── conftest.py
│ ├── test_filtro.py
│ ├── test_deduplicador.py
│ └── test_notificadores.py
├── data/
│ └── vagas.db
├── logs/
├── Dockerfile
├── docker-compose.yml
├── .env.example
└── pyproject.toml
Configuração
# src/jobbot/config.py
from pydantic_settings import BaseSettings
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"]
salario_minimo: float = 5000.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
class Config:
env_file = ".env"
@lru_cache
def get_config() -> Configuracoes:
return Configuracoes()
Domínio: Vaga de Emprego
# src/jobbot/modelos.py
from dataclasses import dataclass, field
from datetime import datetime
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=datetime.utcnow)
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"R$ {self.salario_min:,.0f} – {self.salario_max:,.0f}"
if self.salario_min:
return f"A partir de R$ {self.salario_min:,.0f}"
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
import re
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()
try:
logger.info(f"Coletando vagas do RemoteOK...")
resposta = session.get(self.API_URL, timeout=15)
resposta.raise_for_status()
dados = resposta.json()
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}")
except Exception as e:
logger.error(f"Erro ao coletar RemoteOK: {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
# Extrai salário
salario_min = salario_max = None
salario_str = raw.get("salary", "") or ""
if salario_str:
numeros = re.findall(r"\d+", salario_str.replace(",", ""))
if len(numeros) >= 2:
salario_min = float(numeros[0])
salario_max = float(numeros[1])
elif len(numeros) == 1:
salario_min = float(numeros[0])
# 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"
)
# 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()
)
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 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:
techs = ", ".join(vaga.tecnologias[:5]) if vaga.tecnologias else "N/I"
return (
f"🎯 <b>{vaga.titulo}</b>\n"
f"🏢 {vaga.empresa}\n"
f"📍 {vaga.localidade}\n"
f"💰 {vaga.salario_display}\n"
f"🛠️ {techs}\n"
f"⭐ Score: {vaga.pontuacao:.0f}\n"
f"🔗 <a href='{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 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
template = Template(TEMPLATE_EMAIL)
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
Agendador: Orquestrando Tudo
# src/jobbot/agendador.py
import asyncio
import schedule
import time
import logging
from datetime import datetime
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.utcnow()
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
from datetime import datetime as dt
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"],
)
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.utcnow()
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()
Testes
# 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=8000,
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=12000),
]
resultado = filtrar_vagas(vagas)
if len(resultado) >= 2:
assert resultado[0].pontuacao >= resultado[1].pontuacao
# 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
Docker
# Dockerfile
FROM python:3.12-slim
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY src/ ./src/
COPY .env .
RUN adduser --disabled-password appuser
USER appuser
CMD ["python", "-m", "jobbot.agendador"]
# docker-compose.yml
version: "3.9"
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
Resumo do Projeto
| Componente | Técnicas |
|---|---|
| Coletores | requests, BeautifulSoup, API REST |
| Filtro | scoring, regras de negócio |
| Deduplicação | hash MD5 de conteúdo |
| Armazenamento | SQLite, RepositorioVagas |
| Notificação | Telegram Bot API, SMTP + Jinja2 |
| Agendamento | schedule, asyncio |
| Testes | pytest, MagicMock, fixtures |
| Deploy | Docker, variáveis de ambiente |
Referências e Leituras Complementares
- 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.
Prof. Ricardo Matos — Dominando o Python em 1 Ano · Artigo 50 de 52 Próximo: Artigo 51 — Projeto 4: Dashboard Analítico com Dash e Banco de Dados Real
you asked
Sim