112 lines
3.4 KiB
Python
112 lines
3.4 KiB
Python
"""
|
||
简易 Webhook 接收服务：监听 POST /webhook，打印请求内容与部分头信息。
|
||
|
||
生产请将手机/脚本转发指向萌芽小店后端：`POST {YOUR_API_BASE}/api/webhooks/mengya-pay`，
|
||
并在请求头附带与后端一致的 `X-Webhook-Secret`（对应环境变量 `WEBHOOK_MENGYA_SECRET`）。
|
||
|
||
本地测试环境变量：
|
||
HOST 监听地址，默认 0.0.0.0
|
||
PORT 监听端口，默认 8000
|
||
WEBHOOK_SECRET 若设置，则要求请求头 X-Webhook-Secret 与其一致，否则返回 403
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import secrets
|
||
from datetime import datetime, timezone
|
||
from typing import Annotated
|
||
|
||
from fastapi import FastAPI, Header, HTTPException, Request, Response
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s %(levelname)s %(message)s",
|
||
)
|
||
log = logging.getLogger("webhook")
|
||
|
||
app = FastAPI(title="Webhook 测试接收器", docs_url="/docs")
|
||
|
||
|
||
def _now_iso() -> str:
|
||
return datetime.now(timezone.utc).isoformat()
|
||
|
||
|
||
@app.get("/health")
|
||
def health() -> dict:
|
||
return {"status": "ok", "time": _now_iso()}
|
||
|
||
|
||
@app.post("/webhook")
|
||
async def webhook(
|
||
request: Request,
|
||
response: Response,
|
||
x_webhook_secret: Annotated[str | None, Header(alias="X-Webhook-Secret")] = None,
|
||
) -> dict:
|
||
expected = os.environ.get("WEBHOOK_SECRET")
|
||
if expected is not None:
|
||
if x_webhook_secret is None:
|
||
raise HTTPException(status_code=403, detail="missing X-Webhook-Secret")
|
||
if not secrets.compare_digest(x_webhook_secret, expected):
|
||
raise HTTPException(status_code=403, detail="invalid X-Webhook-Secret")
|
||
|
||
raw = await request.body()
|
||
ct = request.headers.get("content-type", "")
|
||
|
||
try:
|
||
if "application/json" in ct and raw.strip():
|
||
body_preview = json.loads(raw.decode())
|
||
elif raw.strip():
|
||
body_preview = raw.decode(errors="replace")
|
||
if len(body_preview) > 2000:
|
||
body_preview = body_preview[:2000] + "…(truncated)"
|
||
else:
|
||
body_preview = None
|
||
except json.JSONDecodeError:
|
||
body_preview = raw.decode(errors="replace")
|
||
|
||
interesting_headers = {
|
||
k: v
|
||
for k, v in request.headers.items()
|
||
if k.lower()
|
||
in (
|
||
"content-type",
|
||
"user-agent",
|
||
"x-github-event",
|
||
"x-gitlab-token",
|
||
"x-request-id",
|
||
)
|
||
}
|
||
|
||
event_id = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S%f")
|
||
info = {
|
||
"event_id": event_id,
|
||
"time": _now_iso(),
|
||
"client": getattr(request.client, "host", None),
|
||
"method": request.method,
|
||
"path": str(request.url.path),
|
||
"query": dict(request.query_params),
|
||
"content_type": ct or None,
|
||
"interesting_headers": interesting_headers,
|
||
"body": body_preview,
|
||
"body_bytes": len(raw),
|
||
}
|
||
log.info("webhook %s | %s", event_id, json.dumps(info, ensure_ascii=False))
|
||
|
||
response.headers["X-Webhook-Received"] = event_id
|
||
return {"ok": True, "event_id": event_id, "received_at": info["time"]}
|
||
|
||
|
||
def main() -> None:
|
||
import uvicorn
|
||
|
||
host = os.environ.get("HOST", "0.0.0.0")
|
||
port = int(os.environ.get("PORT", "8000"))
|
||
uvicorn.run("server:app", host=host, port=port, reload=False)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|