Newer
Older
from typing import List, Union, Dict, Optional, Type, NewType
from datetime import datetime
from mautrix.types import JSON, SerializableAttrs, SerializerError, Obj, serializer, deserializer
@serializer(datetime)
def datetime_serializer(dt: datetime) -> JSON:
return dt.strftime('%Y-%m-%dT%H:%M:%S%z')
@deserializer(datetime)
def datetime_deserializer(data: JSON) -> datetime:
try:
return datetime.strptime(data, "%Y-%m-%dT%H:%M:%S.%f%z")
except ValueError:
pass
try:
return datetime.strptime(data, "%Y-%m-%dT%H:%M:%S%z")
except ValueError:
pass
try:
return datetime.strptime(data, "%Y-%m-%d %H:%M:%S %z")
except ValueError:
pass
try:
return datetime.strptime(data, "%Y-%m-%dT%H:%M:%SZ")
except ValueError:
pass
try:
return datetime.strptime(data, "%Y-%m-%d %H:%M:%S %Z")
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
144
except ValueError:
pass
raise ValueError(data)
@dataclass
class GitlabLabel(SerializableAttrs['GitlabLabel']):
id: int
title: str
color: str
project_id: int
created_at: datetime
updated_at: datetime
template: bool
description: str
type: str
group_id: int
@dataclass
class GitlabAssignee(SerializableAttrs['GitlabAssignee']):
name: str
username: str
avatar_url: str
@dataclass
class GitlabProject(SerializableAttrs['GitlabProject']):
id: Optional[int] = None
name: Optional[str] = None
description: Optional[str] = None
web_url: Optional[str] = None
avatar_url: Optional[str] = None
git_ssh_url: Optional[str] = None
git_http_url: Optional[str] = None
namespace: Optional[str] = None
visibility_level: Optional[int] = None
path_with_namespace: Optional[str] = None
default_branch: Optional[str] = None
homepage: Optional[str] = None
url: Optional[str] = None
ssh_url: Optional[str] = None
http_url: Optional[str] = None
@dataclass
class GitlabCommit(SerializableAttrs['GitlabCommit']):
id: str
message: str
timestamp: Optional[datetime] = None
url: Optional[str] = None
author: Optional[str] = None
added: Optional[List[str]] = None
modified: Optional[List[str]] = None
removed: Optional[List[str]] = None
@dataclass
class GitlabAuthor(SerializableAttrs['GitlabAuthor']):
name: str
email: str
@dataclass
class GitlabRepository(SerializableAttrs['GitlabRepository']):
name: str
url: Optional[str] = None
description: Optional[str] = None
homepage: Optional[str] = None
git_http_url: Optional[str] = None
git_ssh_url: Optional[str] = None
visibility_level: Optional[int] = None
@dataclass
class GitlabStDiff(SerializableAttrs['GitlabStDiff']):
diff: str
new_path: str
old_path: str
a_mode: str
b_mode: str
new_file: bool
renamed_file: bool
deleted_file: bool
@dataclass
class GitlabSource(SerializableAttrs['GitlabSource']):
name: str
namespace: str
description: Optional[str] = None
web_url: Optional[str] = None
avatar_url: Optional[str] = None
git_ssh_url: Optional[str] = None
git_http_url: Optional[str] = None
visibility_level: Optional[int] = None
path_with_namespace: Optional[str] = None
default_branch: Optional[str] = None
homepage: Optional[str] = None
url: Optional[str] = None
ssh_url: Optional[str] = None
http_url: Optional[str] = None
GitlabTarget = NewType('GitlabTarget', GitlabSource)
GitlabChangeState = NewType('GitlabChangeState', Union[List[GitlabLabel], List[GitlabAssignee],
str, int])
@deserializer(GitlabChangeState)
def deserialize_change_state(val: JSON) -> GitlabChangeState:
if isinstance(val, list):
return [deserialize_change_state(item) for item in val]
elif isinstance(val, dict):
try:
return GitlabLabel.deserialize(val)
except SerializerError:
pass
try:
return GitlabAssignee.deserialize(val)
except SerializerError:
pass
return Obj(**val)
return val
@dataclass
class GitlabChange(SerializableAttrs['GitlabChange']):
previous: GitlabChangeState
current: GitlabChangeState
@dataclass
class GitlabLabelChanges(SerializableAttrs['GitlabLabelChanges']):
previous: List[GitlabLabel]
current: List[GitlabLabel]
@dataclass
class GitlabIssue(SerializableAttrs['GitlabIssue']):
id: int
issue_id: int = attr.ib(metadata={"json": "iid"})
title: str
description: str
author_id: int
project_id: int
created_at: datetime
updated_at: datetime
assignee_id: Optional[int] = None
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
relative_position: Optional[int] = None
branch_name: Optional[str] = None
milestone_id: Optional[int] = None
state: Optional[str] = None
@dataclass
class GitlabSnippet(SerializableAttrs['GitlabSnippet']):
id: int
title: str
content: str
author_id: int
project_id: int
created_at: datetime
updated_at: datetime
file_name: str
expires_at: datetime
type: str
visibility_level: int
@dataclass
class GitlabUser(SerializableAttrs['GitlabUser']):
name: str
username: Optional[str] = None
avatar_url: Optional[str] = None
id: Optional[int] = None
email: Optional[str] = None
@dataclass
class GitlabIssueAttributes(SerializableAttrs['GitlabIssueAttributes']):
id: int
project_id: int
issue_id: int = attr.ib(metadata={"json": "iid"})
state: str
url: str
author_id: int
title: str
description: str
created_at: datetime
updated_at: datetime
updated_by_id: Optional[int] = None
due_date: Optional[datetime] = None
closed_at: Optional[datetime] = None
time_estimate: Optional[int] = None
total_time_spent: Optional[int] = None
human_time_estimate: Optional[str] = None
human_total_time_spent: Optional[str] = None
action: Optional[str] = None
assignee_id: Optional[int] = None
assignee_ids: Optional[List[int]] = None
branch_name: Optional[str] = None
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
confidential: bool = False
duplicated_to_id: Optional[int] = None
moved_to_id: Optional[int] = None
state_id: Optional[int] = None
milestone_id: Optional[int] = None
labels: Optional[List[GitlabLabel]] = None
position: Optional[int] = None
original_position: Optional[int] = None
@dataclass
class GitlabCommentAttributes(SerializableAttrs['GitlabCommentAttributes']):
id: int
note: str
noteable_type: str
project_id: int
url: str
author_id: int
created_at: datetime
updated_at: datetime
updated_by_id: Optional[int] = None
resolved_at: Optional[datetime] = None
resolved_by_id: Optional[int] = None
commit_id: Optional[str] = None
noteable_id: Optional[int] = None
discussion_id: Optional[str] = None
system: Optional[bool] = None
line_code: Optional[str] = None
st_diff: Optional[GitlabStDiff] = None
attachment: Optional[str] = None
position: Optional[int] = None
original_position: Optional[int] = None
@dataclass
class GitlabMergeRequestAttributes(SerializableAttrs['GitlabMergeRequestAttributes']):
id: int
merge_request_id: int = attr.ib(metadata={"json": "iid"})
target_branch: str
source_branch: str
source: GitlabProject
source_project_id: int
target: GitlabProject
target_project_id: int
last_commit: GitlabCommit
author_id: int
title: str
description: str
work_in_progress: bool
url: str
state: str
created_at: datetime
updated_at: datetime
updated_by_id: Optional[int] = None
last_edited_at: Optional[datetime] = None
last_edited_by_id: Optional[int] = None
merge_commit_sha: Optional[str] = None
merge_error: Optional[str] = None
merge_status: Optional[str] = None
merge_user_id: Optional[int] = None
merge_when_pipeline_succeeds: Optional[bool] = False
time_estimate: Optional[int] = None
total_time_spent: Optional[int] = None
human_time_estimate: Optional[str] = None
human_total_time_spent: Optional[str] = None
head_pipeline_id: Optional[int] = None
milestone_id: Optional[int] = None
assignee_id: Optional[int] = None
assignee_ids: Optional[List[int]] = None
assignee: Optional[GitlabUser] = None
action: Optional[str] = None
class GitlabWikiPageAttributes(SerializableAttrs['GitlabWikiAttributes']):
title: str
content: str
format: str
slug: str
url: str
action: str
message: Optional[str] = None
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
@dataclass
class GitlabVariable(SerializableAttrs['GitlabVariable']):
key: str
value: str
@dataclass
class GitlabPipelineAttributes(SerializableAttrs['GitlabPipelineAttributes']):
id: int
ref: str
tag: bool
sha: str
before_sha: str
source: str
status: str
stages: List[str]
created_at: datetime
finished_at: datetime
duration: int
variables: List[GitlabVariable]
@dataclass
class GitlabArtifact(SerializableAttrs['GitlabArtifact']):
filename: str
size: int
@dataclass
class GitlabWiki(SerializableAttrs['GitlabWiki']):
web_url: str
git_ssh_url: str
git_http_url: str
path_with_namespace: str
default_branch: str
@dataclass
class GitlabMergeRequest(SerializableAttrs['GitlabMergeRequest']):
id: int
merge_request_id: int = attr.ib(metadata={"json": "iid"})
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
target_branch: str
source_branch: str
source_project_id: int
assignee_id: int
author_id: int
title: str
created_at: datetime
updated_at: datetime
milestone_id: int
state: str
merge_status: str
target_project_id: int
description: str
position: int
locked_at: datetime
source: GitlabSource
target: GitlabTarget
last_commit: GitlabCommit
work_in_progress: bool
assignee: GitlabAssignee
@dataclass
class GitlabBuild(SerializableAttrs['GitlabBuild']):
id: int
stage: str
name: str
status: str
created_at: datetime
started_at: datetime
finished_at: datetime
when: str
manual: bool
user: GitlabUser
runner: str
artifacts_file: GitlabArtifact
class GitlabEvent:
@property
def has_matrix_message(self) -> bool:
return True
@property
def matrix_message(self) -> Optional[str]:
return "Missing message content"
@property
def matrix_message_edit_id(self) -> Optional[str]:
return None
@dataclass
class GitlabPushEvent(SerializableAttrs['GitlabPushEvent'], GitlabEvent):
object_kind: str
before: str
after: str
ref: str
checkout_sha: str
user_id: int
user_name: str
user_email: str
user_avatar: str
project_id: int
project: GitlabProject
repository: GitlabRepository
commits: List[GitlabCommit]
total_commits_count: int
def format_commit(self, commit: GitlabCommit) -> str:
lines = commit.message.strip().split("\n")
message = lines[0][:80]
if len(lines[0]) > 80:
message += "…"
elif len(lines) > 1:
message += " (…)"
return f"* [{commit.id[:8]}]({self.project.web_url}/commit/{commit.id}): {message}"
@property
def pluralizer(self) -> str:
return "s" if self.total_commits_count != 1 else ""
@property
def branch(self) -> str:
return self.ref.replace("refs/heads/", "")
@property
def has_matrix_message(self) -> bool:
return True
@property
def matrix_message(self) -> str:
branch = self.branch
return (f"[{self.project.namespace} / {self.project.name}] {self.user_name}"
" force pushed to, created or deleted branch"
f" [{branch}]({self.project.web_url}/tree/{branch})")
return (f"[{self.project.namespace} / {self.project.name}] "
f"{self.total_commits_count} new commit{self.pluralizer} "
f"to [{branch}]({self.project.web_url}/tree/{branch}) "
f"by {self.user_name}\n\n"
+ "\n".join(self.format_commit(commit) for commit in reversed(self.commits)))
@dataclass
class GitlabTagEvent(SerializableAttrs['GitlabTagEvent'], GitlabEvent):
object_kind: int
before: str
after: str
ref: str
checkout_sha: str
user_id: int
user_name: str
user_avatar: str
project_id: int
project: GitlabProject
repository: GitlabRepository
commits: List[GitlabCommit]
total_commits_count: int
@property
def tag(self) -> str:
return self.ref.replace("refs/tags/", "")
@property
def has_matrix_message(self) -> bool:
return self.object_kind == "tag_push"
@property
def matrix_message(self) -> Optional[str]:
if self.object_kind != "tag_push":
tag = self.tag
return (f"[{self.project.namespace} / {self.project.name}] {self.user_name} created tag "
f"[{tag}]({self.project.web_url}/tags/{tag}) at commit {self.checkout_sha[:8]}")
def past_tense(action: str) -> str:
if not action:
return action
elif action[-2:-1] != "ed":
if action[-1] == "e":
return f"{action}d"
return f"{action}ed"
return action
@dataclass
class GitlabIssueEvent(SerializableAttrs['GitlabIssueEvent'], GitlabEvent):
object_kind: str
user: GitlabUser
project: GitlabProject
repository: GitlabRepository
assignees: Optional[List[GitlabAssignee]] = None
labels: Optional[List[GitlabLabel]] = None
changes: Optional[Dict[str, GitlabChange]] = None
@property
def has_matrix_message(self) -> bool:
return bool(self.object_attributes.action and self.object_attributes.action != "update")
@property
def matrix_message(self) -> Optional[str]:
action = past_tense(self.object_attributes.action)
if not action or action == "updated":
return (f"[{self.project.namespace} / {self.project.name}] {self.user.name} {action} "
f"{confidential}issue [{self.object_attributes.title} "
f"(#{self.object_attributes.issue_id})]({self.object_attributes.url})")
class GitlabCommentEvent(SerializableAttrs['GitlabCommentEvent'], GitlabEvent):
object_kind: str
user: GitlabUser
project_id: int
project: GitlabProject
repository: GitlabRepository
object_attributes: GitlabCommentAttributes
merge_request: Optional[GitlabMergeRequest] = None
commit: Optional[GitlabCommit] = None
issue: Optional[GitlabIssue] = None
snippet: Optional[GitlabSnippet] = None
@property
def has_matrix_message(self) -> bool:
nt = self.object_attributes.noteable_type
return (nt == "Issue" and self.issue) or (nt == "MergeRequest" and self.merge_request)
@property
def matrix_message(self) -> Optional[str]:
noteable_type = self.object_attributes.noteable_type
if self.issue and noteable_type == "Issue":
note_type = "issue"
id = f"#{self.issue.issue_id}"
elif self.merge_request and noteable_type == "MergeRequest":
note_type = "merge request"
id = f"!{self.merge_request.merge_request_id}"
title = self.merge_request.title
else:
return None
note = "\n".join(f"> {line}" for line in self.object_attributes.note.split("\n"))
return (f"[{self.project.namespace} / {self.project.name}] {self.user.name} "
f"[commented]({self.object_attributes.url}) on {note_type} {title} ({id}):\n\n"
f"{note}")
@dataclass
class GitlabMergeRequestEvent(SerializableAttrs['GitlabMergeRequestEvent'], GitlabEvent):
object_kind: str
user: GitlabUser
project: GitlabProject
repository: GitlabRepository
object_attributes: GitlabMergeRequestAttributes
labels: List[GitlabLabel]
changes: Dict[str, GitlabChange]
@property
def has_matrix_message(self) -> bool:
return self.object_attributes.action != "update"
@property
def matrix_message(self) -> Optional[str]:
action = past_tense(self.object_attributes.action)
if not action or action == "updated" or not self.object_attributes.target:
return (f"[{self.project.namespace} / {self.project.name}] {self.user.name} {action} "
f"merge request [{self.object_attributes.title} "
f"(!{self.object_attributes.merge_request_id})]({self.object_attributes.url})")
class GitlabWikiPageEvent(SerializableAttrs['GitlabWikiPageEvent'], GitlabEvent):
object_kind: str
user: GitlabUser
project: GitlabProject
wiki: GitlabWiki
object_attributes: GitlabWikiPageAttributes
@property
def has_matrix_message(self) -> bool:
return bool(self.object_attributes.action)
@property
def matrix_message(self) -> Optional[str]:
action = past_tense(self.object_attributes.action)
if not action:
return None
return (f"[{self.project.namespace} / {self.project.name}] {self.user.name} {action} "
f"page on wiki [{self.object_attributes.title}]({self.object_attributes.url})")
def pluralize(val: int, unit: str) -> str:
if val == 1:
return f"{val} {unit}"
return f"{val} {unit}s"
def format_duration(seconds: Union[int, float]) -> str:
seconds = round(seconds, 1)
minutes, seconds = divmod(seconds, 60)
hours, minutes = divmod(minutes, 60)
days, hours = divmod(hours, 24)
parts = []
if days > 0:
parts.append(pluralize(days, "day"))
if hours > 0:
parts.append(pluralize(hours, "hour"))
if minutes > 0:
parts.append(pluralize(minutes, "minute"))
if seconds > 0:
parts.append(pluralize(seconds, "second"))
if len(parts) == 1:
return "in " + parts[0]
return "in " + ", ".join(parts[:-1]) + f" and {parts[-1]}"
@dataclass
class GitlabPipelineEvent(SerializableAttrs['GitlabPipelineEvent'], GitlabEvent):
object_attributes: GitlabPipelineAttributes
user: GitlabUser
project: GitlabProject
commit: GitlabCommit
builds: List[GitlabBuild]
@property
def formatted_duration(self) -> str:
return format_duration(self.object_attributes.duration)
@property
def matrix_message_edit_id(self) -> str:
return f"pipeline-{self.object_attributes.id}"
@property
def has_matrix_message(self) -> bool:
return self.object_attributes.status in ("pending", "running", "success", "failed")
@property
def matrix_message(self) -> str:
type = "tag" if self.object_attributes.tag else "branch"
prefix = (f"[{self.project.namespace} / {self.project.name}] "
f"Pipeline {self.object_attributes.id} on {type} {self.object_attributes.ref}")
if self.object_attributes.status == "pending":
return f"{prefix} pending"
elif self.object_attributes.status == "running":
return f"{prefix} started"
builds = "\n".join(f"* [{build.name}:{build.stage} ({build.id})]"
f"({self.project.web_url}/-/jobs/{build.id}) - {build.status}"
for build in self.builds)
if self.object_attributes.status == "success":
return f"{prefix} successfully completed in {self.formatted_duration}\n\n{builds}"
elif self.object_attributes.status == "failed":
return f"{prefix} failed in {self.formatted_duration}\n\n{builds}"
@dataclass
class GitlabJobEvent(SerializableAttrs['GitlabJobEvent'], GitlabEvent):
object_kind: str
ref: str
tag: str
before_sha: str
sha: str
build_id: int
build_name: str
build_stage: str
build_status: str
build_started_at: datetime
build_finished_at: datetime
build_allow_failure: bool
build_failure_reason: str
project_id: int
project_name: str
user: GitlabUser
commit: GitlabCommit
repository: GitlabRepository
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
@property
def formatted_build_duration(self) -> str:
return format_duration(self.build_duration)
@property
def matrix_message_edit_id(self) -> str:
return f"job-{self.build_id}"
@property
def has_matrix_message(self) -> bool:
return self.build_status in ("pending", "running", "skipped", "success", "failed")
@property
def matrix_message(self) -> str:
prefix = (f"[{self.project_name}] Job [{self.build_name}:{self.build_stage} "
f"({self.build_id})]({self.repository.homepage}/-/jobs/{self.build_id}) ")
if self.build_status == "pending":
return f"{prefix} pending"
elif self.build_status == "running":
return f"{prefix} started"
elif self.build_status == "skipped":
return f"{prefix} skipped"
elif self.build_status == "success":
return f"{prefix} successfully completed in {self.formatted_build_duration}"
elif self.build_status == "failed":
return f"{prefix} failed in {self.formatted_build_duration}"
GitlabEventType = Union[Type[GitlabPushEvent],
Type[GitlabTagEvent],
Type[GitlabIssueEvent],
Type[GitlabCommentEvent],
Type[GitlabMergeRequestEvent],
Type[GitlabWikiPageEvent],
Type[GitlabPipelineEvent],
EventParse: Dict[str, GitlabEventType] = {
"Push Hook": GitlabPushEvent,
"Tag Push Hook": GitlabTagEvent,
"Issue Hook": GitlabIssueEvent,
"Confidential Issue Hook": GitlabIssueEvent,
"Note Hook": GitlabCommentEvent,
"Merge Request Hook": GitlabMergeRequestEvent,
"Wiki Page Hook": GitlabWikiPageEvent,
"Pipeline Hook": GitlabPipelineEvent,
"Job Hook": GitlabJobEvent
}