Skip to content

Commit

Permalink
revamp WIP
Browse files Browse the repository at this point in the history
Signed-off-by: Trey <[email protected]>
  • Loading branch information
TreyWW committed Jul 13, 2024
1 parent eab27a5 commit 9caec09
Show file tree
Hide file tree
Showing 7 changed files with 270 additions and 99 deletions.
3 changes: 2 additions & 1 deletion .github/management_bot/pulumi/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@
role=lambda_access_role.arn,
code=pulumi.AssetArchive({".": pulumi.FileArchive("./src")}),
handler="lambda_handler.lambda_handler",
timeout=8,
runtime=lambda_.Runtime.PYTHON3D12,
environment=lambda_.FunctionEnvironmentArgs(
variables={"app_id": config.require_secret("app_id"), "private_key": config.require_secret("private_key")}
Expand Down Expand Up @@ -71,7 +72,7 @@
http_method="POST",
integration_http_method="POST",
type="AWS",
timeout_milliseconds=4000,
timeout_milliseconds=8000,
uri=lambda_func.invoke_arn,
)

Expand Down
108 changes: 108 additions & 0 deletions .github/management_bot/pulumi/src/_types.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,108 @@
from dataclasses import dataclass, fields, field
from typing import Optional, Type, TypeVar, Any, TypedDict

# from github import Github
import github
import github.Repository
import github.Issue
import github.PullRequest
import github.Label
import github.IssueComment
import github.PullRequestComment
import github.NamedUser
from github.PaginatedList import PaginatedList

T = TypeVar("T")


def fill_dataclass_from_dict(dataclass_type: Type[T], data: dict) -> T | None:
if not data:
return None
field_names = {f.name for f in fields(dataclass_type)}
filtered_data = {k: v for k, v in data.items() if k in field_names}
return dataclass_type(**filtered_data)


@dataclass
class User:
id: str
url: str
login: str


@dataclass
class Comment:
id: int
body: str


@dataclass
class Issue:
id: int
number: int


@dataclass
class PullRequest:
id: int
number: int
url: str
state: str
locked: bool
user: User
issue_url: Optional[str] = None


@dataclass
class Repository:
id: int
name: str
full_name: str
owner: User


@dataclass
class Changes:
body: Optional[TypedDict("ChangeProperty", {"from": str})] = None
title: Optional[TypedDict("ChangeProperty", {"from": str})] = None


@dataclass
class Context:
event: dict
action: str
sender: User
issue: Optional[Issue] = None
pull_request: Optional[PullRequest] = None
comment: Optional[Comment] = None
repository: Optional[Repository] = None
changes: Optional[Changes] = None


@dataclass
class Objects:
github: github.MainClass.Github
dict_context: Context
repository: github.Repository.Repository
sender: Optional[github.NamedUser] = None
issue: Optional[github.Issue.Issue] = None
pull_request: Optional[github.PullRequest.PullRequest] = None
labels: PaginatedList[github.Label.Label] = None
comment: Optional[github.IssueComment.IssueComment | github.PullRequestComment.PullRequestComment] = None

def __post_init__(self):
print("post_init")
if not self.repository:
print("no repository, returning")
return

if self.dict_context.issue:
self.issue = self.repository.get_issue(self.dict_context.issue.number)

if self.dict_context.pull_request:
self.pull_request = self.repository.get_pull(self.dict_context.pull_request.number)

if self.pull_request:
self.labels = self.pull_request.get_labels()

self.sender = self.github.get_user(self.dict_context.sender.login)
9 changes: 9 additions & 0 deletions .github/management_bot/pulumi/src/helpers.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
import base64


def encode_private_key(entire_key: str) -> str:
return base64.b64encode(entire_key.encode("ascii")).decode("ascii")


def decode_private_key(raw_private_key) -> str:
return base64.b64decode(raw_private_key.encode("ascii")).decode("ascii")
75 changes: 75 additions & 0 deletions .github/management_bot/pulumi/src/issues/handler.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,75 @@
import os
import re
from textwrap import dedent
import logging
import secrets

import github.Issue

from re import match

import string

import random

logger = logging.getLogger(__name__)
if os.environ.get("AWS_EXECUTION_ENV") is not None:
import _types
else:
from .. import _types


