/
__main__.py
143 lines (118 loc) · 4.36 KB
/
__main__.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
import asyncio
import os
import sys
import traceback
import aiohttp
from aiohttp import web
import cachetools
from gidgethub import aiohttp as gh_aiohttp
from gidgethub import routing
from gidgethub import sansio
from gidgethub import apps
router = routing.Router()
cache = cachetools.LRUCache(maxsize=500)
routes = web.RouteTableDef()
@routes.get("/", name="home")
async def handle_get(request):
return web.Response(text="Hello world")
@routes.post("/webhook")
async def webhook(request):
try:
body = await request.read()
secret = os.environ.get("GH_SECRET")
event = sansio.Event.from_http(request.headers, body, secret=secret)
if event.event == "ping":
return web.Response(status=200)
async with aiohttp.ClientSession() as session:
gh = gh_aiohttp.GitHubAPI(session, "demo", cache=cache)
await asyncio.sleep(1)
await router.dispatch(event, gh)
try:
print("GH requests remaining:", gh.rate_limit.remaining)
except AttributeError:
pass
return web.Response(status=200)
except Exception as exc:
traceback.print_exc(file=sys.stderr)
return web.Response(status=500)
@router.register("installation", action="created")
async def repo_installation_added(event, gh, *args, **kwargs):
installation_id = event.data["installation"]["id"]
installation_access_token = await apps.get_installation_access_token(
gh,
installation_id=installation_id,
app_id=os.environ.get("GH_APP_ID"),
private_key=os.environ.get("GH_PRIVATE_KEY"),
)
sender_name = event.data["sender"]["login"]
for repo in event.data["repositories"]:
repo_full_name = repo["full_name"]
response = await gh.post(
f"/repos/{repo_full_name}/issues",
data={
"title": "Thanks for installing me",
"body": f"You're the best! @{sender_name}",
},
oauth_token=installation_access_token["token"],
)
issue_url = response["url"]
await gh.patch(
issue_url,
data={"state": "closed"},
oauth_token=installation_access_token["token"],
)
@router.register("pull_request", action="opened")
async def pr_opened(event, gh, *args, **kwargs):
issue_url = event.data["pull_request"]["issue_url"]
username = event.data["sender"]["login"]
installation_id = event.data["installation"]["id"]
installation_access_token = await apps.get_installation_access_token(
gh,
installation_id=installation_id,
app_id=os.environ.get("GH_APP_ID"),
private_key=os.environ.get("GH_PRIVATE_KEY"),
)
author_association = event.data["pull_request"]["author_association"]
if author_association == "NONE":
# first time contributor
msg = f"Thanks for your first contribution @{username}"
else:
# seasoned contributor
msg = f"Welcome back, @{username}. You are a {author_association}."
response = await gh.post(
f"{issue_url}/comments",
data={"body": msg},
oauth_token=installation_access_token["token"],
)
@router.register("issue_comment", action="created")
async def issue_comment_created(event, gh, *args, **kwargs):
username = event.data["sender"]["login"]
installation_id = event.data["installation"]["id"]
installation_access_token = await apps.get_installation_access_token(
gh,
installation_id=installation_id,
app_id=os.environ.get("GH_APP_ID"),
private_key=os.environ.get("GH_PRIVATE_KEY"),
)
comments_url = event.data["comment"]["url"]
if username == "Mariatta":
response = await gh.post(
f"{comments_url}/reactions",
data={"content": "heart"},
oauth_token=installation_access_token["token"],
accept="application/vnd.github.squirrel-girl-preview+json",
)
if __name__ == "__main__": # pragma: no cover
app = web.Application()
app.router.add_routes(routes)
port = os.environ.get("PORT")
if port is not None:
port = int(port)
web.run_app(app, port=port)
if __name__ == "__main__": # pragma: no cover
app = web.Application()
app.router.add_routes(routes)
port = os.environ.get("PORT")
if port is not None:
port = int(port)
web.run_app(app, port=port)