-
-
Notifications
You must be signed in to change notification settings - Fork 307
/
Copy pathmain.py
539 lines (447 loc) · 18.1 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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
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
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
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
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
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
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
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
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
"""WakaReadme : WakaTime progress visualizer.
Wakatime Metrics on your Profile Readme.
Title:
```txt
From: 15 February, 2022 - To: 22 February, 2022
````
Byline:
```txt
Total: 34 hrs 43 mins
```
Body:
```txt
Python 27 hrs 29 mins ⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣿⣦⣀⣀⣀⣀⣀ 77.83 %
YAML 2 hrs 14 mins ⣿⣦⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀ 06.33 %
Markdown 1 hr 54 mins ⣿⣤⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀ 05.39 %
TOML 1 hr 48 mins ⣿⣤⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀ 05.11 %
Other 35 mins ⣦⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀⣀ 01.68 %
```
Contents := Title + Byline + Body
"""
# standard
from base64 import b64encode
from dataclasses import dataclass
from datetime import datetime
from functools import partial
import logging as logger
import os
from random import SystemRandom
import re
import sys
from time import sleep
from typing import Any
# external
from faker import Faker
from github import ContentFile, Github, GithubException, InputGitAuthor, Repository
from requests import get as rq_get
from requests.exceptions import RequestException
################### setup ###################
print()
# hush existing loggers
for lgr_name in logger.root.manager.loggerDict:
# to disable log propagation completely set '.propagate = False'
logger.getLogger(lgr_name).setLevel(logger.WARNING)
# somehow github.Requester gets missed out from loggerDict
logger.getLogger("github.Requester").setLevel(logger.WARNING)
# configure logger
logger.basicConfig(
datefmt="%Y-%m-%d %H:%M:%S",
format="[%(asctime)s] ln. %(lineno)-3d %(levelname)-8s %(message)s",
level=logger.DEBUG,
)
try:
if len(sys.argv) == 2 and sys.argv[1] == "--dev":
# get env-vars from .env file for development
from dotenv import load_dotenv
# comment this out to disable colored logging
from loguru import logger
# load from .env before class def gets parsed
load_dotenv()
except ImportError as im_err:
logger.warning(im_err)
################### lib-func ###################
def strtobool(val: str | bool):
"""Strtobool.
PEP 632 https://www.python.org/dev/peps/pep-0632/ is depreciating distutils.
This is from the official source code with slight modifications.
Converts a string representation of truth to `True` or `False`.
Args:
val:
Value to be converted to bool.
Returns:
(Literal[True]):
If `val` is any of 'y', 'yes', 't', 'true', 'on', or '1'.
(Literal[False]):
If `val` is any of 'n', 'no', 'f', 'false', 'off', and '0'.
Raises:
ValueError: If `val` is anything else.
"""
if isinstance(val, bool):
return val
val = val.lower()
if val in {"y", "yes", "t", "true", "on", "1"}:
return True
if val in {"n", "no", "f", "false", "off", "0"}:
return False
raise ValueError(f"invalid
8000
truth value for {val}")
################### data ###################
@dataclass(slots=True)
class WakaInput:
"""WakaReadme Input Env Variables."""
# constants
prefix_length: int = 16
graph_length: int = 25
# mapped environment variables
# # required
gh_token: str | None = os.getenv("INPUT_GH_TOKEN")
waka_key: str | None = os.getenv("INPUT_WAKATIME_API_KEY")
api_base_url: str | None = os.getenv("INPUT_API_BASE_URL", "https://wakatime.com/api")
repository: str | None = os.getenv("INPUT_REPOSITORY")
# # depends
commit_message: str = os.getenv(
"INPUT_COMMIT_MESSAGE", "Updated WakaReadme graph with new metrics"
)
code_lang: str = os.getenv("INPUT_CODE_LANG", "txt")
_section_name: str = os.getenv("INPUT_SECTION_NAME", "waka")
start_comment: str = f"<!--START_SECTION:{_section_name}-->"
end_comment: str = f"<!--END_SECTION:{_section_name}-->"
waka_block_pattern: str = f"{start_comment}[\\s\\S]+{end_comment}"
# # optional
show_title: str | bool = os.getenv("INPUT_SHOW_TITLE") or False
block_style: str = os.getenv("INPUT_BLOCKS", "░▒▓█")
time_range: str = os.getenv("INPUT_TIME_RANGE", "last_7_days")
show_time: str | bool = os.getenv("INPUT_SHOW_TIME") or False
show_total_time: str | bool = os.getenv("INPUT_SHOW_TOTAL") or False
show_masked_time: str | bool = os.getenv("INPUT_SHOW_MASKED_TIME") or False
language_count: str | int = os.getenv("INPUT_LANG_COUNT") or 5
stop_at_other: str | bool = os.getenv("INPUT_STOP_AT_OTHER") or False
ignored_languages: str = os.getenv("INPUT_IGNORED_LANGUAGES", "")
# # optional meta
target_branch: str = os.getenv("INPUT_TARGET_BRANCH", "NOT_SET")
target_path: str = os.getenv("INPUT_TARGET_PATH", "NOT_SET")
committer_name: str = os.getenv("INPUT_COMMITTER_NAME", "NOT_SET")
committer_email: str = os.getenv("INPUT_COMMITTER_EMAIL", "NOT_SET")
author_name: str = os.getenv("INPUT_AUTHOR_NAME", "NOT_SET")
author_email: str = os.getenv("INPUT_AUTHOR_EMAIL", "NOT_SET")
def validate_input(self):
"""Validate Input Env Variables."""
logger.debug("Validating input variables")
if not self.gh_token or not self.waka_key or not self.api_base_url or not self.repository:
logger.error("Invalid inputs")
logger.info("Refer https://github.com/athul/waka-readme")
return False
if len(self.commit_message) < 1:
logger.error("Commit message length must be greater than 1 character long")
return False
try:
self.show_title = strtobool(self.show_title)
self.show_time = strtobool(self.show_time)
self.show_total_time = strtobool(self.show_total_time)
self.show_masked_time = strtobool(self.show_masked_time)
self.stop_at_other = strtobool(self.stop_at_other)
except (ValueError, AttributeError) as err:
logger.error(err)
return False
if not self._section_name.isalnum():
logger.warning("Section name must be in any of [[a-z][A-Z][0-9]]")
logger.debug("Using default section name: waka")
self._section_name = "waka"
self.start_comment = f"<!--START_SECTION:{self._section_name}-->"
self.end_comment = f"<!--END_SECTION:{self._section_name}-->"
self.waka_block_pattern = f"{self.start_comment}[\\s\\S]+{self.end_comment}"
if len(self.block_style) < 2:
logger.warning("Graph block must be longer than 2 characters")
logger.debug("Using default blocks: ░▒▓█")
self.block_style = "░▒▓█"
if self.time_range not in {
"last_7_days",
"last_30_days",
"last_6_months",
"last_year",
"all_time",
}: # "all_time" is un-documented, should it be used?
logger.warning("Invalid time range")
logger.debug("Using default time range: last_7_days")
self.time_range = "last_7_days"
try:
self.language_count = int(self.language_count)
if self.language_count < -1:
raise ValueError
except ValueError:
logger.warning("Invalid language count")
logger.debug("Using default language count: 5")
self.language_count = 5
for option in (
"target_branch",
"target_path",
"committer_name",
"committer_email",
"author_name",
"author_email",
):
if not getattr(self, option):
logger.warning(f"Improper '{option}' configuration")
logger.debug(f"Using default '{option}'")
setattr(self, option, "NOT_SET")
logger.debug("Input validation complete\n")
return True
################### logic ###################
def make_title(dawn: str | None, dusk: str | None, /):
"""WakaReadme Title.
Makes title for WakaReadme.
"""
logger.debug("Making title")
if not dawn or not dusk:
logger.error("Cannot find start/end date\n")
sys.exit(1)
api_dfm, msg_dfm = "%Y-%m-%dT%H:%M:%SZ", "%d %B %Y"
try:
start_date = datetime.strptime(dawn, api_dfm).strftime(msg_dfm)
end_date = datetime.strptime(dusk, api_dfm).strftime(msg_dfm)
except ValueError as err:
logger.error(f"{err}\n")
sys.exit(1)
logger.debug("Title was made\n")
return f"From: {start_date} - To: {end_date}"
def make_graph(block_style: str, percent: float, gr_len: int, lg_nm: str = "", /):
"""WakaReadme Graph.
Makes time graph from the API's data.
"""
logger.debug(f"Generating graph for '{lg_nm or '...'}'")
markers = len(block_style) - 1
proportion = percent / 100 * gr_len
graph_bar = block_style[-1] * int(proportion + 0.5 / markers)
remainder_block = int((proportion - len(graph_bar)) * markers + 0.5)
graph_bar += block_style[remainder_block] if remainder_block > 0 else ""
graph_bar += block_style[0] * (gr_len - len(graph_bar))
logger.debug(f"'{lg_nm or '...'}' graph generated")
return graph_bar
def _extract_ignored_languages():
if not wk_i.ignored_languages:
return ""
temp = ""
for igl in wk_i.ignored_languages.strip().split():
if igl.startswith(('"', "'")):
temp = igl.lstrip('"').lstrip("'")
continue
if igl.endswith(('"', "'")):
igl = f"{temp} {igl.rstrip('"').rstrip("'")}"
temp = ""
yield igl
def prep_content(stats: dict[str, Any], /):
"""WakaReadme Prepare Markdown.
Prepared markdown content from the fetched statistics.
```
"""
logger.debug("Making contents")
contents = ""
# make title
if wk_i.show_title:
contents += make_title(stats.get("start"), stats.get("end")) + "\n\n"
# make byline
if wk_i.show_masked_time and (
total_time := stats.get("human_readable_total_including_other_language")
):
# overrides "human_readable_total"
contents += f"Total Time: {total_time}\n\n"
elif wk_i.show_total_time and (total_time := stats.get("human_readable_total")):
contents += f"Total Time: {total_time}\n\n"
lang_info: list[dict[str, int | float | str]] | None = []
# Check if any language data exists
if not (lang_info := stats.get("languages")):
logger.debug("The API data seems to be empty, please wait for a day")
contents += "No activity tracked"
return contents.rstrip("\n")
# make lang content
pad_len = len(
# comment if it feels way computationally expensive
max((str(lng["name"]) for lng in lang_info), key=len)
# and then do not for get to set `pad_len` to say 13 :)
)
language_count, stop_at_other = int(wk_i.language_count), bool(wk_i.stop_at_other)
if language_count == 0 and not wk_i.stop_at_other:
logger.debug(
"Set INPUT_LANG_COUNT to -1 to retrieve all language"
+ " or specify a positive number (ie. above 0)"
)
return contents.rstrip("\n")
ignored_languages = set(_extract_ignored_languages())
logger.debug(f"Ignoring {', '.join(ignored_languages)}")
for idx, lang in enumerate(lang_info):
lang_name = str(lang["name"])
if lang_name in ignored_languages:
continue
lang_time = str(lang["text"]) if wk_i.show_time else ""
lang_ratio = float(lang["percent"])
lang_bar = make_graph(wk_i.block_style, lang_ratio, wk_i.graph_length, lang_name)
contents += (
f"{lang_name.ljust(pad_len)} "
+ f"{lang_time: <16}{lang_bar} "
+ f"{lang_ratio:.2f}".zfill(5)
+ " %\n"
)
if language_count == -1:
continue
if stop_at_other and (lang_name == "Other"):
break
if idx + 1 >= language_count > 0: # idx starts at 0
break
logger.debug("Contents were made\n")
return contents.rstrip("\n")
def fetch_stats():
"""WakaReadme Fetch Stats.
Returns statistics as JSON string.
"""
attempts = 4
statistic: dict[str, dict[str, Any]] = {}
encoded_key = str(b64encode(bytes(str(wk_i.waka_key), "utf-8")), "utf-8")
logger.debug(f"Pulling WakaTime stats from {' '.join(wk_i.time_range.split('_'))}")
while attempts > 0:
resp_message, fake_ua = "", cryptogenic.choice([str(fake.user_agent()) for _ in range(5)])
# making a request
if (
resp := rq_get(
url=f"{str(wk_i.api_base_url).rstrip('/')}/v1/users/current/stats/{wk_i.time_range}",
headers={
"Authorization": f"Basic {encoded_key}",
"User-Agent": fake_ua,
},
timeout=(30.0 * (5 - attempts)),
)
).status_code != 200:
resp_message += f" • {conn_info}" if (conn_info := resp.json().get("message")) else ""
logger.debug(
f"API response #{5 - attempts}: {resp.status_code} •" + f" {resp.reason}{resp_message}"
)
if resp.status_code == 200 and (statistic := resp.json()):
logger.debug("Fetched WakaTime statistics")
break
logger.debug(f"Retrying in {30 * (5 - attempts )}s ...")
sleep(30 * (5 - attempts))
attempts -= 1
if err := (statistic.get("error") or statistic.get("errors")):
logger.error(f"{err}\n")
sys.exit(1)
print()
return statistic.get("data")
def churn(old_readme: str, /):
"""WakaReadme Churn.
Composes WakaTime stats within markdown code snippet.
"""
# check if placeholder pattern exists in readme
if not re.findall(wk_i.waka_block_pattern, old_readme):
logger.warning(f"Cannot find `{wk_i.waka_block_pattern}` pattern in readme")
return None
# getting contents
if not (waka_stats := fetch_stats()):
logger.error("Unable to fetch data, please rerun workflow\n")
sys.exit(1)
# preparing contents
try:
generated_content = prep_content(waka_stats)
except (AttributeError, KeyError, ValueError) as err:
logger.error(f"Unable to read API data | {err}\n")
sys.exit(1)
print(generated_content, "\n", sep="")
# substituting old contents
new_readme = re.sub(
pattern=wk_i.waka_block_pattern,
repl=f"{wk_i.start_comment}\n\n```{wk_i.code_lang}\n{generated_content}\n```\n\n{wk_i.end_comment}",
string=old_readme,
)
if len(sys.argv) == 2 and sys.argv[1] == "--dev":
logger.debug("Detected run in `dev` mode.")
# to avoid accidentally writing back to Github
# when developing or testing waka-readme
return None
return None if new_readme == old_readme else new_readme
def qualify_target(gh_repo: Repository.Repository):
"""Qualify target repository defaults."""
@dataclass
class TargetRepository:
this: ContentFile.ContentFile
path: str
commit_message: str
sha: str
branch: str
committer: InputGitAuthor | None
author: InputGitAuthor | None
gh_branch = gh_repo.default_branch
if wk_i.target_branch != "NOT_SET":
gh_branch = gh_repo.get_branch(wk_i.target_branch)
target = gh_repo.get_readme()
if wk_i.target_path != "NOT_SET":
target = gh_repo.get_contents(
path=wk_i.target_path,
ref=gh_branch if isinstance(gh_branch, str) else gh_branch.commit.sha,
)
if isinstance(target, list):
raise RuntimeError("Cannot handle multiple files.")
committer, author = None, None
if wk_i.committer_name != "NOT_SET" and wk_i.committer_email != "NOT_SET":
committer = InputGitAuthor(name=wk_i.committer_name, email=wk_i.committer_email)
if wk_i.author_name != "NOT_SET" and wk_i.author_email != "NOT_SET":
author = InputGitAuthor(name=wk_i.author_name, email=wk_i.author_email)
return TargetRepository(
this=target,
path=target.path,
commit_message=wk_i.commit_message,
sha=target.sha,
branch=gh_branch if isinstance(gh_branch, str) else gh_branch.name,
committer=committer,
author=author,
)
def genesis():
"""Run Program."""
logger.debug("Connecting to GitHub")
gh_connect = Github(wk_i.gh_token)
# since a validator is being used earlier, casting
# `wk_i.ENV_VARIABLE` to a string here, is okay
gh_repo = gh_connect.get_repo(str(wk_i.repository))
target = qualify_target(gh_repo)
logger.debug("Decoding readme contents\n")
readme_contents = str(target.this.decoded_content, encoding="utf-8")
if not (new_content := churn(readme_contents)):
logger.info("WakaReadme was not updated")
return
logger.debug("WakaReadme stats has changed")
update_metric = partial(
gh_repo.update_file,
path=target.path,
message=target.commit_message,
content=new_content,
sha=target.sha,
branch=target.branch,
)
if target.committer:
update_metric = partial(update_metric, committer=target.committer)
if target.author:
update_metric = partial(update_metric, author=target.author)
update_metric()
logger.info("Stats updated successfully")
################### driver ###################
if __name__ == "__main__":
# faker data preparation
fake = Faker()
Faker.seed(0)
cryptogenic = SystemRandom()
# initial waka-readme setup
logger.debug("Initialize WakaReadme")
wk_i = WakaInput()
if not wk_i.validate_input():
logger.error("Environment variables are misconfigured\n")
sys.exit(1)
# run
try:
genesis()
except KeyboardInterrupt:
print("\r", end=" ")
logger.error("Interrupt signal received\n")
sys.exit(1)
except RuntimeError as err:
logger.error(f"{type(err).__name__}: {err}\n")
sys.exit(1)
except (GithubException, RequestException) as rq_exp:
logger.critical(f"{rq_exp}\n")
sys.exit(1)
print("\nThanks for using WakaReadme!\n")