def title_handler(context_dicts: _types.Context, context_objs: _types.Objects) -> None:
if not re.match(r"^(bug|idea|implement|cleanup):\s*\S.*", context_objs.issue.title):
logger.info(f"Regex title doesn't match. {context_objs.issue.title} doesnt start with bug|idea|implement|cleanup:")
logger.info(f"Commenting on {context_objs.issue.html_url}")
context_objs.issue.create_comment(
dedent(
f"""
Hi @{context_objs.sender.login},
You have chosen a title that is slightly different to our title standards. Please, if possible, use the format:
(bug, idea, implement or cleanup) : Title
e.g. "bug: xyz page doesn't work"
> If you would like to ignore this message, please reply with the reference `DELREPLY-
{''.join(random.choices(string.ascii_uppercase + string.digits, k=8))}` (you may delete this reply afterwards)
"""
)
)


def delete_reply_handler(context_dicts: _types.Context, context_objs: _types.Objects) -> None:
match = re.search(r"DELREPLY-(.{8})", context_dicts.comment.body)

if not match:
return

logger.info("Deleting comment due to DELREPLY in body")

reference_code = match.group(1)

logger.debug(f"Deleting comment with reference code: {reference_code}")

context_objs.issue: github.Issue.Issue

for comment in context_objs.issue.get_comments():
if f"DELREPLY-{reference_code}" in comment.body.upper():
# comment.delete() # delete users reply comment
context_objs.issue.get_comment(context_dicts.comment.id).delete() # delete bots comment
break


def handler(context_dicts: _types.Context, context_objs: _types.Objects) -> None:
logger.info(f"action: {context_dicts.action}")
match context_dicts.action:
case "opened":
logger.info("Using title handler due to opened issue")
title_handler(context_dicts, context_objs)
case "edited":
if context_dicts.changes.title and context_dicts.changes.title["from"]:
title_handler(context_dicts, context_objs)
case "created":
if context_dicts.comment:
delete_reply_handler(context_dicts, context_objs)
146 changes: 48 additions & 98 deletions .github/management_bot/pulumi/src/lambda_handler.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,26 @@
import json
import os, base64
from textwrap import dedent

import github.GithubException
from github import Github, Issue
from prs import handler as pr_handler
from issues import handler as issue_handler
from helpers import decode_private_key
import _types
from github import Github, Issue, GithubIntegration
from github import Auth

raw_private_key = os.environ.get("private_key")
PRIVATE_KEY = base64.b64decode(raw_private_key).decode("ascii")
import logging

logging.basicConfig()
logging.getLogger().setLevel(logging.DEBUG if os.environ.get("DEBUG") else logging.DEBUG) # todo go back to info
logger = logging.getLogger(__name__)

PRIVATE_KEY = decode_private_key(os.environ.get("private_key"))
APP_ID = os.environ.get("app_id")

REPOSITORY_NAME = "TreyWW/MyFinances"


def check_if_user_perm_issue(issue: dict, sender: dict, repository: dict):
return issue.get("user", {}).get("id") == sender.get("id") or sender.get("id") == repository.get("owner", {}).get("id")
Expand Down Expand Up @@ -45,98 +57,36 @@ def is_trey(sender):


def lambda_handler(event: dict, _):
auth = Auth.AppAuth(APP_ID, PRIVATE_KEY).get_installation_auth(event.get("installation", {}).get("id"))
g = Github(auth=auth)

ACTION = event.get("action", {})
ISSUE = event.get("issue", {})
PR = event.get("pull_request", {})
COMMENT = event.get("comment", {})
SENDER = event.get("sender", {})
REPOSITORY = event.get("repository", {})

repo = g.get_repo(event.get("repository", {}).get("full_name")) if REPOSITORY else {}

if ISSUE or PR:
selected_json: dict = ISSUE or PR

selected_obj = repo.get_issue(number=ISSUE["number"]) if repo and ISSUE else repo.get_pull(number=PR["id"]) if repo and PR else {}

LABELS = selected_json.get("labels")
label_names = {label["name"] for label in LABELS}

if "awaiting-response" in label_names and is_owner(selected_json, SENDER) and COMMENT:
try:
selected_obj.remove_from_labels("awaiting-response")
except github.GithubException:
...

