workflow-scripts/process-repository.py

import os
import sys
import logging

from datetime import datetime, timedelta, timezone

from lib.gitea import Gitea


def main():
    """
    Main function to fetch issues 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"]
        log_level = os.environ.get("LOG_LEVEL", "INFO").upper()
        issue_stale_days = os.environ.get("ISSUE_STALE_DAYS")
        issue_stale_tag = os.environ.get("ISSUE_STALE_TAG")
        issue_exclude_tag = os.environ.get("ISSUE_EXCLUDE_TAG")
        issue_required_tag = os.environ.get("ISSUE_REQUIRED_TAG")
        pull_request_stale_days = os.environ.get("PULL_REQUEST_STALE_DAYS")
        pull_request_stale_tag = os.environ.get("PULL_REQUEST_STALE_TAG")
        pull_request_exclude_tag = os.environ.get("PULL_REQUEST_EXCLUDE_TAG")
        pull_request_required_tag = os.environ.get("PULL_REQUEST_REQUIRED_TAG")
    except KeyError as e:
        print(f">> Error: Missing required environment variable: {e}", file=sys.stderr)
        sys.exit(1)

    # Enable logging
    LOG_LEVELS = {
        "DEBUG": logging.DEBUG,
        "INFO": logging.INFO,
        "WARNING": logging.WARNING,
        "ERROR": logging.ERROR,
    }
    logging.basicConfig(
        level=LOG_LEVELS.get(log_level, logging.INFO),
        format="%(asctime)s - %(levelname)s - %(message)s",
    )
    logger = logging.getLogger(__name__)
    logger.info(f">> Stating process-repository script ...")

    # --- Setup Gitea API ---
    logger.info(f">> Connecting to Gitea ...")
    gitea = Gitea(instance_url, token, log_level)

    # --- Fetch issues ---
    logger.info(f">> Fetching issues ...")
    issues = gitea.get_issues(repository=repository)

    # --- Process issues ---
    if len(issues) < 1:
        logger.info(">> No open issues found")
    else:
        logger.info(f">> Processing {len(issues)} open issues ...")
        for issue in issues:

            # -- Process labels ---
            issue_created_at = datetime.fromisoformat(
                issue["created_at"].replace("Z", "+00:00")
            )
            issue_older_than_date = datetime.now(timezone.utc) - timedelta(
                days=int(issue_stale_days)
            )
            issue_current_labels = {label["name"] for label in issue.get("labels", [])}
            logger.debug(f">> Issue has the following labels: {issue_current_labels}")

            # -- Check required --
            if (not issue_required_tag == None) and (
                not issue_required_tag in issue_current_labels
            ):
                logger.info(
                    f">> Skipping issue #{issue["number"]} because it does not have the required ('{issue_required_tag}') tag"
                )
                continue

            # -- Check exclude --
            if (not issue_exclude_tag == None) and (
                issue_exclude_tag in issue_current_labels
            ):
                logger.info(
                    f">> Skipping issue #{issue["number"]} because it has the exclude ('{issue_exclude_tag}') tag"
                )
                continue

            # -- Apply stale tag --
            if (not issue_stale_tag == None) and (
                issue_created_at < issue_older_than_date
            ):
                logger.debug(
                    f">> Issue was created at {issue_created_at}, which is older then {issue_older_than_date}"
                )

                if issue_stale_tag in issue_current_labels:
                    logger.info(
                        f">> Skipping issue #{issue["number"]} because it already has the stale ('{issue_stale_tag}') tag"
                    )
                    continue

                logger.info(
                    f">> Will tag issue #{issue["number"]} with '{issue_stale_tag}'"
                )

                gitea.post_issue_labels(
                    issue_number=issue["number"],
                    labels=issue_stale_tag,
                    repository=repository,
                )

            logger.info(f">> Finished issue #{issue["number"]}")

        logger.info(">> Finished processing issues")

    # --- Fetch pull requests ---
    pull_requests = gitea.get_pull_requests(repository=repository)

    # --- Process pull requests ---
    if len(pull_requests) < 1:
        logger.info(">> No open pull requests found")
    else:
        logger.info(f">> Processing {len(pull_requests)} open pull requests ...")
        for pull_request in pull_requests:

            # -- Process labels ---
            pull_request_created_at = datetime.fromisoformat(
                pull_request["created_at"].replace("Z", "+00:00")
            )
            pull_request_older_than_date = datetime.now(timezone.utc) - timedelta(
                days=int(pull_request_stale_days)
            )
            pull_request_current_labels = {
                label["name"] for label in pull_request.get("labels", [])
            }
            logger.debug(
                f">> Pull request has the following labels: {pull_request_current_labels}"
            )

            # -- Check required --
            if (not pull_request_required_tag == None) and (
                not pull_request_required_tag in pull_request_current_labels
            ):
                logger.info(
                    f">> Skipping pull request #{pull_request["number"]} because it does not have the required ('{pull_request_required_tag}') tag"
                )
                continue

            # -- Check exclude --
            if (not pull_request_exclude_tag == None) and (
                pull_request_exclude_tag in pull_request_current_labels
            ):
                logger.info(
                    f">> Skipping pull request #{pull_request["number"]} because it has the exclude ('{pull_request_exclude_tag}') tag"
                )
                continue

            # -- Apply stale tag --
            if (not pull_request_stale_tag == None) and (
                pull_request_created_at < pull_request_older_than_date
            ):
                logger.debug(
                    f">> Pull request was created at {pull_request_created_at}, which is older then {pull_request_older_than_date}"
                )

                if pull_request_stale_tag in pull_request_current_labels:
                    logger.info(
                        f">> Skipping pull request #{pull_request["number"]} because it already has the stale ('{pull_request_stale_tag}') tag"
                    )
                    continue

                logger.info(
                    f">> Will tag pull request #{pull_request["number"]} with '{pull_request_stale_tag}'"
                )

                gitea.post_pull_request_labels(
                    pull_request_number=pull_request["number"],
                    labels=pull_request_stale_tag,
                    repository=repository,
                )

            logger.info(f">> Finished pull request #{pull_request["number"]}")

        logger.info(">> Finished processing pull requests")


if __name__ == "__main__":
    main()