知乎Live全文搜索之模型设计和爬虫实现

from sqlalchemy import Column, String, Integer, create_engine, SmallInteger
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

DB_URI = 'sqlite:///user.db'
Base = declarative_base()
engine = create_engine(DB_URI)
Base.metadata.bind = engine
Session = sessionmaker(bind=engine)
session = Session()


class User(Base):
    __tablename__ = 'live_user'

    id = Column(Integer, unique=True, primary_key=True, autoincrement=True)
    speaker_id = Column(String(40), index=True, unique=True)
    name = Column(String(40), index=True, nullable=False)
    gender = Column(SmallInteger, default=2)
    headline = Column(String(200))
    avatar_url = Column(String(100), nullable=False)
    bio = Column(String(200))
    description = Column(String())

    @classmethod
    def add(cls, **kwargs):
        speaker_id = kwargs.get('speaker_id', None)
        if id is not None:
            r = session.query(cls).filter_by(speaker_id=speaker_id).first()
            if r:
                return r
        try:
            r = cls(**kwargs)
            session.add(r)
            session.commit()
        except:
            session.rollback()
            raise
        else:
            return r

Base.metadata.create_all()

In : from elasticsearch_dsl.query import Q

In : Q('multi_match', subject='python').to_dict()
Out: {'multi_match': {'subject': 'python'}}

In : from elasticsearch import Elasticsearch
In : from elasticsearch_dsl import Search, Q

In : s = Search(using=client, index='live')
In : s = s.query('match', subject='python').query(~Q('match', description='量化'))
In : s.execute()
Out: <Response: [<Hit(live/live/789840559912009728): {'subject': 'Python 工程师的入门和进阶', 'feedback_score': 4.5, 'stat...}>]>

from elasticsearch_dsl import DocType, Date, Integer, Text, Float, Boolean
from elasticsearch_dsl.connections import connections
from elasticsearch_dsl.query import SF, Q

from config import SEARCH_FIELDS
from .speaker import User, session

connections.create_connection(hosts=['localhost'])


class Live(DocType):
    id = Integer()
    speaker_id = Integer()
    feedback_score = Float() # 评分
    topic_names = Text(analyzer='ik_max_word')  # 话题标签名字
    seats_taken = Integer()  # 参与人数
    subject = Text(analyzer='ik_max_word')  # 标题
    amount = Float()  # 价格(RMB)
    description = Text(analyzer='ik_max_word')
    status = Boolean()  # public(True)/ended(False)
    starts_at = Date()
    outline = Text(analyzer='ik_max_word')  # Live内容
    speaker_message_count = Integer()
    tag_names = Text(analyzer='ik_max_word')
    liked_num = Integer()

    class Meta:
        index = 'live'

    @classmethod
    def add(cls, **kwargs):
        id = kwargs.pop('id', None)
        if id is None:
            return False
        live = cls(meta={'id': id}, **kwargs)
        live.save()
        return live

...
from models import User, Live, session
from client import ZhihuClient
from utils import flatten_live_dict
from config import SPEAKER_KEYS, LIVE_KEYS

LIVE_API_URL = 'https://api.zhihu.com/lives/{type}?purchasable=0&limit=10&offset={offset}'  # noqa
LIVE_TYPE = frozenset(['ongoing', 'ended'])


class Crawler:
    def __init__(self, max_redirect=10, max_tries=4,
                 max_tasks=10, *, loop=None):
        ...
        for t in LIVE_TYPE:
            for offset in range(self.max_tasks):
                self.add_url(LIVE_API_URL.format(type=t, offset=offset * 10))

        client = ZhihuClient()
        self.headers = {}
        client.auth(self)

paging: {
    is_end: false,
    next: "https://api.zhihu.com/lives/ended?purchasable=0&limit=10&offset=20",
    previous: "https://api.zhihu.com/lives/ended?purchasable=0&limit=10&offset=0"
}

async def parse_link(self, response):
        rs = await response.json()

        if response.status == 200:
            for live in rs['data']:
                speaker = live.pop('speaker')
                speaker_id = speaker['member']['id']
                user = User.add(speaker_id=speaker_id,
                                **flatten_live_dict(speaker, SPEAKER_KEYS))
                live_dict = flatten_live_dict(live, LIVE_KEYS)
                if live_dict['id'] == LAST_INSERT_ID:
                    self._stopped = True
                    return
                live_dict['speaker_id'] = user.id
                live_dict['topic_names'] = ' '.join(
                    [t['name'] for t in live_dict.pop('topics')])
                live_dict['seats_taken'] = live_dict.pop('seats')['taken']
                live_dict['amount'] = live_dict.pop('fee')['amount'] / 100
                live_dict['status'] = live_dict['status'] == 'public'
                live_dict['tag_names'] = ' '.join(
                    set(sum([(t['name'], t['short_name'])
                         for t in live_dict.pop('tags')], ())))
                live_dict['starts_at'] = datetime.fromtimestamp(
                    live_dict['starts_at'])
                Live.add(**live_dict)

            paging = rs['paging']
            if not paging['is_end']:
                next_url = paging['next']
                return paging['next']
        else:
            print('HTTP status_code is {}'.format(response.status))

LIVE_KEYS = ['id', 'feedback_score', 'seats', 'subject', 'fee',
             'description', 'status', 'starts_at', 'outline',
             'speaker_message_count', 'liked_num', 'tags', 'topics']
SEARCH_FIELDS = ['subject', 'description', 'outline', 'tag_names',
                 'topic_names']

...
search = Live.search()
if search.count():
    LAST_INSERT_ID = search.sort('starts_at').execute()[0]._id
else:
    LAST_INSERT_ID = 0


class Crawler:
    def __init__(self, max_redirect=10, max_tries=4,
                 max_tasks=10, *, loop=None): 
        ...
        self._stopped = False

    async def parse_link(self, response):
        rs = await response.json()

        if response.status == 200:
            if live_dict['id'] == LAST_INSERT_ID:
                self._stopped = True
                return 
            ...

   def add_url(self, url, max_redirect=None):
        if max_redirect is None:
            max_redirect = self.max_redirect
        if url not in self.seen_urls and not self._stopped:
            self.seen_urls.add(url)
            self.q.put_nowait((url, max_redirect))