Python e LangChain: Web scraping com IA

1
2
3
4
5
6
7
8
9

pip install playwright \
  beautifulsoup4 \
  python-decouple \
  tiktoken \
  langchain-text-splitters \
  langchain-openai

# Baixa o navegador que o Playwright vai controlar
playwright install chromium

1

OPENAI_API_KEY=sua_chave_aqui

1
2
3
4
5

from decouple import config
import os

# Carrega a chave da API da OpenAI do arquivo .env usando decouple
os.environ['OPENAI_API_KEY'] = config('OPENAI_API_KEY')

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

from pydantic import Field, BaseModel

class Article(BaseModel):
    """
    Representa um artigo com informações estruturadas.
    """
    headline: str = Field(description="O título do artigo")
    description: str | None = Field(default=None, description="Uma breve descrição do artigo, se existir no texto")
    date: str = Field(description="A data em que o artigo foi publicado")
    author: str = Field(description="O autor do artigo")
    link: str = Field(description="O link para o artigo")


class ArticleList(BaseModel):
    articles: list[Article] = Field(description="Todos os artigos encontrados no texto")

1
2
3
4
5
6
7
8
9
10
11
12
13

from playwright.sync_api import sync_playwright

def carregar_html(url):
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()
        page.goto(url, wait_until="networkidle")
        html = page.content()
        browser.close()
    return html

# Lista de URLs que serão carregadas
urls = ["https://pythonacademy.com.br/blog/"]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

from bs4 import BeautifulSoup
from langchain_core.documents import Document

def artigo_para_texto(artigo):
    texto = artigo.get_text(" ", strip=True)
    link = artigo.find("a", href=True)
    return f"{texto} (link: {link['href']})" if link else texto

tags_to_extract = ["article"]

docs_transformed = []
for url in urls:
    soup = BeautifulSoup(carregar_html(url), "html.parser")
    trechos = [artigo_para_texto(tag) for tag in soup.find_all(tags_to_extract)]
    docs_transformed.append(
        Document(page_content="\n\n".join(trechos), metadata={"source": url})
    )

1
2
3
4
5
6
7
8

from langchain_text_splitters import RecursiveCharacterTextSplitter

# Divide o conteúdo extraído em pedaços usando tokenização
splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
  chunk_size=2000,
  chunk_overlap=0
)
splits = splitter.split_documents(documents=docs_transformed)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

from langchain_openai import ChatOpenAI

# Cria uma instância do modelo ChatOpenAI configurado com o gpt-6-luna
llm = ChatOpenAI(model="gpt-6-luna")

# Configura o modelo para retornar respostas no formato da classe ArticleList
structured_llm = llm.with_structured_output(ArticleList)

# Extrai e estrutura os artigos de cada pedaço
extracted_content = []
for split in splits:
    resultado = structured_llm.invoke(split.page_content)
    extracted_content.extend(resultado.articles)

# Converte o conteúdo estruturado em uma lista de dicionários
extracted_content_dict = [article.model_dump() for article in extracted_content]

# Exibe o resultado
print(extracted_content_dict)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

import os

from bs4 import BeautifulSoup
from decouple import config
from langchain_core.documents import Document
from langchain_openai import ChatOpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter
from playwright.sync_api import sync_playwright
from pydantic import Field, BaseModel

os.environ['OPENAI_API_KEY'] = config('OPENAI_API_KEY')

llm = ChatOpenAI(model="gpt-6-luna")

class Article(BaseModel):
    headline: str = Field(description="O título do artigo")
    description: str | None = Field(default=None, description="Uma breve descrição do artigo, se existir no texto")
    date: str = Field(description="A data em que o artigo foi publicado")
    author: str = Field(description="O autor do artigo")
    link: str = Field(description="O link para o artigo")

class ArticleList(BaseModel):
    articles: list[Article] = Field(description="Todos os artigos encontrados no texto")

structured_llm = llm.with_structured_output(ArticleList)

def carregar_html(url):
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()
        page.goto(url, wait_until="networkidle")
        html = page.content()
        browser.close()
    return html

def artigo_para_texto(artigo):
    texto = artigo.get_text(" ", strip=True)
    link = artigo.find("a", href=True)
    return f"{texto} (link: {link['href']})" if link else texto

urls = ["https://pythonacademy.com.br/blog/"]
tags_to_extract = ["article"]

