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

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

使用Certbot自动续签HTTPS证书使用Filebeat采集Nginx日志到ES Python的协程 Python中的异常 Python中的类和对象 Python的函数 Python的数据结构，推导式、迭代器和生成器 Spring AI集成多模态模型 LangChain4j多模态 LangChain Tools工具使用 Python中的模块和包 Python全局环境和虚拟环境（venv） LangChain Prompt提示词工程 LangChain4j Tools工具使用基于Dify搭建AI智能体应用 LangChain4j RAG检索增强生成 Spring AI实现MCP Server Spring AI集成MCP Client LangChain4j Prompt提示词工程 Spring AI使用知识库增强对话功能 Spring AI实现一个智能客服 Spring AI实现一个简单的对话机器人实现MinIO数据的每日备份自己实现一个DNS服务简单理解AI智能体大模型和大模型应用 LangChain开篇 LangChain4j开篇一个解析Excel2007的POI工具类 DataPermissionInterceptor源码解读

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