screenshot_oo.py

import os
import time
import glob
import shutil
import datetime
import requests
from google.cloud import storage
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support.ui import WebDriverWait
from selenium.common.exceptions import NoSuchElementException
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.keys import Keys


# Classe que faz download do pdf no site do PowerBI
class Screenshot:
    def __init__(self):
        self.service_account_json = r"" # Caminho do arquivo JSON de credenciais
        self.chrome_options = Options()
        self.chrome_options.add_argument("--start-maximized") # Parametro para iniciar o navegador maximizado
        self.service = Service("") # Caminho do chromedriver
        self.driver = webdriver.Chrome(service=self.service, options=self.chrome_options) # Inicia o navegador
        self.url = '' # URL do dashboard que será baixado em PDF

    # Método que baixa o pdf do dashboard
    def tirar_screenshot(self):
        os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = self.service_account_json
        self.driver.get(self.url)

        WebDriverWait(self.driver, 20).until(EC.presence_of_element_located((By.ID, "email")))
        element = WebDriverWait(self.driver, 20).until(EC.element_to_be_clickable((By.ID, "email")))
        element.send_keys('') # Email do usuário do PowerBI
        element.send_keys(Keys.RETURN)

        WebDriverWait(self.driver, 20).until(EC.presence_of_element_located((By.ID, "i0118")))
        password_input = self.driver.find_element(By.ID, "i0118")
        password_input.send_keys("") # Senha do usuário do PowerBI
        time.sleep(2)

        try:
            error_element = self.driver.find_element(By.ID, "passwordError")
            if error_element.is_displayed():
                print('Senha incorreta. Aplicação encerrada.')
                self.driver.quit()
                exit()
        except NoSuchElementException:
            pass

        self.driver.find_element(By.ID, "idSIButton9").click()
        time.sleep(3)

        self.driver.find_element(By.ID, "idSIButton9").click()
        time.sleep(30)

        self.driver.find_element(By.ID, "exportMenuBtn").click()
        time.sleep(2)

        botao_exportar = self.driver.find_element(By.CSS_SELECTOR, '[data-testid="export-to-pdf-btn"]')
        botao_exportar.click()
        time.sleep(3)

        self.driver.find_element(By.ID, "okButton").click()
        time.sleep(90)
        self.driver.quit()

        caminho_downloads = os.path.join(os.path.expanduser('~'), 'Downloads') # Caminho do diretório de downloads
        arquivos = glob.glob(os.path.join(caminho_downloads, "**")) # Lista todos os arquivos com um padrão do diretório

        if arquivos: 
            arquivo_recente = max(arquivos, key=os.path.getmtime)
            caminho_destino = os.getcwd()

            for arquivo in os.listdir(caminho_destino):
                if arquivo.endswith('.pdf'):
                    os.remove(os.path.join(caminho_destino, arquivo))
                    
            shutil.copy(arquivo_recente, caminho_destino)
            print(f'Arquivo {arquivo_recente} copiado para {caminho_destino}')
        else:
            print('Nenhum arquivo encontrado com o padrão "Moscatelli"')

        self.envia_bucket(nome_bucket , caminho_arquivo, destino_arquivo_bucket) # Envia o arquivo para o bucket
        print('------------------------- Arquivo enviado para o Bucket! -------------------------')

    # Método que envia o arquivo para o bucket
    def envia_bucket(self, nome_bucket, caminho_arquivo, destino_arquivo):
        storage_client = storage.Client()
        bucket = storage_client.bucket(nome_bucket)
        blob = bucket.blob(destino_arquivo)
        blob.upload_from_filename(caminho_arquivo)
        print(f"File {caminho_arquivo} uploaded to {destino_arquivo}.")

# Classe que envia o PDF para o WhatsApp
class EnviaPDF:
    def __init__(self):
        self.service_account_json = '' # Caminho do arquivo JSON de credenciais
        self.nome_bucket = '' # Nome do bucket
        self.nome_pasta_pdf = '' # Nome da pasta onde o PDF está salvo
        self.storage_client = storage.Client.from_service_account_json(self.service_account_json)
        self.bucket = self.storage_client.bucket(self.nome_bucket)

    # Método que gera a URL do PDF
    def gerar_url_pdf(self):
        pdf_blob = self.bucket.blob(self.nome_pasta_pdf)
        pdf_url_assinado = pdf_blob.generate_signed_url(
            version="v4",
            expiration=datetime.timedelta(hours=24),
            method="GET"
        )
        return pdf_url_assinado

    # Método que envia o PDF para o WhatsApp
    def enviar_whatsapp(self, pdf_url_assinado):
        parte_url = pdf_url_assinado.split('.com/')[1] # Pega a parte da URL após o ".com/" de onde o pdf está salvo
        url_auth = '' # URL do token de autenticação
        headers = {
            "Content-Type": "", # Tipo de conteúdo
            "Authorization": "" # Token de autorização
        }
        data = {
            "grant_type": "", # Tipo de concessão
            "username": "", # Usuário
            "password": "" # Senha do Usuário
        }
        response = requests.post(url_auth, headers=headers, data=data)
        response_json = response.json()
        access_token = response_json.get('access_token')

        url_api = '' # URL da API
        headers = {
            'Content-Type': '', # Tipo de conteúdo
            'from-app': '', # De onde está vindo a requisição
            'category': '', # Categoria
            'Authorization': f'{access_token}', # Token de autorização
            'Cookie': '' # Cookie
        }
        whats_numeros_destino = [] # Números de telefone para enviar o PDF
        for numero in whats_numeros_destino:
            data = {
                "namespace": "", # Namespace
                "mediaId": "", # ID da mídia
                "phone": numero,
                "name": "", # Nome do arquivo
                "link": f"{parte_url}"
            }

            # Envia o PDF para o WhatsApp
            response = requests.post(url_api, headers=headers, json=data)
            print(response.status_code)
            print(response.json())

# Classe que limpa o diretório
class Limpeza:
    def __init__(self):
        self.diretorio = "" # Diretório onde os arquivos serão removidos

    def limpar_diretorio(self):
        for root, dirs, files in os.walk(self.diretorio):
            for file in files:
                if file.endswith(".pdf"): # Remove todos os arquivos com a extensão .pdf
                    file_path = os.path.join(root, file)
                    os.remove(file_path)
                    print(f"Arquivo removido: {file_path}")

        arquivo_especifico = "" # Caminho do arquivo específico que será removido
        if os.path.exists(arquivo_especifico):
            os.remove(arquivo_especifico)
            print(f"Arquivo removido: {arquivo_especifico}")

# Classe principal que executa o robô
class Robo:
    def __init__(self):
        self.screenshot = Screenshot()
        self.envia_pdf = EnviaPDF()
        self.limpeza = Limpeza()

    # Método que executa o robô
    def executar(self):
        self.screenshot.tirar_screenshot()
        pdf_url_assinado = self.envia_pdf.gerar_url_pdf()
        self.envia_pdf.enviar_whatsapp(pdf_url_assinado)
        self.limpeza.limpar_diretorio()
        print("-------------- Limpeza concluida. --------------")

if __name__ == "__main__":
    robo = Robo()
    robo.executar()