Engineering custom email alerts for reddit em 3 passos

Criando alertas de e-mail em tempo real no python com as APIs do Reddit e Gmail.

O problema: sobrecarga de notificações

Se você deseja transformar seu telefone em um daqueles pagers vibratórios de restaurante, eu recomendaria uma câmera Ring Doorbell? A sensibilidade dos sensores e o tráfego pedonal constante significam que meu telefone inevitavelmente vibra, soa ou, durante a temporada de Halloween, toca “The Monster Mash” várias vezes por hora. Uma das outras “vantagens” é o acesso a uma mente coletiva de seus vizinhos mais preocupados (por meio da comunidade Neighbors).

*Não é um endosse pago; mal é uma recomendação

Inevitavelmente, não importa qual tipo de comunidade você viva, você será atraído para um tópico sobre alguém pensando que ouviu tiros. Agora, onde eu moro, há uma explicação perfeitamente razoável para isso. Eu moro a menos de 10 milhas do Disney World e do Universal Studios Orlando. Cada propriedade tem seu próprio show noturno com fogos de artifício que fazem um som de “pop pop” que se assemelha estranhamente a tiros.

Assim como com tiros alucinatórios, os aplicativos que instalamos e nos encontramos quase que atados criam muito ruído na forma de notificações push, e-mail ou SMS. Pior ainda, o conteúdo dessas notificações não é ponderado quanto à magnitude. Tudo é “Urgente” ou “BOMBA”. Mais irritante do que anúncios indiscriminados, no entanto, é a falta de personalização sobre quais alertas eu, como usuário, posso receber.

Como usuário de iOS (e um pequeno acionista minoritário da Apple), minhas opções são permitir todas as notificações ou silenciar todas as notificações.

Mas recentemente, encontrei um caso de uso e um método para criar alertas personalizados a partir de um dos meus notifiers mais barulhentos: o Reddit.

Infelizmente, meu caso de uso não é particularmente nobre.

Não estou, por exemplo, trabalhando para criar um aplicativo que filtre a enxurrada de manchetes mal rotuladas ou desinformadas que inundam o site (talvez um dia!)

Estou simplesmente tentando comprar um relógio.

Antes de elaborar sobre meu caso de uso, quero delinear minha abordagem.

Inicialmente, eu queria ser alertado sempre que houvesse uma nova postagem no subreddit que estou mirando: r/WatchExchange.

Mas, acontece que há muitos usuários postando constantemente, o que cria um ruído que ofusca o modelo específico que estou perseguindo (um Omega Seamaster neo-vintage ref. 2531.80; um primeiro presente de Dia dos Pais para mim mesmo).

Eu poderia, no entanto, ajustar o que vejo usando a API do Reddit.

E, em vez de entupir meu telefone com notificações, posso ser preciso sobre quais alertas estou enviando ao criar um filtro para acionar e-mails automáticos via API do Gmail.

Essencialmente, o processo pode ser reduzido a 3 passos –

• Autenticação/Recuperação de Dados
• Processamento de Dados/Geração de Filtro
• Notificação por E-mail (API do Gmail)

1. Autenticação e Recuperação de Dados

Esse processo, menos o e-mail, é semelhante a um script de produção que escrevi para fazer parsing de metadados para encontrar VMs com falha.

Portanto, mesmo que você não ache que usará esse método para raspar o Reddit, por assim dizer, pode usar isso como um ponto de partida para criar e escalar a infraestrutura de alertas em geral.

Coincidentemente, já falei sobre Reddit e relógios antes.

Como as APIs vão, o Reddit geralmente facilita a extração de uma pequena quantidade de dados. Apenas tenha cuidado ao tentar desenvolver um aplicativo de terceiros sem vários milhões em financiamento para pagar por suas solicitações.

Este é um fluxo de autenticação básico consistindo em–

• Criar um aplicativo que, por sua vez, gerará uma chave e token de API
• “Fazer login” armazenando seu usuário e senha em um payload de dados
• Passar um cabeçalho e seu payload de dados de usuário/senha em uma solicitação POST

