125 lines
4.2 KiB
Python
125 lines
4.2 KiB
Python
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:
|
|
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}/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.add(stale_tag)
|
|
|
|
|
|
# --- Make update with new labels ---
|
|
update_payload = {'labels': list(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()
|