nano_coolify.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""
nano_coolify - checks repositories for new commits and deploys on changes.

Run: python3 ~/nano_coolify.py

Add to cron (every minute):
(crontab -l 2>/dev/null; echo '* * * * * python3 ~/nano_coolify.py >> ~/nano_coolify.log 2>&1') | crontab -
"""

import subprocess
import logging
from dataclasses import dataclass
from pathlib import Path

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
log = logging.getLogger(__name__)

# ============== CONFIG ==============

@dataclass
class Project:
    path: str
    branch: str = "main"
    deploy_cmd: str = "docker compose up -d --build"


PROJECTS = [
    Project(
        path="~/bots/cheramics_qna_bot",
        branch="main",
        deploy_cmd="docker compose up -d --build",
    ),
    # Project(path="~/bots/another_bot", branch="main"),
]

# ====================================


def run(cmd: str, cwd: Path | None = None) -> subprocess.CompletedProcess[str]:
    return subprocess.run(cmd, shell=True, cwd=cwd, capture_output=True, text=True)


def get_local_commit(project_path: Path) -> str | None:
    result = run("git rev-parse HEAD", cwd=project_path)
    if result.returncode != 0:
        log.error(f"git rev-parse failed: {result.stderr}")
        return None
    return result.stdout.strip()


def get_remote_commit(project_path: Path, branch: str) -> str | None:
    result = run(f"git ls-remote origin {branch}", cwd=project_path)
    if result.returncode != 0:
        log.error(f"git ls-remote failed: {result.stderr}")
        return None
    parts = result.stdout.strip().split()
    return parts[0] if parts else None


def pull_and_deploy(project: Project, project_path: Path) -> bool:
    log.info("Pulling changes...")
    pull = run("git pull", cwd=project_path)
    if pull.returncode != 0:
        log.error(f"git pull failed: {pull.stderr}")
        return False

    log.info(f"Deploying: {project.deploy_cmd}")
    deploy = run(project.deploy_cmd, cwd=project_path)
    if deploy.returncode != 0:
        log.error(f"Deploy failed: {deploy.stderr}")
        return False

    log.info("Tagging as deployed...")
    run("git tag -f deployed", cwd=project_path)
    push_tag = run("git push origin deployed --force", cwd=project_path)
    if push_tag.returncode != 0:
        log.warning(f"Failed to push deployed tag: {push_tag.stderr}")

    log.info("Deploy successful")
    return True


def check_project(project: Project) -> None:
    project_path = Path(project.path).expanduser()

    if not project_path.exists():
        log.warning(f"Path does not exist: {project_path}")
        return

    local = get_local_commit(project_path)
    remote = get_remote_commit(project_path, project.branch)

    if not local or not remote:
        return

    if local != remote:
        log.info(f"[{project_path.name}] New commit: {local[:8]} -> {remote[:8]}")
        pull_and_deploy(project, project_path)


def main() -> None:
    for project in PROJECTS:
        try:
            check_project(project)
        except Exception as e:
            log.exception(f"Error checking {project.path}: {e}")


if __name__ == "__main__":
    main()