MeiK / 如何写一个爬虫 - 第二篇

def fetch(
    method: str, 
    url: str, 
    params: dict = None, 
    headers: dict = None
    data: str = None, 
) -> str:

def fetch(
        method: str, url: str, params: dict = None, data: str = None, headers: dict = None
) -> str:
    if url.startswith("http://") is False:
        raise ValueError("url must start with `http://`")

    _tmp = url[7:].split("/", 1)
    if len(_tmp) > 1:
        host = _tmp[0]
        path = "/" + _tmp[1]
    else:
        host = _tmp[0]
        path = "/"

    method = method.upper()
    params = {} if params is None else params
    headers = {} if headers is None else headers

    params_str = "?" if params else ""
    param_count = 0
    for key, value in params.items():
        if param_count != 0:
            params_str += '&'
        params_str += f"{key}={value}"
        param_count += 1

    request_list = [f"{method} {path}{params_str} HTTP/1.0", f"Host: {host}"]
    for key, value in headers.items():
        request_list.append(f"{key}: {value}")

    if data is not None:
        request_list.append(f'Content-Length: {len(data)}')
        request_list.append('')  # 空行，以分割header域与data域
        request_list.append(data)
    request_list.append("\r\n")

    request = "\r\n".join(request_list)

    sock = socket.socket()
    sock.connect((host, 80))
    sock.send(request.encode("ascii"))

    response = b""
    chunk = sock.recv(4096)
    while chunk:
        response += chunk
        chunk = sock.recv(4096)

    sock.close()

    return response.decode()

>>> from ZeroCrawler import fetch
>>> resp = fetch('get', 'http://httpbin.org/get')
>>> resp
'HTTP/1.1 200 OK\r\nAccess-Control-Allow-Credentials: true\r\nAccess-Control-Allow-Origin: *\r\nContent-Type: application/json\r\nDate: Tue, 30 Jul 2019 12:19:26 GMT\r\nReferrer-Policy: no-referrer-when-downgrade\r\nServer: nginx\r\nX-Content-Type-Options: nosniff\r\nX-Frame-Options: DENY\r\nX-XSS-Protection: 1; mode=block\r\nContent-Length: 146\r\nConnection: Close\r\n\r\n{\n  "args": {}, \n  "headers": {\n    "Host": "httpbin.org"\n  }, \n  "origin": "36.110.78.251, 36.110.78.251", \n  "url": "https://httpbin.org/get"\n}\n'

import setuptools

with open("README.md", "r") as fh:
    long_description = fh.read()

setuptools.setup(
    name="ZeroCrawler",
    version="0.0.2",
    author="MeiK2333",
    author_email="[email protected]",
    description="A small example package",
    long_description=long_description,
    long_description_content_type="text/markdown",
    url="https://github.com/MeiK2333/ZeroCrawler",
    packages=setuptools.find_packages(),
)

$ python setup.py install  # or pip install .
Processing /Users/meik/ZeroCrawler
Installing collected packages: ZeroCrawler
  Running setup.py install for ZeroCrawler ... done
Successfully installed ZeroCrawler-0.0.2

$ python -c 'import ZeroCrawler; print(ZeroCrawler.__path__)'  # 此时引入的 ZeroCrawler 是我们正在开发的版本
['/Users/meik/ZeroCrawler/ZeroCrawler']
$ cd ~
$ python -c 'import ZeroCrawler; print(ZeroCrawler.__path__)'  # 此时引入的是我们安装的版本
['/Users/meik/ZeroCrawler/venv/lib/python3.6/site-packages/ZeroCrawler']

def get_body_by_response(response):
    """ 从返回的响应中获得数据正文 """
    return response.split('\r\n\r\n', 1)[1]

import json
import unittest

from ZeroCrawler import fetch
from ZeroCrawler.utils import get_body_by_response


class TestFetch(unittest.TestCase):
    def test_fetch(self):
        resp = fetch("get", "http://httpbin.org/get")
        self.assertTrue(resp.startswith("HTTP/1.1 200 OK"))

    def test_fetch_params(self):
        params = {"key1": "value1", "key2": "value2"}
        resp = fetch("get", "http://httpbin.org/get", params=params)
        data = get_body_by_response(resp)
        data = json.loads(data)
        args = data.get("args")
        for key, value in params.items():
            self.assertTrue(key in args.keys())
            self.assertEqual(value, args[key])

    def test_fetch_headers(self):
        headers = {"User-Agent": "ZeroCrawler"}
        resp = fetch("get", "http://httpbin.org/get", headers=headers)
        data = get_body_by_response(resp)
        data = json.loads(data)
        resp_headers = data.get("headers")
        for key, value in headers.items():
            self.assertTrue(key in resp_headers.keys())
            self.assertEqual(value, resp_headers[key])

    def test_fetch_method(self):
        methods = ["get", "post", "put", "delete", "patch"]
        for method in methods:
            resp = fetch(method, f"http://httpbin.org/{method}")
            data = get_body_by_response(resp)
            data = json.loads(data)
            url = data.get("url")
            self.assertTrue(url.endswith(f"httpbin.org/{method}"))

    def test_fetch_data(self):
        data = "Hello World!"
        resp = fetch("post", "http://httpbin.org/post", data=data)
        resp_data = get_body_by_response(resp)
        resp_data = json.loads(resp_data)
        self.assertEqual(resp_data["data"], data)


if __name__ == "__main__":
    unittest.main()

$ python -m unittest tests/test_*.py
......
----------------------------------------------------------------------
Ran 5 tests in 27.590s

OK