enterprise_digital-platform/digiserver-v2/app/utils/player_build.py

"""Utilities for building/staging the player files on the server.

Admins use the "Build player files" admin page to:
  * clone/refresh the player source code from a git repository into a local
    staged directory (``PLAYER_CODE_DIR``), and
  * write a base ``config/app_config.json`` so the staged code already knows how
    to reach this server.

The SSH deployment flow then ships this staged directory to player devices, so
the version admins build here is exactly what gets deployed.
"""
import os
import json
import shutil
import subprocess
import logging
from datetime import datetime
from typing import Any, Dict, Optional

from app.utils.ssh_deploy import generate_app_config

logger = logging.getLogger(__name__)

# Metadata file name stored in the Flask instance folder.
BUILD_META_FILENAME = 'player_build.json'


def _run_git(args, cwd=None, timeout=300) -> subprocess.CompletedProcess:
    return subprocess.run(
        ['git'] + args,
        cwd=cwd,
        capture_output=True,
        text=True,
        timeout=timeout,
    )


def get_short_head(player_code_dir: str) -> str:
    """Return the short git commit of the staged code, or 'unknown'."""
    try:
        result = _run_git(['-C', player_code_dir, 'rev-parse', '--short', 'HEAD'], timeout=10)
        if result.returncode == 0:
            return result.stdout.strip()
    except Exception:
        pass
    return 'unknown'


def build_player_files(player_code_dir: str, repo_url: str, branch: str = 'main') -> Dict[str, Any]:
    """Clone or refresh the player source into ``player_code_dir``.

    If the directory is already a git checkout of ``repo_url`` it is updated in
    place (fetch + hard reset to the chosen branch). Otherwise it is cloned
    fresh (an existing non-git directory is replaced).

    Returns a dict: ``success`` (bool), ``message`` (str), ``version`` (str),
    ``branch`` (str).
    """
    branch = (branch or 'main').strip()
    repo_url = (repo_url or '').strip()
    if not repo_url:
        return {'success': False, 'message': 'Repository URL is required.', 'version': None, 'branch': branch}

    try:
        git_dir = os.path.join(player_code_dir, '.git')
        is_git_repo = os.path.isdir(git_dir)

        if is_git_repo:
            # Update existing checkout in place.
            fetch = _run_git(['-C', player_code_dir, 'fetch', '--prune', 'origin'])
            if fetch.returncode != 0:
                return {
                    'success': False,
                    'message': f'git fetch failed: {fetch.stderr.strip() or fetch.stdout.strip()}',
                    'version': get_short_head(player_code_dir),
                    'branch': branch,
                }
            # Point origin at the requested URL in case it changed.
            _run_git(['-C', player_code_dir, 'remote', 'set-url', 'origin', repo_url])
            checkout = _run_git(['-C', player_code_dir, 'checkout', branch])
            if checkout.returncode != 0:
                return {
                    'success': False,
                    'message': f'git checkout {branch} failed: {checkout.stderr.strip()}',
                    'version': get_short_head(player_code_dir),
                    'branch': branch,
                }
            reset = _run_git(['-C', player_code_dir, 'reset', '--hard', f'origin/{branch}'])
            if reset.returncode != 0:
                return {
                    'success': False,
                    'message': f'git reset failed: {reset.stderr.strip()}',
                    'version': get_short_head(player_code_dir),
                    'branch': branch,
                }
            action = 'Updated'
        else:
            # Fresh clone. Replace any existing (non-git) directory.
            parent = os.path.dirname(player_code_dir.rstrip('/'))
            os.makedirs(parent, exist_ok=True)
            if os.path.exists(player_code_dir):
                shutil.rmtree(player_code_dir)
            clone = _run_git(['clone', '--branch', branch, repo_url, player_code_dir])
            if clone.returncode != 0:
                return {
                    'success': False,
                    'message': f'git clone failed: {clone.stderr.strip() or clone.stdout.strip()}',
                    'version': None,
                    'branch': branch,
                }
            action = 'Cloned'

        version = get_short_head(player_code_dir)
        logger.info('%s player code from %s (%s) -> %s', action, repo_url, branch, version)
        return {
            'success': True,
            'message': f'{action} player code from {branch} (version {version}).',
            'version': version,
            'branch': branch,
        }
    except subprocess.TimeoutExpired:
        return {'success': False, 'message': 'Git operation timed out.', 'version': None, 'branch': branch}
    except Exception as e:
        logger.exception('build_player_files failed')
        return {'success': False, 'message': f'Build failed: {str(e)}', 'version': None, 'branch': branch}


