<p>看这篇文章前推荐阅读相关的如下文章：</p>    <ol>     <li> <p>使用API登录知乎并获得token</p> </li>     <li> <p>知乎Live全文搜索之模型设计和爬虫实现</p> </li>     <li> <p>知乎Live全文搜索之模型接口</p> </li>    </ol>    <p><strong>抓取话题信息</strong></p>    <p>给新增的Topic提供数据。在parse_live_link中，解析到Live数据中包含了topic的id, 基于这个id拼链接，然后在fetch方法中添加对topic页面的处理，新增parse_topic_link方法：</p>    <pre>  <code class="language-python">TOPIC_API_URL = 'https://api.zhihu.com/topics/{}'  class Crawler:     def __init__(self, max_redirect=10, max_tries=4,                  max_tasks=10, *, loop=None):         ...         self.seen_topics = set()     async def parse_topic_link(self, response):         rs = await response.json()         if response.status == 200:             rs['avatar_url'] = await self.convert_local_image(                 rs['avatar_url'].replace('_s', '_r'))             Topic.add_or_update(**flatten_live_dict(rs, TOPIC_KEYS))     async def parse_live_link(self, response):         ...         topics = live_dict.pop('topics')         for topic in topics:             topic_id = topic['id']             if topic_id not in self.seen_topics:                 self.seen_topics.add(topic_id)                 self.add_url(TOPIC_API_URL.format(topic_id),                              self.max_redirect)         ...     async def fetch(self, url, max_redirect):         try:             if 'api.zhihu.com' in url:                 parse_func = (self.parse_topic_link if 'topics' in url                               else self.parse_live_link)                 next_url = await parse_func(response)             else:                 next_url = await self.parse_zhuanlan_link(response)          ...  </code></pre>    <p>思考下，这是不是一种套路(模式)：</p>    <ol>     <li> <p>初始化一个已完成的url的集合</p> </li>     <li> <p>启动一定量的worker，每个worker都在等待从队列获取要抓取的url</p> </li>     <li> <p>一次性添加足量要抓取的链接到队列中，让每个worker都忙起来(执行前要确认之前没有抓取过)</p> </li>     <li> <p>worker在parse处理中又会解析到新的要抓取的链接，放入队列</p> </li>     <li> <p>worker消费任务，过程中也可能生产任务给自己或者其他worker来消费</p> </li>     <li> <p>全部任务都完成了，通过 self.q.join() 结束</p> </li>     <li> <p>停止这些worker，任务完成</p> </li>    </ol>    <p><strong>修改live灌suggest数据的方式</strong></p>    <p>在上上篇我把相关字段的文本用analyze接口拆分成不同的token成为一个列表赋值给live_suggest，其实完全不需要这样，因为 Completion(analyzer=ik_analyzer) 就是做这个的。gen_suggests用最简单的input+weight就可以：</p>    <pre>  <code class="language-python">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  </code></pre>    <p><strong>下载主讲人头像</strong></p>    <p>小程序开发工具中不能直接使用知乎的图片资源，所以我只能下载下来并生成一个本地的图片地址：</p>    <pre>  <code class="language-python">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'])             ...  </code></pre>    <p>这样User类中的avatar_url最后会变成 static/images/zhihu/v2-4db301967fffa08dfa727ff467170e_s.jpg 这样的地址了。未来我们将让sanic来提供静态资源服务。当然，也可以只存文件名，在接口返回前再去拼图片地址。</p>    <p><strong>抓取专栏信息</strong></p>    <p>知乎Live申请通过之后，主讲人可以写一篇专栏介绍Live，文章中带上Live的链接来导流，知乎Live官方也会收录这个Live的专栏文章。为了让微信小程序的效果更好，我想要抓专栏头图，并且保存专栏链接，希望在小城中能给跳转进去（以证明不可行）。下面我将遍历 知乎Live官方专栏 收录的专栏，解析每个专栏的标题，去ES里面匹配是不是有对应的subject匹配，如果匹配还会验证用户的hash值确保正确，如果没找到还会从Live正文中搜索live的链接的方式来匹配。</p>    <p>看起来很绕，但是没办法啦，因为专栏和live没有明确的匹配关系，目测是知乎2个部门写的不同的系统。</p>    <p>最后要提的是专栏的抓取和live的api不同，它不提供paging键，也就是返回内容中并不告诉你下一页的地址，所以需要我们人工的修改链接，这需要一个转化的函数：</p>    <pre>  <code class="language-python">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)  </code></pre>    <p>这个方法在我实际工作中很常用：</p>    <pre>  <code class="language-python">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'  </code></pre>    <pre>  <code class="language-python">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()  </code></pre>    <p>其中crawl方法中用2次join用来确保 <strong>先抓取全部live信息之后再去抓专栏信息</strong> ，因为得先确保live内容足够完整才能搜索匹配，其次由于parse_live_link和parse_zhuanlan_link都涉及到Live的更新，在并发中容易造成同时更新某些live而触发版本冲突的ConflictError。</p>    <p>我使用 s = s.query(Q('match_phrase', subject=title)) 进行标题匹配，首先我们先聊聊在ES中match和term的区别, 简单的说：</p>    <p><strong>term用于精确查询，match用于全文检索</strong></p>    <p>我们要把标题和Live的subject字段去匹配，但是由于subject设置了analyzer，所以无法使用term。除非新加一个字段，修改成类似cover的那种 Text(index='not_analyzed') 。但是这样新增字段实在有点浪费，用math会把要匹配的标题分词之后挨个去匹配，匹配其中一个或多个的文档就会被搜索出来, 显然不满足「精确」，所以我想到了「短语匹配」(Phrase Matching)。</p>    <p>短语匹配和match查询类似，match_phrase查询首先解析查询字符串产生一个词条列表。但只保留含有了所有搜索词条的文档，并且还要求这些词条的顺序也一致。就是相当于虽然分词了但是词的顺序是有要求的，效果类似于精确匹配。</p>    <p><strong>更新用户举办的Live数量</strong></p>    <p>之前我们给User添加了incr_live_count这个方法，调用一次live_count就+1，由于这个爬虫每次都是重新过一遍，所以需要仅在创建live的时候才更新：</p>    <pre>  <code class="language-python">async def parse_live_link(self, response):     ...     result = await Live.add(**live_dict)                                                                                       if result.meta['version'] == 1:                                                                                                 user.incr_live_count()  </code></pre>    <p>ES每次每次都会返回添加/更新的结果，其中的version字段正好被我们利用。</p>    <p><strong>优化抓取</strong></p>    <p>终于到最后一个小节了。再次道歉，之前分享的爬虫其中有一句检查要抓取的url是不是在self.seen_uls里面的判断，如果已经抓取过就assert抛了异常，这其实造成最后就剩下一个协程在执行任务了。</p>    <p>现在我重构了这部分的内容，大家看代码体会吧：</p>    <pre>  <code class="language-python">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  </code></pre>    <p>这样就既不会重复抓取，也能保证worker都能正常工作。</p>    <p>截止发稿，抓取到的Live 1967个， 话题 656 个 完整抓取一次大概调用约950次API（1967 / 10 + 1967 / 20 + 656）, 在我MacBook上耗时 70 - 90 s。</p>    <p> </p>    <p> </p>    <p>来自：http://mp.weixin.qq.com/s/BWQQSLFYGFMoUgLteTu7tQ</p>    <p> </p>