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

    # def repo():
    #     # 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"] == team_name for repo in repos):
    #             logger.debug(
    #                 f"Gitea repository {team_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 course repository
    #     repo_name = f"course-{course.id}"
    #     url = f"{api_url}/orgs/{GITEA_ORG_NAME}/repos"
    #     payload = {
    #         "auto_init": True,
    #         "default_branch": "main",
    #         "description": f"{course.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("url", None)
    #         course.repository_url = repo_url
    #         course.save(update_fields=["repository_url"])
    #         logger.info(
    #             f"Successfully created Gitea repository for course {course.title}"
    #         )
    #     except Exception as e:
    #         logger.exception(
    #             f"Failed to create Gitea repository 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 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

    # 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("url", None)
        instance.gitea_repo_url = repo_url
        instance.save()
        logger.info(
            f"Successfully created Gitea repository for lesson {instance.title}"
        )
    except Exception as e:
        logger.exception(
            f"Failed to create Gitea repository for lesson {instance.title}: {e}\n{response.text}",
            e,
        )