python 中的 hook

鸡仔说 2021-02-16

3535

鸡仔说：在程序设计中，有很多名词，比如面向对象编程，面向切面编程，面向钩子编程等等。其实他们被设计出来核心作用都是相通的，提高代码的可拓展性，可维护性

hooking，被称作钩子编程，是一种通过拦截程序模块之间消息传递、函数调用等，来改变软件行为的一种技术。这种技术被广泛应用在 AOP 场景中。比如 scrapy 下的 middleware，Flask 和 Django 的自定义请求上下文等。

对于和鸡仔一样的 Pythoner 来说，即使我们不熟悉这个概念，也一定在开发中，有意或无意的使用过该项技术，最简单的比如 Python 的内置函数 sorted：

>>>import string
>>>import random
>>>a = [(random.randint(1, 100), _) for _ in string.ascii_lowercase]
>>>a
[(87, 'a'), (64, 'b'), (60, 'c'), (32, 'd'), (72, 'e'), (37, 'f'), (67, 'g'), (15, 'h'), (76, 'i'), (44, 'j'), (20, 'k'), (54, 'l'), (69, 'm'), (14, 'n'), (30, 'o'), (18, 'p'), (66, 'q'), (57, 'r'), (4, 's'), (91, 't'), (98, 'u'), (87, 'v'), (45, 'w'), (73, 'x'), (47, 'y'), (23, 'z')]
>>>sorted(a, key=lambda x: x[0])  # 根据可迭代对象a中第一项进行排序
[(4, 's'), (14, 'n'), (15, 'h'), (18, 'p'), (20, 'k'), (23, 'z'), (30, 'o'), (32, 'd'), (37, 'f'), (44, 'j'), (45, 'w'), (47, 'y'), (54, 'l'), (57, 'r'), (60, 'c'), (64, 'b'), (66, 'q'), (67, 'g'), (69, 'm'), (72, 'e'), (73, 'x'), (76, 'i'), (87, 'a'), (87, 'v'), (91, 't'), (98, 'u')]

像上述示例一样，Python 中使用钩子的方式非常简单，这得益于 python 中函数是“一等公民”的设计。平常我们在爬虫的时候经常会遇到动态变化的场景，比如发起请求时，需要基于不同的网站，替换不同的 cookies、请求头等。下面就跟着鸡仔一步步实现一个简单的 hook 网络请求工具吧。

01.

需求分析

设计实现通用网络采集工具 fetch，实现 cookies 和 headers 钩子，使得他们能够根据不同的站点替换不同的策略。比如有些网站不需要使用 cookies，就不用传入 cookis。另外，之所以设计成钩子要能够应对未来可能的变化，比如之后又来了一个需求，针对部分网站需要走代理爬取。这就需要我们在不改动原有 hook 的基础上，能够方便的添加新 hook。

02.

结构设计

因为我们设计的维度很简单，仅设计发起请求部分，所以整体的结构非常清晰，如下所示

├── fetch.py
├── hooks
│   ├── __init__.py
│   ├── cookies_hook.py
│   └── headers_hook.py
└── main.py

hooks 文件夹下就是所有的钩子实现，之后如果有需要增加的模块，即按照现有的格式添加 hook 即可。

03.

代码实现

整个设计结构如下，关键就在这个 fetch 部分，它做了如下三件事，校验 req 部分是否存在必要的属性，然后通过钩子函数对 req 进行加工，最后重新打包发起请求

hooks/cookies_hook.py

class CookiesHook:

    __COOKIE_MAP = {
        "www.baidu.com": "BIDUPSID=3D72E33E74AB0913CF112ABEDC1E3464; ",
    }

    @classmethod
    def process(cls, req):
        req.cookie = cls.__COOKIE_MAP.get(req.host, "")
        return req

hooks/headers_hook.py


class HeadersHook:

    __REFER_MAP = {
        "www.baidu.com": "https://www.baidu.com/",
        "www.mzitu.com": "https://www.mzitu.com/"
    }

    __COMMON_HEADERS = {
        "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 11_2_0) AppleWe"
                      "bKit/537.36 (KHTML, like Gecko) Chrome/88.0.4324.150 Safari/537.36"
    }

    @classmethod
    def process(cls, req):
        req.headers["user-agent"] = req.headers.get('user-agent') or cls.__COMMON_HEADERS["user-agent"]
        req.headers["referer"] = cls.__REFER_MAP.get(req.host, "")
        return req

fetch.py


import os
import importlib
import sys
import requests

sys.path.append("./hooks")

all_files = os.listdir("./hooks")

all_hook_files = list(filter(lambda x: x.endswith("_hook.py"), all_files))

all_hook_module = list(map(lambda x: x.replace(".py", ""), all_hook_files))

# 动态加载 hooks 文件夹下的所有 hook
hooks = []
for module_name in all_hook_module:
    hooks.append(importlib.import_module(module_name))


def fetch_req(req):
    # 校验 req 是否存在必须的参数
    if not req.host or not req.url:
        raise ValueError(f"req={req} host or url missed")

    # 这一步就是动态加载所有的 hook 类，这里面方法可以在 hook 文件夹下进行动态增删改，比如你想统计每一个请求的数量，可以增
    # 加一个 statistic_hook.py 文件，并实现其中的 process 函数
    for per_hook in hooks:
        hook_class = getattr(per_hook, dir(per_hook)[0])
        # 动态执行 process 函数
        getattr(hook_class, "process")(req)
    headers = req.headers
    cookie = req.cookie
    if cookie:
        headers["cookie"] = cookie

    if req.method not in ["post", "POST"]:
        print("req get headers>>>>", headers)
        res = requests.get(req.url, headers=headers)
    else:
        res = requests.post(req.url, headers=headers)

    return res.text

main.py


from fetch import fetch_req
from copy import deepcopy


class Req:

    def __init__(self):
        self.host = None
        self.url = None
        self.headers = {}
        self.cookie = None
        self.method = "get"

    def __repr__(self):
        return f'<{self.host}>({self.url})'


base_req = Req()

baidu_req = deepcopy(base_req)
baidu_req.host = "www.baidu.com"
baidu_req.url = "https://www.baidu.com/"
baidu_res = fetch_req(baidu_req)
print(baidu_req)
print("baidu_res>>>", baidu_res)

# test http_org
http_org_req = deepcopy(base_req)
http_org_req.host = "httpbin.org"
http_org_req.url = "https://httpbin.org/"
http_org_res = fetch_req(http_org_req)
print(http_org_req)
print("http_org_res>>>", http_org_res)