Python桌面版Webhook测试接收器与调试工具：本地启动HTTP服务接收回调请求、实时显示请求头与Body、历史记录持久化、一键导出JSON

🚀 实例简介

本工具是一个基于Tkinter的桌面版Webhook测试接收器，专为开发者调试回调接口设计。支持本地启动HTTP服务（默认端口9000），实时接收GET/POST/PUT/DELETE/PATCH等所有方法的Webhook请求，并在界面中展示请求头、请求体等详细信息。所有接收到的请求自动持久化到SQLite数据库，支持历史记录查看与一键导出JSON文件。

💻 核心代码

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Webhook测试接收器与调试工具
- 本地启动HTTP服务接收回调请求
- 实时显示请求头、Body、Method
- 历史记录持久化（SQLite）
- 一键导出JSON
"""
import tkinter as tk
from tkinter import ttk, scrolledtext, messagebox, filedialog
import threading
import json
import time
import sqlite3
import os
from http.server import BaseHTTPRequestHandler, HTTPServer
import platform

CHINESE_FONT = "WenQuanYi Zen Hei" if platform.system() == "Linux" else "Microsoft YaHei"
DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "webhook_history.db")


class WebhookHandler(BaseHTTPRequestHandler):
    """HTTP请求处理器，接收所有方法的Webhook请求"""
    app = None  # 指向主应用，用于更新UI

    def do_GET(self): self._handle()
    def do_POST(self): self._handle()
    def do_PUT(self): self._handle()
    def do_DELETE(self): self._handle()
    def do_PATCH(self): self._handle()

    def _handle(self):
        """统一处理所有请求，记录到数据库并刷新UI"""
        content_len = int(self.headers.get('Content-Length', 0))
        body = self.rfile.read(content_len).decode('utf-8', errors='ignore') if content_len else ''
        headers_dict = {k: v for k, v in self.headers.items()}

        record = {
            "time": time.strftime("%Y-%m-%d %H:%M:%S"),
            "method": self.command,
            "path": self.path,
            "headers": headers_dict,
            "body": body,
            "client": self.client_address[0]
        }

        # 保存到SQLite
        try:
            conn = sqlite3.connect(DB_PATH)
            conn.execute(
                "INSERT INTO history (time, method, path, headers, body, client) VALUES (?,?,?,?,?,?)",
                (record["time"], record["method"], record["path"],
                 json.dumps(headers_dict, ensure_ascii=False), body, record["client"]))
            conn.commit()
            conn.close()
        except Exception as e:
            print(f"DB error: {e}")

        # 回显到UI（线程安全）
        if self.app:
            self.app.root.after(0, self.app._add_record, record)

        self.send_response(200)
        self.send_header('Content-Type', 'application/json')
        self.end_headers()
        self.wfile.write(json.dumps({"status": "ok", "received": True}).encode())

    def log_message(self, format, *args):
        pass  # 静默日志


class WebhookApp:
    def __init__(self, root):
        self.root = root
        self.root.title("Webhook测试接收器与调试工具")
        self.root.geometry("900x650")
        self.server = None
        self.server_thread = None

💡 适用场景

• Webhook回调接口开发与调试
• 第三方平台推送消息接收测试
• API接口请求记录与分析
• 本地开发环境HTTP服务模拟