import requests
import config as cfg

def get_reddit_token():

    auth = requests.auth.HTTPBasicAuth('***********', '************')
    data = {
    'grant_type': 'client_credentials',
    'username': cfg.user,
    'password': cfg.password
    }
    headers = {'User-Agent': 'News/0.0.1'}
    request = requests.post(cfg.base_access_url, auth=auth, data=data, headers=headers)
    token = request.json()['access_token']
    headers = {**headers, **{'Authorization': f"bearer {token}"}}
    
    return headers

Uma vez autenticado, você tem acesso a um monte de endpoints que permitem fazer qualquer coisa, desde extrair texto de postagens até automatizar postagens e respostas.

A solicitação GET é igualmente intuitiva. Uma coisa que vou ressaltar aqui é considerar colocar um “limite” no volume de dados retornados.

import requests

def make_request(url: str):
    
    headers = get_reddit_token()

    request = requests.get(url, headers=headers, params={'limit': '100'})
    
    return request

Como mencionei de forma não muito sutil, o Reddit tem limitado a taxa para desenvolvedores que tentam fazer solicitações excessivas; ele até introduziu uma camada de preços proibitiva para desencorajar aplicativos de terceiros.

Mas se você está apenas tentando raspar algumas postagens, isso não é uma preocupação.

2. Processamento de Dados com Pandas

O Reddit retorna dados JSON em forma de dicionário. Para extrair atributos relevantes, precisamos acessar os valores dentro das chaves “data” e “children”.

Isso me deixa com esses campos:

• Título
• Proporção de upvotes
• Pontuação
• Ups (upvotes)
• Domínio
• Número de Comentários
• Link (permalink)

Como quero usar Pandas >= 2.0, preciso criar um dicionário que vou anexar a uma lista para fácil conversão para um dataframe.

def format_df(end_point):
    
    watch_list = []
    
    for post in end_point.json()['data']['children']:
        watch_list.append({
        'title': post['data']['title'],
        'upvote_ratio': post['data']['upvote_ratio'],
        'score': post['data']['score'],
        'ups': post['data']['ups'],
        'domain': post['data']['domain'],
        'num_comments': post['data']['num_comments'],
        'link': post['data']['permalink']
    })
        
    df = pd.DataFrame(watch_list)
        
    return df

Ao retornar o dataframe, você pode ver uma versão “limpa” tabular das postagens que posso ver no meu aplicativo Reddit.

A partir do “título”, vamos criar 2 novos campos

• Modelo
• Preço

Depois de examinar as postagens, percebi que o modelo quase sempre segue um tag anterior de [WTS] ou seja, Watch To Sell.

Então podemos dividir o texto após o “]” final para obter o modelo.

Para o preço, podemos extrair valores seguindo o “$”.

Refinei meu regex com a ajuda de um agente de IA. Eu sugeriria que você fizesse o mesmo se quiser economizar alguma frustração.

def extract_model(title):
    if "[WTS]" in title:
        parts = title.split("[WTS]")
        if len(parts) > 1:
            model_part = parts[-1].strip()
            if "[" in model_part:
                model = model_part.split("[")[0].strip()
            else:
                model = model_part
            return model
    elif "[WTT]" in title:
        parts = title.split("[WTT]")
        if len(parts) > 1:
            model_part = parts[-1].strip()
            if "(" in model_part:
                model = model_part.split("(")[0].strip()
            else:
                model = model_part
            return model
    return None

# Função para extrair o preço
def extract_price(title):
    price_match = re.search(r'\$\d+(?:\.\d+)?', title)
    if price_match:
        return price_match.group(0)
    return None

def filt(url: str):
    
    watch = make_request(url)
    
    watch_df = format_df(watch)
    
    watch_df['model'] = watch_df['title'].apply(extract_model)
    watch_df['price'] = watch_df['title'].apply(extract_price)
    
    watch_df = watch_df[["title", "model", "price", "link"]]
    
    return watch_df

Retornando o dataframe novamente, temos uma saída mais organizada e granular isolando modelo e preço.