docs_transformed = []
for url in urls:
    soup = BeautifulSoup(carregar_html(url), "html.parser")
    trechos = [artigo_para_texto(tag) for tag in soup.find_all(tags_to_extract)]
    docs_transformed.append(
        Document(page_content="\n\n".join(trechos), metadata={"source": url})
    )

splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
  chunk_size=2000,
  chunk_overlap=0
)

splits = splitter.split_documents(documents=docs_transformed)

extracted_content = []
for split in splits:
    extracted_content.extend(structured_llm.invoke(split.page_content).articles)

extracted_content_dict = [article.model_dump() for article in extracted_content]

print(extracted_content_dict)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

from langchain_openai import ChatOpenAI
from playwright.sync_api import sync_playwright
from pydantic import BaseModel

def carregar_texto(url):
    """Abre a página no Chromium e devolve o texto visível."""
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()
        page.goto(url, wait_until="networkidle")
        texto = page.inner_text("body")
        browser.close()
    return texto

class Produto(BaseModel):
    nome: str
    preco_atual: float
    preco_original: float | None = None
    desconto_percentual: float | None = None
    disponivel: bool

llm = ChatOpenAI(model="gpt-6-luna")
extrator = llm.with_structured_output(Produto)

# Scraping com extração estruturada feita pelo modelo
texto = carregar_texto("https://ecommerce.com/produto")
produto = extrator.invoke(f"Extraia os dados do produto desta página:\n\n{texto}")

print(f"Preço atual: R$ {produto.preco_atual}")
if produto.desconto_percentual:
    print(f"Desconto: {produto.desconto_percentual}%")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

from pydantic import BaseModel

class LinhaTabela(BaseModel):
    empresa: str
    receita: float
    lucro: float
    margem: float

class TabelaFinanceira(BaseModel):
    titulo: str
    linhas: list[LinhaTabela]
    total_empresas: int

# O modelo identifica a tabela no texto e devolve a estrutura pronta
texto = carregar_texto("https://site.com/tabela-financeira")
tabela = llm.with_structured_output(TabelaFinanceira).invoke(texto)

for linha in tabela.linhas:
    print(f"{linha.empresa}: {linha.margem}% de margem")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

from playwright.sync_api import sync_playwright

def scrape_com_interacao(url, username, password):
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=False)
        page = browser.new_page()

        # Login
        page.goto(url)
        page.fill('input[name="username"]', username)
        page.fill('input[name="password"]', password)
        page.click('button[type="submit"]')

        # Aguardar carregamento
        page.wait_for_selector('.dashboard')

        # Scroll infinito
        previous_height = 0
        while True:
            page.evaluate('window.scrollTo(0, document.body.scrollHeight)')
            page.wait_for_timeout(2000)

            new_height = page.evaluate('document.body.scrollHeight')
            if new_height == previous_height:
                break
            previous_height = new_height

        # Extrair conteúdo
        content = page.content()
        browser.close()

        return content

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

import hashlib
import json
from pathlib import Path

def scrape_com_cache(url):
    # Hash da URL como chave de cache
    cache_key = hashlib.md5(url.encode()).hexdigest()
    cache_file = Path(f"cache/{cache_key}.json")

    # Verificar cache
    if cache_file.exists():
        with open(cache_file, encoding="utf-8") as f:
            return json.load(f)

    # Scraping real (custa API)
    texto = carregar_texto(url)
    result = extrator.invoke(texto).model_dump()  # extrator de Produto do caso 1

    # Salvar em cache
    cache_file.parent.mkdir(exist_ok=True)
    with open(cache_file, 'w', encoding="utf-8") as f:
        json.dump(result, f, ensure_ascii=False)

    return result

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

from pydantic import BaseModel

class ListaDeProdutos(BaseModel):
    produtos: list[Produto]  # Produto do caso 1

# Processar múltiplas páginas juntas: uma chamada só, em vez de uma por página
# (as instruções e o schema vão uma única vez)
urls = ["url1", "url2", "url3"]
textos = [carregar_texto(url) for url in urls]

# Combinar antes de processar
combined_content = "\n\n".join(
    f"Página: {url}\n{texto}" for url, texto in zip(urls, textos)
)

# Processar tudo de uma vez
result = llm.with_structured_output(ListaDeProdutos).invoke(combined_content)