DJ-Goanna-Coding/oppo-node
0
1"""2Genesis Boiler - File Auditing and Archive Management System3 4This module provides functionality to audit files from configured source directories,5create JSON inventories, and generate compressed tar archives.6"""7 8import os9import json10import tarfile11import hashlib12from pathlib import Path13from datetime import datetime14from typing import List, Dict, Any15import yaml16 17 18class GenesisBoiler:19 """20 Audits files from configured source directories and creates compressed archives.21 22 This class handles:23 - File system traversal and auditing24 - JSON inventory creation25 - Gzip-compressed tar archive generation26 - File metadata collection (size, hash, modified time)27 """28 29 def __init__(self, config_path: str = "config.yaml"):30 """31 Initialize GenesisBoiler with configuration.32 33 Args:34 config_path: Path to the YAML configuration file35 """36 self.config_path = config_path37 self.config = self._load_config()38 self.inventory = []39 self.audit_enabled = self.config.get('audit', {}).get('enabled', True)40 self.source_dirs = self.config.get('audit', {}).get('source_directories', ['.'])41 self.exclude_patterns = self.config.get('audit', {}).get('exclude_from_audit', [])42 43 def _load_config(self) -> Dict[str, Any]:44 """Load configuration from YAML file."""45 try:46 with open(self.config_path, 'r') as f:47 return yaml.safe_load(f)48 except FileNotFoundError:49 print(f"Config file {self.config_path} not found, using defaults")50 return {}51 52 def _should_exclude(self, path: str) -> bool:53 """Check if a path should be excluded based on patterns."""54 for pattern in self.exclude_patterns:55 if pattern in path or Path(path).match(pattern):56 return True57 return False58 59 def _calculate_file_hash(self, file_path: str) -> str:60 """Calculate SHA256 hash of a file."""61 sha256_hash = hashlib.sha256()62 try:63 with open(file_path, "rb") as f:64 for byte_block in iter(lambda: f.read(4096), b""):65 sha256_hash.update(byte_block)66 return sha256_hash.hexdigest()67 except Exception as e:68 print(f"Error hashing {file_path}: {e}")69 return ""70 71 def audit_files(self) -> List[Dict[str, Any]]:72 """73 Audit files from configured source directories.74 75 Returns:76 List of dictionaries containing file metadata77 """78 self.inventory = []79 80 for source_dir in self.source_dirs:81 source_path = Path(source_dir).resolve()82 83 if not source_path.exists():84 print(f"Source directory {source_dir} does not exist, skipping")85 continue86 87 for root, dirs, files in os.walk(source_path):88 # Filter out excluded directories89 dirs[:] = [d for d in dirs if not self._should_exclude(os.path.join(root, d))]90 91 for file in files:92 file_path = os.path.join(root, file)93 94 if self._should_exclude(file_path):95 continue96 97 try:98 stat_info = os.stat(file_path)99 relative_path = os.path.relpath(file_path, source_path)100 101 file_info = {102 "path": relative_path,103 "full_path": file_path,104 "size": stat_info.st_size,105 "modified": datetime.fromtimestamp(stat_info.st_mtime).isoformat(),106 "hash": self._calculate_file_hash(file_path),107 "source_dir": source_dir108 }109 110 self.inventory.append(file_info)111 except Exception as e:112 print(f"Error processing {file_path}: {e}")113 114 return self.inventory115 116 def write_inventory(self, output_path: str = "inventory.json") -> str:117 """118 Write the file inventory to a JSON file.119 120 Args:121 output_path: Path where the JSON inventory will be written122 123 Returns:124 Path to the created inventory file125 """126 if not self.inventory:127 self.audit_files()128 129 inventory_data = {130 "timestamp": datetime.now().isoformat(),131 "total_files": len(self.inventory),132 "total_size": sum(f["size"] for f in self.inventory),133 "files": self.inventory134 }135 136 with open(output_path, 'w') as f:137 json.dump(inventory_data, f, indent=2)138 139 print(f"Inventory written to {output_path}")140 return output_path141 142 def create_archive(self, archive_path: str = None) -> str:143 """144 Create a gzip-compressed tar archive of the audited files.145 146 Args:147 archive_path: Path for the output archive (default: timestamped)148 149 Returns:150 Path to the created archive151 """152 if archive_path is None:153 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")154 archive_path = f"genesis_archive_{timestamp}.tar.gz"155 156 if not self.inventory:157 self.audit_files()158 159 with tarfile.open(archive_path, "w:gz") as tar:160 for file_info in self.inventory:161 try:162 tar.add(file_info["full_path"], arcname=file_info["path"])163 except Exception as e:164 print(f"Error adding {file_info['path']} to archive: {e}")165 166 print(f"Archive created at {archive_path}")167 return archive_path168 169 def run_full_audit(self, inventory_path: str = "inventory.json",170 archive_path: str = None) -> Dict[str, str]:171 """172 Run complete audit: scan files, write inventory, create archive.173 174 Args:175 inventory_path: Path for the JSON inventory176 archive_path: Path for the tar.gz archive177 178 Returns:179 Dictionary with paths to created files180 """181 print("Starting full audit...")182 183 # Audit files184 files = self.audit_files()185 print(f"Audited {len(files)} files")186 187 # Write inventory188 inv_path = self.write_inventory(inventory_path)189 190 # Create archive if enabled191 arch_path = None192 if self.config.get('audit', {}).get('create_archive', True):193 arch_path = self.create_archive(archive_path)194 195 return {196 "inventory": inv_path,197 "archive": arch_path,198 "file_count": len(files)199 }200 201 202if __name__ == "__main__":203 # Run audit when executed directly204 boiler = GenesisBoiler()205 results = boiler.run_full_audit()206 print(f"\nAudit complete:")207 print(f" Inventory: {results['inventory']}")208 print(f" Archive: {results['archive']}")209 print(f" Files processed: {results['file_count']}")210# DJ GOANNA CODING - GENESIS BOILER (SOVEREIGN EDITION)211# Purpose: Consolidating the 321GB Substrate for TIA-ARCHITECT-CORE212import os213import tarfile214import json215from datetime import datetime216 217class GenesisBoiler:218 def __init__(self):219 self.sources = [220 "./Research/GENESIS_VAULT/", # GDrive Partitions221 "/data/Mapping-and-Inventory-storage/", # HF Persistent222 "./pioneer-trader/vortex_cache/" # Internal Engine Logic223 ]224 self.output_bin = "/data/genesis_monolith.bin"225 self.inventory_path = "./INVENTORY.json"226 227 def audit_territory(self):228 """Map every file before consolidation (Visibility before Velocity)."""229 inventory = {"timestamp": str(datetime.now()), "files": []}230 for src in self.sources:231 if os.path.exists(src):232 try:233 for root, _, files in os.walk(src):234 for f in files:235 inventory["files"].append(os.path.join(root, f))236 except (OSError, PermissionError) as e:237 print(f"[T.I.A.] WARNING: Could not access {src}: {e}")238 239 try:240 with open(self.inventory_path, 'w') as f:241 json.dump(inventory, f, indent=4)242 print(f"[T.I.A.] TERRITORY AUDITED. {len(inventory['files'])} FILES MARKED.")243 except IOError as e:244 print(f"[T.I.A.] ERROR: Could not write inventory file: {e}")245 raise246 247 def boil_and_weld(self):248 """Consolidate sources into the Monolith."""249 print("[T.I.A.] INITIALIZING BOILER... COMPRESSING SUBSTRATE.")250 251 output_dir = os.path.dirname(self.output_bin)252 if output_dir and not os.path.exists(output_dir):253 try:254 os.makedirs(output_dir, exist_ok=True)255 except OSError as e:256 print(f"[T.I.A.] ERROR: Could not create output directory {output_dir}: {e}")257 raise258 259 try:260 with tarfile.open(self.output_bin, "w:gz") as tar:261 for src in self.sources:262 if os.path.exists(src):263 try:264 tar.add(src, arcname=os.path.basename(src))265 except (OSError, PermissionError) as e:266 print(f"[T.I.A.] WARNING: Could not add {src} to archive: {e}")267 print(f"[T.I.A.] BOILER COMPLETE: {self.output_bin} IS READY.")268 except (IOError, tarfile.TarError) as e:269 print(f"[T.I.A.] ERROR: Could not create tarball: {e}")270 raise271 272# FIELD EXECUTION273if __name__ == "__main__":274 boiler = GenesisBoiler()275 boiler.audit_territory()276 boiler.boil_and_weld()277