maint: misc. refactorings, fixes, updates

- puts tests in the correct location of github workflow
- fixes `.env` values not loading via `load_dotenv` - required early loading
- corrects many static type linting errors
- combines all inputs into a single class with validation
- formats markdown & python files, as well as output
- slightly improved log messages, caught potential attribute error
- updates dependencies
This commit is contained in:
Jovial Joe Jayarson
2022-11-30 13:09:00 +05:30
parent 72af24c8af
commit bd7707fc5a
7 changed files with 255 additions and 239 deletions

346
main.py
View File

@@ -54,121 +54,59 @@ from github import GithubException, Github
from faker import Faker
# pylint: disable=logging-fstring-interpolation
################### data ###################
@dataclass(frozen=True, slots=True)
class WakaConstants:
"""
WakaConstants
-------------
"""
prefix_length: int = 16
graph_length: int = 25
section: str = os.getenv("INPUT_SECTION_NAME", "waka")
start_comment: str = f'<!--START_SECTION:{section}-->'
end_comment: str = f'<!--END_SECTION:{section}-->'
waka_block_pattern: str = f'{start_comment}[\\s\\S]+{end_comment}'
def validate_constants(self) -> bool:
if not (self.section):
logger.error('Invalid section name input, refer README')
return False
# pylint: disable = logging-fstring-interpolation
class WakaInput:
"""
WakaInput Env Vars
------------------
"""
def __init__(self) -> None:
"""
WakaInput Initialize
--------------------
"""
# mapped environment variables
# # required
self.gh_token: str = os.getenv('INPUT_GH_TOKEN')
self.waka_key: str = os.getenv('INPUT_WAKATIME_API_KEY')
self.api_base_url: str = os.getenv(
'INPUT_API_BASE_URL', 'https://wakatime.com/api'
)
self.repository: str = os.getenv('INPUT_REPOSITORY')
# # depends
self.commit_message: str = os.getenv(
'INPUT_COMMIT_MESSAGE', 'Updated WakaReadme graph with new metrics'
)
# # optional
self.show_title: str | bool = os.getenv('INPUT_SHOW_TITLE', 'False')
self.block_style: str = os.getenv('INPUT_BLOCKS', '░▒▓█')
self.time_range: str = os.getenv('INPUT_TIME_RANGE', 'last_7_days')
self.show_time: str | bool = os.getenv('INPUT_SHOW_TIME', 'False')
self.show_total_time: str | bool = os.getenv(
'INPUT_SHOW_TOTAL', 'False'
)
self.show_masked_time: str | bool = os.getenv(
'INPUT_SHOW_MASKED_TIME', 'False'
)
def validate_input(self) -> bool:
"""
WakaInput Validate
------------------
"""
if not (self.gh_token and self.waka_key and self.api_base_url and self.repository):
logger.error('Invalid required input(s), refer 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: bool = strtobool(self.show_title)
self.show_time: bool = strtobool(self.show_time)
self.show_total_time: bool = strtobool(self.show_total_time)
self.show_masked_time: bool = strtobool(self.show_masked_time)
except (ValueError, AttributeError) as err:
logger.error(err)
return False
if len(self.block_style) < 2:
logger.warning(
'Block length should be greater than 2 characters long'
)
logger.debug('Using default blocks: ░▒▓█')
# 'all_time' is un-documented, should it be used?
if self.time_range not in {
'last_7_days', 'last_30_days', 'last_6_months', 'last_year', 'all_time'
}:
logger.warning('Invalid time range')
logger.debug('Using default time range: last_7_days')
self.time_range: str = 'last_7_days'
return True
################### setup ###################
def strtobool(val: str) -> bool:
print()
# hush existing loggers
# pylint: disable = no-member # see: https://stackoverflow.com/q/20965287
for lgr_name in logger.root.manager.loggerDict:
# to disable log propagation completely set '.propagate = False'
logger.getLogger(lgr_name).setLevel(logger.WARNING)
# pylint: enable = no-member
# 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) -> bool:
"""
strtobool
---------
PEP 632 https://www.python.org/dev/peps/pep-0632/ is depreciating distutils
PEP 632 https://www.python.org/dev/peps/pep-0632/ is depreciating distutils.
This is from the official source code with slight modifications.
Following code is somewhat shamelessly copied from the original source.
Convert a string representation of truth to True or False.
Converts a string representation of truth to True or False.
- True values are `'y', 'yes', 't', 'true', 'on', and '1'`
- False values are `'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'}:
@@ -180,9 +118,97 @@ def strtobool(val: str) -> bool:
raise ValueError(f'invalid 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'
)
_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
def validate_input(self) -> bool:
"""
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)
except ValueError 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'
logger.debug('Input validation complete\n')
return True
################### logic ###################
def make_title(dawn: str, dusk: str, /) -> str:
def make_title(dawn: str | None, dusk: str | None, /) -> str:
"""
WakaReadme Title
----------------
@@ -190,24 +216,22 @@ def make_title(dawn: str, dusk: str, /) -> str:
Makes title for WakaReadme.
"""
logger.debug('Making title')
if not (dawn or dusk):
logger.error('Cannot find start/end date')
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(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: str, /,
*, lg_nm: str = ''
) -> str:
def make_graph(block_style: str, percent: float, gr_len: int, lg_nm: str = '', /) -> str:
"""
WakaReadme Graph
----------------
@@ -215,19 +239,20 @@ def make_graph(
Makes time graph from the API's data.
"""
logger.debug(f'Generating graph for "{lg_nm or "..."}"')
markers: int = len(block_style) - 1
proportion: float = percent / 100 * gr_len
graph_bar: str = block_style[-1] * int(proportion + 0.5 / markers)
remainder_block: int = int(
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 prep_content(stats: dict | None, /) -> str:
def prep_content(stats: dict[Any, Any], /) -> str:
"""
WakaReadme Prepare Markdown
---------------------------
@@ -235,7 +260,7 @@ def prep_content(stats: dict | None, /) -> str:
Prepared markdown content from the fetched statistics
```
"""
contents: str = ''
contents = ''
# Check if any data exists
if not (lang_info := stats.get('languages')):
@@ -262,18 +287,17 @@ def prep_content(stats: dict | None, /) -> str:
logger.debug('Making contents')
pad_len = len(
# comment if it feels way computationally expensive
max((str(l.get('name')) for l in lang_info), key=len)
max((l.get('name') for l in lang_info), key=len)
# and then don't for get to set pad_len to say 13 :)
)
for idx, lang in enumerate(lang_info):
lang_name: str = lang.get('name')
lang_name = lang.get('name')
# >>> add languages to filter here <<<
# if lang_name in {...}: continue
lang_time: str = lang.get('text') if wk_i.show_time else ''
lang_ratio: float = lang.get('percent')
lang_bar: str = make_graph(
wk_i.block_style, lang_ratio, wk_c.graph_length,
lg_nm=lang_name
lang_time = lang.get('text') if wk_i.show_time else ''
lang_ratio = lang.get('percent')
lang_bar = make_graph(
wk_i.block_style, lang_ratio, wk_i.graph_length, lang_name
)
contents += (
f'{lang_name.ljust(pad_len)} ' +
@@ -287,34 +311,39 @@ def prep_content(stats: dict | None, /) -> str:
return contents.rstrip('\n')
def fetch_stats() -> Any:
def fetch_stats() -> dict[Any, Any] | None:
"""
WakaReadme Fetch Stats
----------------------
Returns statistics as JSON string
"""
attempts, statistic = 4, {}
encoded_key: str = str(b64encode(bytes(wk_i.waka_key, 'utf-8')), 'utf-8')
attempts = 4
statistic: dict[str, dict[Any, 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 = None, cryptogenic.choice(
resp_message, fake_ua = '', cryptogenic.choice(
[str(fake.user_agent()) for _ in range(5)]
)
# making a request
if (resp := rq_get(
url=f'{wk_i.api_base_url.rstrip("/")}/v1/users/current/stats/{wk_i.time_range}',
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 * (5 - attempts)
)).status_code != 200:
resp_message = f'{resp.json().get("message")}'
resp_message += f' {conn_info}' if (
conn_info := resp.json().get("message")
) else ''
logger.debug(
f'API response #{5 - attempts}: {resp.status_code}{resp.reason} {resp_message or ""}'
f'API response #{5 - attempts}: {resp.status_code}{resp.reason}{resp_message}'
)
if resp.status_code == 200 and (statistic := resp.json()):
logger.debug('Fetched WakaTime statistics')
@@ -324,7 +353,7 @@ def fetch_stats() -> Any:
attempts -= 1
if err := (statistic.get('error') or statistic.get('errors')):
logger.error(err)
logger.error(f'{err}\n')
sys.exit(1)
print()
@@ -340,15 +369,18 @@ def churn(old_readme: str, /) -> str | None:
"""
# getting content
if not (waka_stats := fetch_stats()):
logger.error('Unable to fetch data, please rerun workflow')
logger.error('Unable to fetch data, please rerun workflow\n')
sys.exit(1)
# processing content
generated_content = prep_content(waka_stats)
try:
generated_content = prep_content(waka_stats)
except AttributeError as err:
logger.error(f'Unable to read API data | {err}\n')
sys.exit(1)
print(generated_content, '\n', sep='')
new_readme = re.sub(
pattern=wk_c.waka_block_pattern,
repl=f'{wk_c.start_comment}\n\n```text\n{generated_content}\n```\n\n{wk_c.end_comment}',
pattern=wk_i.waka_block_pattern,
repl=f'{wk_i.start_comment}\n\n```text\n{generated_content}\n```\n\n{wk_i.end_comment}',
string=old_readme
)
if len(sys.argv) == 2 and sys.argv[1] == '--dev':
@@ -356,7 +388,7 @@ def churn(old_readme: str, /) -> str | None:
# to avoid accidentally writing back to Github
# when developing and testing WakaReadme
return None
return None if new_readme == old_readme else new_readme
@@ -364,7 +396,8 @@ def genesis() -> None:
"""Run Program"""
logger.debug('Connecting to GitHub')
gh_connect = Github(wk_i.gh_token)
gh_repo = gh_connect.get_repo(wk_i.repository)
# since a validator is being used casting to string here is okay
gh_repo = gh_connect.get_repo(str(wk_i.repository))
readme_file = gh_repo.get_readme()
logger.debug('Decoding readme contents\n')
readme_contents = str(readme_file.decoded_content, encoding='utf-8')
@@ -378,34 +411,11 @@ def genesis() -> None:
)
logger.info('Stats updated successfully')
return
logger.info('WakaReadme was not updated')
################### driver ###################
print()
# hush existing loggers
# pylint: disable = no-member # see: https://stackoverflow.com/q/20965287
for lgr_name in logger.root.manager.loggerDict:
# to disable log propagation completely set '.propagate = False'
logger.getLogger(lgr_name).setLevel(logger.WARNING)
# pylint: enable = no-member
# 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_dotenv()
except ImportError as im_err:
logger.warning(im_err)
if __name__ == '__main__':
@@ -416,21 +426,19 @@ if __name__ == '__main__':
# initial waka-readme setup
logger.debug('Initialize WakaReadme')
wk_c = WakaConstants()
wk_i = WakaInput()
if not (wk_i.validate_input() or wk_i.validate_constants()):
logger.error('Environment variables are misconfigured')
if not wk_i.validate_input():
logger.error('Environment variables are misconfigured\n')
sys.exit(1)
logger.debug('Input validation complete')
# run
try:
genesis()
except KeyboardInterrupt:
print()
logger.error('Interrupt signal received')
logger.error('Interrupt signal received\n')
sys.exit(1)
except (GithubException, RequestException) as rq_exp:
logger.critical(rq_exp)
logger.critical(f'{rq_exp}\n')
sys.exit(1)
print('\nThanks for using WakaReadme!')
print('\nThanks for using WakaReadme!\n')