昨天看到簡(jiǎn)友們推薦了這樣一個(gè)工具：https://js.zhangxiaocai.cn/，輸入簡(jiǎn)書昵稱就可以查詢上榜歷史。

這次帶大家用不同的技術(shù)棧實(shí)現(xiàn)一下這個(gè)工具，并且做一些優(yōu)化。

架構(gòu)設(shè)計(jì)

練手項(xiàng)目，數(shù)據(jù)量也不太大，架構(gòu)不需要特別認(rèn)真。

• 前端
  • 主頁
  • 結(jié)果展示頁
• 后端
  • API 服務(wù)
  • 定時(shí)采集
  • 數(shù)據(jù)庫

項(xiàng)目初始化

建立項(xiàng)目文件夾 BetterRankSearcher，進(jìn)入文件夾，輸入命令：

初始化版本管理：

git init

初始化依賴管理：

poetry init

添加項(xiàng)目依賴：

poetry add sanic pywebio apscheduler httpx pymongo PyYAML JianshuResearchTools
poetry add flake8 mypy yapf types-PyYAML --dev

在 VS Code 版本管理面板中提交更改。

創(chuàng)建開發(fā)分支并切換：

git branch -c dev
git switch dev

啟用該項(xiàng)目需要用到的擴(kuò)展（我的 VS Code 對(duì)新項(xiàng)目默認(rèn)禁用大部分?jǐn)U展），選擇虛擬環(huán)境中的 Python 解釋器。

重載 VS Code,開發(fā)環(huán)境準(zhǔn)備完成。

后端

API

新建 backend 文件夾，在其中新建 api.py，main.py 作為后端程序的入口點(diǎn)。

在 api.py 文件中導(dǎo)入項(xiàng)目依賴，并初始化一個(gè)藍(lán)圖：

from sanic import Blueprint
from sanic.response import json

api = Blueprint("api", url_prefix="/api")

為了便于測(cè)試，我們寫一個(gè)簡(jiǎn)單的 Hello World 函數(shù)：

@api.get("/hello_world")
async def hello_world_handler(request):
    return json({
        "code": 200,
        "message": "Hello World!"d

之后在 main.py 中創(chuàng)建 App，并將 api 藍(lán)圖綁定上去，在 8081 端口啟動(dòng)服務(wù)：

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 部署服務(wù)，在項(xiàng)目根目錄新建 Dockerfile.backend 文件，寫入以下內(nèi)容：

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 文件，寫入以下內(nèi)容：

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

輸入以下命令，導(dǎo)出項(xiàng)目依賴：

poetry export --output requirements.txt --without-hashes
poetry export --output requirements-dev.txt --without-hashes --dev

在項(xiàng)目根目錄下輸入 docker compose up -d，初次構(gòu)建需要下載依賴，速度較慢。

部署完成后，我們打開網(wǎng)絡(luò)請(qǐng)求工具，輸入 localhost:8081/api/hello_world，即可看到服務(wù)端返回的 JSON 信息。

輸入 docker compose down，下線該服務(wù)。

在 backend 中新建 utils 文件夾，創(chuàng)建 db_manager.py 文件，用于連接數(shù)據(jù)庫。

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

我已經(jīng)從服務(wù)器上下載了排行榜數(shù)據(jù)，并導(dǎo)入到 BRSData 數(shù)據(jù)庫的 data 集合中，共有約三萬條。

接下來我們編寫一個(gè) API Route，返回網(wǎng)頁上的“同步時(shí)間”和“數(shù)據(jù)量”信息。

刪除之前的 Hello World 函數(shù)，向 api.py 寫入以下內(nèi)容：

@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
    })

部署服務(wù)，訪問接口，結(jié)果如下：

{
  "code": 200,
  "newest_data_date": "2022-08-10",
  "data_count": 32600
}

同樣的，我們編寫根據(jù)昵稱查找上榜記錄的接口：

@api.post("/query_record")
async def query_record_handler(request):
    if not request.json:
        return json({
            "code": 400,
            "message": "請(qǐng)求必須帶有 JSON Body"
        })

    body = request.json
    name = body.get("name")

    if not name:
        return json({
            "code": 400,
            "message": "缺少參數(shù)"
        })

    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
    })

這里我們對(duì)數(shù)據(jù)進(jìn)行了以日期為倒序的篩選，同時(shí)限制最大返回的數(shù)據(jù)量為 100 條。

我們需要在容器內(nèi)訪問數(shù)據(jù)庫，在 docker-compose.yml 中定義一個(gè)名為 mongodb 的外部網(wǎng)絡(luò)，并將后端容器連接到這個(gè)網(wǎng)絡(luò)上。

同時(shí)，修改 db_manager.py，將數(shù)據(jù)庫 host 更改為 mongodb。

再次部署，訪問接口，結(jié)果如下：

（數(shù)據(jù)為隨機(jī)選取，有刪減）

{
  "code": 200,
  "data": [
    {
      "date": "2022-06-25",
      "ranking": 7,
      "article_title": "我們一起走過",
      "article_url": "http://www.itdecent.cn/p/91f2cd1bed95",
      "reward_to_author": 533.955,
      "reward_total": 1067.911
    },
    {
      "date": "2022-06-09",
      "ranking": 22,
      "article_title": "單純之年",
      "article_url": "http://www.itdecent.cn/p/2e8f7fded713",
      "reward_to_author": 151.058,
      "reward_total": 302.116
    },
  ]
}

