API tools faq
paste
matheus__serpa

MatheusSerpaRaspagem

matheus__serpa
May 30th, 2023
1,057
0
Never
Add comment
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
Python 4.54 KB | Source Code | 0 0
raw download clone embed print report
  1. # This script was written by Matheus S. Serpa
  2. # Contact: [email protected]
  3.  
  4. import pandas as pd
  5. import numpy as np
  6. import requests
  7.  
  8. from bs4 import BeautifulSoup
  9. from datetime import datetime
  10.  
  11. import logging
  12.  
  13. logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(message)s")
  14. logger = logging.getLogger(__name__)
  15.  
  16. # Raspagem de dados de cotações de commodities agrícolas como soja, milho, café, etc.
  17. BASE_URL = "https://www.noticiasagricolas.com.br/cotacoes/"
  18. CROPS_LINK = {
  19.     "Soja": "soja/soja-mercado-fisico-sindicatos-e-cooperativas",
  20.     "Trigo": "trigo/trigo-mercado-fisico",
  21.     "Milho": "milho/milho-mercado-fisico-sindicatos-e-cooperativas",
  22. }
  23. TABLE_COLUMNS = 3
  24.  
  25.  
  26. def extract_data(base_url: str, links: dict, columns: int) -> pd.DataFrame:
  27.     """
  28.    Extracts commodity prices data from noticiasagricolas.com.br
  29.  
  30.    Args:
  31.        base_url (str): Base URL of the website
  32.        links (dict): Dictionary of commodities links
  33.        columns (int): Number of columns in the prices table
  34.  
  35.    Returns:
  36.        pd.DataFrame: Dataframe containing the places and prices data
  37.    """
  38.     df = pd.DataFrame()
  39.  
  40.     for crop, link in links.items():
  41.         url = base_url + link
  42.  
  43.         # Acessa a página e obtém todo conteúdo HTML
  44.         try:
  45.             response = requests.get(url)
  46.             response.raise_for_status()
  47.             html_page = response.text
  48.         except requests.exceptions.RequestException:
  49.             logging.exception(f"Não foi possível acessar a página: {url}")
  50.             continue
  51.  
  52.         # Realiza o parser e busca pela tabela de cotações
  53.         try:
  54.             soup = BeautifulSoup(html_page, "html.parser")
  55.             table_data = soup.find("table", {"class": "cot-fisicas"}).find_all(
  56.                 "td"
  57.             )
  58.         except Exception:
  59.             logging.exception(
  60.                 "Não foi possível encontrar a tabela de cotações"
  61.             )
  62.             continue
  63.  
  64.         # Obtém o nome de todas praças / locais
  65.         places = []
  66.         PLACE_COLUMN = 0
  67.         for element in range(PLACE_COLUMN, len(table_data), columns):
  68.             places.append(table_data[element].text)
  69.  
  70.         # Obtém o preço / quotações de todas praças / locais
  71.         prices = []
  72.         PRICE_COLUMN = 1
  73.         for element in range(PRICE_COLUMN, len(table_data), columns):
  74.             prices.append(table_data[element].text)
  75.  
  76.         # Cria um dataframe com os dados obtidos
  77.         data = {"place": places, "price": prices}
  78.  
  79.         df_crop = pd.DataFrame(data)
  80.  
  81.         # Ajusta o formato da moeda e do nome da cultura
  82.         df_crop["crop"] = crop
  83.         df_crop["currency"] = "BRL"
  84.  
  85.         # Adiciona o dataframe da cultura atual ao dataframe geral
  86.         df = pd.concat([df, df_crop])
  87.  
  88.     return df
  89.  
  90.  
  91. # Limpeza dos dados para adequar ao formato brasileiro e também ajustar tipos dos dados
  92. def data_cleaning(df: pd.DataFrame) -> pd.DataFrame:
  93.     """
  94.    Clean data to adjust to Brazilian format and also adjust data types.
  95.  
  96.    Args:
  97.        df: The DataFrame containing the prices and places.
  98.  
  99.    Returns:
  100.        pd.DataFrame: Cleaned DataFrame.
  101.    """
  102.     # Ajusta o formato do preço
  103.     df["price"] = df["price"].str.replace(",", ".")
  104.     df["price"] = df["price"].apply(
  105.         lambda x: np.nan if x == "s/ cotação" else x
  106.     )
  107.     df["price"] = df["price"].astype(float)
  108.  
  109.     # Remove linhas com valores nulos
  110.     df = df.dropna().reset_index(drop=True)
  111.  
  112.     return df
  113.  
  114.  
  115. def save_data(df: pd.DataFrame, now_str: str) -> bool:
  116.     """
  117.    Save the data to a CSV file.
  118.  
  119.    Args:
  120.        df: The DataFrame containing the prices and places.
  121.        now_str: The current date and time string in the format 'YYYY-MM-DD_HH-MM-SS'.
  122.  
  123.    Returns:
  124.        A boolean value indicating whether the save operation was successful.
  125.    """
  126.     logger.info("Salvando dados...")
  127.     logger.info(f"Shape: {df.shape}")
  128.     logger.info(f"Arquivo: cotacoes_{now_str}.csv")
  129.  
  130.     if df.empty:
  131.         logger.warning("Não há dados a serem salvos")
  132.         return False
  133.     df.to_csv(f"cotacoes_{now_str}.csv", index=False)
  134.     return True
  135.  
  136.  
  137. def main():
  138.     NOW = datetime.now()
  139.     NOW_STR = NOW.strftime("%Y-%m-%d_%H-%M-%S")
  140.  
  141.     logger.info("Iniciando raspagem de dados...")
  142.  
  143.     df = extract_data(BASE_URL, CROPS_LINK, TABLE_COLUMNS)
  144.  
  145.     df = data_cleaning(df)
  146.  
  147.     save_data(df, NOW_STR)
  148.  
  149.     NOW = datetime.now()
  150.     logger.info("Raspagem de dados finalizada...")
  151.  
  152.  
  153. if __name__ == "__main__":
  154.     main()
  155.  
Advertisement
Add Comment
Please, Sign In to add comment
Public Pastes
We use cookies for various purposes including analytics. By continuing to use Pastebin, you agree to our use of cookies as described in the Cookies Policy.  OK, I Understand
Not a member of Pastebin yet?
Sign Up, it unlocks many cool features!