import requests import json import datetime from dotenv import load_dotenv from os import getenv from month_view import * import sys load_dotenv() ACCESS_TOKEN = getenv("TOKEN") URL = getenv("URL") USERNAMES = ["kalinom6"] CERT = getenv("CERT_LOCATION") WEEKS_BACK = 0 MONTH_VIEW = False MONTHS_BACK = 0 VERBOSE = True def get_flag_value(flag): return flag.split("=")[1] if "--monthview" in sys.argv: MONTH_VIEW = True for arg in sys.argv: if "--weeksback" in arg and not MONTH_VIEW: WEEKS_BACK = int(get_flag_value(arg)) if "--monthsback" in arg and MONTH_VIEW: MONTHS_BACK = int(get_flag_value(arg)) if "--username" in arg: USERNAMES[0] = get_flag_value(arg) if ',' in USERNAMES[0]: USERNAMES = USERNAMES[0].split(',') if "--verbose" in sys.argv: VERBOSE = True headers = { "Authorization": f"Bearer {ACCESS_TOKEN}", "Content-Type": "application/json" } def send_request(url, method, payload=None): if method == "GET": response = requests.get(url, headers=headers, data=json.dumps(payload), verify=CERT) if method == "POST": response = requests.post(url, headers=headers, data=json.dumps(payload), verify=CERT) if response.status_code == 200: return response.json() print(f"Error: {response.status_code}, {response.text}") def get_issues(username): jql = f"assignee WAS '{username}'" payload = { "jql": jql, "startAt": 0, "maxResults": 999, "fields": [ "key", "summary" ] } data = send_request(f'{URL}/search', "POST", payload) if data: tasks = [] for issue in data.get("issues", []): task = { "task_id": issue["key"], "summary": issue["fields"]["summary"], "task_link": f"https://globaljira.roche.com/browse/{issue['key']}" } tasks.append(task) return tasks def get_all_worklogs(issues): worklogs = [] for issue in issues: data = send_request(f'{URL}/issue/{issue["task_id"]}/worklog', "GET") if data: for log in data.get("worklogs", []): worklog = { "ticket_id": issue["task_id"], "jira_link": issue["task_link"], "time_spent": log["timeSpent"], "date": log["started"], "author": log["author"]["name"] } if worklog["author"] in USERNAMES: worklogs.append(worklog) worklogs.sort(key=date_getter) return worklogs def date_getter(worklog): return worklog['date'] def get_days(day): if WEEKS_BACK: day = day - datetime.timedelta(weeks=WEEKS_BACK) year, weeknum, day_of_week = day.isocalendar() while day_of_week != 1: day = day - datetime.timedelta(days=1) year, weeknum, day_of_week = day.isocalendar() days = [] current_month = day.month while day_of_week <= 7: if day.month != current_month: break # Stop collecting if month changes days.append(str(day)) day = day + datetime.timedelta(days=1) day_of_week += 1 days_second_part = [] if day_of_week <= 7: # This condition means the loop ended because the month changed next_month = day.month while day_of_week <= 7 and day.month == next_month: days_second_part.append(str(day)) day = day + datetime.timedelta(days=1) day_of_week += 1 return days, days_second_part def get_worklogs_for_days(worklogs, days): timeperiod_tickets = [] for worklog in worklogs: worklog_date = datetime.datetime.strptime( worklog["date"].split('T')[0], '%Y-%m-%d').date() if str(worklog_date) in days: timeperiod_tickets.append(worklog) return timeperiod_tickets def main(): print('The script starts its work...') issues = [] worklogs = [] if MONTH_VIEW: for username in USERNAMES: issues = get_issues(username) all_worklogs = get_all_worklogs(issues) days = get_days_of_month(MONTHS_BACK) print(all_worklogs) worklogs = get_worklogs_for_days(all_worklogs, days) print(worklogs) month_view = get_month_view(worklogs) print(month_view) return for username in USERNAMES: issues = get_issues(username) all_worklogs = get_all_worklogs(issues) today = datetime.date.today() days, days_second_part = get_days(today) worklogs = get_worklogs_for_days(all_worklogs, days) omnimat_string = "" print(worklogs) for worklog in worklogs: if worklog["ticket_id"] not in omnimat_string: omnimat_string += worklog["ticket_id"] + '\n' if days_second_part: second_omnimat_string = "" worklogs_second_part = get_worklogs_for_days(all_worklogs, days_second_part) for worklog in worklogs: if worklog["ticket_id"] not in second_omnimat_string: second_omnimat_string += worklog["ticket_id"] + '\n' if VERBOSE: print('Issues of the user:') print(json.dumps(issues, indent=4)) print(f"Worklogs for the time period:") print(json.dumps(worklogs, indent=4)) print("Days:") print(days) print(f"Omnimat string:") print(omnimat_string) if days_second_part: if VERBOSE: print(f"Worklogs for the weeks 2nd part:") print(json.dumps(worklogs_second_part, indent=4)) print("Days:") print(days_second_part) print(f"Omnimat string:") print(second_omnimat_string) if __name__ == '__main__': main()