數(shù)據(jù)采集

接下來，我們編寫數(shù)據(jù)自動(dòng)采集模塊，在 backend 文件夾下新建 data_fetcher.py 文件。

這里我們直接在 JFetcher 相關(guān)采集任務(wù)的基礎(chǔ)上修改，將其縮減成單文件。

我們希望采集任務(wù)在每天早上八點(diǎn)自動(dòng)執(zhí)行，并將采集到的數(shù)據(jù)存入數(shù)據(jù)庫中。

修改我們的 main.py 文件，加入采集任務(wù)相關(guān)代碼：

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)

到這里，后端部分開發(fā)完成。

前端

主頁

新建 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 文件，該文件內(nèi)容和 backend 部署文件的唯一區(qū)別是 COPY 語句從 backend 變?yōu)榱?frontend。

向 docker-compose.yml 文件中添加以下內(nèi)容：

  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

創(chuàng)建 index_page.py 和 result_page.py，分別對(duì)應(yīng) index 和 result 頁面，這時(shí)，我們可以將 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)

在主頁面中，我們輸出該頁的標(biāo)題，并創(chuàng)建一個(gè)搜索框，將它的值綁定到 name 變量：

def index():
    """簡(jiǎn)書排行榜搜索
    """
    put_markdown("# 簡(jiǎn)書排行榜搜索")

    put_row([
        put_input("name", placeholder="請(qǐng)輸入簡(jiǎn)書昵稱進(jìn)行搜索"),
        put_button("搜索", color="success", onclick=on_search_button_clicked)
    ], size=r"60% 40%")

在下方輸出一些介紹信息，并通過對(duì)后端 API 的訪問，獲取數(shù)據(jù)更新時(shí)間和數(shù)據(jù)量。

為主頁面的搜索按鈕創(chuàng)建一個(gè)回調(diào)函數(shù)，函數(shù)中獲取 name 的值，與 URL 拼接后跳轉(zhuǎn)到結(jié)果頁。

結(jié)果頁

結(jié)果頁中獲取查詢參數(shù)鍵值對(duì)，對(duì) API 發(fā)起請(qǐng)求，這里我們需要用到一個(gè)映射表：

DATA_HEADER_MAPPING = [
    ("上榜日期", "date"),
    ("排名", "ranking"),
    ("文章", "article_title"),
    ("作者收益", "reward_to_author"),
    ("總收益", "reward_total"),
    ("鏈接", "article_url")
]

該表定義了 API 數(shù)據(jù)和表頭的映射關(guān)系，之后，我們可以通過以下代碼顯示我們的表格：

put_table(
    tdata=data["data"],
    header=DATA_HEADER_MAPPING
)

完成所有代碼編寫后，重新部署程序。

此處有一個(gè)安全問題需要留意：我們需要避免用戶直接訪問 API。

在 Docker 中，位于同一網(wǎng)絡(luò)的容器可以互相訪問，因此，我們將 docker-compose.yml 文件的網(wǎng)絡(luò)定義部分改為如下內(nèi)容：

networks:
  mongodb:
    external: true
  internal:

這樣，我們就定義了一個(gè)名為 internal 的內(nèi)部網(wǎng)絡(luò)，它會(huì)在部署時(shí)被 Docker 自動(dòng)創(chuàng)建。

之后，將應(yīng)用中所有用到 IP 的位置全部替換成服務(wù)名，對(duì)我們來說是 backend。

由于這一邏輯在服務(wù)端進(jìn)行，客戶端將無法看到我們的 API 路徑，只能獲得 PyWebIO 框架的 WebSocket 通信內(nèi)容。

至此，我們用不到三百行代碼實(shí)現(xiàn)了這個(gè)服務(wù)。

效果展示

主頁

查詢結(jié)果頁

Lighthouse 測(cè)試

性能指標(biāo)

（測(cè)試基于本地服務(wù)器進(jìn)行，僅供參考）

結(jié)語

因?yàn)槭蔷毷猪?xiàng)目，代碼自然不會(huì)特別規(guī)范，我也想到了幾個(gè)點(diǎn)需要優(yōu)化：

• 輸入時(shí)實(shí)時(shí)提示匹配項(xiàng)
• 數(shù)據(jù)更新時(shí)間和總數(shù)據(jù)量可以每天刷新一次，無需頻繁請(qǐng)求數(shù)據(jù)庫
• 支持通過個(gè)人主頁鏈接搜索
• 顯示一些統(tǒng)計(jì)信息（一共上榜幾次、最高排名、獲得的總收益）

這個(gè)項(xiàng)目將會(huì)合并到簡(jiǎn)書小工具集中，會(huì)加入更多新功能，簡(jiǎn)書小工具集也會(huì)在近期進(jìn)行一次升級(jí)，對(duì)首頁的用戶體驗(yàn)和性能進(jìn)行優(yōu)化。

本項(xiàng)目在 GitHub 上開源：https://github.com/FHU-yezi/BetterRankSearcher。

同時(shí)對(duì)原服務(wù)的開發(fā)者表示感謝。