jira-worklog-data/main.py

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"]  # Modify
CERT = getenv("CERT_LOCATION")
WEEKS_BACK = -1
CHECK_MONTH_BACK = False
MONTH_VIEW = False
VERBOSE = False
if "--monthview" in sys.argv:
    MONTH_VIEW = True
    if "--monthback" in sys.argv:
        CHECK_MONTH_BACK = True
else:
    for arg in sys.argv:
        if "--weeksback" in arg:
            WEEKS_BACK = int(arg.split("=")[1])
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": 15,
        "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_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_sort)
    return worklogs


def date_sort(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_week_tickets(worklogs, days):
    this_week_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:
            this_week_tickets.append(worklog)
    return this_week_tickets

def main():
    print('The script starts its work...')

    if MONTH_VIEW:
        for username in USERNAMES:
            issues = get_issues(username)
            worklogs = get_worklogs(issues)
            days = get_days_of_month(CHECK_MONTH_BACK)
            worklogs = get_month_tickets(worklogs, days, CHECK_MONTH_BACK)
            omnimat_string = get_omnimat_string(worklogs)
            print(omnimat_string)
    else:
        for username in USERNAMES:
            issues = get_issues(username)
            worklogs = get_worklogs(issues)
            today = datetime.date.today()
            days, days_second_part = get_days(today)
            week_worklogs = get_week_tickets(worklogs, days)
            omnimat_string = ""

            for worklog in week_worklogs:
                if worklog["ticket_id"] not in omnimat_string:
                    omnimat_string += worklog["ticket_id"] + '\n'

            omnimat_string = ""
            if days_second_part:
                second_omnimat_string = ""
                worklogs_second_part = get_week_tickets(worklogs, days_second_part)
                for worklog in week_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 week:")
                print(json.dumps(week_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(omnimat_string)


if __name__ == '__main__':
    main()