modules/copy_tracker.py

from __future__ import annotations
import logging
import json
import os
import platform
import time
import uuid
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
DATA_DIR = ROOT / 'data'
TRACKER_PATH = DATA_DIR / 'copy_tracker.json'
DISABLE_MARKER = '.friday_disabled'
COPY_ID_FILE = '.friday_copy_id.json'

def _now_text() -> str:
    return time.strftime('%Y-%m-%d %H:%M:%S')

def _load() -> dict[str, Any]:
    try:
        if TRACKER_PATH.exists():
            data = json.loads(TRACKER_PATH.read_text(encoding='utf-8'))
            if isinstance(data, dict):
                data.setdefault('events', [])
                return data
    except Exception as _friday_exc:
        logging.getLogger(__name__).debug('suppressed optional failure', exc_info=_friday_exc)
    return {'events': []}

def _save(data: dict[str, Any]) -> None:
    try:
        DATA_DIR.mkdir(parents=True, exist_ok=True)
        TRACKER_PATH.write_text(json.dumps(data, indent=2), encoding='utf-8')
    except Exception as _friday_exc:
        logging.getLogger(__name__).debug('suppressed optional failure', exc_info=_friday_exc)

def record_copy(target: str | os.PathLike[str], copy_type: str, result: dict[str, Any] | None=None) -> dict[str, Any]:
    target_path = Path(target).resolve()
    copy_id = uuid.uuid4().hex[:12].upper()
    event = {'id': copy_id, 'type': str(copy_type or 'unknown'), 'status': 'active', 'target': str(target_path), 'device_name': platform.node() or 'unknown-device', 'date': _now_text(), 'timestamp': time.time(), 'result': result or {}}
    data = _load()
    data.setdefault('events', []).append(event)
    _save(data)
    try:
        target_path.mkdir(parents=True, exist_ok=True)
        (target_path / COPY_ID_FILE).write_text(json.dumps(event, indent=2), encoding='utf-8')
    except Exception as _friday_exc:
        logging.getLogger(__name__).debug('suppressed optional failure', exc_info=_friday_exc)
    try:
        from core.gui_bridge import update as gui_update
        gui_update(last_copy_event=event)
    except Exception as _friday_exc:
        logging.getLogger(__name__).debug('suppressed optional failure', exc_info=_friday_exc)
    return event

def _already_tracked(target: Path) -> bool:
    target_text = str(target.resolve()).lower()
    for event in list_copies():
        if str(event.get('target', '')).lower() == target_text:
            return True
    return False

def scan_removable_copies() -> list[dict[str, Any]]:
    found: list[dict[str, Any]] = []
    if os.name != 'nt':
        return found
    try:
        import ctypes
        get_drive_type = ctypes.windll.kernel32.GetDriveTypeW
        for letter in 'DEFGHIJKLMNOPQRSTUVWXYZ':
            drive = Path(f'{letter}:\\')
            if get_drive_type(str(drive)) != 2:
                continue
            for candidate in (drive / 'FRIDAY', drive / 'FRIDAY_Copy'):
                if not candidate.exists():
                    continue
                if not ((candidate / '.portable_friday').exists() or (candidate / COPY_ID_FILE).exists()):
                    continue
                if _already_tracked(candidate):
                    continue
                found.append(record_copy(candidate, 'detected_removable', {'detected': True}))
    except Exception as _friday_exc:
        logging.getLogger(__name__).debug('suppressed optional failure', exc_info=_friday_exc)
    return found

def watch_copy_tracker(speak=None, interval: float=30.0) -> None:
    last_notice = 0.0
    while True:
        time.sleep(interval)
        try:
            found = scan_removable_copies()
            if found and speak and (time.time() - last_notice > 60.0):
                last_notice = time.time()
                speak(f"JARVIS copy detected on removable storage. Copy id {found[-1].get('id')}.")
        except Exception as _friday_exc:
            logging.getLogger(__name__).debug('suppressed optional failure', exc_info=_friday_exc)

def list_copies() -> list[dict[str, Any]]:
    events = _load().get('events', [])
    return events if isinstance(events, list) else []

def revoke_copy(copy_id_or_target: str) -> tuple[bool, str]:
    needle = (copy_id_or_target or '').strip()
    if not needle:
        return (False, 'No copy id or path provided.')
    data = _load()
    events = data.get('events', [])
    if not isinstance(events, list):
        events = []
    selected = None
    for event in events:
        if not isinstance(event, dict):
            continue
        if str(event.get('id', '')).lower() == needle.lower() or needle.lower() in str(event.get('target', '')).lower():
            selected = event
            break
    if selected is None:
        return (False, 'I could not find that copy in the tracker.')
    target = Path(str(selected.get('target') or ''))
    try:
        target.mkdir(parents=True, exist_ok=True)
        (target / DISABLE_MARKER).write_text(json.dumps({'disabled_at': _now_text(), 'copy_id': selected.get('id'), 'reason': 'Revoked by owner from source system.'}, indent=2), encoding='utf-8')
        selected['status'] = 'revoked'
        selected['revoked_at'] = _now_text()
        _save(data)
        return (True, f"Copy {selected.get('id')} has been locked. It will refuse to start from {target}.")
    except Exception as _friday_exc:
        logging.getLogger(__name__).debug('suppressed optional failure', exc_info=_friday_exc)
        return (False, f'I found the copy, but could not write the lock marker: {_friday_exc}')

def disabled_reason(root: Path | None=None) -> str:
    base = root or ROOT
    marker = base / DISABLE_MARKER
    if not marker.exists():
        return ''
    try:
        data = json.loads(marker.read_text(encoding='utf-8'))
        return str(data.get('reason') or 'This JARVIS copy has been disabled by the owner.')
    except Exception as _friday_exc:
        logging.getLogger(__name__).debug('suppressed optional failure', exc_info=_friday_exc)
        return 'This JARVIS copy has been disabled by the owner.'