from flask import Flask, request from gb_secrets import credentials import logging, requests, os import re LOGLEVEL = os.environ.get('LOGLEVEL', 'INFO').upper() logging.basicConfig(level=LOGLEVEL) app = Flask(__name__) session = requests.Session() session.post("https://www.planet-casio.com/Fr/compte", data=credentials) def make_url_user(user): return f"[url=https://gitea.planet-casio.com/{user['username']}]{user['username']}[/url]" def make_url_repository(repository): fork = " (fork)" if repository["fork"] else "" return f"[url={repository['html_url']}]{repository['name']}{fork}[/url]" @app.route("/", methods=["GET", "POST"]) def main(): global session data = request.json event = request.headers['X-Gitea-Event'] msg = None if data["repository"]["private"] or data["repository"]["mirror"]: return "IGN" url_repository = make_url_repository(data["repository"]) if event == "push" and data["commits"] != []: # Heuristic to handle pull request merges by detecting default merge message RE_MERGE = r"^Merge pull request '(.+)' \(#([0-9]+)\)"#from .* into" m = re.match(RE_MERGE, data["commits"][0]["message"]) if len(data["commits"]) > 1 and m: pr_url = data["repository"]["html_url"] + "/pulls/" + m[2] pr_url = f"[url={pr_url}]{m[1]}[/url]" data["commits"] = data["commits"][1:] else: pr_url = None commits = data["commits"] total_commits = data.get("total_commits", len(commits)) commit_count = f"{total_commits} commit{'s' if total_commits > 1 else ''}" message = "[inlinecode]" + commits[0]["message"].split('\n',1)[0] + "[/inlinecode]" if total_commits <= 1: others = "" elif total_commits == 2: others = " (et 1 autre commit)" else: others = f" (et {total_commits-1} autres commits)" url_user = make_url_user(data['pusher']) # Pull request merge if pr_url: msg = f"{url_user} a fusionné la PR {pr_url}: {message}{others}" # Ref doesn't have a previous commit = new branch elif all(c == "0" for c in data["before"]) and data["ref"].startswith("refs/heads/"): branch = data["ref"][11:] url_branch = data["repository"]["html_url"] + f"/src/branch/{branch}" url_branch = f"[url={url_branch}]{branch}[/url]" msg = f"{url_user} a créé une nouvelle branche {url_branch} dans {url_repository} : {message}{others}" # Pre-existing branch else: if data["ref"].startswith("refs/heads/"): branch = " (branche " + data["ref"][11:] + ")" else: branch = "" url_commits = f"[url={data['compare_url']}]{commit_count}[/url]" msg = f"{url_user} a poussé {url_commits} dans {url_repository}{branch} : {message}{others}" if event == "push" and data["ref"].startswith("refs/tags/"): url_user = make_url_user(data['pusher']) tag_name = data["ref"][10:] url_tag = data["repository"]["html_url"] + f"/src/tag/{tag_name}" url_tag = f"[url={url_tag}]{tag_name}[/url]" if all(c == '0' for c in data["before"]): msg = f"{url_user} a créé le tag {url_tag} dans {url_repository}" else: msg = f"{url_user} a mis à jour le tag {url_tag} dans {url_repository}" if event == "issues": url_user = make_url_user(data['sender']) url_issue = f"[url={data['issue']['html_url']}]{data['issue']['title']}[/url]" if data['action'] == "closed": msg = f"{url_user} a fermé le ticket {url_issue} dans {url_repository}" elif data['action'] in ["created", "opened"]: msg = f"{url_user} a créé un ticket {url_issue} dans {url_repository}" elif data['action'] == "edited": msg = f"{url_user} a édité le ticket {url_issue} dans {url_repository}" else: return "IGN" if event == "issue_comment": url_user = make_url_user(data['sender']) url_issue = f"[url={data['issue']['html_url']}]{data['issue']['title']}[/url]" if data['action'] == "created": msg = f"{url_user} a répondu au ticket {url_issue} dans {url_repository}" else: return "IGN" if event == "pull_request" and data["action"] == "opened": url_user = make_url_user(data["sender"]) url_pr = f"[url={data['pull_request']['url']}]{data['pull_request']['title']}[/url]" msg = f"{url_user} a créé une PR: {url_pr} dans {url_repository}" if event == "repository" and data["action"] == "created": url_user = make_url_user(data['sender']) msg = f"{url_user} a créé un nouveau dépôt {url_repository}" if event == "repository" and data["action"] == "deleted": return "IGN" if event == "push" and all(c == '0' for c in data["after"]): return "IGN" if event == "push" and not data["ref"].startswith("refs/tags/") \ and all(c == '0' for c in data["before"]) and data["commits"] == []: return "IGN" if event == "pull_request" and data["action"] in ["assigned", "closed", "synchronized", "edited"]: return "IGN" if event == "pull_request_approved" and data["action"] == "reviewed": url_user = make_url_user(data['sender']) url_pr = f"[url={data['pull_request']['url']}]{data['pull_request']['title']}[/url]" msg = f"{url_user} a approuvé la PR {url_pr} dans {url_repository}" if event == "pull_request_rejected" and data["action"] == "reviewed": url_user = make_url_user(data['sender']) url_pr = f"[url={data['pull_request']['url']}]{data['pull_request']['title']}[/url]" msg = f"{url_user} a demandé des modifications sur la PR {url_pr} dans {url_repository}" if event == "fork": return "IGN" if event == "create" and data["ref_type"] in ["tag", "branch"]: return "IGN" if event == "delete": return "IGN" if event == "release" and data["action"] == "updated": return "IGN" if not msg: msg = f"Événement {event}" if "repository" in data: msg += " sur " + make_url_repository(data["repository"]) msg += "." if msg: msg = f"[gray][i]{msg}[/i][/gray]" app.logger.info(msg) r = session.post("https://www.planet-casio.com/Fr/shoutbox/api/post", data={"message": msg, "channel": "projets"}) if r.status_code == 200: return "OK" else: app.logger.warn(f"Error: {r.status_code}") return f"ERR {r.status_code}" return "ACK"