当前位置：首页>前端>正文

带你做一个更好的上榜查询工具

前端2024-04-02 08:31:07

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

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

查看全文

https://www.xamrdz.com/web/2vj1883008.html

架构设计

项目初始化

后端

API

数据采集

前端

主页

结果页

效果展示

结语

相关文章：