Spaces:
Running
Running
File size: 6,198 Bytes
a31f556 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 | 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.'
|