File.py

"""
File.py — Enhanced canonical file value object.

Drop-in replacement/extension for the File class in data.py.
All raw I/O lives here. Callers use File, never pathlib/open directly.

Error callbacks:
    on_read_error(cb)       cb(path, error) called on any read failure
    on_write_error(cb)      cb(path, error) called on any write failure
    on_permission_error(cb) cb(path, error) called specifically on PermissionError/WinError 5

Usage:
    f = File('C:/path/to/file.txt')
    f.on_read_error(lambda p, e: print(f'read failed: {e}'))
    text = f.readText()
    ok = f.writeText('hello')
    lockers = f.getLockingProcesses()
    f.killLockingProcesses()
    f.setMtime()  # touch
    meta = f.getMetadata()
    print(f.sha1Hex())
    print(f.base64Encode())
"""

from __future__ import annotations

import base64
import hashlib
import os
import shutil
import sys
import time
from pathlib import Path
from typing import Any, Callable

EMPTY_STRING = ''

# ---------------------------------------------------------------------------
# Exception recorder — mirrors data.py's recordException signature so this
# file can be used standalone OR inside the PyEncoder project unchanged.
# ---------------------------------------------------------------------------
try:
    from data import recordException as _recordException  # type: ignore
except ImportError as exc:
    print(f'[WARN:File.recordException-import] {type(exc).__name__}: {exc}', file=sys.stderr)
    def _recordException(context: str, error=None, *, handled: bool = True, source: str = 'File.py') -> int:  # type: ignore
        print(f'[EXCEPTION:{context}] {type(error).__name__}: {error}', file=sys.stderr)
        return 0


def _record(context: str, error=None) -> int:
    return _recordException(context, error, source='File.py')


# ---------------------------------------------------------------------------
# Lock detection — uses restartmgr on Windows (pip install restartmgr).
# Falls back gracefully when the package is not installed.
# ---------------------------------------------------------------------------
def _who_locks(path: Path) -> list[dict[str, Any]]:
    """Return list of {pid, name, type} dicts for processes locking *path*."""
    try:
        from restartmgr import who_locks  # type: ignore
        return [
            {'pid': int(p.pid), 'name': str(p.app_name), 'type': str(p.app_type.name)}
            for p in who_locks(path)
        ]
    except ImportError as exc:
        _record('File._who_locks.restartmgr-import', exc)
        return []
    except Exception as exc:
        _record('File._who_locks', exc)
        return []


# ---------------------------------------------------------------------------
# Permission helpers
# ---------------------------------------------------------------------------
def _get_permissions_posix(path: Path) -> dict[str, Any]:
    st = path.stat()
    mode = st.st_mode
    return {
        'mode_octal': oct(mode & 0o777),
        'readable': os.access(path, os.R_OK),
        'writable': os.access(path, os.W_OK),
        'executable': os.access(path, os.X_OK),
        'uid': st.st_uid,
        'gid': st.st_gid,
    }


def _get_permissions_windows(path: Path) -> dict[str, Any]:
    result: dict[str, Any] = {
        'readable': os.access(path, os.R_OK),
        'writable': os.access(path, os.W_OK),
        'executable': os.access(path, os.X_OK),
    }
    try:
        import subprocess
        out = subprocess.run(
            ['icacls', str(path)],
            capture_output=True, text=True, timeout=5,
        )
        result['icacls'] = out.stdout.strip()
    except Exception as exc:
        _record('File._get_permissions_windows', exc)
    return result


