From 23ca5e4b2da1360d84f8f352276461a0fdb9e1f8 Mon Sep 17 00:00:00 2001 From: Alex Lebens Date: Tue, 10 Jun 2025 16:09:03 -0500 Subject: [PATCH] add scripts --- scripts/process-issues.py | 123 ++++++++++++++++++++++++++++++ scripts/process-pull-requests.py | 126 +++++++++++++++++++++++++++++++ 2 files changed, 249 insertions(+) create mode 100644 scripts/process-issues.py create mode 100644 scripts/process-pull-requests.py diff --git a/scripts/process-issues.py b/scripts/process-issues.py new file mode 100644 index 0000000..e049d86 --- /dev/null +++ b/scripts/process-issues.py @@ -0,0 +1,123 @@ +import os +import sys +import requests +from datetime import datetime, timedelta, timezone + +def main(): + """ + Main function to fetch issues from a Gitea repository and applies a tag to them. + """ + # --- Get configuration from environment variables --- + try: + instance_url = os.environ['INSTANCE_URL'].rstrip('/') + repository = os.environ['REPOSITORY'] + token = os.environ['TOKEN'] + stale_days = os.environ.get('STALE_DAYS') + stale_tag = os.environ.get('STALE_TAG', 'stale') + exclude_tag = os.environ.get('EXCLUDE_TAG') + required_tag = os.environ.get('REQUIRED_TAG') + except KeyError as e: + print(f"Error: Missing required environment variable: {e}", file=sys.stderr) + sys.exit(1) + + + # --- Switch off checks if env is empty --- + if required_tag == None: + enable_required_tag = False + print(f">> No optional required tag set.") + else: + enable_required_tag = True + + if exclude_tag == None: + enable_exclude_tag= False + print(f">> No optional exclusive tag set.") + else: + enable_exclude_tag = True + + if stale_days == None: + enable_stale_tag= False + print(f">> No value for stale days.") + else: + older_than_date = datetime.now(timezone.utc) - timedelta(days=stale_days) + enable_stale_tag = True + + + # --- Check if any actions are enabled --- + if not any([enable_stale_tag]) == True: + print(f">> No actions enabled, exiting.") + sys.exit(1) + + + # --- Set up API headers and base URL --- + headers = { + 'Authorization': f'token {token}', + 'Accept': 'application/json', + 'Content-Type': 'application/json' + } + base_api_url = f"{instance_url}/api/v1/repos/{repository}/issues" + params = {'state': 'open'} + + + # --- 3. Fetch open issues --- + print(">> Fetching issues ...") + try: + response = requests.get(base_api_url, headers=headers, params=params, timeout=30) + response.raise_for_status() + issues = response.json() + except requests.exceptions.RequestException as e: + print(f"Error fetching issues: {e}", file=sys.stderr) + sys.exit(1) + + if not issues: + print(">> No open issues found, exiting.") + sys.exit(0) + + print(f">> Processing {len(issues)} open issues ...") + + + # --- Process Pull Requests --- + for issue in issues: + issue_number = issue['number'] + issue_created_at = datetime.fromisoformat(issue['created_at'].replace('Z', '+00:00')) + + + # -- Process labels --- + issue_current_labels = {label['name'] for label in issue.get('labels', [])} + update_labels = issue_current_labels + + if enable_required_tag: + if not required_tag in issue_current_labels: + print(f">> Skipping issue #{issue_number} because it does not have the '{required_tag}' tag.") + continue + + if enable_exclude_tag: + if exclude_tag in issue_current_labels: + print(f">> Skipping issue #{issue_number} because it has the '{exclude_tag}' tag.") + continue + + if enable_stale_tag: + if issue_created_at < older_than_date: + if stale_tag in issue_current_labels: + print(f">> Skipping issue #{issue_number} because it already has the '{stale_tag}' tag.") + continue + + print(f">> Will tag issue #{issue_number} with '{stale_tag}'") + update_labels = list(update_labels | {stale_tag}) + + + # --- Make update with new labels --- + update_payload = {'labels': update_labels} + update_url = f"{base_api_url}/{issue_number}/labels" + + print(f">> Updating Issue #{issue_number} ...") + try: + update_response = requests.put(update_url, headers=headers, json=update_payload, timeout=30) + update_response.raise_for_status() + print(f">> Successfully updated labels for issue #{issue_number}") + except requests.exceptions.RequestException as e: + print(f"Error updating labels for issue #{issue_number}: {e}", file=sys.stderr) + + print(">> Finished processing issues.") + +if __name__ == "__main__": + main() diff --git a/scripts/process-pull-requests.py b/scripts/process-pull-requests.py new file mode 100644 index 0000000..cb8c6ae --- /dev/null +++ b/scripts/process-pull-requests.py @@ -0,0 +1,126 @@ +import os +import sys +import requests +from datetime import datetime, timedelta, timezone + +def main(): + """ + Main function to fetch pull requests from a Gitea repository and process them. + """ + # --- Get configuration from environment variables --- + try: + instance_url = os.environ['INSTANCE_URL'].rstrip('/') + repository = os.environ['REPOSITORY'] + token = os.environ['TOKEN'] + stale_days = os.environ.get('STALE_DAYS') + stale_tag = os.environ.get('STALE_TAG', 'stale') + exclude_tag = os.environ.get('EXCLUDE_TAG') + required_tag = os.environ.get('REQUIRED_TAG') + except KeyError as e: + print(f"Error: Missing required environment variable: {e}", file=sys.stderr) + sys.exit(1) + + print(f">> Processing pull requests ...") + + + # --- Switch off checks if env is empty --- + if required_tag == None: + enable_required_tag = False + print(f">> No optional required tag set.") + else: + enable_required_tag = True + + if exclude_tag == None: + enable_exclude_tag= False + print(f">> No optional exclusive tag set.") + else: + enable_exclude_tag = True + + if stale_days == None: + enable_stale_tag= False + print(f">> No value for stale days.") + else: + stale_days = int(stale_days) + older_than_date = datetime.now(timezone.utc) - timedelta(days=stale_days) + enable_stale_tag = True + + + # --- Check if any actions are enabled --- + if not any([enable_stale_tag]) == True: + print(f">> No actions enabled, exiting.") + sys.exit(1) + + + # --- Set up API headers and base URL --- + headers = { + 'Authorization': f'token {token}', + 'Accept': 'application/json', + 'Content-Type': 'application/json' + } + base_api_url = f"{instance_url}/api/v1/repos/{repository}/pulls" + params = {'state': 'open'} + + + # --- Fetch pull requests --- + print(">> Fetching pull requests ...") + try: + response = requests.get(base_api_url, headers=headers, params=params, timeout=30) + response.raise_for_status() + pull_requests = response.json() + except requests.exceptions.RequestException as e: + print(f"Error fetching pull requests: {e}", file=sys.stderr) + sys.exit(1) + + if not pull_requests: + print(">> No open pull requests found, exiting.") + sys.exit(0) + + print(f">> Processing {len(pull_requests)} open pull requests ...") + + + # --- Process Pull Requests --- + for pr in pull_requests: + pr_number = pr['number'] + pr_created_at = datetime.fromisoformat(pr['created_at'].replace('Z', '+00:00')) + + + # -- Process labels --- + pr_current_labels = {label['name'] for label in pr.get('labels', [])} + update_labels = pr_current_labels + + if enable_required_tag: + if not required_tag in pr_current_labels: + print(f">> Skipping PR #{pr_number} because it does not have the '{required_tag}' tag.") + continue + + if enable_exclude_tag: + if exclude_tag in pr_current_labels: + print(f">> Skipping PR #{pr_number} because it has the '{exclude_tag}' tag.") + continue + + if enable_stale_tag: + if pr_created_at < older_than_date: + if stale_tag in pr_current_labels: + print(f">> Skipping issue #{pr_number} because it already has the '{stale_tag}' tag.") + continue + + print(f">> Will tag PR #{pr_number} with '{stale_tag}'") + update_labels = list(update_labels | {stale_tag}) + + + # --- Make update with new labels --- + update_payload = {'labels': update_labels} + update_url = f"{base_api_url}/{pr_number}/labels" + + print(f">> Updating PR #{pr_number} ...") + try: + update_response = requests.put(update_url, headers=headers, json=update_payload, timeout=30) + update_response.raise_for_status() + print(f">> Successfully updated labels for PR #{pr_number}") + except requests.exceptions.RequestException as e: + print(f"Error updating labels for PR #{pr_number}: {e}", file=sys.stderr) + + print(">> Finished processing pull requests.") + +if __name__ == "__main__": + main()