站内搜索
经过友人的一再催促,终于打算在端午开始写站内搜索。至于为什么要自己写而不是用别的——谷歌站内搜索没法定制,百度的看都没看,反正百度不录我博客。
一开始以为站内搜索很复杂,我做好了花整整三天的时间来搞的准备,然而Python并不打算给我这个机会(Python大法好),Whoosh+Jieba两个库就满足了我的需求。
按照惯例,首先pip install whoosh jieba
装好两个库。然后可以愉快的写代码了。
索引
索引的建立和常规的ORM很像
import os from jieba.analyse import ChineseAnalyzer from whoosh import fields, index # 使用结巴中文分词 analyzer = ChineseAnalyzer() class Index: def __init__(self): # 定义索引schema,确定索引字段 schema = fields.Schema( path=fields.ID(unique=True, stored=True), title=fields.TEXT(stored=True, analyzer=analyzer), content=fields.TEXT(stored=True, analyzer=analyzer), ) path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'index') # 初始化索引对象 if index.exists_in(path): self.ix = index.open_dir(path) else: if not os.path.exists(path): os.mkdir(path) self.ix = index.create_in(path, schema)
顾名思义,unique
就是指定唯一标识(像是数据库的主键?)。stored=True
约定能够被搜索(如果为False,就不能在结果里显示这个字段了)。analyzer
是用来约定分词器的,默认的分词器是英文的,这里我把它替换成Jieba的默认中文分词了,也可以自定义专业分词库来针对专业内容。
增删改
一开始我是把增和改分开的,然后我心血来潮看了看源码发现
def update_document(self, **fields): # Delete the set of documents matching the unique terms unique_fields = self._unique_fields(fields) if unique_fields: with self.searcher() as s: uniqueterms = [(name, fields[name]) for name in unique_fields] docs = s._find_unique(uniqueterms) for docnum in docs: self.delete_document(docnum) # Add the given fields self.add_document(**fields)
??? 我把你当Update,你就给我delete然后add是吗?
于是最终写成了这样
def update(self, *, path, title, content) -> None: """ 增加对应的文章, 如果对应的path存在,则更新. """ writer = self.ix.writer() writer.update_document(path=path, title=title, content=content) writer.commit() def delete(self, text, fieldname="path") -> None: """ 删除对应的文章, fieldname默认为path """ self.ix.delete_by_term(fieldname, text)
更新文章和删除文章。跟SQL十分像有么有
- 获取游标
writer = self.ix.writer()
- 执行语句
writer.update_document(path=path, title=title, content=content)
- 提交执行
writer.commit()
本来delete也该这样写的,但ix的delete_by_term
方法已经帮我们封装好了这三段,所以直接用就行了。
查
def search(self, *args, strict=False) -> list: """ 默认使用 OR 连接搜索条件. strict为真时使用 AND 连接搜索条件 return: 返回一个列表, 包含所有搜索结果的字典. """ with self.ix.searcher() as searcher: if strict: query = QueryParser("content", self.ix.schema).parse(" AND ".join(args)) else: query = QueryParser("content", self.ix.schema).parse(" OR ".join(args)) results = searcher.search(query) return [{"link":result.fields()["path"], "score":result.score, "title":result.fields()["title"]} for result in results]
查的时候肯定不是只有一个词语来查的,所以我们不写死参数。在参数多于一个的时候,就会自动使用QueryParser的连结字符进行连接并查询。
将查询结果进行for in
遍历,单个结果会有种种方法可以用,在这里我们了解三个常用的。
-
.fields()
fields()
会返回一个包含所有stored=True
的字段的字典。 -
highlight()
顾名思义是高亮击中的搜索词汇 -
.score
.score
是搜索结果的权重,在没有增加其他的判断规则的时候,就是搜索词在文章里的匹配度。
爬虫
爬虫部分让我考虑了挺久的,最后还是选择了最简单通用的方式——sitemap。
我的爬虫策略是定时抓取sitemap,然后与已保存的的sitemap进行比较。如果有新增或者更改就进行抓取并入库。
Web
这个部分是最快的了,因为都二次封装好了,只需要调调API就行了。
from multiprocessing import Process import time from tools.crawler import Crawler from tools.index import Index from sanic import Sanic from sanic.response import json app = Sanic() index = Index() @app.listener('before_server_start') def start_crawler(app, loop): def run(): while True: C = Crawler(index) C.run() time.sleep(3600) P = Process(target=run) P.daemon = True P.start() @app.route("/") async def main(request): try: keywords = request.raw_args.get("keyword").split("+") except AttributeError: return json([], headers={ "Access-Control-Allow-Origin": "*", }) return json(index.search(*keywords), headers={ "Access-Control-Allow-Origin": "*", })
闲言碎语
经我的尝试,这玩意在使用的时候,无论是几个查询,都是占用大约两百Mb的内存。这种内存占用还是可以接受的,我的站内搜索就是架设在1G1h的机子上。有兴趣的可以体验一下 https://abersheeran.com/404.html