# spider_event_loop.py
import time
import os
import socket
from urllib.parse import urlparse
from collections import deque
from selectors import DefaultSelector, EVENT_WRITE, EVENT_READ
# selector = DefaultSelector()
urls = ['https://p9-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/9bc51bc53f634bf79b5de5c8b9810817~tplv-k3u1fbpfcp-watermark.image',
        'https://p1-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/2b95ac8571ba403180743495ed56e492~tplv-k3u1fbpfcp-watermark.image',
        'https://p6-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/459c6c73887a4206a33b13ef23988809~tplv-k3u1fbpfcp-watermark.image',
        'https://p6-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/80701ade82d345cd8aa0b08fef008fe1~tplv-k3u1fbpfcp-watermark.image',
        'https://p6-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/99855ed1f79c456f9fdcd83ce5cff4f2~tplv-k3u1fbpfcp-watermark.image',
        'https://p1-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/0c4bd38c2fe3434587025748d051362e~tplv-k3u1fbpfcp-watermark.image',
        'https://p1-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/80cd2a1d165b43beb6234d893ce391e2~tplv-k3u1fbpfcp-watermark.image',
        'https://p3-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/ea18df95f26f4314a0a36d11d4a067d0~tplv-k3u1fbpfcp-watermark.image',
        'https://p1-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/095c342a796c411cad1800396746bdaf~tplv-k3u1fbpfcp-watermark.image',
        'https://p6-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/d3769e4d468a4f64a1e2879f94ad742b~tplv-k3u1fbpfcp-watermark.image'
]
# 事件循環(huán)實(shí)現(xiàn)
class EventLoop:
    def __init__(self):
     	# 整個(gè)流程是否
        self._stopped = False
        self._ready = deque()
        self._selector = DefaultSelector()
    # 向 _ready 隊(duì)列中加入回調(diào)方法
    def call_soon(self, callback, *args):
        # 將事件添加到隊(duì)列里
        handle = Handle(callback, self, *args)
        self._ready.append(handle)
    # 套接字注冊(cè)讀事件
    def add_writer(self, fd, callback, *args):
        # 將回調(diào)方法封裝成Handle
        handle = Handle(callback, self, *args)
        try:
            # 檢查有沒(méi)有注冊(cè)過(guò)
            key = self._selector.get_key(fd)
        except KeyError:
            # 沒(méi)有注冊(cè)過(guò)，進(jìn)行寫事件注冊(cè)
            self._selector.register(fd, EVENT_WRITE, handle)
        else:
            # 注冊(cè)過(guò)，添加寫事件
            mask = key.events
            self._selector.modify(fd, mask | EVENT_WRITE, handle)
    # 套接字注銷讀事件
    def remove_writer(self, fd):
        try:
            # 檢查有沒(méi)有注冊(cè)過(guò)
            self._selector.get_key(fd)
        except KeyError:
            # 沒(méi)有，直接返回
            return
        # 注銷事件
        self._selector.unregister(fd)
    # 套接字注冊(cè)寫事件
    def add_reader(self, fd, callback, *args):
        # 將回調(diào)方法，封裝成Handle
        handle = Handle(callback, self, *args)
        try:
            # 檢查是否注冊(cè)過(guò)
            key = self._selector.get_key(fd)
        except KeyError:
            # 沒(méi)有注冊(cè)過(guò)，進(jìn)行讀事件注冊(cè)
            self._selector.register(fd, EVENT_READ, handle)
        else:
            # 注冊(cè)過(guò)，添加讀事件
            mask = key.events
            self._selector.modify(fd, mask | EVENT_READ, handle)
    # 套接字注銷寫事件
    def remove_reader(self, fd):
        try:
            # 檢查有沒(méi)有注冊(cè)過(guò)
            self._selector.get_key(fd)
        except KeyError:
            # 沒(méi)有，直接返回
            return
        # 注銷事件
        self._selector.unregister(fd)
    # 處理事件
    def _process_events(self, event_list):
        for key, mask in event_list:
            fileobj, handle = key.fileobj, key.data
            if mask & EVENT_READ:
                self.remove_reader(fileobj)
            if mask & EVENT_WRITE:
                self.remove_writer(fileobj)
            # 注冊(cè)的事件發(fā)生，將注冊(cè)時(shí)的handle放入隊(duì)列中
            self._ready.append(handle)
    # 運(yùn)行事件隊(duì)列，執(zhí)行回調(diào)方法
    def run_once(self):
        # 獲取發(fā)生事件的所有文件描述符
        event_list = self._selector.select()
        self._process_events(event_list)
        while self._ready:
            # 將隊(duì)列中的handle取出，執(zhí)行回調(diào)方法
            handle = self._ready.popleft()
            handle.run()
    # 無(wú)限循環(huán)，直到stopped
    def run_forever(self):
        while True:
            # 執(zhí)行事件
            self.run_once()
            if self._stopped:
                break
    # 執(zhí)行協(xié)程方法，直到事件循環(huán)執(zhí)行結(jié)束
    def run_until_complete(self, future):
        # 在 future 中添加回調(diào)方法，在 future set_value 的時(shí)候執(zhí)行這個(gè)回調(diào)方法
        future.add_done_callback(self._run_until_complete_cb)
        # 開始無(wú)限循環(huán)
        self.run_forever()
        # 返回 future 的值
        return future.value
    # 結(jié)束時(shí)的回調(diào)方法
    def _run_until_complete_cb(self, future):
        # 結(jié)束無(wú)限循環(huán)
        self.close()
    # 關(guān)閉方法
    def close(self):
        self._stopped = True
    # 創(chuàng)建 Future 實(shí)例
    def create_future(self):
        return Future(loop=self)
    # 創(chuàng)建 Task 實(shí)例
    def create_task(self, coro):
        return Task(coro, loop=self)

