利用Python实现Hexo站点的持续集成
引言
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密钥设置在服务器的上用于拉取我们私有仓库的内容。
"如果文章对您有帮助,可以请作者喝杯咖啡吗?"

微信支付

支付宝
利用Python实现Hexo站点的持续集成
https://blog.liuzijian.com/post/af1f30e3-c846-650e-9a3f-34e326bf950d.html