Spaces:
Running
Running
File size: 2,833 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 | import os
import shutil
import sqlite3
import logging
from pathlib import Path
async def redact_env_file(env_path: Path):
if not env_path.exists():
return
with open(env_path, 'r', encoding='utf-8') as f:
lines = f.readlines()
with open(env_path, 'w', encoding='utf-8') as f:
for line in lines:
stripped = line.strip()
if not stripped or stripped.startswith('#'):
f.write(line)
continue
if '=' in line:
key = line.split('=', 1)[0]
f.write(f"{key}=\n")
else:
f.write(line)
async def truncate_sqlite_table(db_path: Path, table_name: str):
if not db_path.exists():
return
try:
with sqlite3.connect(db_path) as conn:
# Check if table exists first
cursor = conn.cursor()
cursor.execute(f"SELECT name FROM sqlite_master WHERE type='table' AND name='{table_name}'")
if cursor.fetchone():
conn.execute(f"DELETE FROM {table_name}")
conn.commit()
except Exception as e:
logging.error(f"Failed to truncate {table_name} in {db_path}: {e}")
async def delete_file(file_path: Path):
if file_path.exists():
try:
if file_path.is_file():
os.remove(file_path)
else:
shutil.rmtree(file_path)
except Exception as e:
import logging; logging.getLogger(__name__).error(f"Swallowed exception: {e}")
async def strip_personal_data(staging_dir: Path) -> Path:
sanitized_dir = staging_dir.parent / (staging_dir.name + "_sanitized")
if sanitized_dir.exists():
shutil.rmtree(sanitized_dir)
shutil.copytree(staging_dir, sanitized_dir)
# 1. Redact environments and delete Playwright session
code_dir = sanitized_dir / "F.R.I.D.A.Y - OMEGA"
if code_dir.exists():
await redact_env_file(code_dir / ".env")
await delete_file(code_dir / "session_store.json")
await delete_file(code_dir / "scratch") # Clean scratch too
# 2. Truncate personal data from SQLite databases
db_locations = [
code_dir / "memory.db",
sanitized_dir / "JARVIS_OS_APPDATA" / "memory.db",
sanitized_dir / "JARVIS_OS_APPDATA" / "nexus.db",
sanitized_dir / "memory.db"
]
for db_path in db_locations:
if db_path.exists():
await truncate_sqlite_table(db_path, "vault_credentials")
await truncate_sqlite_table(db_path, "conversation_history")
await truncate_sqlite_table(db_path, "upgrade_history")
await truncate_sqlite_table(db_path, "overwatch_logs")
return sanitized_dir
|