Skip to content

Instantly share code, notes, and snippets.

@zeandrade

zeandrade/automacao_web_crawler.ipynb

Created August 29, 2025 15:47
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save zeandrade/12f9f3fee22f7b3214b6228350d376b5 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save zeandrade/12f9f3fee22f7b3214b6228350d376b5 to your computer and use it in GitHub Desktop.
Download ZIP
ferramenta
Display the source blob
Display the rendered blob
Raw
automacao_web_crawler.ipynb
{
"cells": [
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"id": "BYMtmPVipQLP"
},
"outputs": [],
"source": [
"! pip install aiohttp[speedups]"
]
},
{
"cell_type": "code",
"execution_count": 9,
"metadata": {},
"outputs": [],
"source": [
"import json\n",
"from collections import defaultdict\n",
"from typing import Dict, List, Any\n",
"import time\n",
"import aiohttp\n",
"import asyncio"
]
},
{
"cell_type": "code",
"execution_count": 19,
"metadata": {},
"outputs": [],
"source": [
"# --- Configurações\n",
"# Substitua por seu usuário e senha reais\n",
"USUARIO_PADRAO = \"josejr\"\n",
"SENHA_PADRAO = \"senha\"\n",
"# URLs dos endpoints\n",
"URL_AUTENTICACAO = \"https://api.zeroandrade.com.br/pinata/autenticar\"\n",
"URL_CONSULTA = \"https://api.zeroandrade.com.br/pinata/consultar\"\n",
"\n",
"NUMERO_MAXIMO_CONSULTAS = 30 # Ajuste este valor conforme necessário"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "RslBSO66p-2W",
"outputId": "a164a883-d36e-4b52-bc04-cd2cf40ae521"
},
"outputs": [],
"source": [
"\n",
"async def teste():\n",
"\n",
" async with aiohttp.ClientSession() as session:\n",
" async with session.get('http://api.zeroandrade.com.br') as response:\n",
" print('Massa')\n",
" print(\"Status:\", response.status)\n",
" print(\"Content-type:\", response.headers['content-type'])\n",
"\n",
" html = await response.text()\n",
" print(\"Body:\", html[:50], \"...\")\n",
"\n",
"# Use await main() instead of asyncio.run(main()) in environments like Colab\n",
"await teste()"
]
},
{
"cell_type": "code",
"execution_count": 21,
"metadata": {},
"outputs": [],
"source": [
"\n",
"\n",
"\n",
"# Gera uma lista de parâmetros com o número máximo ajustável\n",
"parametros_consulta = [f\"param_{i:04d}\" for i in range(NUMERO_MAXIMO_CONSULTAS)]\n",
"\n",
"# --- Funções Assíncronas\n",
"\n",
"async def autenticar(session: aiohttp.ClientSession, usuario: str, senha: str) -> str:\n",
" \"\"\"\n",
" Realiza a autenticação e retorna o token de acesso.\n",
" \"\"\"\n",
" payload = {\"usuario\": usuario, \"senha\": senha}\n",
" headers = {\"Content-Type\": \"application/json\"}\n",
" \n",
" print(\"Tentando autenticar...\")\n",
" try:\n",
" async with session.post(URL_AUTENTICACAO, json=payload, headers=headers) as response:\n",
" if response.status == 200:\n",
" data = await response.json()\n",
" print(\"Autenticação bem-sucedida.\")\n",
" return data.get(\"token\")\n",
" elif response.status == 401:\n",
" print(\"Erro de autenticação: Usuário ou senha incorretos.\")\n",
" return None\n",
" else:\n",
" print(f\"Erro inesperado na autenticação: Status {response.status}\")\n",
" return None\n",
" except aiohttp.ClientError as e:\n",
" print(f\"Erro de conexão na autenticação: {e}\")\n",
" return None\n",
"\n",
"async def consultar_item(session: aiohttp.ClientSession, item: str, token: str) -> Dict[str, Any]:\n",
" \"\"\"\n",
" Realiza uma única consulta para um item e retorna a resposta.\n",
" \"\"\"\n",
" headers = {\n",
" \"Content-Type\": \"application/json\",\n",
" \"X-Auth\": token\n",
" }\n",
" payload = {\"parametro\": item}\n",
" \n",
" try:\n",
" async with session.post(URL_CONSULTA, json=payload, headers=headers) as response:\n",
" if response.status == 403:\n",
" print(f\"Consulta bloqueada para o item '{item}'. Interrompendo...\")\n",
" return {\"status\": 403, \"item\": item}\n",
" \n",
" data = await response.json()\n",
" return {\"status\": response.status, \"data\": data, \"item\": item}\n",
" except aiohttp.ClientError as e:\n",
" print(f\"Erro de conexão na consulta para o item '{item}': {e}\")\n",
" return {\"status\": \"error\", \"item\": item}\n",
"\n",
"async def processar_lote(session: aiohttp.ClientSession, lote: List[str], token: str, resultados: defaultdict, nao_encontrados: List) -> bool:\n",
" \"\"\"\n",
" Processa um lote de consultas em paralelo.\n",
" Retorna True se deve continuar, False se uma resposta 403 foi recebida.\n",
" \"\"\"\n",
" tasks = [consultar_item(session, item, token) for item in lote]\n",
" respostas = await asyncio.gather(*tasks)\n",
" \n",
" for resposta in respostas:\n",
" item = resposta[\"item\"]\n",
" status = resposta[\"status\"]\n",
" \n",
" if status == 403:\n",
" print(\"Resposta 403 recebida. Processamento interrompido.\")\n",
" return False # Indica que o loop principal deve parar\n",
" \n",
" if status == 200:\n",
" dados = resposta[\"data\"]\n",
" resultados[item] = dados\n",
" # print(f\"Item '{item}' encontrado. Total de encontrados: {len(resultados)}\")\n",
" elif status == 404 or status == 400:\n",
" nao_encontrados.append(item)\n",
" else:\n",
" print(f\"Status inesperado para o item '{item}': {status}\")\n",
" \n",
" return True # Indica que o loop principal pode continuar\n",
"\n",
"async def main():\n",
" \"\"\"\n",
" Função principal que coordena a autenticação e as consultas em lotes.\n",
" \"\"\"\n",
" resultados = defaultdict(dict)\n",
" nao_encontrados = []\n",
" \n",
" async with aiohttp.ClientSession() as session:\n",
" # Etapa 1: Autenticação\n",
" token = await autenticar(session, USUARIO_PADRAO, SENHA_PADRAO)\n",
" \n",
" if not token:\n",
" print(\"Não foi possível obter o token. Encerrando o programa.\")\n",
" return\n",
"\n",
" # Etapa 2: Processamento dos lotes\n",
" lotes_processados = 0\n",
" total_lotes = len(parametros_consulta) // 100 + (len(parametros_consulta) % 100 > 0)\n",
"\n",
" for i in range(0, len(parametros_consulta), 100):\n",
" if i > 0:\n",
" print(\"Aguardando 2 segundos para o próximo lote...\")\n",
" await asyncio.sleep(2)\n",
" \n",
" lote = parametros_consulta[i:i+100]\n",
" lotes_processados += 1\n",
" print(f\"Processando lote {lotes_processados}/{total_lotes} (itens {i+1} a {min(i+100, len(parametros_consulta))})...\")\n",
" \n",
" deve_continuar = await processar_lote(session, lote, token, resultados, nao_encontrados)\n",
" if not deve_continuar:\n",
" print(\"Processamento interrompido devido à resposta 403.\")\n",
" break\n",
" \n",
" print(\"\\n--- Processo Concluído ---\")\n",
" print(f\"Total de itens encontrados: {len(resultados)}\")\n",
" print(f\"Total de itens não encontrados: {len(nao_encontrados)}\")\n",
" print(f\"Lotes processados: {lotes_processados}\")\n",
"\n",
" # Exibindo os resultados (opcional)\n",
" print(\"\\nResultados encontrados (amostra):\")\n",
" for item, dados in list(resultados.items())[:5]:\n",
" print(f\" - Item '{item}': {dados}\")\n",
"\n",
" print(\"\\nItens não encontrados (amostra):\")\n",
" print(nao_encontrados[:10])\n",
"\n",
" print(\"\\n--- Estruturas de Dados Finais ---\")\n",
" print(\"Variável 'resultados' (defaultdict):\")\n",
" print(len(resultados))\n",
" print(\"Variável 'nao_encontrados' (lista):\")\n",
" print(len(nao_encontrados))\n",
" \n",
"# Inicia a execução da função principal\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Executar o ATAQUE!"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"await main()"
]
}
],
"metadata": {
"colab": {
"provenance": []
},
"kernelspec": {
"display_name": "Python 3",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.10.12"
}
},
"nbformat": 4,
"nbformat_minor": 0
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment