Automatización Web Híbrida

Isaac de la Peña
Algonauta
Published in
7 min readAug 31, 2020

En nuestro artículo anterior, Automatización Web, explicamos las opciones disponibles cuando se trata de interactuar programáticamente con contenido basado en la web, dependiendo del nivel de complejidad requerido para la tarea en cuestión: desde transacciones en bruto, contenido estructurado, sesiones, navegadores, hasta control de tráfico a través de proxies. Si eres nuevo en este campo, te recomiendo que comiences con esa lectura ya que proporciona una introducción paso a paso y muchos ejemplos en Python.

De hecho, todo lo que necesitas está ahí … si no fuera por el hecho de que muchos propietarios de contenido odian la automatización web (a menudo porque quieren implementar tácticas discriminatorias de precios y obligarte a suscribir su costosa API premium para estas funciones) e intentan dificultarla por todos los medios posibles. El más popular son las pantallas captcha en las que se te presenta un desafío que requiere capacidades cognitivas humanas para resolverlo y continuar con la navegación.

Pantalla de Captcha de Instagram como Ejemplo

La Trampa para Ratones

Por supuesto puedes implementar contramedidas en tu código para evitar la activación de tales captchas, tanto tradicionales como basadas en inteligencia artificial: reducir la velocidad de tus acciones, distribuir las solicitudes de manera desigual, añadir movimientos y clics esporádicos del ratón… pero no importa cuán cuidadosamente el roedor camine de puntillas por el sitio web, lo más probable es que eventualmente caiga en una trampa u otra, porque en este juego del gato y el ratón siempre tiene la ventaja el felino: las condiciones de activación precisas nos son desconocidas, pueden cambiar con el tiempo, y algunos sitios incluso implementan captchas incondicionales en momentos inesperados “por si acaso”, similares a las evaluaciones aleatorias de la TSA en los aeropuertos.

Y mientras que la primera generación de captchas (por ejemplo, “escriba los dígitos que ve en esta imagen”) podría ser derrotada implementando algoritmos de IA específicos para ese contexto, la versión de captchas ampliamente utilizada hoy en día tiene un nivel de sofisticación (por ejemplo, “encuentre objetos específicos en esta serie de imágenes, las cuales cambian con el tiempo“) que hacen que cualquier intento de automatización sea demasiado costoso en tiempo y poco práctico.

El Captcha de Dropbox en que se nos pide que “pongamos de pie al animal”

Por tanto parece que estamos condenados a este lema: cuando el captcha aparece, el algoritmo fenece. Nuestro proceso se detiene, nuestras tareas se quedan sin completar y surge la frustración. Pero hay una alternativa: si los captchas requieren cognición humana … ¡entonces pongamos un humano en la ecuación!. Eso es lo que llamamos Automatización Web Híbrida: un sistema que se ejecuta, en su mayor parte, de forma independiente, pero cuando se enfrenta a una situación inesperada (como una pantalla de captcha) solicita la ayuda de un agente humano y espera pacientemente hasta que todo esté claro para reanudar con normalidad sus operaciones en lugar de colapsar en el acto.

Ejemplo: Automatizando Instagram

Para que nuestra explicación sea lo más práctica posible, vamos a aplicar la automatización híbrida en Python al caso de uso particular de descargar todas las imágenes de un perfil de Instagram de nuestra elección.

Es importante recordar, sin embargo, que la “automatización web” va más allá de la mera recopilación de contenidos, y también incluye la posibilidad de interactuar con las páginas web rellenando formularios, proporcionando datos y activando servicios. Es decir, interacción autónoma bidireccional. Pero para nuestros propósitos el web scraping es el caso más simple de retratar, una especie de MVP.

En esencia lo que tenemos que hacer es crear una envoltura alrededor de nuestros métodos utilizando el Patrón de Diseño de Proxy, de modo que no los llamemos directamente, sino siempre a través del proxy. Así en lugar de:

def do_something(param1):
driver.get(param1)

Escribiremos nuestras funcionalidades así:

def do_something(param1):
proxy(_do_something, param1)
def proxy(fun, param1=None):
try:
return fun(param1)
except:
pass # La gestión manual de errores va aquí!
def _do_something(param1)
driver.get(param1)

Lo que se lee como: cuando solicitamos do_something, llama al proxy, que llama al método interno _do_something, que a su vez ejecuta la funcionalidad requerida del navegador. Si la tarea falla en algún momento, el proceso retrocede al proxy donde se detiene (y llama al humano usando, por ejemplo, señales visuales y auditivas) hasta que se gestiona la eventualidad.

Humanos y Máquinas trabajando juntos

Código de Prueba

El código completo se ha publicado en el mismo repositorio de Git utilizado en el artículo anterior de Automatización Web, utilizando Selenium Chrome como nuestro controlador web programático:

https://github.com/isaacdlp/scraphacks

Primero tenemos que iniciar sesión en Instagram. Nuestro código admite tanto la entrada de credenciales como la recuperación de cookies almacenadas para reutilizar sesiones (la frecuencia de inicio de sesión suele ser uno de los criterios de activación de captchas y, por lo tanto, nos cubrimos las espaldas minimizando la necesidad de volver a autenticarse):

