84 lines
3.2 KiB
Python
84 lines
3.2 KiB
Python
"""Gitea webhook: принимает push-события, обновляет кэш, уведомляет подписчиков."""
|
||
import hmac
|
||
import hashlib
|
||
import json
|
||
from datetime import datetime
|
||
from flask import request, jsonify
|
||
from sqlmodel import Session, select
|
||
from ..db import engine
|
||
from ..models import Project
|
||
from ..helpers import (
|
||
_project_subscriber_ids, notify_users,
|
||
)
|
||
from .. import config, gitea_api
|
||
|
||
|
||
def _verify_signature(raw_body, sig_header):
|
||
if not config.GITEA_WEBHOOK_SECRET:
|
||
return True
|
||
if not sig_header:
|
||
return False
|
||
expected = hmac.new(
|
||
config.GITEA_WEBHOOK_SECRET.encode("utf-8"),
|
||
raw_body, hashlib.sha256).hexdigest()
|
||
return hmac.compare_digest(expected, sig_header)
|
||
|
||
|
||
def register(app):
|
||
|
||
@app.route("/webhooks/gitea", methods=["POST"])
|
||
def gitea_webhook():
|
||
raw = request.get_data() or b""
|
||
sig = request.headers.get("X-Gitea-Signature", "")
|
||
if not _verify_signature(raw, sig):
|
||
return jsonify({"ok": False, "error": "invalid_signature"}), 401
|
||
|
||
event_type = request.headers.get("X-Gitea-Event", "")
|
||
try:
|
||
payload = json.loads(raw.decode("utf-8"))
|
||
except Exception:
|
||
return jsonify({"ok": False, "error": "bad_json"}), 400
|
||
|
||
repo_full = ((payload.get("repository") or {}).get("full_name") or "").strip()
|
||
if not repo_full:
|
||
return jsonify({"ok": False, "error": "no_repo"}), 400
|
||
|
||
# Инвалидируем кэш gitea_api
|
||
dropped = gitea_api.invalidate_cache(repo_full)
|
||
|
||
# Найдём проект(ы), привязанные к этому репо
|
||
linked_projects = []
|
||
with Session(engine) as db_session:
|
||
for p in db_session.exec(select(Project)).all():
|
||
repo = p.gitea_repo or (gitea_api.extract_repo_from_url(p.gitea_url) if p.gitea_url else None)
|
||
if repo == repo_full:
|
||
linked_projects.append({"id": p.id, "title": p.title})
|
||
|
||
# Уведомления о push
|
||
if event_type == "push":
|
||
commits = payload.get("commits") or []
|
||
pusher = (payload.get("pusher") or {}).get("username") or ""
|
||
branch = (payload.get("ref") or "").replace("refs/heads/", "") or "?"
|
||
|
||
if commits:
|
||
n = len(commits)
|
||
first_msg = (commits[0].get("message") or "").split("\n")[0][:100]
|
||
title = f"Новый push в {repo_full} ({n} коммит{'ов' if n > 1 else ''})"
|
||
body = f"Ветка: {branch}\nАвтор: {pusher}\nПервый коммит: {first_msg}"
|
||
with Session(engine) as s:
|
||
for lp in linked_projects:
|
||
sub_ids = _project_subscriber_ids(s, lp["id"])
|
||
if sub_ids:
|
||
notify_users(sub_ids, title, body,
|
||
link_url=f"/projects/{lp['id']}",
|
||
kind="info")
|
||
|
||
# Gitea ждёт 200-204
|
||
return jsonify({
|
||
"ok": True,
|
||
"event": event_type,
|
||
"repo": repo_full,
|
||
"cache_invalidated_keys": dropped,
|
||
"linked_projects": linked_projects,
|
||
})
|