import os import json import time import shutil import hashlib from pathlib import Path from typing import Dict, Optional, Any, List, Tuple from config import ( CONFIG_FILE, MODEL_DIR, EXPORT_DIR, DEFAULT_MEMORY_LIMIT_GB, DEFAULT_MODEL, ) from utils import console, Theme, error_logger, debug_logger # ========================================================================== # Constants # ========================================================================== CONFIG_VERSION = 2 CONFIG_BACKUP_DIR = "backups" MAX_CONFIG_BACKUPS = 5 MIN_CONFIG_BACKUP_INTERVAL_SEC = 30.0 # Nilai default yang dipakai saat migrasi config lama. _TRAINING_DEFAULTS: Dict[str, Any] = { "base_model": DEFAULT_MODEL, "learning_rate": 3e-5, "batch_size": 4, "num_epochs": 6, "max_length": 256, "warmup_steps": 100, "warmup_ratio": 0.1, "weight_decay": 0.01, "gradient_accumulation_steps": 1, "validation_split": 0.1, "seed": 42, "save_steps": 500, "logging_steps": 50, "early_stopping_patience": 3, "fp16": False, "bf16": False, "max_samples_limit": 20000, "data_augmentation": { "enabled": False, "split_long": True, "split_max_chars": 300, }, "use_peft": False, "use_8bit": False, "peft_config": { "r": 8, "alpha": 16, "dropout": 0.05, "target_modules": "auto", }, "freeze_embeddings": False, "dynamic_grad_accumulation": False, "gradient_clip_value": 1.0, "eval_gen_samples": 16, "load_best_model_at_end": True, "max_context_tokens": 768, "num_proc": 1, } _GENERATION_DEFAULTS: Dict[str, Any] = { "max_new_tokens": 150, "temperature": 0.8, "top_p": 0.9, "top_k": 50, "repetition_penalty": 1.1, "no_repeat_ngram_size": 3, "max_context_length": 768, "stop_sequences": [], } # Rentang nilai yang dianggap valid (min, max), inclusive. _TRAINING_RANGES: Dict[str, Tuple[float, float]] = { "learning_rate": (1e-8, 1e-1), "batch_size": (1, 4096), "num_epochs": (1, 1000), "max_length": (16, 32768), "warmup_steps": (0, 1000000), "warmup_ratio": (0.0, 1.0), "weight_decay": (0.0, 1.0), "gradient_accumulation_steps": (1, 1024), "validation_split": (0.0, 0.9), "seed": (0, 2147483647), "save_steps": (1, 1000000), "logging_steps": (1, 1000000), "early_stopping_patience": (0, 1000), "max_samples_limit": (1, 100000000), "gradient_clip_value": (0.0, 100.0), "eval_gen_samples": (0, 10000), "max_context_tokens": (64, 131072), "num_proc": (1, 128), } _GENERATION_RANGES: Dict[str, Tuple[float, float]] = { "max_new_tokens": (1, 8192), "temperature": (0.01, 2.0), "top_p": (0.0, 1.0), "top_k": (0, 1000), "repetition_penalty": (1.0, 3.0), "no_repeat_ngram_size": (0, 20), "max_context_length": (64, 131072), } # Nama file yang dianggap sebagai penanda komponen model. _WEIGHT_FILE_NAMES = { "pytorch_model.bin", "model.safetensors", "tf_model.h5", "adapter_model.bin", "adapter_model.safetensors", } _TOKENIZER_FILE_NAMES = { "tokenizer.json", "tokenizer_config.json", "vocab.json", "vocab.txt", "merges.txt", "special_tokens_map.json", "spiece.model", } # ========================================================================== # Presets # ========================================================================== TRAINING_PRESETS: Dict[str, Dict[str, Any]] = { "fast": { "learning_rate": 5e-5, "batch_size": 8, "num_epochs": 3, "max_length": 128, "gradient_accumulation_steps": 1, "warmup_ratio": 0.05, "save_steps": 200, "logging_steps": 25, "early_stopping_patience": 2, }, "balanced": { "learning_rate": 3e-5, "batch_size": 4, "num_epochs": 6, "max_length": 256, "gradient_accumulation_steps": 1, "warmup_ratio": 0.1, "save_steps": 500, "logging_steps": 50, "early_stopping_patience": 3, }, "quality": { "learning_rate": 2e-5, "batch_size": 2, "num_epochs": 12, "max_length": 512, "gradient_accumulation_steps": 4, "warmup_ratio": 0.15, "save_steps": 500, "logging_steps": 50, "early_stopping_patience": 5, }, "low_vram": { "learning_rate": 3e-5, "batch_size": 1, "num_epochs": 6, "max_length": 128, "gradient_accumulation_steps": 4, "warmup_ratio": 0.1, "save_steps": 200, "logging_steps": 25, "early_stopping_patience": 3, "use_peft": True, "use_8bit": True, "freeze_embeddings": True, "dynamic_grad_accumulation": True, }, } GENERATION_PRESETS: Dict[str, Dict[str, Any]] = { "balanced": { "temperature": 0.8, "top_p": 0.9, "top_k": 50, "repetition_penalty": 1.1, "no_repeat_ngram_size": 3, "max_new_tokens": 150, }, "creative": { "temperature": 1.1, "top_p": 0.95, "top_k": 80, "repetition_penalty": 1.05, "no_repeat_ngram_size": 2, "max_new_tokens": 200, }, "precise": { "temperature": 0.4, "top_p": 0.85, "top_k": 30, "repetition_penalty": 1.2, "no_repeat_ngram_size": 4, "max_new_tokens": 120, }, "coding": { "temperature": 0.2, "top_p": 0.9, "top_k": 40, "repetition_penalty": 1.15, "no_repeat_ngram_size": 3, "max_new_tokens": 256, }, } # ========================================================================== # ModelMetadata # ========================================================================== class ModelMetadata: def __init__( self, model_name: str, base_model: str, dataset_path: str, dataset_format: str, dataset_size: int, training_samples: int, validation_samples: int, created_date: str, status: str = "trained", description: str = "", version: str = "1.0.0", hash: str = "", tags: list = None, training_env: dict = None, metrics: dict = None, config: dict = None, ): self.model_name = model_name self.base_model = base_model self.dataset_path = dataset_path self.dataset_format = dataset_format self.dataset_size = dataset_size self.training_samples = training_samples self.validation_samples = validation_samples self.created_date = created_date self.status = status self.description = description self.version = version self.hash = hash self.tags = tags or [] self.training_env = training_env or {} self.metrics = metrics or {} self.config = config or {} def to_dict(self) -> Dict: return { "model_name": self.model_name, "base_model": self.base_model, "dataset_path": self.dataset_path, "dataset_format": self.dataset_format, "dataset_size": self.dataset_size, "training_samples": self.training_samples, "validation_samples": self.validation_samples, "created_date": self.created_date, "status": self.status, "description": self.description, "version": self.version, "hash": self.hash, "tags": self.tags, "training_env": self.training_env, "metrics": self.metrics, "config": self.config, } @classmethod def from_dict(cls, data: Dict) -> "ModelMetadata": return cls( model_name=data.get("model_name", ""), base_model=data.get("base_model", ""), dataset_path=data.get("dataset_path", ""), dataset_format=data.get("dataset_format", ""), dataset_size=data.get("dataset_size", 0), training_samples=data.get("training_samples", 0), validation_samples=data.get("validation_samples", 0), created_date=data.get("created_date", ""), status=data.get("status", "trained"), description=data.get("description", ""), version=data.get("version", "1.0.0"), hash=data.get("hash", ""), tags=data.get("tags", []), training_env=data.get("training_env", {}), metrics=data.get("metrics", {}), config=data.get("config", {}), ) # ========================================================================== # ModelManager # ========================================================================== class ModelManager: def __init__(self): # Inisialisasi atribut yang dipakai oleh save_config/backup SEBELUM # load_config() dipanggil, supaya tidak ada AttributeError. self._last_backup_time: float = 0.0 self.config_file = CONFIG_FILE self.config: Dict = self.load_config() # ------------------------------------------------------------------ # Config load / create / migration # ------------------------------------------------------------------ def load_config(self) -> Dict: if os.path.exists(self.config_file): try: with open(self.config_file, "r", encoding="utf-8") as f: data = json.load(f) except (json.JSONDecodeError, IOError) as e: console.print(Theme.warning(f"Config error: {e}, using default")) error_logger.error(f"Config load failed: {e}") return self.create_default_config() if not isinstance(data, dict): console.print( Theme.warning("Config bukan dict, menggunakan default") ) return self.create_default_config() migrated, was_migrated = self._migrate_config(data) if was_migrated: try: self.save_config(migrated) debug_logger.debug("Config migrated and saved") except Exception as e: # Jangan crash hanya karena gagal simpan hasil migrasi. error_logger.error(f"Config migration save failed: {e}") # Validasi hanya untuk memberi tahu, tidak menghentikan load. self._validate_and_log(migrated, prefix="load_config") return migrated else: return self.create_default_config() def create_default_config(self) -> Dict: import torch config: Dict[str, Any] = { "config_version": CONFIG_VERSION, "current_model": None, "models": {}, "memory_limit_gb": DEFAULT_MEMORY_LIMIT_GB, "multi_gpu": False, "training_config": dict(_TRAINING_DEFAULTS, **{ "fp16": True if torch.cuda.is_available() else False, }), "generation_config": dict(_GENERATION_DEFAULTS), } # Deep-copy untuk field nested agar tidak ada shared reference. config["training_config"]["data_augmentation"] = dict( _TRAINING_DEFAULTS["data_augmentation"] ) config["training_config"]["peft_config"] = dict( _TRAINING_DEFAULTS["peft_config"] ) self.save_config(config) return config def _migrate_config(self, cfg: Dict) -> Tuple[Dict, bool]: """ Migrasi config lama ke versi terbaru. Return (config_baru, apakah_berubah). Tidak menghapus field yang sudah ada. """ changed = False try: version = int(cfg.get("config_version", 1)) except (TypeError, ValueError): version = 1 if version < CONFIG_VERSION: # Pastikan top-level field penting ada. for key, value in [ ("current_model", None), ("models", {}), ("memory_limit_gb", DEFAULT_MEMORY_LIMIT_GB), ("multi_gpu", False), ]: if key not in cfg: cfg[key] = value changed = True # Migrasi training_config tc = cfg.get("training_config") if not isinstance(tc, dict): tc = {} changed = True for key, default_value in _TRAINING_DEFAULTS.items(): if key not in tc: if isinstance(default_value, dict): tc[key] = dict(default_value) else: tc[key] = default_value changed = True cfg["training_config"] = tc # Migrasi generation_config gc = cfg.get("generation_config") if not isinstance(gc, dict): gc = {} changed = True for key, default_value in _GENERATION_DEFAULTS.items(): if key not in gc: if isinstance(default_value, list): gc[key] = list(default_value) else: gc[key] = default_value changed = True cfg["generation_config"] = gc cfg["config_version"] = CONFIG_VERSION changed = True return cfg, changed # ------------------------------------------------------------------ # Validation # ------------------------------------------------------------------ def _check_range_dict( self, cfg: Dict, ranges: Dict[str, Tuple[float, float]], issues: List[str], label: str, ) -> None: for key, (lo, hi) in ranges.items(): if key not in cfg: continue val = cfg[key] # bool adalah subclass int - lewati agar tidak salah nilai. if isinstance(val, bool): issues.append(f"{label}.{key} bertipe bool, diharapkan angka") continue if not isinstance(val, (int, float)): issues.append( f"{label}.{key} bertipe {type(val).__name__}, diharapkan angka" ) continue if val < lo or val > hi: issues.append( f"{label}.{key}={val} di luar rentang [{lo}, {hi}]" ) def _validate_training_config(self, cfg: Dict) -> List[str]: issues: List[str] = [] if not isinstance(cfg, dict): return ["training_config bukan dict"] self._check_range_dict(cfg, _TRAINING_RANGES, issues, "training_config") # Cek nilai yang tidak masuk rentang numerik tapi boolean/logika. for key in ("fp16", "bf16"): if key in cfg and not isinstance(cfg[key], bool): issues.append(f"training_config.{key} bukan boolean") for key in ( "use_peft", "use_8bit", "freeze_embeddings", "dynamic_grad_accumulation", "load_best_model_at_end", ): if key in cfg and not isinstance(cfg[key], bool): issues.append(f"training_config.{key} bukan boolean") bm = cfg.get("base_model") if bm is not None and not isinstance(bm, str): issues.append("training_config.base_model bukan string") return issues def _validate_generation_config(self, cfg: Dict) -> List[str]: issues: List[str] = [] if not isinstance(cfg, dict): return ["generation_config bukan dict"] self._check_range_dict(cfg, _GENERATION_RANGES, issues, "generation_config") ss = cfg.get("stop_sequences") if ss is not None and not isinstance(ss, list): issues.append("generation_config.stop_sequences bukan list") return issues def _validate_global_config(self, cfg: Dict) -> List[str]: issues: List[str] = [] if "memory_limit_gb" in cfg: ml = cfg["memory_limit_gb"] if isinstance(ml, bool) or not isinstance(ml, (int, float)): issues.append("memory_limit_gb bukan angka") elif ml < 0 or ml > 4096: issues.append(f"memory_limit_gb={ml} di luar rentang [0, 4096]") if "multi_gpu" in cfg and not isinstance(cfg["multi_gpu"], bool): issues.append("multi_gpu bukan boolean") if "models" in cfg and not isinstance(cfg["models"], dict): issues.append("models bukan dict") return issues def validate_config( self, cfg: Optional[Dict] = None, log: bool = False ) -> Dict[str, List[str]]: """ Validasi config. Return dict berisi daftar issue per kategori. Tidak crash — hanya melaporkan. """ if cfg is None: cfg = self.config report = { "global": self._validate_global_config(cfg), "training": self._validate_training_config( cfg.get("training_config", {}) ), "generation": self._validate_generation_config( cfg.get("generation_config", {}) ), } if log: total = sum(len(v) for v in report.values()) if total == 0: console.print(Theme.success("Config valid")) else: console.print(Theme.warning(f"Config issues: {total}")) for category, items in report.items(): for item in items: console.print(f" [{category}] {item}") return report def _validate_and_log(self, cfg: Dict, prefix: str = "") -> None: try: report = self.validate_config(cfg, log=False) except Exception as e: debug_logger.debug(f"Validation error: {e}") return for category, items in report.items(): for item in items: msg = f"{prefix + ': ' if prefix else ''}[{category}] {item}" debug_logger.debug(msg) # ------------------------------------------------------------------ # Backup # ------------------------------------------------------------------ def _backup_config(self) -> None: """ Buat backup config lama sebelum ditimpa. Throttled: minimal MIN_CONFIG_BACKUP_INTERVAL_SEC detik antar backup. """ if not os.path.exists(self.config_file): return now = time.time() if now - self._last_backup_time < MIN_CONFIG_BACKUP_INTERVAL_SEC: return try: os.makedirs(CONFIG_BACKUP_DIR, exist_ok=True) ts = time.strftime("%Y%m%d_%H%M%S") # Tambahkan mikrodetik agar tidak tabrakan. backup_path = os.path.join( CONFIG_BACKUP_DIR, f"config_backup_{ts}_{int((now % 1) * 1000):03d}.json", ) shutil.copy2(self.config_file, backup_path) self._last_backup_time = now self._prune_backups(CONFIG_BACKUP_DIR) except (OSError, shutil.Error) as e: error_logger.error(f"Config backup failed: {e}") def _prune_backups(self, backup_dir: str) -> None: """Simpan hanya MAX_CONFIG_BACKUPS backup terbaru.""" try: entries = [ f for f in os.listdir(backup_dir) if f.startswith("config_backup_") and f.endswith(".json") ] if len(entries) <= MAX_CONFIG_BACKUPS: return entries.sort( key=lambda f: os.path.getmtime(os.path.join(backup_dir, f)) ) to_remove = entries[:-MAX_CONFIG_BACKUPS] for name in to_remove: try: os.remove(os.path.join(backup_dir, name)) except OSError: pass except OSError: pass # ------------------------------------------------------------------ # Save # ------------------------------------------------------------------ def save_config(self, config: Dict = None) -> None: if config is None: config = self.config # Pastikan field versi selalu ada. if isinstance(config, dict) and "config_version" not in config: config["config_version"] = CONFIG_VERSION self._backup_config() temp_path = f"{self.config_file}.tmp" try: with open(temp_path, "w", encoding="utf-8") as f: json.dump(config, f, indent=2, ensure_ascii=False) os.replace(temp_path, self.config_file) except Exception as e: error_logger.error(f"Config save error: {e}") raise # ------------------------------------------------------------------ # Existing API (backward-compatible) # ------------------------------------------------------------------ def get_current_model(self) -> Optional[str]: return self.config.get("current_model") def set_current_model(self, model_id: str) -> None: self.config["current_model"] = model_id self.save_config() def get_model_info(self, model_id: str) -> Optional[Dict]: return self.config["models"].get(model_id) def list_models(self) -> Dict: return self.config["models"] def add_model(self, model_id: str, metadata: ModelMetadata) -> None: self.config["models"][model_id] = metadata.to_dict() self.save_config() def delete_model(self, model_id: str) -> bool: if model_id in self.config["models"]: del self.config["models"][model_id] if self.config["current_model"] == model_id: self.config["current_model"] = None self.save_config() model_path = os.path.join(MODEL_DIR, model_id) if os.path.exists(model_path): try: shutil.rmtree(model_path) except OSError as e: error_logger.error(f"Gagal hapus folder model: {e}") return False return True return False def get_training_config(self) -> Dict: return self.config.get("training_config", {}) def update_training_config(self, **kwargs) -> None: for key, value in kwargs.items(): if key in self.config["training_config"]: self.config["training_config"][key] = value self.save_config() def get_generation_config(self) -> Dict: return self.config.get("generation_config", {}) def update_generation_config(self, **kwargs) -> None: for key, value in kwargs.items(): if key in self.config["generation_config"]: self.config["generation_config"][key] = value self.save_config() def is_multi_gpu(self) -> bool: return self.config.get("multi_gpu", False) # ------------------------------------------------------------------ # Presets # ------------------------------------------------------------------ def list_training_presets(self) -> List[str]: return sorted(TRAINING_PRESETS.keys()) def list_generation_presets(self) -> List[str]: return sorted(GENERATION_PRESETS.keys()) def apply_training_preset(self, preset_name: str) -> bool: """ Terapkan preset ke training_config. Hanya mengubah key yang sudah ada di training_config. """ preset = TRAINING_PRESETS.get(preset_name) if preset is None: console.print( Theme.error(f"Preset training tidak dikenal: {preset_name}") ) return False tc = self.config.get("training_config", {}) applied: List[str] = [] skipped: List[str] = [] for key, value in preset.items(): if key in tc: tc[key] = value applied.append(key) else: skipped.append(key) self.save_config() console.print( Theme.success(f"Preset training '{preset_name}' diterapkan " f"({len(applied)} parameter)") ) if skipped: console.print( Theme.dim(f" Dilewati (tidak ada di config): {skipped}") ) return True def apply_generation_preset(self, preset_name: str) -> bool: preset = GENERATION_PRESETS.get(preset_name) if preset is None: console.print( Theme.error(f"Preset generation tidak dikenal: {preset_name}") ) return False gc = self.config.get("generation_config", {}) applied: List[str] = [] skipped: List[str] = [] for key, value in preset.items(): if key in gc: gc[key] = value applied.append(key) else: skipped.append(key) self.save_config() console.print( Theme.success(f"Preset generation '{preset_name}' diterapkan " f"({len(applied)} parameter)") ) if skipped: console.print( Theme.dim(f" Dilewati (tidak ada di config): {skipped}") ) return True # ------------------------------------------------------------------ # Model health & statistics # ------------------------------------------------------------------ def _scan_dir(self, path: str) -> Tuple[int, int]: """Return (total_size_bytes, file_count) untuk folder.""" total = 0 count = 0 try: for root, _dirs, files in os.walk(path): for name in files: full = os.path.join(root, name) try: total += os.path.getsize(full) count += 1 except OSError: continue except OSError: pass return total, count def check_model_health(self, model_id: str) -> Dict[str, Any]: """ Cek apakah model masih valid. Return dict tanpa melempar exception. """ result: Dict[str, Any] = { "model_id": model_id, "path": None, "exists": False, "is_dir": False, "size_bytes": 0, "file_count": 0, "has_config": False, "has_tokenizer": False, "has_weights": False, "in_config": model_id in self.config.get("models", {}), "is_current": self.config.get("current_model") == model_id, "status": "missing", "issues": [], } path = os.path.join(MODEL_DIR, model_id) result["path"] = path if not os.path.exists(path): result["issues"].append("model directory tidak ditemukan") return result if not os.path.isdir(path): result["issues"].append("path ada tapi bukan directory") return result result["exists"] = True result["is_dir"] = True size, count = self._scan_dir(path) result["size_bytes"] = size result["file_count"] = count try: entries = os.listdir(path) except OSError as e: result["issues"].append(f"tidak bisa list directory: {e}") entries = [] lower_entries = {name.lower() for name in entries} result["has_config"] = "config.json" in lower_entries result["has_tokenizer"] = any( name in lower_entries for name in _TOKENIZER_FILE_NAMES ) result["has_weights"] = any( name in lower_entries for name in _WEIGHT_FILE_NAMES ) if count == 0: result["issues"].append("directory kosong") result["status"] = "empty" return result if not result["has_weights"]: result["issues"].append("file bobot model tidak ditemukan") if not result["has_tokenizer"]: result["issues"].append("file tokenizer tidak ditemukan") if not result["has_config"]: result["issues"].append("config.json tidak ditemukan") if result["issues"]: result["status"] = "warning" else: result["status"] = "ok" return result def get_model_stats(self, model_id: Optional[str] = None) -> Dict[str, Any]: """ Ambil statistik model. Jika model_id None, gunakan model aktif. """ if model_id is None: model_id = self.get_current_model() stats: Dict[str, Any] = { "model_id": model_id, "path": None, "exists": False, "total_size": 0, "total_size_human": "0 B", "file_count": 0, "is_current": False, "metadata": None, } if not model_id: stats["error"] = "Tidak ada model aktif" return stats path = os.path.join(MODEL_DIR, model_id) stats["path"] = path stats["exists"] = os.path.isdir(path) stats["is_current"] = self.config.get("current_model") == model_id stats["metadata"] = self.config.get("models", {}).get(model_id) if stats["exists"]: size, count = self._scan_dir(path) stats["total_size"] = size stats["total_size_human"] = self._human_size(size) stats["file_count"] = count return stats def get_all_models_stats(self) -> List[Dict[str, Any]]: """Statistik semua model yang terdaftar di config.""" return [ self.get_model_stats(mid) for mid in self.config.get("models", {}).keys() ] @staticmethod def _human_size(num_bytes: int) -> str: size = float(num_bytes) for unit in ("B", "KB", "MB", "GB", "TB"): if size < 1024: return f"{size:.2f} {unit}" size /= 1024 return f"{size:.2f} PB" # ------------------------------------------------------------------ # Model hash # ------------------------------------------------------------------ def compute_model_hash( self, model_id: str, chunk_size: int = 1024 * 1024, max_total_bytes: int = 0, ) -> Optional[str]: """ Hitung SHA256 dari isi folder model secara chunked (streaming). max_total_bytes = 0 berarti tanpa batas. Return None kalau tidak bisa dihitung. """ path = os.path.join(MODEL_DIR, model_id) if not os.path.isdir(path): return None if chunk_size <= 0: chunk_size = 1024 * 1024 hasher = hashlib.sha256() total_read = 0 try: rel_files: List[str] = [] for root, _dirs, files in os.walk(path): for name in files: full = os.path.join(root, name) rel = os.path.relpath(full, path).replace("\\", "/") rel_files.append(rel) rel_files.sort() for rel in rel_files: full = os.path.join(path, rel.replace("/", os.sep)) try: size = os.path.getsize(full) except OSError: continue # Selalu masukkan nama + ukuran untuk stabilitas. hasher.update(rel.encode("utf-8")) hasher.update(b"\0") hasher.update(str(size).encode("utf-8")) hasher.update(b"\0") if max_total_bytes > 0 and total_read + size > max_total_bytes: # Kalau melebihi batas, hanya hash sebagian. remaining = max_total_bytes - total_read if remaining <= 0: continue read_bytes = remaining else: read_bytes = size try: with open(full, "rb") as fh: remaining_to_read = read_bytes while remaining_to_read > 0: chunk = fh.read(min(chunk_size, remaining_to_read)) if not chunk: break hasher.update(chunk) remaining_to_read -= len(chunk) total_read += len(chunk) except OSError as e: error_logger.error(f"Hash read error ({rel}): {e}") continue hasher.update(b"\0") except OSError as e: error_logger.error(f"Hash scan error: {e}") return None return hasher.hexdigest() def update_model_hash(self, model_id: str) -> Optional[str]: """Hitung hash dan simpan ke metadata model.""" h = self.compute_model_hash(model_id) if h is None: return None if model_id in self.config.get("models", {}): self.config["models"][model_id]["hash"] = h self.save_config() return h # ------------------------------------------------------------------ # Clone / export # ------------------------------------------------------------------ def clone_model( self, source_id: str, dest_id: str, overwrite: bool = False ) -> bool: """ Clone model dari source_id ke dest_id di dalam MODEL_DIR. Tidak menghapus source. Metadata ikut disalin. """ if not source_id or not dest_id: console.print(Theme.error("ID tidak boleh kosong")) return False if source_id == dest_id: console.print(Theme.warning("Source dan destination sama")) return False src_path = os.path.join(MODEL_DIR, source_id) dst_path = os.path.join(MODEL_DIR, dest_id) if not os.path.isdir(src_path): console.print(Theme.error(f"Source tidak ada: {src_path}")) return False if os.path.exists(dst_path): if not overwrite: console.print( Theme.error(f"Destination sudah ada: {dst_path}") ) return False try: shutil.rmtree(dst_path) except OSError as e: error_logger.error(f"Gagal hapus destination: {e}") console.print(Theme.error(f"Gagal hapus destination: {e}")) return False try: os.makedirs(MODEL_DIR, exist_ok=True) shutil.copytree(src_path, dst_path) except (OSError, shutil.Error) as e: error_logger.error(f"Clone gagal: {e}") console.print(Theme.error(f"Clone gagal: {e}")) return False # Salin metadata kalau ada. src_meta = self.config.get("models", {}).get(source_id) if isinstance(src_meta, dict): try: new_meta = json.loads(json.dumps(src_meta)) except (TypeError, ValueError): new_meta = dict(src_meta) new_meta["model_name"] = dest_id new_meta["created_date"] = time.strftime("%Y-%m-%d %H:%M:%S") tags = list(new_meta.get("tags", []) or []) if "cloned" not in tags: tags.append("cloned") new_meta["tags"] = tags self.config["models"][dest_id] = new_meta self.save_config() console.print( Theme.success(f"Model di-clone: {source_id} -> {dest_id}") ) return True def export_model_metadata( self, model_id: str, dest_path: Optional[str] = None ) -> Optional[str]: """ Export metadata model ke file JSON. Kalau dest_path None, gunakan folder EXPORT_DIR. Return path file yang ditulis, atau None kalau gagal. """ meta = self.config.get("models", {}).get(model_id) if meta is None: console.print(Theme.error(f"Metadata tidak ada untuk: {model_id}")) return None if dest_path is None: try: os.makedirs(EXPORT_DIR, exist_ok=True) except OSError as e: error_logger.error(f"Gagal buat EXPORT_DIR: {e}") return None dest_path = os.path.join(EXPORT_DIR, f"{model_id}_metadata.json") payload = { "exported_at": time.strftime("%Y-%m-%d %H:%M:%S"), "config_version": self.config.get("config_version", CONFIG_VERSION), "model_id": model_id, "is_current": self.config.get("current_model") == model_id, "metadata": meta, } try: with open(dest_path, "w", encoding="utf-8") as f: json.dump(payload, f, indent=2, ensure_ascii=False) except OSError as e: error_logger.error(f"Export metadata gagal: {e}") console.print(Theme.error(f"Export metadata gagal: {e}")) return None console.print(Theme.success(f"Metadata diexport: {dest_path}")) return dest_path def export_model_files( self, model_id: str, dest_dir: str, overwrite: bool = False ) -> bool: """ Copy seluruh folder model ke dest_dir (di luar MODEL_DIR). Tidak menghapus source. """ src_path = os.path.join(MODEL_DIR, model_id) if not os.path.isdir(src_path): console.print(Theme.error(f"Model tidak ada: {src_path}")) return False try: abs_src = os.path.abspath(src_path) abs_dst = os.path.abspath(dest_dir) if abs_dst == abs_src or abs_dst.startswith(abs_src + os.sep): console.print( Theme.error("Destination tidak boleh di dalam source") ) return False except OSError: pass if os.path.exists(dest_dir): if not overwrite: console.print( Theme.error(f"Destination sudah ada: {dest_dir}") ) return False try: shutil.rmtree(dest_dir) except OSError as e: error_logger.error(f"Gagal hapus destination: {e}") return False try: parent = os.path.dirname(dest_dir) if parent: os.makedirs(parent, exist_ok=True) shutil.copytree(src_path, dest_dir) except (OSError, shutil.Error) as e: error_logger.error(f"Export model gagal: {e}") console.print(Theme.error(f"Export model gagal: {e}")) return False console.print(Theme.success(f"Model diexport ke: {dest_dir}")) return True