def _login(self, site):
# [...]
elif site == "instagram":
if not self._cookies(site):
self._print("Login to instagram")
self.browser.get("https://www.instagram.com")
self.wait()
login_form = self.browser.find_element_by_css_selector("article form") login_email = login_form.find_element_by_name("username")
login_email.send_keys(creds["username"])
login_pass = login_form.find_element_by_name("password")
login_pass.send_keys(creds["password"])
login_pass.send_keys(Keys.ENTER) self.wait(5)
self.browser.find_element_by_css_selector("nav a[href='/%s/']" % creds["username"])
if self.use_cookies:
cookies = self.browser.get_cookies()
with open("%sscrap.cookie" % site, "w") as f:
json.dump(cookies, f, indent=2)
return True

Después, nuestra implementación de la función do_something (llamada _instagram) es bastante simple y se puede encontrar a continuación:

def _instagram(self, url):
props = self._base(url)
media = [] try:
self.browser.execute_script("document.querySelector('article a').click()")
while True:
self.wait()
try:
image = self.browser.find_element_by_css_selector("article.M9sTE img[decoding='auto']")
srcset = image.get_attribute("srcset")
srcs = [src.split(" ") for src in srcset.split(",")]
srcs.sort(reverse=True, key=lambda x: int(x[1][:-1]))
src = srcs[0][0]
media.append({"type" : "jpg", "src" : src})
except:
try :
video = self.browser.find_element_by_css_selector("article.M9sTE video")
src = video.get_attribute("src")
media.append({"type": "mpg", "src": src})
except:
pass
try:
self.browser.execute_script("document.querySelector('a.coreSpriteRightPaginationArrow').click()")
except:
break
except:
pass props["Media"] = media
return props

Básicamente sigue esta rutina:

  • Navega hasta el perfil de destino
  • Visita todos los elementos multimedia de forma secuencial, del último al primero (porque Instagram, como muchos otros sitios sociales, el contenido más antiguo sólo se carga una vez que se el usuario desciende en la página).
  • Toma la URL única de cada elemento multimedia (desacoplando así la recopilación de elementos de la descarga real, nuevamente una estrategia para la prevención de captchas).
  • Devuelve la lista de elementos multimedia en la propiedad “Media”.

Ten en cuenta que, a pesar de su simplicidad, el ejemplo se ha ampliado ya para poder manejar tanto imágenes como vídeos.

Ejecutando la Automatización Híbrida (detrás) para descargar mi propio perfil de Instagram (delante)

Envoltorio Híbrido Reusable

Lo más interesante del ejemplo anterior es que tanto el método de login como el de extracción utilizan la misma función proxy. A saber, esta:

def _proxy(self, fun, var = None):
if self.browser:
successful = False
while not successful:
try:
return fun(var)
except Exception as e:
if self.interactive:
exc_type, exc_obj, exc_tb = sys.exc_info()
print("ERROR '%s' at line %s" % (e, exc_tb.tb_lineno))
cmd = self.default_cmd
if not cmd:
props = {"loop": 0}
if self.audible:
thread = threading.Thread(target=self._play, args=(props,))
thread.start()
cmd = input("*(r)epeat, (c)ontinue, (a)bort or provide new url? ")
props["loop"] = self.max_loop
if cmd == "r" or cmd == "":
pass
elif cmd == "c":
successful = True
elif cmd == "a":
raise e
else:
var = cmd
else:
raise e

La reutilización es el punto principal: el código anterior maneja los errores con elegancia sin importar cuál fue la tarea original en cuestión, repite un sonido un número configurable de veces (para llamar la atención del humano sin ser molesto en caso de que esté ocupado con otros asuntos) y presenta una línea de comandos con las opciones de volver a probar la última URL, cambiar a una nueva URL, pasar a la siguiente o terminar el proceso en el caso de que el error no se haya podido solventar.

Incluso en esta última situación (finalización forzada), implementar la automatización híbrida también ayuda enormemente, ya que todavía vemos una sesión interactiva en el navegador actual para poder comprender y depurar el problema; en lugar de una sesión bloqueada, un navegador cerrado y una excepción arcana en la línea de comandos.

Ayúdame para que pueda ayudarte mejor, humano :)

Además, podemos encapsular nuestro proxy en un objeto de Python para portar el código y sus acciones a diferentes proyectos de automatización web. Eso es precisamente lo que hemos hecho en la carpeta scrapper de nuestra demostración:

https://github.com/isaacdlp/scraphacks/tree/master/scrapper

Ahora, los requisitos para completar nuestro ejemplo particular de Instagram se han simplificado mucho. Sólo tenemos que: primero llamar al objeto que acabamos de crear y luego descargar los elementos multimedia específicos como mejor nos parezca. El código que mostramos a continuación se puede encontrar en el archivo socialscrap.py:

https://github.com/isaacdlp/scraphacks/blob/master/socialscrap.py

from scrapper import *
import requests as req
target = "isaacdlp"folder = "download/%s" % target
if not os.path.exists(folder):
os.mkdir(folder)
scrapper = Scrapper()
try:
scrapper.start()
scrapper.login("instagram")
props = scrapper.instagram("https://www.instagram.com/%s" % target)
finally:
scrapper.stop()
for i, prop in enumerate(props["Media"], start = 1):
res = req.get(prop["src"])
if res.status_code != 200:
break
with open("%s/%s-%s.%s" % (folder, target, i, prop["type"]), 'wb') as bout:
bout.write(res.content)

Más allá de eso, consulta el archivo __init__.py para obtener más detalles sobre la implementación del contenedor. Verás que incluye otras funciones avanzadas como la gestión generalizada de cookies, el scroll de las páginas y las capturas de pantalla de sitios web. Siéntete libre de hacer tuyo el código, adaptarlo a tus necesidades y ampliarlo para admitir otros casos de uso.

¡De nada! 😃

--

--