昨天看到简友们推荐了这样一个工具：https://js.zhangxiaocai.cn/，输入简书昵称就可以查询上榜历史。
这次带大家用不同的技术栈实现一下这个工具，并且做一些优化。
架构设计
练手项目，数据量也不太大，架构不需要特别认真。
- 前端
- 主页
- 结果展示页
- 后端
- API 服务
- 定时采集
- 数据库
项目初始化
建立项目文件夹 BetterRankSearcher，进入文件夹，输入命令：
初始化版本管理：
git init
初始化依赖管理：
poetry init
添加项目依赖：
poetry add sanic pywebio apscheduler httpx pymongo PyYAML JianshuResearchTools
poetry add flake8 mypy yapf types-PyYAML --dev
在 VS Code 版本管理面板中提交更改。
创建开发分支并切换：
git branch -c dev
git switch dev
启用该项目需要用到的扩展（我的 VS Code 对新项目默认禁用大部分扩展），选择虚拟环境中的 Python 解释器。
重载 VS Code,开发环境准备完成。
后端
API
新建 backend 文件夹，在其中新建 api.py，main.py 作为后端程序的入口点。
在 api.py 文件中导入项目依赖，并初始化一个蓝图：
from sanic import Blueprint
from sanic.response import json
api = Blueprint("api", url_prefix="/api")
为了便于测试，我们写一个简单的 Hello World 函数：
@api.get("/hello_world")
async def hello_world_handler(request):
return json({
"code": 200,
"message": "Hello World!"d
之后在 main.py 中创建 App，并将 api 蓝图绑定上去，在 8081 端口启动服务：
from sanic import Sanic
from api import api
app = Sanic(__name__)
app.blueprint(api)
app.run(host="0.0.0.0", port=8081, access_log=False)
我们希望使用 Docker 部署服务，在项目根目录新建 Dockerfile.backend 文件，写入以下内容：
FROM python:3.8.10-slim
ENV TZ Asia/Shanghai
WORKDIR /app
COPY requirements.txt .
RUN pip install \
-r requirements.txt \
--no-cache-dir \
--quiet \
-i https://mirrors.aliyun.com/pypi/simple
COPY backend .
CMD ["python", "main.py"]
之后新建 docker-compose.yml 文件，写入以下内容：
version: "3"
services:
backend:
image: betterranksearcher-backend:0.1.0
build:
dockerfile: Dockerfile.backend
ports:
- "8081:8081"
environment:
- PYTHONUNBUFFERED=1
deploy:
resources:
limits:
cpus: "0.5"
memory: 256M
restart_policy:
condition: on-failure
delay: 5s
max_attempts: 3
stop_grace_period: 1s
输入以下命令，导出项目依赖：
poetry export --output requirements.txt --without-hashes
poetry export --output requirements-dev.txt --without-hashes --dev
在项目根目录下输入 docker compose up -d，初次构建需要下载依赖，速度较慢。
部署完成后，我们打开网络请求工具，输入 localhost:8081/api/hello_world，即可看到服务端返回的 JSON 信息。
输入 docker compose down，下线该服务。
在 backend 中新建 utils 文件夹，创建 db_manager.py 文件，用于连接数据库。
from pymongo import MongoClient
def init_DB():
connection: MongoClient = MongoClient(
"127.0.0.1", 27017
)
db = connection.BRSData
return db
db = init_DB()
data = db.data
我已经从服务器上下载了排行榜数据，并导入到 BRSData 数据库的 data 集合中，共有约三万条。
接下来我们编写一个 API Route，返回网页上的“同步时间”和“数据量”信息。
删除之前的 Hello World 函数，向 api.py 写入以下内容：
@api.post("/data_info")
async def data_info_handler(request):
newest_data_date = list(
data.find({}, {"_id": 0, "date": 1})
.sort("date", -1)
.limit(1)
)[0]["date"]
newest_data_date = str(newest_data_date).split()[0]
data_count = data.count_documents({})
return json({
"code": 200,
"newest_data_date": newest_data_date,
"data_count": data_count
})
部署服务，访问接口，结果如下：
{
"code": 200,
"newest_data_date": "2022-08-10",
"data_count": 32600
}
同样的，我们编写根据昵称查找上榜记录的接口：
@api.post("/query_record")
async def query_record_handler(request):
if not request.json:
return json({
"code": 400,
"message": "请求必须带有 JSON Body"
})
body = request.json
name = body.get("name")
if not name:
return json({
"code": 400,
"message": "缺少参数"
})
if data.count_documents({"author.name": name}) == 0:
return json({
"code": 400,
"message": "用户不存在或无上榜记录"
})
data_list = []
for item in data.find({"author.name": name}).sort("date", -1).limit(100):
data_list.append({
"date": str(item["date"]).split()[0],
"ranking": item["ranking"],
"article_title": item["article"]["title"],
"article_url": item["article"]["url"],
"reward_to_author": item["reward"]["to_author"],
"reward_total": item["reward"]["total"],
})
return json({
"code": 200,
"data": data_list
})
这里我们对数据进行了以日期为倒序的筛选，同时限制最大返回的数据量为 100 条。
我们需要在容器内访问数据库，在 docker-compose.yml 中定义一个名为 mongodb 的外部网络，并将后端容器连接到这个网络上。
同时，修改 db_manager.py，将数据库 host 更改为 mongodb。
再次部署，访问接口，结果如下：
（数据为随机选取，有删减）
{
"code": 200,
"data": [
{
"date": "2022-06-25",
"ranking": 7,
"article_title": "我们一起走过",
"article_url": "https://www.jianshu.com/p/91f2cd1bed95",
"reward_to_author": 533.955,
"reward_total": 1067.911
},
{
"date": "2022-06-09",
"ranking": 22,
"article_title": "单纯之年",
"article_url": "https://www.jianshu.com/p/2e8f7fded713",
"reward_to_author": 151.058,
"reward_total": 302.116
},
]
}
数据采集
接下来，我们编写数据自动采集模块，在 backend 文件夹下新建 data_fetcher.py 文件。
这里我们直接在 JFetcher 相关采集任务的基础上修改，将其缩减成单文件。
我们希望采集任务在每天早上八点自动执行，并将采集到的数据存入数据库中。
修改我们的 main.py 文件，加入采集任务相关代码：
from apscheduler.schedulers.background import BackgroundScheduler
from sanic import Sanic
from api import api
from data_fetcher import main_fetcher
from utils.cron_helper import CronToKwargs
scheduler = BackgroundScheduler()
scheduler.add_job(main_fetcher, "cron", **CronToKwargs("0 0 8 1/1 * *"))
scheduler.start()
app = Sanic(__name__)
app.blueprint(api)
app.run(host="0.0.0.0", port=8081, access_log=False)
到这里，后端部分开发完成。
前端
主页
新建 frontend 文件夹，在 main.py 中写入以下代码：
from pywebio import start_server
from pywebio.output import put_text
def index():
put_text("Hello World!")
start_server([index], host="0.0.0.0", port=8080)
新建 Dockerfile.frontend 文件，该文件内容和 backend 部署文件的唯一区别是 COPY 语句从 backend 变为了 frontend。
向 docker-compose.yml 文件中添加以下内容：
frontend:
image: betterranksearcher-frontend:0.1.0
build:
dockerfile: Dockerfile.frontend
ports:
- "8080:8080"
environment:
- PYTHONUNBUFFERED=1
deploy:
resources:
limits:
cpus: "0.5"
memory: 256M
restart_policy:
condition: on-failure
delay: 5s
max_attempts: 3
stop_grace_period: 1s
创建 index_page.py 和 result_page.py，分别对应 index 和 result 页面，这时，我们可以将 main.py 改写成这样：
from pywebio import start_server
from index_page import index
from result_page import result
start_server([index, result], host="0.0.0.0", port=8080)
在主页面中，我们输出该页的标题，并创建一个搜索框，将它的值绑定到 name 变量：
def index():
"""简书排行榜搜索
"""
put_markdown("# 简书排行榜搜索")
put_row([
put_input("name", placeholder="请输入简书昵称进行搜索"),
put_button("搜索", color="success", onclick=on_search_button_clicked)
], size=r"60% 40%")
在下方输出一些介绍信息，并通过对后端 API 的访问，获取数据更新时间和数据量。
为主页面的搜索按钮创建一个回调函数，函数中获取 name 的值，与 URL 拼接后跳转到结果页。
结果页
结果页中获取查询参数键值对，对 API 发起请求，这里我们需要用到一个映射表：
DATA_HEADER_MAPPING = [
("上榜日期", "date"),
("排名", "ranking"),
("文章", "article_title"),
("作者收益", "reward_to_author"),
("总收益", "reward_total"),
("链接", "article_url")
]
该表定义了 API 数据和表头的映射关系，之后，我们可以通过以下代码显示我们的表格：
put_table(
tdata=data["data"],
header=DATA_HEADER_MAPPING
)
完成所有代码编写后，重新部署程序。
此处有一个安全问题需要留意：我们需要避免用户直接访问 API。
在 Docker 中，位于同一网络的容器可以互相访问，因此，我们将 docker-compose.yml 文件的网络定义部分改为如下内容：
networks:
mongodb:
external: true
internal:
这样，我们就定义了一个名为 internal 的内部网络，它会在部署时被 Docker 自动创建。
之后，将应用中所有用到 IP 的位置全部替换成服务名，对我们来说是 backend。
由于这一逻辑在服务端进行，客户端将无法看到我们的 API 路径，只能获得 PyWebIO 框架的 WebSocket 通信内容。
至此，我们用不到三百行代码实现了这个服务。
效果展示
（测试基于本地服务器进行，仅供参考）
结语
因为是练手项目，代码自然不会特别规范，我也想到了几个点需要优化：
- 输入时实时提示匹配项
- 数据更新时间和总数据量可以每天刷新一次，无需频繁请求数据库
- 支持通过个人主页链接搜索
- 显示一些统计信息（一共上榜几次、最高排名、获得的总收益）
这个项目将会合并到简书小工具集中，会加入更多新功能，简书小工具集也会在近期进行一次升级，对首页的用户体验和性能进行优化。
本项目在 GitHub 上开源：https://github.com/FHU-yezi/BetterRankSearcher。
同时对原服务的开发者表示感谢。