kursy-mirror/home/signals.py

import logging as lg
import os

import requests
from django.conf import settings
from django.db.models.signals import post_save
from django.dispatch import receiver

from home.models.pages import CoursePage, ModuleLessonPage

GITEA_ORG_NAME = "Studio77"

logger = lg.getLogger(__name__)


@receiver(post_save, sender=CoursePage)
def create_gitea_team_on_course_creation(sender, instance, created, **kwargs):
    if not instance.live:
        logger.debug(
            f"Course {instance.title} is not live, skipping Gitea team creation"
        )
        return

    course = instance
    team_name = f"course-{course.id}"
    api_url = getattr(settings, "GITEA_URL", None)

    if not api_url:
        logger.debug("GITEA_URL is not set, skipping Gitea team creation")
        return

    # check if team already exists
    try:
        response = requests.get(
            f"{api_url}/orgs/{GITEA_ORG_NAME}/teams",
            timeout=5,
            headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
        )
        response.raise_for_status()
        teams = response.json()
        if any(team["name"] == team_name for team in teams):
            logger.info(f"Gitea team {team_name} already exists, skipping creation")
            return
    except Exception as e:
        logger.exception(
            f"Failed to check existing Gitea teams: {e}\n{response.text}",
            e,
        )
        return

    url = f"{api_url}/orgs/{GITEA_ORG_NAME}/teams"
    payload = {
        "can_create_org_repo": False,
        "description": f"Team for course {course.title}",
        "includes_all_repositories": False,
        "name": team_name,
        "permission": "read",
        "units": [
            # "repo.actions",
            "repo.code",
            # "repo.issues",
            # "repo.ext_issues",
            # "repo.wiki",
            # "repo.ext_wiki",
            # "repo.pulls",
            # "repo.releases",
            # "repo.projects",
            # "repo.ext_wiki",
        ],
    }
    try:
        response = requests.post(
            url,
            json=payload,
            timeout=5,
            headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
        )
        response.raise_for_status()
        logger.info(
            f"Successfully created Gitea team `{team_name}` for course {course.title}"
        )
    except Exception as e:
        logger.exception(
            f"Failed to create Gitea team for course {course.title}: {e}\n{response.text}",
            e,
        )


@receiver(post_save, sender=ModuleLessonPage)
def create_gitea_repo_on_lesson_creation(
    sender, instance: ModuleLessonPage, created, **kwargs
):
    if not instance.live:
        logger.debug(
            f"Lesson {instance.title} is not live, skipping Gitea repository creation"
        )
        return

    course = instance.module.course
    repo_name = f"course-{course.id}-lesson-{instance.id}"

    if not instance.create_gitea_repo:
        logger.debug(
            f"Lesson {instance.title} is not marked for Gitea repository creation, skipping"
        )
        return

    if not course.live:
        logger.debug(
            f"Course {course.title} is not live, skipping Gitea repository creation for lesson {instance.title}"
        )
        return

    api_url = getattr(settings, "GITEA_URL", None)

    if not api_url:
        logger.debug("GITEA_URL is not set, skipping Gitea repository creation")
        return

    def create_repository() -> None:
        # check if repository already exists
        try:
            response = requests.get(
                f"{api_url}/orgs/{GITEA_ORG_NAME}/repos",
                timeout=5,
                headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
            )
            response.raise_for_status()
            repos = response.json()
            if any(repo["name"] == repo_name for repo in repos):
                logger.info(
                    f"Gitea repository {repo_name} already exists, skipping creation"
                )
                return
        except Exception as e:
            logger.exception(
                f"Failed to check existing Gitea repositories: {e}\n{response.text}",
                e,
            )
            return

        # create lesson repository
        url = f"{api_url}/orgs/{GITEA_ORG_NAME}/repos"
        payload = {
            "auto_init": True,
            "default_branch": "main",
            "description": f"{instance.module.course} – {instance.module}: {instance.title}",
            "name": repo_name,
            "private": True,
        }

        try:
            response = requests.post(
                url,
                json=payload,
                timeout=5,
                headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
            )
            response.raise_for_status()
            repo_url = response.json().get("html_url")

            instance.gitea_repo_url = repo_url
            instance.save(update_fields=["gitea_repo_url"])

            logger.info(
                f"Successfully created Gitea repository for lesson {instance.title}: {repo_url}"
            )
        except Exception as e:
            logger.exception(
                f"Failed to create Gitea repository for lesson {instance.title}: {e}\n{response.text}",
                e,
            )

    def add_repository_to_team() -> None:
        # add repository to course team
        team_name = f"course-{course.id}"
        try:
            response = requests.get(
                f"{api_url}/orgs/{GITEA_ORG_NAME}/teams",
                timeout=5,
                headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
            )
            response.raise_for_status()
            teams = response.json()

            team = next((team for team in teams if team["name"] == team_name), None)

            if not team:
                logger.error(
                    f"Gitea team {team_name} not found when trying to add repository {repo_name}"
                )
                return

            team_id = team["id"]
            add_repo_url = (
                f"{api_url}/teams/{team_id}/repos/{GITEA_ORG_NAME}/{repo_name}"
            )
            add_response = requests.put(
                add_repo_url,
                timeout=5,
                headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
            )
            add_response.raise_for_status()
            logger.info(
                f"Successfully added repository {repo_name} to Gitea team {team_name}"
            )
        except Exception as e:
            logger.exception(
                f"Failed to add repository {repo_name} to Gitea team {team_name}: {e}\n{response.text}",
                e,
            )

    create_repository()
    add_repository_to_team()