引言

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 osimport subprocessfrom flask import Flask, request, jsonifyimport gitapp = 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"}), 500if __name__ == "__main__":    app.run(host="0.0.0.0", port=5000)

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

import osimport subprocessfrom flask import Flask, request, jsonifyimport gitimport threadingapp = 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 osimport subprocessfrom flask import Flask, request, jsonifyimport gitimport threadingapp = 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密钥设置在服务器的上用于拉取我们私有仓库的内容。

对象有一本做数独的书，里面满满的都是题目，然而，看到这前三关直接不淡定了，这尼玛是人能做出来的。

试了半天无过之后，果断还是决定上代码了。

就这么个破玩意儿，愣是跑了十几分钟才出来。

import tkinter as tk
from tkinter import ttk, messagebox, filedialog
import threading
import queue
import time
import json
import copy
import concurrent.futures
import os
import multiprocessing

# --- 数独核心逻辑 ---
def is_valid(board, row, col, num):
    for i in range(9):
        if board[row][i] == num or board[i][col] == num:
            return False
    start_row, start_col = 3 * (row // 3), 3 * (col // 3)
    for i in range(3):
        for j in range(3):
            if board[start_row + i][start_col + j] == num:
                return False
    return True

def solve_sudoku_all_with_progress(board, branch_id, progress_queue):
    solutions = []
    visited = [0]
    def backtrack(bd, pos=0):
        if pos == 81:
            solutions.append(copy.deepcopy(bd))
            return
        row, col = divmod(pos, 9)
        if bd[row][col] != 0:
            backtrack(bd, pos + 1)
        else:
            for num in range(1, 10):
                if is_valid(bd, row, col, num):
                    bd[row][col] = num
                    visited[0] += 1
                    if visited[0] % 1000 == 0:
                        progress_queue.put((branch_id, visited[0]))
                    backtrack(bd, pos + 1)
                    bd[row][col] = 0
    backtrack(board)
    # 结束时推送最终节点数
    progress_queue.put((branch_id, visited[0]))
    return solutions, visited[0]

def worker(new_board, branch_id, progress_queue):
    return solve_sudoku_all_with_progress(new_board, branch_id, progress_queue)

class SudokuGUI:
    def __init__(self, root):
        self.root = root
        self.root.title('多进程数独求解器')
        self.root.geometry('520x650')
        self.root.resizable(False, False)
        self.entries = [[None for _ in range(9)] for _ in range(9)]
        self._build_ui()
        self.progress_queue = queue.Queue()
        self._poll_progress()

    def _build_ui(self):
        self.root.configure(bg='#f7f7fa')
        style = ttk.Style()
        style.theme_use('clam')
        style.configure('TButton', font=('Segoe UI', 12), padding=6)
        style.configure('TLabel', font=('Segoe UI', 11), background='#f7f7fa')
        style.configure('TProgressbar', thickness=18, troughcolor='#e0e0e0', background='#4caf50')
        title = ttk.Label(self.root, text='多进程数独求解器', font=('Segoe UI', 22, 'bold'))
        title.pack(pady=(18, 8))
        frame = tk.Frame(self.root, bg='#f7f7fa')
        frame.pack(pady=8)
        for i in range(9):
            for j in range(9):
                e = tk.Entry(frame, width=2, font=('Consolas', 22), justify='center', bd=2, relief='ridge', bg='#fff', fg='#222')
                e.grid(row=i, column=j, padx=(2 if j % 3 == 0 else 0, 2), pady=(2 if i % 3 == 0 else 0, 2))
                self.entries[i][j] = e
        btn_frame = tk.Frame(self.root, bg='#f7f7fa')
        btn_frame.pack(pady=10)
        ttk.Button(btn_frame, text='求解所有解', command=self.solve).grid(row=0, column=0, padx=8)
        ttk.Button(btn_frame, text='清空', command=self.clear).grid(row=0, column=1, padx=8)
        ttk.Button(btn_frame, text='保存', command=self.save).grid(row=0, column=2, padx=8)
        ttk.Button(btn_frame, text='加载', command=self.load).grid(row=0, column=3, padx=8)
        self.progress_var = tk.DoubleVar()
        self.progress_bar = ttk.Progressbar(self.root, variable=self.progress_var, maximum=1, length=400)
        self.progress_bar.pack(pady=(10, 2))
        self.progress_label = ttk.Label(self.root, text='进度: 0/0  节点: 0  速度: -- 节点/秒')
        self.progress_label.pack()
        self.status_label = ttk.Label(self.root, text='', font=('Segoe UI', 10))
        self.status_label.pack(pady=(8, 0))
        self.thread_label = ttk.Label(self.root, text='线程状态: 空闲', font=('Segoe UI', 10))
        self.thread_label.pack(pady=(2, 0))

    def get_board(self):
        board = []
        for i in range(9):
            row = []
            for j in range(9):
                val = self.entries[i][j].get()
                if val == '':
                    row.append(0)
                else:
                    try:
                        num = int(val)
                        if 1 <= num <= 9:
                            row.append(num)
                        else:
                            raise ValueError
                    except ValueError:
                        messagebox.showerror('输入错误', f'第{i+1}行第{j+1}列输入无效')
                        return None
            board.append(row)
        return board

    def clear(self):
        for i in range(9):
            for j in range(9):
                self.entries[i][j].delete(0, tk.END)
        self.progress_var.set(0)
        self.progress_label.config(text='进度: 0/0  节点: 0  速度: -- 节点/秒')
        self.status_label.config(text='')
        self.thread_label.config(text='线程状态: 空闲')

    def save(self):
        data = [[self.entries[i][j].get() for j in range(9)] for i in range(9)]
        file_path = filedialog.asksaveasfilename(defaultextension='.json', filetypes=[('JSON文件', '*.json')])
        if file_path:
            with open(file_path, 'w', encoding='utf-8') as f:
                json.dump(data, f, ensure_ascii=False)
            self.status_label.config(text='已保存')

    def load(self):
        file_path = filedialog.askopenfilename(defaultextension='.json', filetypes=[('JSON文件', '*.json')])
        if file_path:
            with open(file_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
            for i in range(9):
                for j in range(9):
                    self.entries[i][j].delete(0, tk.END)
                    if data[i][j]:
                        self.entries[i][j].insert(0, data[i][j])
            self.status_label.config(text='已加载')

    def solve(self):
        board = self.get_board()
        if board is None:
            return
        self.progress_var.set(0)
        self.progress_label.config(text='进度: 0/0  节点: 0  速度: -- 节点/秒')
        self.status_label.config(text='正在求解...')
        self.thread_label.config(text='线程状态: 启动中...')
        self.disable_inputs()
        self._parallel_solve(board)

    def disable_inputs(self):
        for i in range(9):
            for j in range(9):
                self.entries[i][j].config(state='disabled')

    def enable_inputs(self):
        for i in range(9):
            for j in range(9):
                self.entries[i][j].config(state='normal')

    def _parallel_solve(self, board):
        # 找到第一个空格
        first_empty = None
        for i in range(9):
            for j in range(9):
                if board[i][j] == 0:
                    first_empty = (i, j)
                    break
            if first_empty:
                break
        if not first_empty:
            self.root.after(0, lambda: self.thread_label.config(text='线程状态: 空闲'))
            self.on_solve_done([copy.deepcopy(board)])
            return
        i, j = first_empty
        candidates = [num for num in range(1, 10) if is_valid(board, i, j, num)]
        total_tasks = len(candidates)
        finished = [0]
        all_solutions = []
        all_visited = [0]
        branch_nodes = {idx: 0 for idx in range(total_tasks)}
        start_time = time.time()
        self.root.after(0, lambda: self.thread_label.config(text='线程状态: 运行中 (多进程)'))
        # 用multiprocessing.Manager().Queue()跨进程通信
        from multiprocessing import Manager
        mp_manager = Manager()
        mp_queue = mp_manager.Queue()
        def manager():
            try:
                with concurrent.futures.ProcessPoolExecutor(max_workers=os.cpu_count()) as executor:
                    futures = []
                    for idx, num in enumerate(candidates):
                        new_board = copy.deepcopy(board)
                        new_board[i][j] = num
                        futures.append(executor.submit(worker, new_board, idx, mp_queue))
                    # 进度监听线程
                    def progress_listener():
                        while finished[0] < total_tasks:
                            try:
                                branch_id, nodes = mp_queue.get(timeout=0.1)
                                branch_nodes[branch_id] = nodes
                                all_visited[0] = sum(branch_nodes.values())
                                elapsed = time.time() - start_time
                                speed = all_visited[0] / elapsed if elapsed > 0 else 0
                                self.progress_queue.put((finished[0], total_tasks, speed, all_visited[0]))
                            except Exception:
                                pass
                    listener_thread = threading.Thread(target=progress_listener, daemon=True)
                    listener_thread.start()
                    for idx, fut in enumerate(futures):
                        result, visited = fut.result()
                        all_solutions.extend(result)
                        finished[0] += 1
                        branch_nodes[idx] = visited
                        all_visited[0] = sum(branch_nodes.values())
                        elapsed = time.time() - start_time
                        speed = all_visited[0] / elapsed if elapsed > 0 else 0
                        self.progress_queue.put((finished[0], total_tasks, speed, all_visited[0]))
                    listener_thread.join(timeout=0.1)
                self.root.after(0, lambda: self.thread_label.config(text='线程状态: 已完成 (多进程)'))
                self.root.after(0, lambda: self.on_solve_done(all_solutions))
            except Exception as e:
                import traceback
                traceback.print_exc()
                self.root.after(0, lambda e=e: messagebox.showerror('主进程异常', str(e)))
        threading.Thread(target=manager, daemon=True).start()

    def _poll_progress(self):
        try:
            while True:
                v, t, s, nodes = self.progress_queue.get_nowait()
                self._update_progress(v, t, s, nodes)
        except queue.Empty:
            pass
        self.root.after(50, self._poll_progress)

    def _update_progress(self, finished, total, speed, nodes):
        progress = finished / total if total > 0 else 0
        speed_str = f"{speed:.1f}" if speed > 0 else "--"
        self.progress_var.set(progress)
        self.progress_label.config(text=f'进度: {finished}/{total}  节点: {nodes}  速度: {speed_str} 节点/秒')

    def on_solve_done(self, solutions):
        self.enable_inputs()
        if not solutions:
            self.status_label.config(text='无解')
            self.thread_label.config(text='线程状态: 空闲')
            messagebox.showinfo('结果', '无解')
        else:
            self.status_label.config(text=f'共找到 {len(solutions)} 个解')
            self.thread_label.config(text='线程状态: 空闲')
            self.show_solution_window(solutions)

    def show_solution_window(self, solutions):
        win = tk.Toplevel(self.root)
        win.title(f'共找到{len(solutions)}个解')
        win.geometry('420x420')
        idx = tk.IntVar(value=0)
        original = [[self.entries[i][j].get() != '' for j in range(9)] for i in range(9)]
        sol_entries = [[tk.Entry(win, width=2, font=('Consolas', 18), justify='center', state='readonly') for _ in range(9)] for _ in range(9)]
        for i in range(9):
            for j in range(9):
                sol_entries[i][j].grid(row=i, column=j, padx=(2 if j % 3 == 0 else 0, 2), pady=(2 if i % 3 == 0 else 0, 2))
        def show(idx_val):
            for i in range(9):
                for j in range(9):
                    sol_entries[i][j].config(state='normal')
                    sol_entries[i][j].delete(0, tk.END)
                    sol_entries[i][j].insert(0, str(solutions[idx_val][i][j]))
                    if original[i][j]:
                        sol_entries[i][j].config(fg='#e53935')
                    else:
                        sol_entries[i][j].config(fg='#388e3c')
                    sol_entries[i][j].config(state='readonly')
        btn_prev = ttk.Button(win, text='上一解', command=lambda: (idx.set((idx.get()-1)%len(solutions)), show(idx.get())))
        btn_next = ttk.Button(win, text='下一解', command=lambda: (idx.set((idx.get()+1)%len(solutions)), show(idx.get())))
        btn_prev.grid(row=9, column=2, columnspan=2, pady=8)
        btn_next.grid(row=9, column=5, columnspan=2, pady=8)
        if len(solutions) == 1:
            btn_prev.config(state='disabled')
            btn_next.config(state='disabled')
        show(0)

if __name__ == "__main__":
    multiprocessing.freeze_support()  # for Windows compatibility
    root = tk.Tk()
    app = SudokuGUI(root)
    root.mainloop()

The post 数独求解工具 appeared first on obaby@mars.