Usaremos isso para construir um filtro para determinar se o modelo está nas últimas 100 postagens.

omega_filter = watch_df["model"].str.contains("Omega Seamaster 2531.80", na=False)

omega_df = watch_df[omega_filter]

E quando for incluído, meu inbox será bombardeado.

3. Notificação por E-mail com API do Gmail

A parte do e-mail pode ser um pouco complicada, então vou dividi-la.

Primeiro, quero esclarecer a autenticação. Normalmente, quando você acessa uma API do Google, você faria autenticação usando credenciais associadas à sua conta de serviço. Mas a API do Gmail requer a criação de um aplicativo baseado na web que, por sua vez, terá credenciais diferentes.

Você pode, na verdade, criar essas credenciais programaticamente e verificá-las em um fluxo OAuth2 para autenticação MFA de uma única vez.

Solicitações subsequentes referenciariam o arquivo armazenado. Você pode (e deve) também criar lógica para usar um token de atualização se o token existente expirar.

creds = None
if os.path.exists('token.json'):
  creds = Credentials.from_authorized_user_file('token.json', ['https://www.googleapis.com/auth/gmail.send']if not creds or not creds.valid:
  if creds and creds.expired and creds.refresh_token:
      creds.refresh(Request())
  else:
      flow = InstalledAppFlow.from_client_secrets_file('credentials.json', ['https://www.googleapis.com/auth/gmail.send']) # Substitua pelo caminho do seu arquivo de credenciais
      creds = flow.run_local_server(port=0)
  with open('token.json', 'w') as token:
      token.write(creds.to_json())

Criar o corpo do e-mail requer a criação de um serviço baseado em descoberta. Este é o mesmo método usado ao tentar acessar aplicações como Cloud Storage, Drive ou Sheets.

Enviar uma mensagem requer:

• Para
• Assunto
• Mensagem

Eu criei um assunto: “Alerta: Relógio(s) Omega Desejado(s) Encontrado(s)!”

try:
  service = build('gmail', 'v1', credentials=creds)
  message = MIMEText(body)
  message['to'] = recipient_email
  message['subject'] = subject
  raw_message = base64.urlsafe_b64encode(message.as_bytes()).decode()
  send_message = {'raw': raw_message}
  message = (service.users().messages().send(userId="me", body=send_message).execute())
  print(f"E-mail enviado para {recipient_email} com ID da mensagem: {message['id']}")
except Exception as error:
    print(f'Ocorreu um erro: {error}')

E adicione as postagem e o link como variáveis f-string.

“Os seguintes relógios Omega que correspondem aos seus critérios foram encontrados:

{watch_list}

Confira-os – https://oauth.reddit.com/{link}!

if not target_watch.empty:
  watch_list = "\n".join(target_watch['title'].tolist())
  link = "\n".join(target_watch["link"].tolist())
  email_subject = "Alerta: Relógio(s) Omega Desejado(s) Encontrado(s)!"
  email_body = f"Os seguintes relógios Omega que correspondem aos seus critérios foram encontrados:

{watch_list}

Confira-os - https://oauth.reddit.com/{link}!"
else:
   print("Nenhum relógio Omega correspondente encontrado.")

Então, quando meu filtro encontra uma correspondência, a lógica if/else garante que uma mensagem seja enviada para minha caixa de entrada, para que eu possa inspecionar instantaneamente uma listagem.

Embora eu tenha aplicado essa abordagem a um hobby inócuo (e nada relevante ao negócio), essa é uma forma “divertida” de demonstrar como os sistemas de alerta funcionam em uma escala menor.

Básicamente, criamos um filtro para detectar anomalias (ou relógios desejados) dentro dos dados de origem de forma recorrente (horária, a cada 15 minutos etc.) quando um erro ou fonte de dados atende à condição especificada pelo filtro, enviamos um alerta para qualquer um na “necessidade de saber”.

Com o suficiente de ajustes, ao contrário do debate sobre fogos de artifício/tiro, não devemos ter muitos falsos alarmes.