Coleta de Dados de Ações da Bolsa de Valores: Extração Dinâmica via Ajax e Armazenamento Estruturado

Implementação e Código Principal

Esta seção aborda a coleta automatizada de dados de ações das listas de Shanghai e Shenzhen (A-shares) do portal East Money. Os principais desafios incluem a captura de dados carregados dinamicamente via Ajax, a evasão de mecanismos anti-scraping, o processamento em lote de múltiplos setores e o armazenamento estruturado. A arquitetura adota uma abordagem orientada a objetos, dividindo o scraper em módulos para inicialização do navegador e banco de dados, parsing de dados, persistência e paginação.

Configuração do Navegador e Mecanismos de Espera

Para contornar a detecção de automação, o navegador é configurado com opções personalizadas, incluindo a alteração do User-Agent e a desativação de extensões de automação. A propriedade navigator.webdriver é redefinida via Chrome DevTools Protocol. Esperas implícitas e explícitas garantem que elementos dinâmicos sejam carregados antes da interação.

class ColetorDadosBolsa:
    def __init__(self):
        # Configuração avançada do navegador para evitar bloqueios
        opcoes_navegador = webdriver.ChromeOptions()
        opcoes_navegador.add_argument('--headless')  # Modo headless para ambientes sem interface
        opcoes_navegador.add_argument('--disable-gpu')
        opcoes_navegador.add_experimental_option("excludeSwitches", ["enable-automation"])
        opcoes_navegador.add_argument('user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36')

        self.navegador = webdriver.Chrome(options=opcoes_navegador)
        self.navegador.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument", {
            "source": "Object.defineProperty(navigator, 'webdriver', {get: () => false})"
        })
        
        # Configuração de tempos de espera
        self.navegador.implicitly_wait(15)
        self.espera = WebDriverWait(self.navegador, 25)

        # Inicialização do banco de dados e mapeamento de setores
        self.conexao_db = self._conectar_banco()
        self.cursor_db = self.conexao_db.cursor()
        self.setores_mapeados = {
            "Bolsa_Completa": "hs_a_board",
            "Shanghai_A": "sh_a_board",
            "Shenzhen_A": "sz_a_board"
        }

Conexão com Banco de Dados e Definição do Esquema

A conexão com o MySQL é estabelecida utilizando variáveis de ambiente para credenciais. A tabela de ações é criada com campos que correspondem aos dados extraídos, incluindo chaves únicas para evitar duplicidades.

def _conectar_banco(self):
    conexao = pymysql.connect(
        host=os.environ.get('DB_HOST', 'localhost'),
        port=int(os.environ.get('DB_PORT', 3306)),
        user=os.environ.get('DB_USER'),
        password=os.environ.get('DB_PASS'),
        database=os.environ.get('DB_NAME', 'market_data'),
        charset='utf8mb4'
    )
    return conexao

def _criar_tabela_acoes(self):
    sql_criacao = """
    CREATE TABLE IF NOT EXISTS info_acoes (
        registro_id INT AUTO_INCREMENT PRIMARY KEY,
        codigo_acao VARCHAR(15) NOT NULL,
        nome_acao VARCHAR(60),
        preco_atual DECIMAL(12,2),
        variacao_percentual VARCHAR(15),
        volume_negociado VARCHAR(40),
        setor VARCHAR(50) NOT NULL,
        data_coleta TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
        UNIQUE KEY idx_acao_setor (codigo_acao, setor)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
    """
    self.cursor_db.execute(sql_criacao)
    self.conexao_db.commit()

Lógica de Navegação e Extração de Dados

A coleta por setor envolve navegação para a página específica, localização da tabela de ações por cabeçalhos (como "Código" e "Nome"), e processamento de linhas. A paginação é tratada identificando o botão "Próxima página" dentro de um contêiner específico, monitorando seu estado de desativação para encerrar o loop.

def coletar_dados_por_setor(self, nome_setor, parametro_setor):
    url_alvo = f"http://quote.eastmoney.com/center/gridlist.html#{parametro_setor}"
    self.navegador.get(url_alvo)
    self.navegador.execute_script("window.scrollTo(0, 500);")
    time.sleep(2)

    try:
        tabela = self.espera.until(
            EC.presence_of_element_located((By.XPATH, "//table[.//th[contains(text(),'Código')] and .//th[contains(text(),'Nome')]]"))
        )
        self.espera.until(EC.presence_of_all_elements_located((By.XPATH, "//table/tbody/tr")))
        
        paginas_processadas = 0
        while True:
            linhas = tabela.find_elements(By.XPATH, ".//tbody/tr")
            dados_coletados = 0
            for linha in linhas:
                dados_acao = self._extrair_dados_linha(linha, nome_setor)
                if dados_acao and self._salvar_no_banco(dados_acao):
                    dados_coletados += 1
            
            # Lógica de paginação
            try:
                container_paginas = self.espera.until(
                    EC.presence_of_element_located((By.CSS_SELECTOR, "div.paginate-container"))
                )
                botao_proximo = container_paginas.find_element(By.XPATH, ".//a[contains(text(),'Próximo')]")
                if "disabled" in botao_proximo.get_attribute("class"):
                    break
                botao_proximo.click()
                time.sleep(3)
                paginas_processadas += 1
            except (TimeoutException, NoSuchElementException):
                break
    except TimeoutException:
        print(f"Falha ao carregar tabela para {nome_setor}")

