jira-worklog-data/main.py

import requests
import json
import datetime
from dotenv import load_dotenv
from os import getenv
from dates 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
DATE_FROM = None
DATE_TO = None

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 "--date-from" in arg:
        DATE_FROM = datetime.datetime.strptime(
        get_flag_value(arg), '%Y-%m-%d').date()
    if "--date-to" in arg:
        DATE_TO = datetime.datetime.strptime(
        get_flag_value(arg), '%Y-%m-%d').date()
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 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 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
    if DATE_FROM and DATE_TO:
        for username in USERNAMES:
            issues = get_issues(username)
            all_worklogs = get_all_worklogs(issues)
            days = get_days_for_range(DATE_FROM, DATE_TO)
            print(all_worklogs)
            worklogs = get_worklogs_for_days(all_worklogs, days)
            print(worklogs)
        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, WEEKS_BACK)
        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()