def write_base_config(
    player_code_dir: str,
    server_ip: str,
    port: str,
    use_https: bool = True,
    verify_ssl: bool = False,
    orientation: str = 'Landscape',
    max_resolution: str = '1920x1080',
) -> Dict[str, Any]:
    """Write a base ``config/app_config.json`` into the staged player code.

    ``screen_name`` and ``quickconnect_key`` are left blank on purpose: they are
    per-player and get filled in by the SSH deploy step for each device.
    """
    try:
        config_dir = os.path.join(player_code_dir, 'config')
        os.makedirs(config_dir, exist_ok=True)
        content = generate_app_config(
            server_ip=server_ip,
            port=str(port),
            screen_name='',
            quickconnect_code='',
            orientation=orientation,
            use_https=use_https,
            verify_ssl=verify_ssl,
            max_resolution=max_resolution,
        )
        config_path = os.path.join(config_dir, 'app_config.json')
        with open(config_path, 'w', encoding='utf-8') as f:
            f.write(content)
        logger.info('Wrote base player config -> %s', config_path)
        return {'success': True, 'message': 'Base player config written.', 'path': config_path}
    except Exception as e:
        logger.exception('write_base_config failed')
        return {'success': False, 'message': f'Failed to write config: {str(e)}'}


def load_build_settings(meta_path: str) -> Optional[Dict[str, Any]]:
    """Load saved build settings from ``meta_path`` (or None if absent/invalid)."""
    try:
        if os.path.isfile(meta_path):
            with open(meta_path, 'r', encoding='utf-8') as f:
                return json.load(f)
    except Exception as e:
        logger.warning('Could not read build settings: %s', e)
    return None


def save_build_settings(meta_path: str, data: Dict[str, Any]) -> bool:
    """Persist build settings to ``meta_path``."""
    try:
        os.makedirs(os.path.dirname(meta_path), exist_ok=True)
        with open(meta_path, 'w', encoding='utf-8') as f:
            json.dump(data, f, indent=2)
        return True
    except Exception as e:
        logger.warning('Could not save build settings: %s', e)
        return False


def get_player_server_settings(meta_path: str) -> Optional[Dict[str, Any]]:
    """Return the saved server address settings for deployment, if available.

    Returns a dict with ``server_ip``, ``port`` (str), ``use_https`` (bool) and
    ``verify_ssl`` (bool), or None when no usable build settings are saved.
    """
    settings = load_build_settings(meta_path)
    if not settings:
        return None
    server_ip = (settings.get('server_ip') or '').strip()
    if not server_ip:
        return None
    return {
        'server_ip': server_ip,
        'port': str(settings.get('port') or ('443' if settings.get('use_https', True) else '80')),
        'use_https': bool(settings.get('use_https', True)),
        'verify_ssl': bool(settings.get('verify_ssl', False)),
    }


def make_build_record(repo_url, branch, server_ip, port, use_https, verify_ssl,
                      orientation, max_resolution, version, built_by) -> Dict[str, Any]:
    """Assemble the metadata record to persist after a build."""
    return {
        'repo_url': repo_url,
        'branch': branch,
        'server_ip': server_ip,
        'port': str(port),
        'use_https': bool(use_https),
        'verify_ssl': bool(verify_ssl),
        'orientation': orientation,
        'max_resolution': max_resolution,
        'built_version': version,
        'built_at': datetime.utcnow().isoformat(timespec='seconds') + 'Z',
        'built_by': built_by,
    }