# Handle 類，對(duì)回調(diào)方法做封裝
class Handle:
    def __init__(self, callback, loop, *args):
        self._callback = callback
        self._args = args
    # 執(zhí)行回調(diào)方法
    def run(self):
        self._callback(*self._args)

# Future 類
class Future:
    def __init__(self, loop=None):
        self.value = None
        # 將要執(zhí)行的回調(diào)方法
        self._step_func = []
        # 和事件循環(huán)關(guān)聯(lián)
        if loop:
            self._loop = loop
        else:
            self._loop = get_event_loop()
    def add_done_callback(self, func):
        # 添加回調(diào)方法
        self._step_func.append(func)
    def set_value(self, value):
        # 設(shè)置值
        self.value = value
        for func in self._step_func:
            # 將回調(diào)方法添加到事件循環(huán)的 _ready 隊(duì)列中
            # 在下次事件循環(huán)中執(zhí)行回調(diào)方法
            self._loop.call_soon(func, self)
    # 實(shí)現(xiàn) __iter__ 方法，F(xiàn)uture 類的實(shí)例為可迭代對(duì)象
    def __iter__(self):
        # 該語(yǔ)句起到暫停協(xié)程的作用，并返回實(shí)例本身
        yield self
        # 該語(yǔ)句定義的返回值會(huì)賦給 yield from 語(yǔ)句等號(hào)前面的變量
        return self.value

# Task 類，繼承 Future 類， 也是可迭代的
class Task(Future):
    def __init__(self, coro, loop=None):
        super().__init__(loop=loop)
        self.coro = coro
        # step 方法直接放入事件循環(huán)的隊(duì)列中進(jìn)行執(zhí)行
        # 激活協(xié)程方法運(yùn)行
        self._loop.call_soon(self.step, self)
    def step(self, future):
        try:
            # 向協(xié)程發(fā)送數(shù)據(jù)，驅(qū)動(dòng)協(xié)程執(zhí)行
            # 直到遇到 yield ，返回新的 Future實(shí)例
            new_futrue = self.coro.send(future.value)
        except StopIteration as exc:
            # 在協(xié)程方法執(zhí)行到結(jié)束，或者 return 之后，拋出 StopIteration 異常
            # 由于 Task 也是 Future， 在協(xié)程執(zhí)行完之后，最后 Task 執(zhí)行自己的回調(diào)方法
            self.set_value(exc.value)
            return
        # 將 step 加入回調(diào)列表，等待下次驅(qū)動(dòng)執(zhí)行
        # 在 Future 執(zhí)行 set_value 時(shí)，又會(huì)執(zhí)行 step 方法，再次驅(qū)動(dòng)協(xié)程執(zhí)行
        new_futrue.add_done_callback(self.step)

class AsyncSocket:
    def __init__(self, loop=None):
        # 綁定事件循環(huán)
        if loop:
            self._loop = loop
        else:
            self._loop = get_event_loop()
        self.sock = socket.socket()
        self.sock.setblocking(False)
    # 該方法用于向服務(wù)器發(fā)送連接請(qǐng)求并注冊(cè)監(jiān)聽套接字的可寫事件
    def connect(self, address):
        # 由 loop 創(chuàng)建 Future 實(shí)例
        f = self._loop.create_future()
        try:
            self.sock.connect(address)
        except BlockingIOError:
            pass
        # 注冊(cè)寫事件，在連接成功事件發(fā)生之后，調(diào)用回調(diào)方法
        self._loop.add_writer(self.sock.fileno(), self._connect_cb, f)
        # 暫停執(zhí)行，等待寫事件發(fā)生
        yield from f
    # 回調(diào)方法，連接事件發(fā)生
    def _connect_cb(self, future):
        # 設(shè)置 Future 值，并且驅(qū)動(dòng)協(xié)程繼續(xù)執(zhí)行
        future.set_value(None)
    # 向服務(wù)器發(fā)送獲取圖片的請(qǐng)求
    def send(self, data):
        self.sock.send(data)
    # 該方法會(huì)多次執(zhí)行，以獲取服務(wù)器返回的數(shù)據(jù)片段
    def read(self):
        f = self._loop.create_future()
        # 注冊(cè)讀事件，在可讀事件發(fā)生之后，調(diào)用回調(diào)方法
        self._loop.add_reader(self.sock.fileno(), self._read_cb, f, self.sock)
        # 暫停執(zhí)行，等待讀事件發(fā)生
        yield from f
        # 返回最后的值
        return f.value
    # 回調(diào)方法，讀事件發(fā)生
    def _read_cb(self, future, sock):
        # 套接字讀取 4096 字節(jié)的數(shù)據(jù)，設(shè)置 Future 值，并且驅(qū)動(dòng)協(xié)程繼續(xù)執(zhí)行
        future.set_value(sock.recv(4096))
    # 讀取所有數(shù)據(jù)
    def read_all(self):
        data = b''
        while True:
            # 不斷讀取 sock 的數(shù)據(jù)
            value = yield from self.read()
            if value:
                data += value
            else:
                return data
    # 關(guān)閉客戶端套接字
    def close(self):
        self.sock.close()