Coleta de Cursos da Plataforma MOOC: Login Simulado e Extração Multidimensional

Abordagem e Implementação

Este módulo foca na extração de informações de cursos da plataforma China MOOC. Os desafios incluem o tratamento de formulários de login dentro de iframes, a alternância estável entre abas do navegador, a aálise de cards de cursos dinamicamente carregados e o armazenamento em banco de dados relacional.

Configuração do Ambiente e Automação do Login

O scraper é projetado para lidar com a detecção de automação, utilizando múltiplos XPaths para localizar elementos de login, e um mecanismo robusto para alternar para iframes que contêm formulários de autenticação.

class ColetorCursosMooc:
    def __init__(self):
        self.opcoes = webdriver.ChromeOptions()
        self.opcoes.add_argument('--disable-popup-blocking')
        self.opcoes.add_argument('--window-size=1280,720')
        # Adicionar outras configurações de anti-detecção similares...
        
        self.driver = webdriver.Chrome(options=self.opcoes)
        self.espera_geral = WebDriverWait(self.driver, 30)
        self.localizadores_login = {
            'iframe': (By.XPATH, "//iframe[contains(@src,'login')]"),
            'campo_usuario': (By.CSS_SELECTOR, "input[type='text']"),
            'campo_senha': (By.CSS_SELECTOR, "input[type='password']"),
            'botao_login': (By.CSS_SELECTOR, "button.login-btn")
        }

Estrutura do Banco de Dados para Cursos

Uma tabela dedicada aramzena detalhes dos cursos, como nome, instituição, instrutor, contagem de participantes e descrição.

def inicializar_banco_cursos(self):
    conexao = pymysql.connect(**parametros_conexao)
    cursor = conexao.cursor()
    sql = """
    CREATE TABLE IF NOT EXISTS cursos_mooc (
        curso_id INT AUTO_INCREMENT PRIMARY KEY,
        titulo VARCHAR(255),
        instituicao VARCHAR(255),
        instrutor VARCHAR(255),
        numero_participantes VARCHAR(50),
        descricao TEXT,
        url_curso VARCHAR(500) UNIQUE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
    """
    cursor.execute(sql)
    conexao.commit()

Navegação e Extração de Dados dos Cursos

A coleta é realizada percorrendo uma página de listagem, simulando rolagem para carregar mais cards, e extraindo informações básicas. Em seguida, para cada curso, uma nova aba é aberta para coletar detalhes adicionais da página de detalhes.

def processar_pagina_listagem(self, url_listagem):
    self.driver.get(url_listagem)
    self.espera_geral.until(EC.presence_of_element_located((By.CSS_SELECTOR, "div.curso-card")))
    
    cursos_coletados = []
    for _ in range(3):  # Rolar algumas vezes para carregar mais cursos
        self.driver.execute_script("window.scrollBy(0, window.innerHeight);")
        time.sleep(2)
        
        cards = self.driver.find_elements(By.CSS_SELECTOR, "div.curso-card")
        for card in cards:
            titulo = card.find_element(By.CSS_SELECTOR, "h3.titulo").text
            instituicao = card.find_element(By.CSS_SELECTOR, "span.instituicao").text
            url_detalhes = card.find_element(By.TAG_NAME, "a").get_attribute("href")
            
            # Abrir nova aba para detalhes
            self.driver.execute_script(f"window.open('{url_detalhes}');")
            self.driver.switch_to.window(self.driver.window_handles[-1])
            
            # Extrair detalhes adicionais
            try:
                descricao = self.espera_geral.until(
                    EC.presence_of_element_located((By.CSS_SELECTOR, "div.descricao"))
                ).text[:500]
                
                dados_curso = {
                    'titulo': titulo,
                    'instituicao': instituicao,
                    'descricao': descricao,
                    'url_curso': url_detalhes
                }
                self._salvar_curso_db(dados_curso)
                cursos_coletados.append(dados_curso)
            finally:
                self.driver.close()
                self.driver.switch_to.window(self.driver.window_handles[0])
    
    return cursos_coletados

Experimento de Análise de Dados em Tempo Real

Esta seção descreve um pipeline de dados usando tecnologias de big data. Um script Python é empregado para gerar dados de teste no nó mestre do MRS. O Kafka é configurado como sistema de mensagens, com clientes Flume coletando os dados. Análises em tempo real são executadas por meio de jobs do Flink no DLI, permitindo o processamento de streams de dados financeiros ou educacionais.