91 lines
2.5 KiB
Python
91 lines
2.5 KiB
Python
from kython import load_json_file
|
|
from typing import Dict, List, Union, Any, NamedTuple
|
|
from datetime import datetime
|
|
import logging
|
|
|
|
import os
|
|
|
|
BPATH = "/L/backups/github-events"
|
|
|
|
def get_logger():
|
|
return logging.getLogger('github-provider')
|
|
|
|
def iter_files():
|
|
for f in os.listdir(BPATH):
|
|
if f.endswith('.json'):
|
|
yield os.path.join(BPATH, f)
|
|
|
|
def iter_events():
|
|
for f in list(sorted(iter_files())):
|
|
yield f
|
|
|
|
class Event(NamedTuple):
|
|
dt: datetime
|
|
summary: str
|
|
|
|
def _get_summary(e) -> str:
|
|
tp = e['type']
|
|
pl = e['payload']
|
|
rname = e['repo']['name']
|
|
if tp == 'ForkEvent':
|
|
return f"forked {rname}"
|
|
elif tp == 'PushEvent':
|
|
return f"pushed to {rname}"
|
|
elif tp == 'WatchEvent':
|
|
return f"watching {rname}"
|
|
elif tp == 'CreateEvent':
|
|
return f"created {rname}"
|
|
elif tp == 'PullRequestEvent':
|
|
pr = pl['pull_request']
|
|
action = pl['action']
|
|
link = pr['html_url']
|
|
title = pr['title']
|
|
return f"{action} PR {title} {link}"
|
|
elif tp == "IssuesEvent":
|
|
action = pl['action']
|
|
iss = pl['issue']
|
|
link = iss['html_url']
|
|
title = iss['title']
|
|
return f"{action} issue {title} {link}"
|
|
elif tp == "IssueCommentEvent":
|
|
com = pl['comment']
|
|
link = com['html_url']
|
|
iss = pl['issue']
|
|
title = iss['title']
|
|
return f"commented on issue {title} {link}"
|
|
elif tp == "ReleaseEvent":
|
|
action = pl['action']
|
|
rel = pl['release']
|
|
tag = rel['tag_name']
|
|
link = rel['html_url']
|
|
return f"{action} {rname} [{tag}] {link}"
|
|
elif tp in (
|
|
"DeleteEvent",
|
|
"PublicEvent",
|
|
):
|
|
return tp # TODO ???
|
|
else:
|
|
return tp
|
|
|
|
def get_events():
|
|
logger = get_logger()
|
|
|
|
events: Dict[str, Any] = {}
|
|
for f in iter_events():
|
|
jj = load_json_file(f)
|
|
for e in jj:
|
|
eid = e['id']
|
|
prev = events.get(eid, None)
|
|
if prev is not None:
|
|
if prev != e:
|
|
# a = prev['payload']
|
|
# b = e['payload']
|
|
# TODO err... push_id has changed??? wtf??
|
|
logger.error(f"Mismatch in \n{e}\n vs \n{prev}")
|
|
events[eid] = e
|
|
# TODO utc?? localize
|
|
ev = [Event(
|
|
dt=datetime.strptime(d['created_at'], '%Y-%m-%dT%H:%M:%SZ'),
|
|
summary=_get_summary(d),
|
|
) for d in events.values()]
|
|
return sorted(ev, key=lambda e: e.dt)
|