Uso de programación asíncrona para acceso concurrente: aiohttp
La mayoría de las veces, simplemnete necesitarás acceder a una URL específica, usando directamente el método asíncrono de aiohttp.
Sin embargo, a veces es necesario acceder a múltiples niveles de URL donde el siguiente nivel depende del resultado del anterior, requiriendo esperar a que la primera respuesta se complete antes de continuar con la siguiente. Por ejemplo, al obtener proxies (como en el segundo ejemplo).
1. Accceso concurrente a URL individuales
Acceso directo a sitios web a través de direcciones proxy.
Esto implica múltiples solicitudes concurrentes a sitios de proxy, quienes se encargan de devolver los datos.
Código de ejemplo:
import asyncio
import aiohttp
import random
import threading
import uuid
import traceback
import time
import json
from bs4 import BeautifulSoup
URL_BASE = 'https://es.wikipedia.org/wiki/{}'
LISTA_AGENTES_USUARIO = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.99 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36 Edge/16.16299',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.5219.102 Safari/537.38',
# Agregar más User-Agent
]
# Versión simple
async def ejecucion_simple():
async with aiohttp.ClientSession() as sesion:
respuesta = await sesion.get("http://www.google.com")
print(respuesta.status)
# --------------------------- Versión concurrente ----------------------------
# Límite de concurrencia
limite_concurrencia = asyncio.Semaphore(30)
async def procesador_url(indice, sesion, url):
lista_agentes = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/93.0.4577.63 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/95.0.4638.69 Safari/537.36",
# Agregar más agentes de usuario
]
# Configuración del proxy
servidor_proxy = "proxy.example.com:8080"
usuario_proxy = "usuario_proxy"
contrasena_proxy = "contrasena_proxy"
configuracion_proxy = "http://%(usuario)s:%(contrasena)s@%(servidor)s:%(puerto)s" % {
"servidor" : servidor_proxy.split(":")[0],
"puerto" : servidor_proxy.split(":")[1],
"usuario" : usuario_proxy,
"contrasena" : contrasena_proxy,
}
encabezados = {
"Host": "es.wikipedia.org",
"User-Agent": random.choice(lista_agentes),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
"Accept-Language": "es-ES,es;q=0.8,en;q=0.5",
"Accept-Encoding": "gzip, deflate, br",
"Referer": "https://es.wikipedia.org/",
"Connection": "keep-alive",
"Upgrade-Insecure-Requests": "1",
"Sec-Fetch-Dest": "document",
"Sec-Fetch-Mode": "navigate",
"Sec-Fetch-Site": "same-origin",
"Sec-Fetch-User": "?1",
"Pragma": "no-cache",
"Cache-Control": "no-cache",
}
# Configuración del túnel proxy con IP aleatoria
tunnel_proxy = "{}".format(random.randint(1, 10000))
identificador_unico = str(uuid.uuid4())
estado = "1"
try:
async with limite_concurrencia:
contador = 0
sesion = aiohttp.ClientSession()
async with sesion.get(url.replace("\n",""), headers = encabezados, proxy=configuracion_proxy, proxy_headers={"Proxy-Tunnel":tunnel_proxy}) as respuesta:
contenido = await respuesta.text()
print(str(threading.current_thread().ident) + ":Tarea:" + str(indice) + ":Completada:Estado:" + str(respuesta.status))
print('------------------------------')
print(encabezados["User-Agent"] + ": Resultado " + str(respuesta.status))
if "Verificación de Wikipedia" in contenido or "Acceso anómalo" in contenido:
print(url + " Requiere verificación: " + encabezados["User-Agent"])
estado = "1"
elif "La página no existe" in contenido:
print(url + " Página no encontrada")
estado = "2"
elif "Tiempo de espera agotado del proxy" in contenido:
estado = "6"
elif "407 Autenticación de proxy requerida" in contenido:
estado = "7"
elif "Demasiadas solicitudes" in contenido:
print(url + " Límite de solicitudes alcanzado")
estado = "5"
else:
print(url + " Acceso exitoso " + identificador_unico)
analizador = BeautifulSoup(contenido, 'html.parser')
with open("F:\\datos_wikipedia\\html\\" + identificador_unico + ".txt", 'w', encoding="utf-8") as archivo:
archivo.write(contenido)
estado = "3"
return [identificador_unico, url, estado]
except:
print(traceback.format_exc())
estado = "4"
return [identificador_unico, url, estado]
async def funcion_principal():
lista_resultados = []
archivo_entrada = open("F:\\datos_wikipedia\\urls.txt", 'r', encoding="utf-8")
linea = archivo_entrada.readable()
indice = 0
while linea:
if linea is True:
linea = archivo_entrada.readline()
continue
lineas = linea.split("\t")
tarea = asyncio.ensure_future(procesador_url(indice, None, url=lineas[2]))
lista_resultados.append(tarea)
indice += 1
linea = archivo_entrada.readline()
if indice % 3000 == 0:
contador_resultados = 1
archivo_resultados = open("F:\\datos_wikipedia\\resultados.txt", 'a', encoding="utf-8")
for dato in await asyncio.gather(*lista_resultados):
archivo_resultados.write(dato[0] + "\t" + dato[1] + "\t" + dato[2] + "\n")
if contador_resultados > 5000:
archivo_resultados.flush()
contador_resultados += 1
archivo_resultados.close()
lista_resultados = []
await asyncio.gather(*lista_resultados)
if __name__ == '__main__':
inicio = time.time()
asyncio.get_event_loop().run_until_complete(funcion_principal())
fin = time.time()
print("Tiempo transcurrido:")
print(fin - inicio)
2. Acceso concurrente a URL dependientes en múltiples niveles
Es decir: primero se accede a la primera URL, y una vez obtenida su respuesta, se procede con el acceso a la segunda capa de URL.
Código de ejemplo:
import asyncio
import aiohttp
import random
import threading
import uuid
import traceback
import time
import json
from bs4 import BeautifulSoup
URL_BASE = 'https://es.wikipedia.org/wiki/{}'
LISTA_AGENTES_USUARIO = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.99 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36 Edge/16.16299',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.5219.102 Safari/537.38',
]
# Obtener proxies de un servicio externo
async def obtener_proxy():
async with aiohttp.ClientSession() as sesion:
intento = 0
resultado = None
while intento < 4:
try:
print(str(intento) + " Iniciando obtención de proxy")
url_servicio = "http://api.proxy-service.com/getip?cantidad=1&tipo=2"
dominios_posibles = ["proxy1.", "proxy2.", "proxy3."]
# Seleccionar aleatoriamente un dominio para obtener proxies
prefijo = random.choice(dominios_posibles)
url_final = url_servicio.replace("api.", prefijo)
respuesta = await sesion.get(url_final)
texto_respuesta = await respuesta.text()
datos_json = json.loads(texto_respuesta)
if datos_json["codigo"] == 0:
resultado = datos_json["datos"][0]["ip"], datos_json["datos"][0]["puerto"]
break
except:
print(traceback.format_exc())
finally:
intento += 1
print(resultado)
return resultado
# Versión concurrente
limite_concurrencia = asyncio.Semaphore(30)
async def procesador_url_con_proxy(indice, sesion, url):
lista_agentes = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/93.0.4577.63 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/95.0.4638.69 Safari/537.36",
# Agregar más agentes de usuario
]
encabezados = {
"Host": "es.wikipedia.org",
"User-Agent": random.choice(lista_agentes),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
"Accept-Language": "es-ES,es;q=0.8,en;q=0.5",
"Accept-Encoding": "gzip, deflate, br",
"Referer": "https://es.wikipedia.org/",
"Connection": "keep-alive",
"Upgrade-Insecure-Requests": "1",
"Sec-Fetch-Dest": "document",
"Sec-Fetch-Mode": "navigate",
"Sec-Fetch-Site": "same-origin",
"Sec-Fetch-User": "?1",
"Pragma": "no-cache",
"Cache-Control": "no-cache",
}
identificador_unico = str(uuid.uuid4())
estado = "1"
try:
async with limite_concurrencia:
contador = 0
sesion = aiohttp.ClientSession()
# !!!! Acceder a la primera capa de URL, obtener resultados y luego acceder a la segunda
lista_proxies = await obtener_proxy()
proxy_servidor = 'http://{0}:{1}'.format(lista_proxies[0], lista_proxies[1])
# Acceder a la segunda capa de URL
async with sesion.get(url.replace("\n",""), headers=encabezados, proxy=proxy_servidor, ssl=False) as respuesta:
contenido = await respuesta.text()
print(str(threading.current_thread().ident) + ":Tarea:" + str(indice) + ":Completada:Estado:" + str(respuesta.status))
print('------------------------------')
print(encabezados["User-Agent"] + ": Resultado " + str(respuesta.status))
if "Verificación de Wikipedia" in contenido or "Acceso anómalo" in contenido:
print(url + " Requiere verificación: " + encabezados["User-Agent"])
estado = "1"
elif "La página no existe" in contenido:
print(url + " Página no encontrada")
estado = "2"
elif "Tiempo de espera agotado del proxy" in contenido:
estado = "6"
elif "407 Autenticación de proxy requerida" in contenido:
estado = "7"
elif "Demasiadas solicitudes" in contenido:
print(url + " Límite de solicitudes alcanzado")
estado = "5"
else:
print(url + " Acceso exitoso " + identificador_unico)
analizador = BeautifulSoup(contenido, 'html.parser')
with open("F:\\datos_wikipedia\\html\\" + identificador_unico + ".txt", 'w', encoding="utf-8") as archivo:
archivo.write(contenido)
estado = "3"
return [identificador_unico, url, estado]
except:
print(traceback.format_exc())
estado = "4"
return [identificador_unico, url, estado]
async def funcion_principal():
lista_resultados = []
archivo_entrada = open("F:\\datos_wikipedia\\urls.txt", 'r', encoding="utf-8")
linea = archivo_entrada.readable()
indice = 0
while linea:
if linea is True:
linea = archivo_entrada.readline()
continue
lineas = linea.split("\t")
tarea = asyncio.ensure_future(procesador_url_con_proxy(indice, None, url=lineas[2]))
lista_resultados.append(tarea)
indice += 1
linea = archivo_entrada.readline()
if indice % 3000 == 0:
contador_resultados = 1
archivo_resultados = open("F:\\datos_wikipedia\\resultados.txt", 'a', encoding="utf-8")
for dato in await asyncio.gather(*lista_resultados):
archivo_resultados.write(dato[0] + "\t" + dato[1] + "\t" + dato[2] + "\n")
if contador_resultados > 5000:
archivo_resultados.flush()
contador_resultados += 1
archivo_resultados.close()
lista_resultados = []
await asyncio.gather(*lista_resultados)
if __name__ == '__main__':
inicio = time.time()
asyncio.get_event_loop().run_until_complete(funcion_principal())
fin = time.time()
print("Tiempo transcurrido:")
print(fin - inicio)