利用Python实现Hexo站点的持续集成

Liu Zijian's Blog | 一个技术博客

Liu Zijian · 2024-10-15 · via Liu Zijian's Blog | 一个技术博客

引言

Hexo博客编写完文章需要从头构建并重新上传生成的静态文件到服务器，不能增量更新十分不便，每次上传的文件也越来越大，于是一个比较好的解决办法是在提交文章、资源和配置后，让服务器自动去从git远程仓库拉取最新的提交到服务器上的本地git仓库中，并部署。使用Jenkins等持续集成工具比较繁琐也消耗资源，于是我选择用python编写一个基于flask的webhook脚本来实现。

具体实现流程是：文章提交到git代码托管平台的远程仓库后，远程仓库发送HTTP回调请求给服务器上的webhook脚本，脚本程序根据回调请求的参数，先调用git拉取远程仓库最新提交内容，再去调用npm构建和部署最新版hexo博客到nginx下。

这里的git代码托管平台我选择Gitee：https://gitee.com/

1. 安装pip和python

首先，确保你已经安装了python。如果没有安装，可以使用以下命令来安装python和pip：

1.1 检查python版本

python3 --version

如果你已经安装了python 3.x版本，可以跳过安装python的步骤。否则，继续安装：

1.2 安装python3

sudo yum install python3 -y  # 适用于 CentOS 或其他 RHEL 系统

1.3 安装pip

安装pip的方法：

sudo yum install python3-pip -y  # CentOS/RHEL 系统

安装完成后，确认pip是否已经成功安装：

pip3 --version

2. 使用pip安装依赖

一旦 pip 安装好，你可以使用以下命令来安装需要的库：

pip3 install flask gitpython

3.编写脚本

vim webhook.py

import os
import subprocess
from flask import Flask, request, jsonify
import git

app = Flask(__name__)

# 配置你的本地仓库路径和构建命令
REPO_PATH = "/path/to/your/hexo/blog"
PUBLIC_PATH = os.path.join(REPO_PATH, 'public')

# 拉取代码的函数
def pull_code():
    try:
        repo = git.Repo(REPO_PATH)
        origin = repo.remotes.origin
        origin.pull()
        return True
    except Exception as e:
        print(f"Failed to pull code: {e}")
        return False

# 构建 Hexo 站点的函数
def build_hexo():
    try:
        # 执行 Hexo 命令
        subprocess.run(["npm", "run", "build"], cwd=REPO_PATH, check=True)
        
        return True
    except subprocess.CalledProcessError as e:
        print(f"Failed to build Hexo: {e}")
        return False

@app.route("/webhook", methods=["POST"])
def webhook():
    # 验证请求是否来自 Gitee
    if request.headers.get("X-Gitee-Token") != "": #这里改成你设置的密码
        return jsonify({"message": "Unauthorized"}), 401

    # 获取事件类型，确保是 push 事件
    event = request.headers.get("X-Gitee-Event")
    if event != "Push Hook":
        return jsonify({"message": "Not a push event"}), 400

    # 拉取代码并构建
    if pull_code() and build_hexo():
        return jsonify({"message": "Hexo build success"}), 200
    else:
        return jsonify({"message": "Failed to pull or build"}), 500

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

代码优化，加入线程控制，防止webhook链接被并发调用后，两个hook任务线程同时执行出现安全问题。

import os
import subprocess
from flask import Flask, request, jsonify
import git
import threading

app = Flask(__name__)

# 配置你的本地仓库路径和构建命令
REPO_PATH = "/blog"
PUBLIC_PATH = os.path.join(REPO_PATH, 'public')

lock = threading.Lock()
is_building = False  # 标志位，用于指示是否有任务正在进行

# 拉取代码的函数
def pull_code():
    try:
        repo = git.Repo(REPO_PATH)
        origin = repo.remotes.origin
        origin.pull()
        return True
    except Exception as e:
        print(f"Failed to pull code: {e}")
        return False

