homelab/.gitea/workflows/deploy-plan.py

#!/usr/bin/env python3
"""Calculate selected components against the last fully successful deploy."""

import hashlib
import json
import re
import subprocess
from pathlib import Path


def output(*args, **kwargs):
    return subprocess.check_output(args, text=True, **kwargs).strip()  # noqa: S603


def tracked(repo):
    return output('git', '-C', str(repo), 'ls-files').splitlines()


def helm_releases(repo):
    text = (repo / '.gitea/workflows/deploy-lib.sh').read_text()
    return [line.split('|') for line in re.findall(r'^  "([^"\n]+\|[^"\n]+)"$', text, re.MULTILINE)]


def inventory(repo):
    files = tracked(repo)
    k8s = sorted(
        {f.split('/k8s/')[0] for f in files if '/k8s/' in f and (repo / f.split('/k8s/')[0] / 'k8s/active').is_file()}
    )
    compose = sorted(
        {
            str(Path(f).parent)
            for f in files
            if Path(f).name in ('compose.yaml', 'compose.yml') and (repo / Path(f).parent / 'active').is_file()
        }
    )
    return {'k8s': k8s, 'compose': compose}


def file_hash(path):
    return hashlib.sha256(path.read_bytes()).hexdigest() if path.is_file() else 'missing'


def make_plan(repo, config_repo, release, previous, mode, live_helm):
    active = inventory(repo)
    all_services = set(active['k8s'] + active['compose'])
    helm_inputs = {}
    helm_selected = []
    for name, chart, namespace, version, values, marker in helm_releases(repo):
        if not (repo / marker).is_file():
            continue
        value_path = repo / values if (repo / values).is_file() else config_repo / values
        if not value_path.is_file():
            raise ValueError(f'Missing Helm values: {values}')
        stamp = hashlib.sha256(f'{chart}|{version}|{file_hash(value_path)}'.encode()).hexdigest()
        helm_inputs[name] = stamp
        live = next((h for h in live_helm if h['name'] == name and h['namespace'] == namespace), None)
        if (
            mode == 'full'
            or previous is None
            or previous.get('helm_inputs', {}).get(name) != stamp
            or live is None
            or live.get('status') != 'deployed'
            or live.get('chart') != f'{chart.split("/")[-1]}-{version}'
        ):
            helm_selected.append(name)
    local_inputs = {}
    for service in all_services:
        candidates = [config_repo / service / '.env']
        if service in active['compose']:
            candidates.append(config_repo / '.env')
            cfg = config_repo / service / 'config'
            if cfg.is_dir():
                candidates.extend(
                    p for p in cfg.rglob('*') if p.is_file() and p.suffix in ('.yaml', '.yml', '.json', '.conf')
                )
        local_inputs[service] = hashlib.sha256(
            '\n'.join(f'{p.relative_to(config_repo)}:{file_hash(p)}' for p in sorted(candidates)).encode()
        ).hexdigest()
    if previous is None:
        if mode == 'changed':
            raise ValueError('No successful baseline; run deploy in full mode first')
        changed = set(all_services)
        removed = []
    else:
        paths = output('git', '-C', str(repo), 'diff', '--name-only', previous['sha'], release['sha']).splitlines()
        changed = {service for service in all_services for path in paths if path.startswith(service + '/')}
        if any(path.startswith('.gitea/') for path in paths):
            changed |= all_services
        changed |= {s for s in all_services if previous.get('local_inputs', {}).get(s) != local_inputs[s]}
        for file in tracked(repo):
            owners = {service for service in all_services if file.startswith(service + '/')}
            if not owners or not file.endswith(('.yaml', '.yml')):
                continue
            text = (repo / file).read_text()
            if any(
                image in text and previous.get('images', {}).get(image) != digest
                for image, digest in release['images'].items()
            ):
                changed |= owners
        removed = sorted(
            set(previous.get('active', {}).get('k8s', []) + previous.get('active', {}).get('compose', []))
            - all_services
        )
        removed += [path for path in paths if '/k8s/' in path and not (repo / path).exists()]
    if mode == 'full':
        changed = set(all_services)
    dependencies = json.loads((repo / '.gitea/deploy-dependencies.json').read_text())
    while True:
        expanded = changed | {dependent for service in changed for dependent in dependencies.get(service, [])}
        if expanded == changed:
            break
        changed = expanded
    return {
        'version': 1,
        'sha': release['sha'],
        'images': release['images'],
        'active': active,
        'selected': {kind: sorted(set(services) & changed) for kind, services in active.items()},
        'helm': helm_selected,
        'helm_inputs': helm_inputs,
        'local_inputs': local_inputs,
        'removed': sorted(set(removed)),
        'full_smoke': mode == 'full' or 'traefik' in changed,
    }