xyvs/wakapi-readme-stats
1
0
Fork 0
Code Issues 2 Pull Requests 1 Actions Packages Projects Releases 1 Wiki Activity

download manager improved to handle and cache all kinds of requests

Browse Source
This commit is contained in:
pseusys
2023-02-13 14:59:43 +01:00
parent c2f1fdc82d
commit 887ff2bdc8
6 changed files with 354 additions and 355 deletions
Download Patch File Download Diff File Expand all files Collapse all files

436
main.py
View File

@@ -5,15 +5,15 @@ import re
import os
import base64
from asyncio import run
from typing import Dict
from pytz import timezone
import pytz
import requests
from github import Github, GithubException, InputGitAuthor
from github import Github, GithubException, InputGitAuthor, AuthenticatedUser
import datetime
from string import Template
from download_manager import init_download_manager
from download_manager import init_download_manager, DownloadManager
from loc import LinesOfCode
import time
import traceback
@@ -58,107 +58,11 @@ commit_email = os.getenv('INPUT_COMMIT_EMAIL')
show_total_code_time = os.getenv('INPUT_SHOW_TOTAL_CODE_TIME')
symbol_version = os.getenv('INPUT_SYMBOL_VERSION').strip()
show_waka_stats = 'y'
# The GraphQL query to get commit data.
userInfoQuery = """
{
viewer {
login
email
id
}
}
"""
createContributedRepoQuery = Template("""query {
user(login: "$username") {
repositoriesContributedTo(last: 100, includeUserRepositories: true) {
nodes {
isFork
name
owner {
login
}
}
}
}
}
""")
createCommittedDateQuery = Template("""
query {
repository(owner: "$owner", name: "$name") {
defaultBranchRef {
target {
... on Commit {
history(first: 100, author: { id: "$id" }) {
edges {
node {
committedDate
}
}
}
}
}
}
}
}
""")
get_loc_url = Template("""/repos/$owner/$repo/stats/code_frequency""")
get_profile_view = Template("""/repos/$owner/$repo/traffic/views?per=week""")
get_profile_traffic = Template("""/repos/$owner/$repo/traffic/popular/referrers""")
truthy = ['true', '1', 't', 'y', 'yes']
def run_v3_api(query):
request = requests.get('https://api.github.com' + query, headers=headers)
if request.status_code == 200:
return request.json()
else:
raise Exception(
"Query failed to run by returning code of {}. {},... {}".format(request.status_code, query,
str(request.json())))
repositoryListQuery = Template("""
{
user(login: "$username") {
repositories(orderBy: {field: CREATED_AT, direction: ASC}, last: 100, affiliations: [OWNER, COLLABORATOR, ORGANIZATION_MEMBER], isFork: false) {
totalCount
edges {
node {
object(expression:"master") {
... on Commit {
history (author: { id: "$id" }){
totalCount
}
}
}
primaryLanguage {
color
name
id
}
stargazers {
totalCount
}
collaborators {
totalCount
}
createdAt
name
owner {
id
login
}
nameWithOwner
}
}
}
location
createdAt
name
}
}
""")
translate: Dict[str, str]
user: AuthenticatedUser
def millify(n):
@@ -172,14 +76,6 @@ def millify(n):
return '{:.0f}{}'.format(n / 10 ** (3 * millidx), millnames[millidx])
def run_query(query):
request = requests.post('https://api.github.com/graphql', json={'query': query}, headers=headers)
if request.status_code == 200:
return request.json()
else:
raise Exception("Query failed to run by returning code of {}. {}".format(request.status_code, query))
def make_graph(percent: float):
'''Make progress graph from API graph'''
if (symbol_version == '1'): # version 1
@@ -223,14 +119,10 @@ def make_commit_list(data: list):
return '\n'.join(data_list)
def generate_commit_list(tz):
async def generate_commit_list(tz):
string = ''
result = run_query(userInfoQuery) # Execute the query
username = result["data"]["viewer"]["login"]
id = result["data"]["viewer"]["id"]
# print("user {}".format(username))
result = run_query(createContributedRepoQuery.substitute(username=username))
result = await DownloadManager.get_remote_graphql("repositories_contributed_to", username=user.login)
nodes = result["data"]["user"]["repositoriesContributedTo"]["nodes"]
repos = [d for d in nodes if d['isFork'] is False]
@@ -248,42 +140,35 @@ def generate_commit_list(tz):
Sunday = 0
for repository in repos:
result = run_query(
createCommittedDateQuery.substitute(owner=repository["owner"]["login"], name=repository["name"], id=id))
try:
committed_dates = result["data"]["repository"]["defaultBranchRef"]["target"]["history"]["edges"]
for committedDate in committed_dates:
date = datetime.datetime.strptime(committedDate["node"]["committedDate"],
"%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=pytz.utc).astimezone(
timezone(tz))
hour = date.hour
weekday = date.strftime('%A')
if 6 <= hour < 12:
morning += 1
if 12 <= hour < 18:
daytime += 1
if 18 <= hour < 24:
evening += 1
if 0 <= hour < 6:
night += 1
result = await DownloadManager.get_remote_graphql("repository_committed_dates", owner=repository["owner"]["login"], name=repository["name"], id=user.node_id)
committed_dates = result["data"]["repository"]["defaultBranchRef"]["target"]["history"]["edges"]
for committedDate in committed_dates:
date = datetime.datetime.strptime(committedDate["node"]["committedDate"], "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=pytz.utc).astimezone(timezone(tz))
hour = date.hour
weekday = date.strftime('%A')
if 6 <= hour < 12:
morning += 1
if 12 <= hour < 18:
daytime += 1
if 18 <= hour < 24:
evening += 1
if 0 <= hour < 6:
night += 1
if weekday == "Monday":
Monday += 1
if weekday == "Tuesday":
Tuesday += 1
if weekday == "Wednesday":
Wednesday += 1
if weekday == "Thursday":
Thursday += 1
if weekday == "Friday":
Friday += 1
if weekday == "Saturday":
Saturday += 1
if weekday == "Sunday":
Sunday += 1
except Exception as ex:
if str(ex) != "'NoneType' object is not subscriptable":
print("Exception occurred " + str(ex))
if weekday == "Monday":
Monday += 1
if weekday == "Tuesday":
Tuesday += 1
if weekday == "Wednesday":
Wednesday += 1
if weekday == "Thursday":
Thursday += 1
if weekday == "Friday":
Friday += 1
if weekday == "Saturday":
Saturday += 1
if weekday == "Sunday":
Sunday += 1
sumAll = morning + daytime + evening + night
sum_week = Sunday + Monday + Tuesday + Friday + Saturday + Wednesday + Thursday
@@ -328,60 +213,54 @@ def generate_commit_list(tz):
return string
def get_waka_time_stats():
async def get_waka_time_stats():
stats = ''
request = requests.get(
f"https://wakatime.com/api/v1/users/current/stats/last_7_days?api_key={waka_key}")
no_activity = translate["No Activity Tracked This Week"]
if request.status_code == 401 or request.status_code != 200:
print("Error With WAKA time API returned " + str(request.status_code) + " Response " + str(request.json()))
else:
data = request.json()
if showCommit.lower() in truthy:
stats = stats + generate_commit_list(tz=data['data']['timezone']) + '\n\n'
data = await DownloadManager.get_remote_json("waka_latest")
if showCommit.lower() in truthy:
stats = stats + await generate_commit_list(data['data']['timezone']) + '\n\n'
if showTimeZone.lower() in truthy or showLanguage.lower() in truthy or showEditors.lower() in truthy or \
showProjects.lower() in truthy or showOs.lower() in truthy:
stats += '📊 **' + translate['This Week I Spend My Time On'] + '** \n\n'
stats += '```text\n'
if showTimeZone.lower() in truthy or showLanguage.lower() in truthy or showEditors.lower() in truthy or showProjects.lower() in truthy or showOs.lower() in truthy:
stats += '📊 **' + translate['This Week I Spend My Time On'] + '** \n\n'
stats += '```text\n'
if showTimeZone.lower() in truthy:
tzone = data['data']['timezone']
stats = stats + '⌚︎ ' + translate['Timezone'] + ': ' + tzone + '\n\n'
if showTimeZone.lower() in truthy:
tzone = data['data']['timezone']
stats = stats + '⌚︎ ' + translate['Timezone'] + ': ' + tzone + '\n\n'
if showLanguage.lower() in truthy:
if len(data['data']['languages']) == 0:
lang_list = no_activity
else:
lang_list = make_list(data['data']['languages'])
stats = stats + '💬 ' + translate['Languages'] + ': \n' + lang_list + '\n\n'
if showLanguage.lower() in truthy:
if len(data['data']['languages']) == 0:
lang_list = no_activity
else:
lang_list = make_list(data['data']['languages'])
stats = stats + '💬 ' + translate['Languages'] + ': \n' + lang_list + '\n\n'
if showEditors.lower() in truthy:
if len(data['data']['editors']) == 0:
edit_list = no_activity
else:
edit_list = make_list(data['data']['editors'])
stats = stats + '🔥 ' + translate['Editors'] + ': \n' + edit_list + '\n\n'
if showEditors.lower() in truthy:
if len(data['data']['editors']) == 0:
edit_list = no_activity
else:
edit_list = make_list(data['data']['editors'])
stats = stats + '🔥 ' + translate['Editors'] + ': \n' + edit_list + '\n\n'
if showProjects.lower() in truthy:
if len(data['data']['projects']) == 0:
project_list = no_activity
else:
# Re-order the project list by percentage
data['data']['projects'] = sorted(data['data']['projects'], key=lambda x: x["percent"],
if showProjects.lower() in truthy:
if len(data['data']['projects']) == 0:
project_list = no_activity
else:
# Re-order the project list by percentage
data['data']['projects'] = sorted(data['data']['projects'], key=lambda x: x["percent"],
reverse=True)
project_list = make_list(data['data']['projects'])
stats = stats + '🐱‍💻 ' + translate['Projects'] + ': \n' + project_list + '\n\n'
project_list = make_list(data['data']['projects'])
stats = stats + '🐱‍💻 ' + translate['Projects'] + ': \n' + project_list + '\n\n'
if showOs.lower() in truthy:
if len(data['data']['operating_systems']) == 0:
os_list = no_activity
else:
os_list = make_list(data['data']['operating_systems'])
stats = stats + '💻 ' + translate['operating system'] + ': \n' + os_list + '\n\n'
if showOs.lower() in truthy:
if len(data['data']['operating_systems']) == 0:
os_list = no_activity
else:
os_list = make_list(data['data']['operating_systems'])
stats = stats + '💻 ' + translate['operating system'] + ': \n' + os_list + '\n\n'
stats += '```\n\n'
stats += '```\n\n'
return stats
@@ -393,14 +272,12 @@ def generate_language_per_repo(result):
if repo['node']['primaryLanguage'] is None:
continue
language = repo['node']['primaryLanguage']['name']
color_code = repo['node']['primaryLanguage']['color']
total += 1
if language not in language_count.keys():
language_count[language] = {}
language_count[language]['count'] = 1
else:
language_count[language]['count'] = language_count[language]['count'] + 1
language_count[language]['color'] = color_code
data = []
sorted_labels = list(language_count.keys())
sorted_labels.sort(key=lambda x: language_count[x]['count'], reverse=True)
@@ -420,44 +297,43 @@ def generate_language_per_repo(result):
return '**' + title + '** \n\n' + '```text\n' + make_list(data) + '\n\n```\n'
def get_yearly_data():
repository_list = run_query(repositoryListQuery.substitute(username=username, id=id))
loc = LinesOfCode(id, username, ghtoken, repository_list, ignored_repos_name)
yearly_data = loc.calculateLoc()
async def get_yearly_data():
repository_list = await DownloadManager.get_remote_graphql("user_repository_list", username=user.login, id=user.node_id)
loc = LinesOfCode(user, ghtoken, repository_list, ignored_repos_name)
yearly_data = await loc.calculateLoc()
if showLocChart.lower() in truthy:
loc.plotLoc(yearly_data)
await loc.plotLoc(yearly_data)
return yearly_data
def get_line_of_code():
repositoryList = run_query(repositoryListQuery.substitute(username=username, id=id))
loc = LinesOfCode(id, username, ghtoken, repositoryList, ignored_repos_name)
yearly_data = loc.calculateLoc()
async def get_line_of_code() -> str:
repositoryList = await DownloadManager.get_remote_graphql("user_repository_list", username=user.login, id=user.node_id)
loc = LinesOfCode(user, ghtoken, repositoryList, ignored_repos_name)
yearly_data = await loc.calculateLoc()
total_loc = sum(
[yearly_data[year][quarter][lang] for year in yearly_data for quarter in yearly_data[year] for lang in
yearly_data[year][quarter]])
return millify(int(total_loc))
def get_short_info(github):
async def get_short_info():
string = '**🐱 ' + translate['My GitHub Data'] + '** \n\n'
user_info = github.get_user()
if user_info.disk_usage is None:
if user.disk_usage is None:
disk_usage = humanize.naturalsize(0)
print("Please add new github personal access token with user permission")
else:
disk_usage = humanize.naturalsize(user_info.disk_usage)
request = requests.get('https://github-contributions.vercel.app/api/v1/' + user_info.login)
if request.status_code == 200 and len(request.json()['years']) > 0:
this_year_data = request.json()['years'][0]
disk_usage = humanize.naturalsize(user.disk_usage)
data = await DownloadManager.get_remote_json("github_stats")
if len(data['years']) > 0:
this_year_data = data['years'][0]
total = this_year_data['total']
year = this_year_data['year']
string += '> 🏆 ' + translate['Contributions in the year'] % (humanize.intcomma(total), year) + '\n > \n'
string += '> 📦 ' + translate["Used in GitHub's Storage"] % disk_usage + ' \n > \n'
is_hireable = user_info.hireable
public_repo = user_info.public_repos
private_repo = user_info.owned_private_repos
is_hireable = user.hireable
public_repo = user.public_repos
private_repo = user.owned_private_repos
if private_repo is None:
private_repo = 0
if is_hireable:
@@ -475,52 +351,45 @@ def get_short_info(github):
return string
def get_stats(github):
async def get_stats(github) -> str:
'''Gets API data and returns markdown progress'''
stats = ''
repositoryList = run_query(repositoryListQuery.substitute(username=username, id=id))
repositoryList = await DownloadManager.get_remote_graphql("user_repository_list", username=user.login, id=user.node_id)
if show_loc.lower() in truthy or showLocChart.lower() in truthy:
# This condition is written to calculate the lines of code because it is heavy process soo needs to be calculate once this will reduce the execution time
yearly_data = get_yearly_data()
await get_yearly_data()
if show_total_code_time.lower() in truthy:
request = requests.get(
f"https://wakatime.com/api/v1/users/current/all_time_since_today?api_key={waka_key}")
if request.status_code == 401:
print("Error With WAKA time API returned " + str(request.status_code) + " Response " + str(request.json()))
elif "text" not in request.json()["data"]:
print("User stats are calculating. Try again later.")
else:
data = request.json()
stats += '![Code Time](http://img.shields.io/badge/' + quote(
str("Code Time")) + '-' + quote(str(
data['data']['text'])) + '-blue)\n\n'
data = await DownloadManager.get_remote_json("waka_all")
stats += '![Code Time](http://img.shields.io/badge/' + quote(
str("Code Time")) + '-' + quote(str(
data['data']['text'])) + '-blue)\n\n'
if show_profile_view.lower() in truthy:
data = run_v3_api(get_profile_view.substitute(owner=username, repo=username))
data = github.get_repo(f"{user.login}/{user.login}").get_views_traffic(per="week")
stats += '![Profile Views](http://img.shields.io/badge/' + quote(str(translate['Profile Views'])) + '-' + str(
data['count']) + '-blue)\n\n'
if show_loc.lower() in truthy:
stats += '![Lines of code](https://img.shields.io/badge/' + quote(
str(translate['From Hello World I have written'])) + '-' + quote(
str(get_line_of_code())) + '%20' + quote(str(translate['Lines of code'])) + '-blue)\n\n'
str(await get_line_of_code())) + '%20' + quote(str(translate['Lines of code'])) + '-blue)\n\n'
if show_short_info.lower() in truthy:
stats += get_short_info(github)
stats += await get_short_info()
if show_waka_stats.lower() in truthy:
stats += get_waka_time_stats()
stats += await get_waka_time_stats()
if showLanguagePerRepo.lower() in truthy:
stats = stats + generate_language_per_repo(repositoryList) + '\n\n'
if showLocChart.lower() in truthy:
stats += '**' + translate['Timeline'] + '**\n\n'
branch_name = github.get_repo(f'{username}/{username}').default_branch
stats = stats + '![Chart not found](https://raw.githubusercontent.com/' + username + '/' + username + '/' + branch_name + '/charts/bar_graph.png) \n\n'
branch_name = github.get_repo(f'{user.login}/{user.login}').default_branch
stats = stats + '![Chart not found](https://raw.githubusercontent.com/' + user.login + '/' + user.login + '/' + branch_name + '/charts/bar_graph.png) \n\n'
if show_updated_date.lower() in truthy:
now = datetime.datetime.utcnow()
@@ -530,10 +399,6 @@ def get_stats(github):
return stats
# def star_me():
# requests.put("https://api.github.com/user/starred/anmol098/waka-readme-stats", headers=headers)
def decode_readme(data: str):
'''Decode the contents of old readme'''
decoded_bytes = base64.b64decode(data)
@@ -546,53 +411,52 @@ def generate_new_readme(stats: str, readme: str):
return re.sub(listReg, stats_in_readme, readme)
if __name__ == '__main__':
run(init_download_manager())
async def main():
global translate, user
if ghtoken is None:
raise Exception('Token not available')
user = Github(ghtoken).get_user()
print(f"Current user: {user.login}")
await init_download_manager(waka_key, ghtoken, user)
try:
start_time = datetime.datetime.now().timestamp() * 1000
if ghtoken is None:
raise Exception('Token not available')
g = Github(ghtoken)
headers = {"Authorization": "Bearer " + ghtoken}
user_data = run_query(userInfoQuery) # Execute the query
if "errors" in user_data:
raise Exception(user_data)
username = user_data["data"]["viewer"]["login"]
id = user_data["data"]["viewer"]["id"]
email = user_data["data"]["viewer"]["email"]
print("Username " + username)
repo = g.get_repo(f"{username}/{username}")
contents = repo.get_readme()
try:
with open(os.path.join(os.path.dirname(__file__), 'translation.json'), encoding='utf-8') as config_file:
data = json.load(config_file)
translate = data[locale]
except Exception as e:
print("Cannot find the Locale choosing default to english")
translate = data['en']
waka_stats = get_stats(g)
# star_me()
rdmd = decode_readme(contents.content)
new_readme = generate_new_readme(stats=waka_stats, readme=rdmd)
if commit_by_me.lower() in truthy:
committer = InputGitAuthor(username or commit_username, email or commit_email)
else:
committer = InputGitAuthor(
commit_username or 'readme-bot',
commit_email or '41898282+github-actions[bot]@users.noreply.github.com'
)
if new_readme != rdmd:
try:
repo.update_file(path=contents.path, message=commit_message,
content=new_readme, sha=contents.sha, branch=branchName,
committer=committer)
except:
repo.update_file(path=contents.path, message=commit_message,
content=new_readme, sha=contents.sha, branch='main',
committer=committer)
print("Readme updated")
end_time = datetime.datetime.now().timestamp() * 1000
print("Program processed in {} miliseconds.".format(round(end_time - start_time, 0)))
with open(os.path.join(os.path.dirname(__file__), 'translation.json'), encoding='utf-8') as config_file:
data = json.load(config_file)
translate = data[locale]
except Exception as e:
traceback.print_exc()
print("Exception Occurred " + str(e))
print("Cannot find the Locale choosing default to english")
translate = data['en']
g = Github(ghtoken)
waka_stats = await get_stats(g)
repo = g.get_repo(f"{user.login}/{user.login}")
contents = repo.get_readme()
rdmd = decode_readme(contents.content)
new_readme = generate_new_readme(stats=waka_stats, readme=rdmd)
if commit_by_me.lower() in truthy:
committer = InputGitAuthor(user.login or commit_username, user.email or commit_email)
else:
committer = InputGitAuthor(
commit_username or 'readme-bot',
commit_email or '41898282+github-actions[bot]@users.noreply.github.com'
)
if new_readme != rdmd:
try:
repo.update_file(path=contents.path, message=commit_message,
content=new_readme, sha=contents.sha, branch=branchName,
committer=committer)
except:
repo.update_file(path=contents.path, message=commit_message,
content=new_readme, sha=contents.sha, branch='main',
committer=committer)
print("Readme updated")
if __name__ == '__main__':
start_time = datetime.datetime.now().timestamp() * 1000
run(main())
end_time = datetime.datetime.now().timestamp() * 1000
print(f"Program processed in {round(end_time - start_time, 0)} miliseconds.")