coletor-cotacao-cravil/App.py

#!/usr/bin/env python3

import requests
from bs4 import BeautifulSoup
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
import re
from datetime import datetime
import time


INFLUXDB_URL = "http://xx.xxx.xx.xxx:8086"
INFLUXDB_TOKEN = "XXXXXXXXXXXX"
INFLUXDB_ORG = "XXXXX"
INFLUXDB_BUCKET = "XXXXX"

CRAVIL_URL = "https://www.cravil.com.br/cotacoes/"

INTERVALO_COLETA = 300


def parse_valor(valor_str: str) -> float:
    if not valor_str:
        return 0.0
    valor_limpo = re.sub(r'[^\d,.]', '', valor_str)
    valor_limpo = valor_limpo.replace('.', '').replace(',', '.')
    try:
        return float(valor_limpo)
    except ValueError:
        return 0.0


def coletar_cotacoes() -> list[dict]:
    cotacoes = []
    
    try:
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }
        response = requests.get(CRAVIL_URL, headers=headers, timeout=30)
        response.raise_for_status()
        
        soup = BeautifulSoup(response.content, 'html.parser')
        
        # xpath: /html/body/main/section[2]/div/div/div/div[1]/ul/li  (lista com todos os valores)
        section = soup.select('main section')
        
        if len(section) >= 2:
            ul = section[1].find('ul')
            if ul:
                lis = ul.find_all('li')
                
                for li in lis:
                    try:
                        h3 = li.find('h3')
                        titulo = h3.get_text(strip=True) if h3 else None
                        
                        divs = li.find_all('div', recursive=False)
                        valor_str = None
                        
                        for div in divs:
                            inner_divs = div.find_all('div', recursive=False)
                            if len(inner_divs) >= 2:
                                valor_div = inner_divs[1].find_all('div', recursive=False)
                                if len(valor_div) >= 2:
                                    valor_str = valor_div[1].get_text(strip=True)
                                    break
                                elif inner_divs[1]:
                                    valor_str = inner_divs[1].get_text(strip=True)
                                    break
                        
                        if not valor_str:
                            for elem in li.find_all(string=re.compile(r'R?\$?\s*\d+[,\.]\d+')):
                                valor_str = elem.strip()
                                break
                        
                        if titulo and valor_str:
                            valor = parse_valor(valor_str)
                            cotacoes.append({
                                'titulo': titulo,
                                'valor': valor,
                                'valor_raw': valor_str
                            })
                            print(f"  ✓ {titulo}: {valor_str} -> {valor}")
                    
                    except Exception as e:
                        print(f"  ✗ Erro ao processar item: {e}")
                        continue

        if not cotacoes:
            print("Tentando método alternativo de extração...")
            items = soup.select('li')
            for li in items:
                h3 = li.find('h3')
                if h3:
                    titulo = h3.get_text(strip=True)
                    texto = li.get_text()
                    match = re.search(r'R?\$?\s*([\d.,]+)', texto)
                    if match:
                        valor_str = match.group(0)
                        valor = parse_valor(valor_str)
                        if valor > 0:
                            cotacoes.append({
                                'titulo': titulo,
                                'valor': valor,
                                'valor_raw': valor_str
                            })
                            print(f"  ✓ {titulo}: {valor_str} -> {valor}")
                            
    except requests.RequestException as e:
        print(f"Erro ao acessar {CRAVIL_URL}: {e}")
    
    return cotacoes


def enviar_influxdb(cotacoes: list[dict]):
    if not cotacoes:
        print("Nenhuma cotação para enviar.")
        return
    
    try:
        client = InfluxDBClient(
            url=INFLUXDB_URL,
            token=INFLUXDB_TOKEN,
            org=INFLUXDB_ORG
        )
        
        write_api = client.write_api(write_options=SYNCHRONOUS)
        
        for cotacao in cotacoes:
            produto_tag = cotacao['titulo'].lower().replace(' ', '_')
            produto_tag = re.sub(r'[^a-z0-9_]', '', produto_tag)
            
            point = (
                Point("cotacao_agricola")
                .tag("produto", produto_tag)
                .tag("produto_nome", cotacao['titulo'])
                .tag("fonte", "cravil")
                .field("valor_saco", cotacao['valor'])
                .field("valor_raw", cotacao['valor_raw'])
            )
            
            write_api.write(bucket=INFLUXDB_BUCKET, record=point)
            print(f"  → Enviado: {cotacao['titulo']}")
        
        client.close()
        print(f"\n✓ {len(cotacoes)} cotações enviadas para InfluxDB2")
        
    except Exception as e:
        print(f"Erro ao enviar para InfluxDB: {e}")


def main():
    print("=" * 50)
    print("COLETOR DE COTAÇÕES AGRÍCOLAS - CRAVIL")
    print(f"Intervalo de coleta: {INTERVALO_COLETA // 60} minutos")
    print("=" * 50)
    
    ciclo = 1
    while True:
        print(f"\n{'─' * 50}")
        print(f"📅 CICLO #{ciclo} - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print(f"{'─' * 50}")
        
        print(f"\n📥 Coletando cotações de {CRAVIL_URL}...")
        cotacoes = coletar_cotacoes()
        
        if cotacoes:
            print(f"\n📊 Total de {len(cotacoes)} cotações coletadas")
            print(f"\n📤 Enviando para InfluxDB2 ({INFLUXDB_URL})...")
            enviar_influxdb(cotacoes)
        else:
            print("\n⚠ Nenhuma cotação encontrada!")
        
        proxima_coleta = datetime.now().timestamp() + INTERVALO_COLETA
        proxima_hora = datetime.fromtimestamp(proxima_coleta).strftime('%H:%M:%S')
        print(f"\n⏰ Próxima coleta em {INTERVALO_COLETA // 60} minutos ({proxima_hora})")
        print("   (Ctrl+C para encerrar)")
        
        try:
            time.sleep(INTERVALO_COLETA)
        except KeyboardInterrupt:
            print("\n\n🛑 Coleta encerrada pelo usuário.")
            break
        
        ciclo += 1


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\n\n🛑 Programa encerrado.")