""" Sheet management — Python port of sheet_helpers.php. File-based JSON storage, one metadata file + one data file per sheet. """ import json import logging import os import re import time import random from datetime import datetime, timezone from typing import List, Optional, Dict from ..config_runtime import server_config logger = logging.getLogger(__name__) METADATA_FILE = os.path.join(server_config.DATA_DIR, 'sheets_metadata.json') def _safe_user(user_id: str) -> str: """Sanitise user_id for use in filenames.""" return re.sub(r'[^a-zA-Z0-9_\-]', '_', user_id) def _sheet_path(user_id: str, sheet_id: str) -> str: return os.path.join(server_config.SHEETS_DIR, f"{_safe_user(user_id)}_{sheet_id}.json") def _load_metadata() -> Dict: if not os.path.exists(METADATA_FILE): return {} try: with open(METADATA_FILE, 'r') as f: return json.load(f) except Exception: return {} def _save_metadata(meta: Dict): with open(METADATA_FILE, 'w') as f: json.dump(meta, f, indent=2) def get_user_sheets(user_id: str) -> List[Dict]: meta = _load_metadata() return meta.get(user_id, []) def create_sheet(user_id: str, name: str, data: List[dict] = None, client_id: str = '') -> Dict: if data is None: data = [] sheet_id = str(int(time.time())) + str(random.randint(100, 999)) now = datetime.now(timezone.utc).isoformat() sheet_meta = { 'id': sheet_id, 'name': name or f"Untitled Sheet — {datetime.now().strftime('%Y-%m-%d %H:%M')}", 'created': now, 'modified': now, 'itemCount': len(data), 'user': user_id, 'client_id': client_id, } # Write data file path = _sheet_path(user_id, sheet_id) with open(path, 'w') as f: json.dump(data, f, indent=2) # Update metadata meta = _load_metadata() meta.setdefault(user_id, []).append(sheet_meta) _save_metadata(meta) return sheet_meta def load_sheet_data(user_id: str, sheet_id: str) -> Optional[List[dict]]: path = _sheet_path(user_id, sheet_id) if not os.path.exists(path): return None try: with open(path, 'r') as f: return json.load(f) except Exception: return None def update_sheet(user_id: str, sheet_id: str, data: List[dict]) -> bool: path = _sheet_path(user_id, sheet_id) with open(path, 'w') as f: json.dump(data, f, indent=2) # Update metadata counts meta = _load_metadata() if user_id in meta: for sheet in meta[user_id]: if sheet['id'] == sheet_id: sheet['modified'] = datetime.now(timezone.utc).isoformat() sheet['itemCount'] = len(data) break _save_metadata(meta) return True def delete_sheet(user_id: str, sheet_id: str): path = _sheet_path(user_id, sheet_id) if os.path.exists(path): os.remove(path) meta = _load_metadata() if user_id in meta: meta[user_id] = [s for s in meta[user_id] if s['id'] != sheet_id] _save_metadata(meta) def rename_sheet(user_id: str, sheet_id: str, new_name: str) -> bool: meta = _load_metadata() if user_id not in meta: return False for sheet in meta[user_id]: if sheet['id'] == sheet_id: sheet['name'] = new_name sheet['modified'] = datetime.now(timezone.utc).isoformat() _save_metadata(meta) return True return False def duplicate_sheet(user_id: str, sheet_id: str) -> Optional[Dict]: data = load_sheet_data(user_id, sheet_id) if data is None: return None meta = _load_metadata() original_name = "Copy of Sheet" for sheet in meta.get(user_id, []): if sheet['id'] == sheet_id: original_name = f"Copy of {sheet['name']}" break return create_sheet(user_id, original_name, data) def get_sheet_client_id(user_id: str, sheet_id: str) -> Optional[str]: """Return the client_id associated with a sheet, or None.""" meta = _load_metadata() for sheet in meta.get(user_id, []): if sheet['id'] == sheet_id: return sheet.get('client_id') or None return None def set_sheet_client_id(user_id: str, sheet_id: str, client_id: str): """Update the client_id on an existing sheet.""" meta = _load_metadata() if user_id in meta: for sheet in meta[user_id]: if sheet['id'] == sheet_id: sheet['client_id'] = client_id _save_metadata(meta) return True return False def generate_next_id(data: List[dict]) -> str: """Generate the next DEL-NNN id.""" max_id = 0 for row in data: num_str = row.get('Number', '').replace('DEL-', '') try: n = int(num_str) if n > max_id: max_id = n except ValueError: pass return f"DEL-{str(max_id + 1).zfill(3)}"