if COMMENT and (is_trey(SENDER) or is_owner(selected_json, SENDER)):
if ACTION == "created": # sent comment
msg = COMMENT.get("body", "")
msg_stripped = msg.strip().split(" ")
msg_len = len(msg_stripped)

match msg_stripped[0]:
case "/add_label":
if not msg_len == 2:
send_error(
selected_obj,
sender=SENDER["login"],
body=COMMENT["body"],
msg_len=msg_len,
required=2,
example_cmd="add_label bug",
)

return g.close()

selected_obj.add_to_labels(msg_stripped[1])
selected_obj.create_comment(f"Okay @{SENDER['login']}, I have added the label '{msg_stripped[1]}'")
case "/add_labels":
selected_obj.add_to_labels(*msg_stripped[1:])
selected_obj.create_comment(f"Okay @{SENDER['login']}, I have added the labels \"{', '.join(msg_stripped[1:])}\"")
case "/remove_label":
if not msg_len == 2:
send_error(
selected_obj,
sender=SENDER["login"],
body=COMMENT["body"],
msg_len=msg_len,
required=2,
example_cmd="remove_label bug",
)

return g.close()

selected_obj.remove_from_labels(msg_stripped[1])
selected_obj.create_comment(f"Okay @{SENDER['login']}, I have removed the label \"{msg_stripped[1]}\"")
case "/remove_labels":
selected_obj.remove_from_labels(*msg_stripped[1:])
selected_obj.create_comment(f"Okay @{SENDER['login']}, I have removed the labels \"{', '.join(msg_stripped[1:])}\"")
case _:
selected_obj.create_comment(
dedent(
f"""
Hi @{SENDER["login"]},
<details><summary>My available commands:</summary>
<p>
| Command | Description | Arg Types | Example |
|---------|-------------|--------|--------|
| /add_label | Adds one label | string | /add_label bug |
| /add_labels | Adds multiple labels | list[string] | /add_label bug enhancement |
| /remove_label | Removes one label | string | /remove_label bug |
| /remove_labels | Removes multiple labels | list[string] | /remove_labels bug enhancement |
</p>
</details>
"""
)
)
# elif PR:
# match ACTION:
# case "labeled":
#

return {"statusCode": 200, "body": {}}
auth = Auth.AppAuth(APP_ID, PRIVATE_KEY)
gi = GithubIntegration(auth=auth)
g: Github = gi.get_installations()[0].get_github_for_installation()

logger.debug(event)

context_dicts = _types.Context(
event=event,
action=event.get("action", ""),
issue=_types.fill_dataclass_from_dict(_types.Issue, event.get("issue", {})),
pull_request=_types.fill_dataclass_from_dict(_types.PullRequest, event.get("pull_request", {})),
comment=_types.fill_dataclass_from_dict(_types.Comment, event.get("comment", {})),
sender=_types.fill_dataclass_from_dict(_types.User, event.get("sender", {})),
repository=_types.fill_dataclass_from_dict(_types.Repository, event.get("repository", {})),
changes=_types.fill_dataclass_from_dict(_types.Changes, event.get("changes", {})),
)

logger.debug(context_dicts)

context_objs = _types.Objects(github=g, dict_context=context_dicts, repository=g.get_repo(context_dicts.repository.full_name))

logger.debug(context_objs)

if context_dicts.pull_request:
logger.debug("Using PR handler")
pr_handler.handler(context_dicts, context_objs)
elif context_dicts.issue:
logger.debug("Using issue handler")
issue_handler.handler(context_dicts, context_objs)
else:
logger.debug("Using no handler; invalid request.")

return {"statusCode": 200, "body": json.dumps({})}
15 changes: 15 additions & 0 deletions .github/management_bot/pulumi/src/prs/actions.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,15 @@
def add_label():
if not msg_len == 2:
send_error(
selected_obj,
sender=SENDER["login"],
body=COMMENT["body"],
msg_len=msg_len,
required=2,
example_cmd="add_label bug",
)

return g.close()

selected_obj.add_to_labels(msg_stripped[1])
selected_obj.create_comment(f"Okay @{SENDER['login']}, I have added the label '{msg_stripped[1]}'")
Loading

0 comments on commit 9caec09

Please sign in to comment.