-
-
Notifications
You must be signed in to change notification settings - Fork 2
Expand file tree
/
Copy pathconversion_to_python.py
More file actions
141 lines (116 loc) · 5.81 KB
/
Copy pathconversion_to_python.py
File metadata and controls
141 lines (116 loc) · 5.81 KB
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
import textwrap
full_conversion_path = "/mnt/data/true_contributors_mixin_full.py"
# Full class implementation with mocks for GitHub API interactions
class_code = textwrap.dedent('''\
# true_contributors_mixin_full.py
from typing import List, Dict, Callable, Any
from collections import defaultdict
class GitHubAPIStub:
"""Stub class to simulate GitHub API structure used in the mixin."""
def __init__(self):
self.repos = self.RepoStub()
self.issues = self.IssueStub()
class RepoStub:
def listCommits(self, params):
return [{"author": {"id": 1, "login": "alice"}},
{"author": {"id": 2, "login": "bob"}}]
def listContributors(self, params):
return {"status": 204, "headers": {"status": "204 No Content"}}
def listForOrg(self, params):
return [
{"owner": {"login": "org"}, "name": "repo1"},
{"owner": {"login": "org"}, "name": "repo2"}
]
class IssueStub:
def listCommentsForRepo(self, params):
return [{"user": {"id": 1, "login": "alice"}},
{"user": {"id": 3, "login": "charlie"}}]
def paginate(self, func, params):
# Simulated pagination logic (returns direct result)
return func(params)
class TrueContributorsMixin:
def __init__(self):
self.api = GitHubAPIStub()
self.repos = self.api.repos
self.issues = self.api.issues
self.paginate = self.api.paginate
def listCommitCommentContributorsForOrg(self, parameters):
return self._listForOrgHelper(self.listCommitCommentContributors, parameters)
def listCommitContributorsForOrg(self, parameters):
return self._listForOrgHelper(self.listCommitContributors, parameters)
def listContributorsForOrg(self, parameters):
return self._listForOrgHelper(self._listContributors, parameters)
def listCommentContributorsForOrg(self, parameters):
return self._listForOrgHelper(self.listCommentContributors, parameters)
def listCommitCommentContributors(self, parameters):
params = self._createParamsFromObject(["owner", "repo", "since"], parameters)
contributors = (
self.listCommitContributors(params)
if "since" in params
else self._listContributors(params)
)
comments = self.listCommentContributors(params)
return self._aggregateContributors(contributors + comments)
def listCommitContributors(self, parameters):
params = self._createParamsFromObject(
["owner", "repo", "sha", "path", "since", "until"], parameters
)
try:
commits = self.paginate(self.repos.listCommits, params)
except Exception as err:
if getattr(err, 'status', None) != 409 or str(err) != "Git Repository is empty.":
raise
commits = []
return self._aggregateContributions(commits, "author")
def listCommentContributors(self, parameters):
params = self._createParamsFromObject(["owner", "repo", "since"], parameters)
comments = self.paginate(self.issues.listCommentsForRepo, params)
return self._aggregateContributions(comments, "user")
def _listForOrgHelper(self, endpoint: Callable, parameters: Dict):
valid_endpoints = [
self.listCommentContributors,
self._listContributors,
self.listCommitContributors,
self.listCommitCommentContributors,
]
if endpoint not in valid_endpoints:
raise TypeError("Unexpected endpoint function provided.")
org_params = self._createParamsFromObject(["org", "type"], parameters)
repos = self.paginate(self.repos.listForOrg, org_params)
contributors = []
for repo in repos:
params = {"owner": repo["owner"]["login"], "repo": repo["name"], **parameters}
contributors += endpoint(params)
return self._aggregateContributors(contributors)
def _listContributors(self, parameters):
try:
return self.paginate(self.repos.listContributors, parameters)
except Exception:
res = self.repos.listContributors(parameters)
if res.get("status") != 204 or res.get("headers", {}).get("status") != "204 No Content":
raise
return []
def _aggregateContributions(self, items, key):
if not key:
raise ReferenceError("Missing key for contribution.")
contributors = [
{**item[key], "contributions": 1}
for item in items if key in item and item[key]
]
return self._aggregateContributors(contributors)
def _aggregateContributors(self, contributors: List[Dict]):
counts = defaultdict(lambda: {"contributions": 0})
for contributor in contributors:
uid = contributor["id"]
if "login" in contributor:
counts[uid]["login"] = contributor["login"]
counts[uid]["id"] = uid
counts[uid]["contributions"] += contributor["contributions"]
return sorted(counts.values(), key=lambda c: c["contributions"], reverse=True)
def _createParamsFromObject(self, keys: List[str], params: Dict) -> Dict:
return {k: params[k] for k in keys if k in params}
''')
# Write the full Python conversion to disk
with open(full_conversion_path, "w") as f:
f.write(class_code)
full_conversion_path