# 构建 Hexo 站点的函数
def build_hexo():
    try:
        # 执行 Hexo 的清理和生成命令
        subprocess.run(["npm", "run", "build"], cwd=REPO_PATH, check=True)
        #subprocess.run(["hexo", "generate"], cwd=REPO_PATH, check=True)
        return True
    except subprocess.CalledProcessError as e:
        print(f"Failed to build Hexo: {e}")
        return False

@app.route("/webhook", methods=["POST"])
def webhook():

    global is_building

    # 验证请求是否来自 Gitee
    if request.headers.get("X-Gitee-Token") != "":
        return jsonify({"message": "Unauthorized"}), 401

    # 获取事件类型，确保是 push 事件
    event = request.headers.get("X-Gitee-Event")
    if event != "Push Hook":
        return jsonify({"message": "Not a push event"}), 400

    if is_building:
        return jsonify({"message": "Build in progress, try again later"}), 429

    with lock:
        is_building = True  # 设置标志位为 True，表示任务开始

        try:
            # 拉取代码并构建
            if pull_code() and build_hexo():
                return jsonify({"message": "Hexo build success"}), 200
            else:
                return jsonify({"message": "Failed to pull or build"}), 500

        finally:
            is_building = False  # 重置标志位，表示任务结束

            

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

代码还可以进一步优化，只对master分支提交推送进行触发构建。

import os
import subprocess
from flask import Flask, request, jsonify
import git
import threading

app = Flask(__name__)

# 配置你的本地仓库路径和构建命令
REPO_PATH = "/blog"
PUBLIC_PATH = os.path.join(REPO_PATH, 'public')

lock = threading.Lock()
is_building = False  # 标志位，用于指示是否有任务正在进行

# 拉取代码的函数
def pull_code():
    try:
        repo = git.Repo(REPO_PATH)
        origin = repo.remotes.origin
        origin.pull()
        return True
    except Exception as e:
        print(f"Failed to pull code: {e}")
        return False

# 构建 Hexo 站点的函数
def build_hexo():
    try:
        # 执行 Hexo 的清理和生成命令
        subprocess.run(["npm", "run", "build"], cwd=REPO_PATH, check=True)
        #subprocess.run(["hexo", "generate"], cwd=REPO_PATH, check=True)
        return True
    except subprocess.CalledProcessError as e:
        print(f"Failed to build Hexo: {e}")
        return False

@app.route("/webhook", methods=["POST"])
def webhook():

    global is_building

    # 验证请求是否来自 Gitee
    if request.headers.get("X-Gitee-Token") != "":
        return jsonify({"message": "Unauthorized"}), 401

    # 获取事件类型，确保是 push 事件
    event = request.headers.get("X-Gitee-Event")
    if event != "Push Hook":
        return jsonify({"message": "Not a push event"}), 400

    # 解析推送数据
    payload = request.get_json()
    if not payload:
        return jsonify({"message": "Invalid payload"}), 400

    # 检查是否是 master 分支的推送
    ref = payload.get("ref")
    if ref != "refs/heads/master":
        return jsonify({"message": "Not a master branch push, ignored"}), 200

    if is_building:
        return jsonify({"message": "Build in progress, try again later"}), 429

    with lock:
        is_building = True  # 设置标志位为 True，表示任务开始

        try:
            # 拉取代码并构建
            if pull_code() and build_hexo():
                return jsonify({"message": "Hexo build success"}), 200
            else:
                return jsonify({"message": "Failed to pull or build"}), 500

        finally:
            is_building = False  # 重置标志位，表示任务结束

            

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

4.执行脚本

执行前，服务器还需要安装好node、npm、git并配置好环境变量

nohup python3 webhook.py &

5.配置hook到gitee

设置好签名（密码），设置回调地址，勾选两项

除此外，还要将自己gitee账户相匹配的ssh密钥设置在服务器的上用于拉取我们私有仓库的内容。

此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。原文来自 — 版权归原作者所有。

推荐订阅源

Liu Zijian's Blog | 一个技术博客

引言

1. 安装pip和python

1.1 检查python版本

1.2 安装python3

1.3 安装pip

2. 使用pip安装依赖

3.编写脚本

4.执行脚本

5.配置hook到gitee