# 爬蟲類
class Crawler:
    def __init__(self, url):
        self._url = url
        self.url = urlparse(url)
        self.response = b''
    def fetch(self):
        self.time = time.time()
        # AsyncSocket 類的實(shí)例對(duì)象負(fù)責(zé)完成數(shù)據(jù)獲取的工作
        sock = AsyncSocket()
        # 向服務(wù)器發(fā)送連接請(qǐng)求，協(xié)程會(huì)暫停到嵌套協(xié)程中的某個(gè) yield from 處
        yield from sock.connect((self.url.netloc, 80))
        data = 'GET {0} HTTP/1.1\r\nHost: {1}\r\nConnection: close\r\n\r\n \
                '.format(self.url.path, self.url.netloc)
        # 發(fā)送請(qǐng)求數(shù)據(jù)
        sock.send(data.encode())
        # 讀取全部的數(shù)據(jù)
        self.response = yield from sock.read_all()
        # 關(guān)閉 socket
        sock.close()
        # 將下載的圖片寫入文件
        with open('pic/{}'.format(self.url.path[1:].split('/')[-1]), 'wb') as f:
            f.write(self.response.split(b'\r\n\r\n')[1])
        return "URL: {0}, 耗時(shí): {1:.3f}s".format(self._url, time.time() - self.time)

# 事件循環(huán)，全局變量
_event_loop = None
# 獲取事件循環(huán)，這里是要獲取同一個(gè)全局實(shí)例
def get_event_loop():
    global _event_loop
    if _event_loop is None:
        # 生成一個(gè)新的事件循環(huán)實(shí)例
        _event_loop = EventLoop()
    return _event_loop
# 收集所有的 task
def gather(tasks, loop=None):
    # 使用 Future 類 收集 所有 tasks 的結(jié)果
    outer = Future(loop=loop)
    # tasks 數(shù)量
    count = len(tasks)
    nfinished = 0
    # 收集結(jié)果
    results = []
    # 回調(diào)方法
    def _gather_cb(f):
        nonlocal nfinished
        # 完成數(shù)量
        nfinished += 1
        # 收集結(jié)果
        results.append(f.value)
        if nfinished == count:
            # 都完成之后，outer 設(shè)置值，同時(shí)執(zhí)行回調(diào)方法
            outer.set_value(results)
    for task in tasks:
        # 所有的 task 都添加回調(diào)方法
        # task 中協(xié)程方法執(zhí)行完成時(shí)， 在 step 方法中會(huì)拋出 StopIteration 異常
        # 這時(shí)候 task 會(huì)執(zhí)行 set_value 方法，同時(shí)會(huì)執(zhí)行 _gather_cb 回調(diào)方法
        task.add_done_callback(_gather_cb)
    # 將 Future 實(shí)例返回
    return outer

def main():
    os.system('mkdir -p pic')
    start = time.time()
    loop = get_event_loop()
    tasks = []
    for url in urls:
        # 爬蟲實(shí)例
        crawler = Crawler(url)
        # 將 fetch 方法封裝成 task 實(shí)例
        tasks.append(Task(crawler.fetch()))        
    # gather 方法將收集所有的 task 結(jié)果，并且返回 Futrue 實(shí)例，
    # Futrue 實(shí)例在 run_until_complete 方法中添加了 _run_until_complete_cb 回調(diào)方法
    # 在所有的 task 執(zhí)行結(jié)束之后，F(xiàn)uture 實(shí)例執(zhí)行 set_value 方法，同時(shí)執(zhí)行回調(diào)方法 _run_until_complete_cb
    # 在 _run_until_complete_cb 方法中執(zhí)行了 close 方法，無(wú)限循環(huán)結(jié)束，整個(gè)流程結(jié)束
    # 返回 results 的值
    results = loop.run_until_complete(gather(tasks))
    for res in results:
        print(res)
if __name__ == '__main__':
    main()