scripts/create_naughty_list.py [16:103]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - def setup_argparser() -> argparse.ArgumentParser: """Set up the argument parser for the script""" argparser = argparse.ArgumentParser(description='Create list of overdue projects') argparser.add_argument('-b', '--baseurl', help='Base URL of the environment to run the script against') argparser.add_argument('-t', '--timestamp', help='Date to filter records before (yyyy-mm-dd)') return argparser def get_token() -> str: """Set token from environment variable""" token = os.environ.get("PLUTO_TOKEN") if token == None: print("No token found. Exiting script...") sys.exit() return token def create_urls(base_url): commission_list_url = f"{base_url}/pluto-core/api/pluto/commission/list" project_list_url = f"{base_url}/pluto-core/api/project/list" return commission_list_url, project_list_url def get_headers(token: str) -> dict: return { "Content-Type": "application/json", "Authorization": f"Bearer {token}", } def api_put_request(url, headers, json_body, max_retries=5): backoff_factor = 2 for retry in range(max_retries): try: with requests.put(url, headers=headers, data=json_body, verify=False) as response: response.raise_for_status() return response.json() except (requests.exceptions.HTTPError, requests.exceptions.RequestException) as e: if retry == max_retries - 1: raise wait_time = backoff_factor ** retry print(f"An error occurred: {e}. Retrying in {wait_time} seconds...") time.sleep(wait_time) def get_filtered_commission_records(timestamp, headers, commission_list_url) -> list: request_body = { "match": "W_CONTAINS", "completionDateBefore": timestamp } json_body = json.dumps(request_body) records = [] try: json_content = api_put_request(commission_list_url, headers, json_body) total_records = json_content["count"] total_pages = (total_records + MAX_RECORDS_PER_PAGE - 1) // MAX_RECORDS_PER_PAGE start_at = 0 for page in range(1, total_pages + 1): print(f"Loading commission page: {page}") response = api_put_request( f"{commission_list_url}?startAt={start_at}&length={MAX_RECORDS_PER_PAGE}", headers, json_body, ) records.extend(response["result"]) start_at += MAX_RECORDS_PER_PAGE except requests.exceptions.RequestException as e: print(e) raise Exception("An error occurred. Exiting script...") return records def get_projects_by_user(records, headers, project_list_url) -> dict: user_projects = defaultdict(list) user_project_count = defaultdict(int) for record in records: commission_id = record['id'] print(f"Getting projects for commission ID: {commission_id}") try: json_content = api_put_request( project_list_url, headers, json.dumps({"match": "W_EXACT", "commissionId": commission_id}), ) for project in json_content["result"]: if project['status'] == "In Production": user = project['user'] - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - scripts/create_naughtylist_csv.py [52:139]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - def setup_argparser() -> argparse.ArgumentParser: """Set up the argument parser for the script""" argparser = argparse.ArgumentParser(description='Create list of overdue projects') argparser.add_argument('-b', '--baseurl', help='Base URL of the environment to run the script against') argparser.add_argument('-t', '--timestamp', help='Date to filter records before (yyyy-mm-dd)') return argparser def get_token() -> str: """Set token from environment variable""" token = os.environ.get("PLUTO_TOKEN") if token == None: print("No token found. Exiting script...") sys.exit() return token def create_urls(base_url): commission_list_url = f"{base_url}/pluto-core/api/pluto/commission/list" project_list_url = f"{base_url}/pluto-core/api/project/list" return commission_list_url, project_list_url def get_headers(token: str) -> dict: return { "Content-Type": "application/json", "Authorization": f"Bearer {token}", } def api_put_request(url, headers, json_body, max_retries=5): backoff_factor = 2 for retry in range(max_retries): try: with requests.put(url, headers=headers, data=json_body, verify=False) as response: response.raise_for_status() return response.json() except (requests.exceptions.HTTPError, requests.exceptions.RequestException) as e: if retry == max_retries - 1: raise wait_time = backoff_factor ** retry print(f"An error occurred: {e}. Retrying in {wait_time} seconds...") time.sleep(wait_time) def get_filtered_commission_records(timestamp, headers, commission_list_url) -> list: request_body = { "match": "W_CONTAINS", "completionDateBefore": timestamp } json_body = json.dumps(request_body) records = [] try: json_content = api_put_request(commission_list_url, headers, json_body) total_records = json_content["count"] total_pages = (total_records + MAX_RECORDS_PER_PAGE - 1) // MAX_RECORDS_PER_PAGE start_at = 0 for page in range(1, total_pages + 1): print(f"Loading commission page: {page}") response = api_put_request( f"{commission_list_url}?startAt={start_at}&length={MAX_RECORDS_PER_PAGE}", headers, json_body, ) records.extend(response["result"]) start_at += MAX_RECORDS_PER_PAGE except requests.exceptions.RequestException as e: print(e) raise Exception("An error occurred. Exiting script...") return records def get_projects_by_user(records, headers, project_list_url) -> dict: user_projects = defaultdict(list) user_project_count = defaultdict(int) for record in records: commission_id = record['id'] print(f"Getting projects for commission ID: {commission_id}") try: json_content = api_put_request( project_list_url, headers, json.dumps({"match": "W_EXACT", "commissionId": commission_id}), ) for project in json_content["result"]: if project['status'] == "In Production": user = project['user'] - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -