working script
This commit is contained in:
146
main.py
146
main.py
@@ -1,59 +1,107 @@
|
|||||||
from ssl import SSLCertVerificationError
|
|
||||||
import requests
|
import requests
|
||||||
from requests.auth import HTTPBasicAuth
|
|
||||||
import json
|
import json
|
||||||
from dotenv import load_dotenv
|
import datetime
|
||||||
import os
|
|
||||||
|
|
||||||
# Before running the script make sure the following:
|
ACCESS_TOKEN = "" # Modify
|
||||||
# 1. The correct root certificate is in the script directory and specified
|
USERNAMES = ["kalinom6"] # Modify
|
||||||
# under the global variables
|
URL = "https://globaljira.roche.com/rest/api/2"
|
||||||
# 2. You've created the .env file with the following format:
|
CERT = 'Roche G3 Root CA.crt'
|
||||||
# JIRA_USER=<your username>
|
CHECK_WEEK_BACK = False
|
||||||
# JIRA_PASSWORD=<your password>
|
|
||||||
# 3. You have installed necessary packages from the requirements.txt file
|
|
||||||
|
|
||||||
load_dotenv()
|
|
||||||
|
|
||||||
JIRA_URL = 'https://globaljira.roche.com/rest/api/2/issue/picker'
|
|
||||||
USERNAME = os.getenv("JIRA_USER")
|
|
||||||
PASSWORD = os.getenv("JIRA_PASSWORD")
|
|
||||||
ROOT_CERT = ".\\Roche G3 Root CA.crt"
|
|
||||||
PRINT_RESPONSE_ON_ERR = False
|
|
||||||
|
|
||||||
headers = {
|
headers = {
|
||||||
'Content-Type': 'application/json'
|
"Authorization": f"Bearer {ACCESS_TOKEN}",
|
||||||
|
"Content-Type": "application/json"
|
||||||
}
|
}
|
||||||
|
|
||||||
# For this query for user kalinom6 there are 6 results, 5 of them are correct,
|
def send_request(url, method, payload=None):
|
||||||
# but one is certainly not associated with the user in any way according to
|
if method == "GET":
|
||||||
# the web application. The result set is lacking too, there should be way
|
response = requests.get(url, headers=headers, data=json.dumps(payload), verify=CERT)
|
||||||
# more issues assigned to this user
|
if method == "POST":
|
||||||
query = {
|
response = requests.post(url, headers=headers, data=json.dumps(payload), verify=CERT)
|
||||||
'query': f'assignee = {USERNAME} AND status not in (Closed, Done)'
|
if response.status_code == 200:
|
||||||
}
|
return response.json()
|
||||||
|
print(f"Error: {response.status_code}, {response.text}")
|
||||||
|
|
||||||
print(f'{USERNAME}, {PASSWORD}')
|
def get_issues(username):
|
||||||
|
jql = f"assignee = '{username}'"
|
||||||
|
payload = {
|
||||||
|
"jql": jql,
|
||||||
|
"startAt": 0,
|
||||||
|
"maxResults": 50,
|
||||||
|
"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
|
||||||
|
|
||||||
try:
|
def get_worklogs(issues):
|
||||||
response = requests.get(
|
worklogs = []
|
||||||
JIRA_URL,
|
for issue in issues:
|
||||||
auth=HTTPBasicAuth(USERNAME, PASSWORD),
|
data = send_request(f'{URL}/issue/{issue["task_id"]}/worklog', "GET")
|
||||||
headers=headers,
|
if data:
|
||||||
params=query,
|
for log in data.get("worklogs", []):
|
||||||
verify=ROOT_CERT
|
worklog = {
|
||||||
)
|
"ticket_id": issue["task_id"],
|
||||||
except SSLCertVerificationError:
|
"time_spent": log["timeSpent"],
|
||||||
print("SSL verification failed. Please make sure you're connected to the VPN and the correct root certificate is included.")
|
"date": log["started"],
|
||||||
|
"author": log["author"]["name"]
|
||||||
|
}
|
||||||
|
if worklog["author"] in USERNAMES:
|
||||||
|
worklogs.append(worklog)
|
||||||
|
return worklogs
|
||||||
|
|
||||||
if response.status_code == 200:
|
def get_days():
|
||||||
print("JQL query executed successfully!")
|
today = datetime.date.today()
|
||||||
issues = json.dumps(json.loads(response.text), sort_keys=True, indent=4, separators=(",", ": "))
|
if CHECK_WEEK_BACK:
|
||||||
print(issues)
|
today = today - datetime.timedelta(weeks=1)
|
||||||
else:
|
year, weeknum, day_of_week = today.isocalendar()
|
||||||
print(f"Failed to execute JQL query. Status code: {response.status_code}")
|
days = []
|
||||||
if response.status_code == 403:
|
days.append(str(today))
|
||||||
print("If you're sure you've set up the environment variables correctly\n \
|
while day_of_week > 1:
|
||||||
Please login to Jira in the browser to solve a CAPTCHA.")
|
today = today - datetime.timedelta(days=1)
|
||||||
if PRINT_RESPONSE_ON_ERR:
|
day_of_week -= 1
|
||||||
print("Response:", response.text)
|
days.append(str(today))
|
||||||
|
return days
|
||||||
|
|
||||||
|
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()
|
||||||
|
current_month = get_max_month(days)
|
||||||
|
worklog_month = str(worklog_date).split('-')[1]
|
||||||
|
if str(worklog_date) in days and int(worklog_month) == current_month:
|
||||||
|
this_week_tickets.append(worklog)
|
||||||
|
return this_week_tickets
|
||||||
|
|
||||||
|
def get_max_month(days):
|
||||||
|
max = 0
|
||||||
|
for day in days:
|
||||||
|
month = int(day.split('-')[1])
|
||||||
|
if month > max:
|
||||||
|
max = month
|
||||||
|
return max
|
||||||
|
|
||||||
|
def main():
|
||||||
|
for username in USERNAMES:
|
||||||
|
issues = get_issues(username)
|
||||||
|
print('Issues:')
|
||||||
|
print(json.dumps(issues, indent=4))
|
||||||
|
worklogs = get_worklogs(issues)
|
||||||
|
days = get_days()
|
||||||
|
worklogs = get_week_tickets(worklogs, days)
|
||||||
|
omnimat_string = ""
|
||||||
|
for worklog in worklogs:
|
||||||
|
omnimat_string += worklog["ticket_id"] + '\n'
|
||||||
|
print(omnimat_string)
|
||||||
|
if __name__ == '__main__':
|
||||||
|
main()
|
||||||
Reference in New Issue
Block a user