知乎Live全文搜索之完成爬虫

def gen_suggests(topics, tags, outline, username, subject):
    suggests = [{'input': item, 'weight': weight}
                for item, weight in ((topics, 10), (subject, 5), (outline, 3),
                                     (tags, 3), (username, 2)) if item]
    return suggests

import os

IMAGE_FOLDER = 'static/images/zhihu'
if not os.path.exists(IMAGE_FOLDER):
    os.mkdir(IMAGE_FOLDER) 

class Crawler:
    ...
    async def convert_local_image(self, pic):
        pic_name = pic.split('/')[-1]
        path = os.path.join(IMAGE_FOLDER, pic_name)
        if not os.path.exists(path):
            async with self.session.get(pic) as resp:
                content = await resp.read()
                with open(path, 'wb') as f:
                    f.write(content)
        return path

    async def parse_live_link(self, response):
        ...
        for live in rs['data']:
            ...
            speaker = live.pop('speaker')
            speaker_id = speaker['member']['id']
            speaker['member']['avatar_url'] = await self.convert_local_image(  # noqa
                    speaker['member']['avatar_url'])
            ...

from urllib.parse import urlparse, parse_qsl, urlunparse, urlencode

def get_next_url(url):
    url_parts = list(urlparse(url))
    query = dict(parse_qsl(url_parts[4]))
    query['offset'] = int(query['offset']) + int(query['limit'])
    url_parts[4] = urlencode(query)
    return urlunparse(url_parts)

In : get_next_url('http://dongwm.com?offset=10&limit=20')
Out: 'http://dongwm.com?offset=30&limit=20'

In : get_next_url('http://dongwm.com?offset=20&limit=30')
Out: 'http://dongwm.com?offset=50&limit=30'

ZHUANLAN_API_URL = 'https://zhuanlan.zhihu.com/api/columns/zhihulive/posts?limit=20&offset={offset}'
LIVE_REGEX = re.compile(r'<a href="https://(www.)?zhihu.com/lives/(\d+)(.*)?"')  # noqa


class Crawler:
    def __init__(self, max_redirect=10, max_tries=4,
                 max_tasks=10, *, loop=None):
        ...
        self.seen_zhuanlan = set()
    ...
    async def parse_zhuanlan_link(self, response):
        posts = await response.json()

        if response.status == 200 and posts:
            for post in posts:
                cover = post['titleImage']
                if not cover:
                    continue
                s = Live.search()
                title = post['title']
                for sep in ('－', '—'):
                    if sep in title:
                        title = title.split(sep)[-1].strip()
                speaker_id = post['author']['hash']
                zid = post['url'].split('/')[-1]
                s = s.query(Q('match_phrase', subject=title))
                lives = await s.execute()
                for live in lives:
                    if live.speaker and live.speaker.speaker_id == speaker_id:
                        await self.update_live(zid, cover, live)
                        break
                else:
                    match = LIVE_REGEX.search(post['content'])
                    if match:
                        live_id = match.group(2)
                        try:
                            live = await Live.get(live_id)
                        except NotFoundError:
                            pass
                        else:
                            await self.update_live(zid, cover, live)

            return get_next_url(response.url)

    async def update_live(self, zid, cover, live):
        if live.id in self.seen_zhuanlan:
            return
        self.seen_zhuanlan.add(live.id)
        zhuanlan_url = ZHUANLAN_URL.format(zid)
        cover = await self.convert_local_image(cover)
        await live.update(cover=cover, zhuanlan_url=zhuanlan_url)

    def add_zhuanlan_urls(self):
        for offset in range(self.max_tasks):
            self.add_url(ZHUANLAN_API_URL.format(offset=offset * 20)) 

    async def crawl(self):
        self.__workers = [asyncio.Task(self.work(), loop=self.loop)
                          for _ in range(self.max_tasks)]
        self.t0 = time.time()
        await self.q.join()
        self.add_zhuanlan_urls()
        await self.q.join()
        self.t1 = time.time()
        for w in self.__workers:
            w.cancel()

class ZhihuClient: 
    def refresh_token(self):
        data = LOGIN_DATA.copy()
        data['grant_type'] = 'refresh_token'
        data['refresh_token'] = self._token.refresh_token
        gen_signature(data)
        auth = ZhihuOAuth(self._token)
        self.save_token(auth, data)

    def save_token(self, auth, data):
        res = self._session.post(LOGIN_URL, auth=auth, data=data)
        try:
            json_dict = res.json()
            if 'error' in json_dict:
                raise LoginException(json_dict['error']['message'])
            self._token = ZhihuToken.from_dict(json_dict)
        except (ValueError, KeyError) as e:
            raise LoginException(str(e))
        else:
            ZhihuToken.save_file(self.token_file, json_dict)

class Crawler:
    ...
    async def check_token(self):
        async with self.session.get(
                LIVE_API_URL.format(type='ended', offset=0)) as resp:
            if resp.status == 401:
                self.client.refresh_token()

    async def crawl(self):
        await self.check_token()
        ...

PS: 今天试用好像不对

async def parse_live_link(self, response):
    ...
    result = await Live.add(**live_dict)                                                                                   
    if result.meta['version'] == 1:                                                                                        
        user.incr_live_count()

class Crawler:
    def __init__(self, max_redirect=10, max_tries=4,
                 max_tasks=10, *, loop=None):
        self.__stopped = {}.fromkeys(['ended', 'ongoing', 'posts'], False) 

    async def fetch(self, url, max_redirect):
        ...
        if next_url is not None:
            self.add_url(next_url, max_redirect)
        else:
            # 如果live或者知乎官方专栏接口不再返回下一页，这个类型的任务就停止
            for type in self.__stopped:
                if type in url:
                    self.__stopped[type] = True

    async def work(self):
        try:
            while 1:
                url, max_redirect = await self.q.get()
                if url in self.seen_urls:
                    type = url.split('/')[-1].split('?')[0]
                    # 如果一个接口返回的next_url已经抓去过，自动添加next_url的下一页
                    # 除非这个类型的任务停止状态为True
                    if not type.isdigit() and not self.__stopped[type]:
                        self.add_url(get_next_url(url), max_redirect)
                await self.fetch(url, max_redirect)
                self.q.task_done()
                asyncio.sleep(1)
        except asyncio.CancelledError:
            pass