昨天看到简友们推荐了这样一个工具：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 通信内容。

至此，我们用不到三百行代码实现了这个服务。

效果展示

主页 查询结果页 Lighthouse 测试 性能指标

（测试基于本地服务器进行，仅供参考）

结语

因为是练手项目，代码自然不会特别规范，我也想到了几个点需要优化：

• 输入时实时提示匹配项
• 数据更新时间和总数据量可以每天刷新一次，无需频繁请求数据库
• 支持通过个人主页链接搜索
• 显示一些统计信息（一共上榜几次、最高排名、获得的总收益）

这个项目将会合并到简书小工具集中，会加入更多新功能，简书小工具集也会在近期进行一次升级，对首页的用户体验和性能进行优化。

本项目在 GitHub 上开源：https://github.com/FHU-yezi/BetterRankSearcher。

同时对原服务的开发者表示感谢。