add scripts

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()

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()