kam-mobile/tools/common.py

"""Small, dependency-free utilities used by the Android-only probe."""
from __future__ import annotations
import hashlib, json, os, re, shlex, signal, subprocess, time
from pathlib import Path, PurePosixPath
from datetime import datetime, timezone

ROOT = Path(__file__).resolve().parents[1]
COMMIT = '962fb850324b79bbe5480e1c2087d98f83a60917'
REPOSITORY = 'reyandme/kam_remake'
CPUS = {'aarch64': ('arm64-v8a', 183, 2, 8), 'arm': ('armeabi-v7a', 40, 1, 4),
        'x86_64': ('x86_64', 62, 2, 8)}
CURRENT_OUTPUT_DIR = ROOT / 'out/current'
CURRENT_BUILD_METADATA = CURRENT_OUTPUT_DIR / 'build.json'
CURRENT_PACKAGE_METADATA = CURRENT_OUTPUT_DIR / 'package.json'
CURRENT_APK = CURRENT_OUTPUT_DIR / 'kam-mobile.apk'
CURRENT_BUILD_SCHEMA = 'kam-android-build/2'
CURRENT_PACKAGE_SCHEMA = 'kam-android-package/1'

class ProbeError(RuntimeError):
    pass

def stamp() -> str:
    return datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')

def write_json(path: Path, value: object) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(path.name + '.tmp')
    tmp.write_text(json.dumps(value, indent=2, ensure_ascii=False) + '\n', encoding='utf-8')
    os.replace(tmp, path)

def sha256(path: Path) -> str:
    h = hashlib.sha256()
    with path.open('rb') as f:
        for b in iter(lambda: f.read(1024 * 1024), b''): h.update(b)
    return h.hexdigest()

def git_blob(data: bytes) -> str:
    return hashlib.sha1(b'blob ' + str(len(data)).encode() + b'\0' + data).hexdigest()

def safe_relative(name: str) -> str:
    p = PurePosixPath(name)
    if not name or p.is_absolute() or '\\' in name or ':' in name or '\x00' in name:
        raise ProbeError('Unsafe relative path: ' + repr(name))
    if any(c in ('', '.', '..') for c in name.split('/')) or any(ord(c) < 32 for c in name):
        raise ProbeError('Unsafe relative path: ' + repr(name))
    return str(p)

def run_command(args: list[str], cwd: Path, log: Path, timeout: int = 1800) -> dict:
    """No shell, no swallowed exit codes, kill the whole process group on timeout."""
    started = time.monotonic()
    log.parent.mkdir(parents=True, exist_ok=True)
    with log.open('w', encoding='utf-8') as out:
        out.write('$ ' + shlex.join(args) + '\n'); out.flush()
        try:
            p = subprocess.Popen(args, cwd=cwd, stdout=out, stderr=subprocess.STDOUT,
                                 start_new_session=True)
            try:
                code = p.wait(timeout=timeout)
                status = 'PASS' if code == 0 else 'FAIL'
            except (subprocess.TimeoutExpired, KeyboardInterrupt) as e:
                os.killpg(p.pid, signal.SIGKILL); p.wait()
                if isinstance(e, KeyboardInterrupt): raise
                out.write('\nTIMEOUT: entire process group terminated\n')
                status, code = 'BLOCKED', 124
        except FileNotFoundError as e:
            out.write(str(e)); status, code = 'BLOCKED', 127
    return {'status': status, 'exit_code': code, 'seconds': round(time.monotonic()-started, 3),
            'command': args, 'log': str(log.relative_to(ROOT)) if log.is_relative_to(ROOT) else str(log)}

def first_errors(log: str, limit: int = 12) -> list[str]:
    rows = [s.strip() for s in log.splitlines() if re.search(
        r'\b(fatal|error|undefined reference|cannot find|can.t find|not found)\b', s, re.I)]
    return rows[:limit]