# ---------------------------------------------------------------------------
# File
# ---------------------------------------------------------------------------
class File:
    """
    Canonical file value object. All project file I/O must go through this class.

    Constructor
    -----------
    File(path)
    File(path, kind='audio', language='python', mime='text/plain')

    Error callbacks — set before the operation you want to intercept:
        .on_read_error(cb)       cb(path: Path, error: Exception)
        .on_write_error(cb)      cb(path: Path, error: Exception)
        .on_permission_error(cb) cb(path: Path, error: Exception)
    """

    def __init__(
        self,
        path: str | Path,
        *,
        kind: str = 'file',
        language: str = '',
        mime: str = '',
    ) -> None:
        self.path = Path(path)
        self.kind = str(kind or 'file')
        self.language = str(language or EMPTY_STRING)
        self.mime = str(mime or EMPTY_STRING)
        self._on_read_error: Callable[[Path, Exception], None] | None = None
        self._on_write_error: Callable[[Path, Exception], None] | None = None
        self._on_permission_error: Callable[[Path, Exception], None] | None = None

    # ------------------------------------------------------------------
    # Callback registration
    # ------------------------------------------------------------------

    def on_read_error(self, cb: Callable[[Path, Exception], None]) -> 'File':
        self._on_read_error = cb
        return self

    def on_write_error(self, cb: Callable[[Path, Exception], None]) -> 'File':
        self._on_write_error = cb
        return self

    def on_permission_error(self, cb: Callable[[Path, Exception], None]) -> 'File':
        self._on_permission_error = cb
        return self

    def onPermissionError(self, cb: Callable[[Path, Exception], None]) -> 'File':  # noqa: N802
        """CamelCase alias used by launcher/startup code and File-class whitepapers."""
        return self.on_permission_error(cb)

    def _fire_read_error(self, exc: Exception) -> None:
        if isinstance(exc, PermissionError):
            if self._on_permission_error:
                self._on_permission_error(self.path, exc)
        if self._on_read_error:
            self._on_read_error(self.path, exc)

    def _fire_write_error(self, exc: Exception) -> None:
        if isinstance(exc, PermissionError):
            if self._on_permission_error:
                self._on_permission_error(self.path, exc)
        if self._on_write_error:
            self._on_write_error(self.path, exc)

    # ------------------------------------------------------------------
    # Identity / metadata properties
    # ------------------------------------------------------------------

    @property
    def name(self) -> str:
        return self.path.name

    @property
    def stem(self) -> str:
        return self.path.stem

    @property
    def suffix(self) -> str:
        return self.path.suffix.lower()

    @property
    def exists(self) -> bool:
        return self.path.exists()

    @property
    def size(self) -> int:
        """File size in bytes. Returns 0 if file does not exist or stat fails."""
        try:
            return int(self.path.stat().st_size)
        except Exception as exc:
            _record('File.size', exc)
            return 0

    @property
    def mtime(self) -> float:
        """Last-modified timestamp as a Unix float. Returns 0.0 on failure."""
        try:
            return float(self.path.stat().st_mtime)
        except Exception as exc:
            _record('File.mtime', exc)
            return 0.0

    # Legacy alias — keeps existing callers working
    @property
    def modified(self) -> float:
        return self.mtime

    def setMtime(self, timestamp: float | None = None) -> bool:  # noqa: N802
        """
        Set last-modified time. Defaults to now.
        Wraps os.utime — works on both POSIX and Windows.
        """
        try:
            t = float(timestamp) if timestamp is not None else time.time()
            os.utime(self.path, (t, t))
            return True
        except Exception as exc:
            _record('File.setMtime', exc)
            self._fire_write_error(exc)
            return False

    # ------------------------------------------------------------------
    # Full metadata
    # ------------------------------------------------------------------

    def getMetadata(self, field: str | None = None) -> Any:  # noqa: N802
        """
        Return file metadata.

        getMetadata()           -> dict with all fields
        getMetadata('size')     -> int
        getMetadata('mtime')    -> float
        getMetadata('missing')  -> None  (never raises)

        Fields: size, mtime, atime, ctime, mode_octal, readable, writable,
                executable, uid (POSIX), gid (POSIX), icacls (Windows).
        """
        try:
            st = self.path.stat()
            data: dict[str, Any] = {
                'size': int(st.st_size),
                'mtime': float(st.st_mtime),
                'atime': float(st.st_atime),
                'ctime': float(st.st_ctime),
                'readable': os.access(self.path, os.R_OK),
                'writable': os.access(self.path, os.W_OK),
                'executable': os.access(self.path, os.X_OK),
            }
            if os.name == 'nt':
                data.update(_get_permissions_windows(self.path))
            else:
                data.update(_get_permissions_posix(self.path))
            if field is not None:
                return data.get(field)
            return data
        except Exception as exc:
            _record('File.getMetadata', exc)
            return None if field else {}

    def setMetadata(self, field: str, value: Any) -> bool:  # noqa: N802
        """
        Set a metadata field.

        Supported fields:
            'mtime'   -> calls setMtime(value)
            'atime'   -> sets access time via os.utime
            On POSIX: 'mode' -> calls os.chmod(path, value)
        """
        try:
            if field == 'mtime':
                return self.setMtime(float(value))
            if field == 'atime':
                st = self.path.stat()
                os.utime(self.path, (float(value), st.st_mtime))
                return True
            if field == 'mode' and os.name != 'nt':
                os.chmod(self.path, int(value))
                return True
            _record('File.setMetadata', ValueError(f'unsupported field: {field!r}'))
            return False
        except Exception as exc:
            _record('File.setMetadata', exc)
            self._fire_write_error(exc)
            return False

    # ------------------------------------------------------------------
    # Permissions
    # ------------------------------------------------------------------

    def getPermissions(self) -> dict[str, Any]:  # noqa: N802
        """Return permission info. See getMetadata() for full dict."""
        try:
            if os.name == 'nt':
                return _get_permissions_windows(self.path)
            return _get_permissions_posix(self.path)
        except Exception as exc:
            _record('File.getPermissions', exc)
            return {}

    def setPermissions(self, mode: int | str) -> bool:  # noqa: N802
        """
        Set file permissions.

        Accepts octal ints like 0o644 and strings like '644' or '0o644'.
        Uses os.chmod on every platform so Windows only changes the supported
        file mode bits instead of granting broad ACLs.
        """
        try:
            parsed_mode = self._parsePermissionsMode(mode)
            os.chmod(self.path, parsed_mode)
            return True
        except Exception as exc:
            _record('File.setPermissions', exc)
            self._fire_permission_error(exc)
            return False

    def _parsePermissionsMode(self, mode: int | str) -> int:  # noqa: N802
        if isinstance(mode, str):
            stripped = mode.strip().lower()
            if stripped.startswith('0o'):
                return int(stripped, 8)
            if stripped and all(ch in '01234567' for ch in stripped) and 3 <= len(stripped) <= 4:
                return int(stripped, 8)
            return int(stripped, 10)
        return int(mode)

    def _is_permission_error(self, exc: Exception) -> bool:
        return isinstance(exc, PermissionError) or getattr(exc, 'winerror', None) == 5 or getattr(exc, 'errno', None) in {13, 1}

    def _fire_permission_error(self, exc: Exception) -> None:
        if self._on_permission_error and self._is_permission_error(exc):
            try:
                self._on_permission_error(self.path, exc)
            except Exception as callback_exc:
                _record('File.onPermissionError-callback', callback_exc)

    # ------------------------------------------------------------------
    # Lock management (Windows — requires: pip install restartmgr)
    # ------------------------------------------------------------------

    def getLockingProcesses(self) -> list[dict[str, Any]]:  # noqa: N802
        """
        Return list of processes currently locking this file.
        Each item: {'pid': int, 'name': str, 'type': str}
        Returns [] on non-Windows or if restartmgr is not installed.
        """
        if os.name != 'nt':
            return []
        return _who_locks(self.path)

    def isLocked(self) -> bool:  # noqa: N802
        """True if any process currently holds a lock on this file."""
        return len(self.getLockingProcesses()) > 0

    def killLockingProcesses(self, *, only_known_children: bool = False) -> list[dict[str, Any]]:  # noqa: N802
        """
        Kill all processes locking this file via taskkill /PID /T /F.

        only_known_children=True: only kill if the locker's PID is a child of
        the current process (safer for shared-environment use).

        Returns the list of processes that were targeted.
        """
        import subprocess
        lockers = self.getLockingProcesses()
        current_pid = os.getpid()
        if only_known_children:
            try:
                import psutil  # type: ignore
                children = {p.pid for p in psutil.Process(current_pid).children(recursive=True)}
                lockers = [l for l in lockers if l['pid'] in children]
            except ImportError:
                pass
        for proc in lockers:
            pid = int(proc.get('pid') or 0)
            if pid <= 0 or pid == current_pid:
                continue
            print(f'[WARN:file-lock] {self.path} locked by PID={pid} name={proc.get("name")}')
            subprocess.run(['taskkill', '/PID', str(pid), '/T', '/F'], check=False)
        return lockers

    def waitUntilUnlocked(self, timeout: float = 10.0, interval: float = 0.25) -> bool:  # noqa: N802
        """
        Poll until the file is no longer locked or timeout expires.
        Returns True if unlocked, False if timeout was reached.
        """
        deadline = time.monotonic() + float(timeout)
        while time.monotonic() < deadline:
            if not self.isLocked():
                return True
            time.sleep(float(interval))
        return not self.isLocked()

    # ------------------------------------------------------------------
    # Read operations
    # ------------------------------------------------------------------

    def readText(self, encoding: str = 'utf-8', errors: str = 'replace') -> str:  # noqa: N802
        try:
            return self.path.read_text(encoding=encoding, errors=errors)  # file-io-ok: canonical File wrapper owns raw read_text
        except Exception as exc:
            _record('File.readText', exc)
            self._fire_read_error(exc)
            return EMPTY_STRING

    def readBytes(self) -> bytes:  # noqa: N802
        try:
            return self.path.read_bytes()  # file-io-ok: canonical File wrapper owns raw read_bytes
        except Exception as exc:
            _record('File.readBytes', exc)
            self._fire_read_error(exc)
            return b''

    def readLines(self, encoding: str = 'utf-8', errors: str = 'replace') -> list[str]:  # noqa: N802
        """Return file content split into lines (newlines stripped)."""
        text = self.readText(encoding=encoding, errors=errors)
        return text.splitlines()

    # ------------------------------------------------------------------
    # Write operations
    # ------------------------------------------------------------------

    def writeText(self, text: str, encoding: str = 'utf-8') -> bool:  # noqa: N802
        try:
            self.path.parent.mkdir(parents=True, exist_ok=True)
            self.path.write_text(str(text or EMPTY_STRING), encoding=encoding)  # file-io-ok: canonical File wrapper owns raw write_text
            return True
        except Exception as exc:
            _record('File.writeText', exc)
            self._fire_write_error(exc)
            return False

    def writeBytes(self, payload: bytes | bytearray | memoryview) -> bool:  # noqa: N802
        try:
            self.path.parent.mkdir(parents=True, exist_ok=True)
            self.path.write_bytes(bytes(payload or b''))  # file-io-ok: canonical File wrapper owns raw write_bytes
            return True
        except Exception as exc:
            _record('File.writeBytes', exc)
            self._fire_write_error(exc)
            return False

    def appendText(self, text: str, encoding: str = 'utf-8', errors: str = 'replace') -> bool:  # noqa: N802
        try:
            self.path.parent.mkdir(parents=True, exist_ok=True)
            with self.path.open('a', encoding=encoding, errors=errors) as handle:  # file-io-ok: canonical File wrapper owns raw append open
                handle.write(str(text or EMPTY_STRING))
            return True
        except Exception as exc:
            _record('File.appendText', exc)
            self._fire_write_error(exc)
            return False

    def open(self, mode: str = 'r', **kwargs):  # file-io-ok: canonical File wrapper owns raw open
        try:
            if any(flag in str(mode or '') for flag in ('w', 'a', 'x', '+')):
                self.path.parent.mkdir(parents=True, exist_ok=True)
            return self.path.open(mode, **kwargs)
        except Exception as exc:
            _record('File.open', exc)
            self._fire_read_error(exc) if 'r' in mode else self._fire_write_error(exc)
            raise

    # ------------------------------------------------------------------
    # Hashing
    # ------------------------------------------------------------------

    def md5Hex(self) -> str:  # noqa: N802
        try:
            return hashlib.md5(self.readBytes()).hexdigest()
        except Exception as exc:
            _record('File.md5Hex', exc)
            return EMPTY_STRING

    def sha1Hex(self) -> str:  # noqa: N802
        try:
            return hashlib.sha1(self.readBytes()).hexdigest()
        except Exception as exc:
            _record('File.sha1Hex', exc)
            return EMPTY_STRING

    def sha256Hex(self) -> str:  # noqa: N802
        try:
            return hashlib.sha256(self.readBytes()).hexdigest()
        except Exception as exc:
            _record('File.sha256Hex', exc)
            return EMPTY_STRING

    # ------------------------------------------------------------------
    # Encoding
    # ------------------------------------------------------------------

    def base64Encode(self) -> str:  # noqa: N802
        """Return base64-encoded content of the file as a UTF-8 string."""
        try:
            return base64.b64encode(self.readBytes()).decode('utf-8')
        except Exception as exc:
            _record('File.base64Encode', exc)
            return EMPTY_STRING

    def base64Decode(self, encoded: str) -> bool:  # noqa: N802
        """Write base64-encoded string *encoded* as raw bytes to this file."""
        try:
            return self.writeBytes(base64.b64decode(encoded.encode('utf-8')))
        except Exception as exc:
            _record('File.base64Decode', exc)
            self._fire_write_error(exc)
            return False

    # ------------------------------------------------------------------
    # File operations
    # ------------------------------------------------------------------

    def copyTo(self, target: str | Path) -> bool:  # noqa: N802
        try:
            destination = Path(target)
            destination.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(str(self.path), str(destination))  # file-io-ok: canonical File wrapper owns raw copy2
            return True
        except Exception as exc:
            _record('File.copyTo', exc)
            self._fire_write_error(exc)
            return False

    def moveTo(self, target: str | Path) -> bool:  # noqa: N802
        """Atomic rename/move. Uses os.rename for same-filesystem moves."""
        try:
            destination = Path(target)
            destination.parent.mkdir(parents=True, exist_ok=True)
            shutil.move(str(self.path), str(destination))  # file-io-ok: canonical File wrapper owns raw move
            self.path = destination
            return True
        except Exception as exc:
            _record('File.moveTo', exc)
            self._fire_write_error(exc)
            return False

    def delete(self) -> bool:
        """Delete the file. Returns True if deleted or already gone."""
        try:
            self.path.unlink(missing_ok=True)
            return True
        except Exception as exc:
            _record('File.delete', exc)
            self._fire_write_error(exc)
            return False

    def zeroOut(self) -> bool:  # noqa: N802
        """Overwrite content with null bytes (secure pre-delete wipe)."""
        try:
            size = self.size
            return self.writeBytes(b'\x00' * size)
        except Exception as exc:
            _record('File.zeroOut', exc)
            self._fire_write_error(exc)
            return False

    def deleteSecure(self) -> bool:  # noqa: N802
        """Zero out then delete."""
        return self.zeroOut() and self.delete()

    # ------------------------------------------------------------------
    # Repr
    # ------------------------------------------------------------------

    def __repr__(self) -> str:
        return f'File({str(self.path)!r})'

    def __str__(self) -> str:
        return str(self.path)

    def __eq__(self, other: object) -> bool:
        if isinstance(other, File):
            return self.path == other.path
        return NotImplemented

    def __hash__(self) -> int:
        return hash(self.path)


# ---------------------------------------------------------------------------
# Subclasses — mirrors existing data.py subclasses
# ---------------------------------------------------------------------------

class CSSFile(File):
    def __init__(self, path: str | Path) -> None:
        super().__init__(path, kind='css', language='css', mime='text/css')


class JSFile(File):
    def __init__(self, path: str | Path) -> None:
        super().__init__(path, kind='js', language='javascript', mime='application/javascript')


class HTMLFile(File):
    def __init__(self, path: str | Path) -> None:
        super().__init__(path, kind='html', language='html', mime='text/html')


class AudioFile(File):
    def __init__(self, path: str | Path) -> None:
        super().__init__(path, kind='audio', mime='audio/wav')


class ImageFile(File):
    def __init__(self, path: str | Path) -> None:
        super().__init__(path, kind='image')


class FontFile(File):
    def __init__(self, path: str | Path) -> None:
        super().__init__(path, kind='font')