"""Analyzer for file path patterns (local_filepath and s3_filepath).""" from collections import Counter, defaultdict from typing import List, Dict, Any import re import logging from .base_analyzer import BaseAnalyzer logger = logging.getLogger(__name__) class PathsAnalyzer(BaseAnalyzer): """Analyze file path patterns and anomalies.""" def __init__(self, output_dir): """Initialize PathsAnalyzer. Args: output_dir: Output directory for results """ super().__init__("paths", output_dir) def extract_extension(self, filepath: str, max_parts: int = 2) -> str: """Extract up to max_parts extension components from filepath. Args: filepath: Full file path or filename max_parts: Maximum number of extension parts to extract (default 2) Returns: Extension string (e.g., "mp3", "mp3.mp3", "tar.gz", or "no_ext") Examples: "file.mp3.mp3" -> "mp3.mp3" "file.tar.gz" -> "tar.gz" "file.mp3" -> "mp3" "file" -> "no_ext" """ filename = filepath.split("/")[-1] if "." not in filename: return "no_ext" parts = filename.split(".") # Get filename without extension (everything before first dot) # and extension parts (everything after first dot) if len(parts) < 2: return "no_ext" # Get last max_parts extension components ext_parts = parts[-max_parts:] if len(parts) > max_parts else parts[1:] return ".".join(ext_parts) def extract_s3_pattern(self, s3_path: str) -> tuple: """Extract pattern from S3 path.""" # Pattern: s3://bucket/path/components/file.ext if not s3_path.startswith("s3://"): return ("invalid", None, None) parts = s3_path[5:].split("/") if len(parts) < 2: return ("invalid", None, None) bucket = parts[0] # Get main category (e.g., 'bundles/v0/discogs') if len(parts) > 3: category = "/".join(parts[1:4]) else: category = "/".join(parts[1:-1]) # Get file extension (now supporting multi-part extensions) ext = self.extract_extension(parts[-1]) return (bucket, category, ext) def process_chunk(self, records: List[Dict[str, Any]]) -> Dict[str, Any]: """Process a chunk of records for path analysis.""" local_dirs = Counter() s3_patterns = Counter() s3_buckets = Counter() s3_categories = Counter() s3_file_extensions = Counter() local_file_extensions = Counter() sources = Counter() audio_types = Counter() audio_type_local_ext_combos = Counter() records_with_local = 0 records_with_s3 = 0 anomalies = [] for record in records: # Analyze local_filepath if "local_filepath" in record: local_path = record["local_filepath"] if local_path: records_with_local += 1 # Extract directory dir_path = "/".join(local_path.split("/")[:-1]) local_dirs[dir_path] += 1 # Extract file extension local_ext = self.extract_extension(local_path) local_file_extensions[local_ext] += 1 # Check for anomalies (not in expected directory) if not local_path.startswith("/app2/suno/data/"): anomalies.append( {"type": "local_path", "id": record.get("id", "unknown"), "path": local_path} ) # Analyze s3_filepath if "s3_filepath" in record: s3_path = record["s3_filepath"] if s3_path: records_with_s3 += 1 bucket, category, ext = self.extract_s3_pattern(s3_path) s3_patterns[f"{bucket}/{category}"] += 1 s3_buckets[bucket] += 1 if category: s3_categories[category] += 1 s3_file_extensions[ext] += 1 # Track source field if "source" in record and record["source"]: sources[record["source"]] += 1 # Track audio_type field if "audio_type" in record and record["audio_type"]: audio_types[record["audio_type"]] += 1 # Track audio_type + local_file_extension combinations if "local_filepath" in record and record["local_filepath"]: local_ext = self.extract_extension(record["local_filepath"]) audio_type = record.get("audio_type", "unknown") if audio_type: combo_key = f"{audio_type}|{local_ext}" audio_type_local_ext_combos[combo_key] += 1 return { "local_dirs": dict(local_dirs), "s3_patterns": dict(s3_patterns), "s3_buckets": dict(s3_buckets), "s3_categories": dict(s3_categories), "s3_file_extensions": dict(s3_file_extensions), "local_file_extensions": dict(local_file_extensions), "sources": dict(sources), "audio_types": dict(audio_types), "audio_type_local_ext_combos": dict(audio_type_local_ext_combos), "records_with_local": records_with_local, "records_with_s3": records_with_s3, "anomalies": anomalies, "total_records": len(records), } def aggregate(self, chunk_results: List[Dict[str, Any]]) -> Dict[str, Any]: """Aggregate results from all chunks.""" from ..utils.stats_utils import aggregate_counters all_local_dirs = [] all_s3_patterns = [] all_s3_buckets = [] all_s3_categories = [] all_s3_extensions = [] all_local_extensions = [] all_sources = [] all_audio_types = [] all_audio_type_local_ext_combos = [] all_anomalies = [] total_with_local = 0 total_with_s3 = 0 total_records = 0 for chunk in chunk_results: all_local_dirs.append(chunk.get("local_dirs", {})) all_s3_patterns.append(chunk.get("s3_patterns", {})) all_s3_buckets.append(chunk.get("s3_buckets", {})) all_s3_categories.append(chunk.get("s3_categories", {})) all_s3_extensions.append(chunk.get("s3_file_extensions", {})) all_local_extensions.append(chunk.get("local_file_extensions", {})) all_sources.append(chunk.get("sources", {})) all_audio_types.append(chunk.get("audio_types", {})) all_audio_type_local_ext_combos.append(chunk.get("audio_type_local_ext_combos", {})) all_anomalies.extend(chunk.get("anomalies", [])) total_with_local += chunk.get("records_with_local", 0) total_with_s3 += chunk.get("records_with_s3", 0) total_records += chunk.get("total_records", 0) # Aggregate total_local_dirs = aggregate_counters(all_local_dirs) total_s3_patterns = aggregate_counters(all_s3_patterns) total_s3_buckets = aggregate_counters(all_s3_buckets) total_s3_categories = aggregate_counters(all_s3_categories) total_s3_extensions = aggregate_counters(all_s3_extensions) total_local_extensions = aggregate_counters(all_local_extensions) total_sources = aggregate_counters(all_sources) total_audio_types = aggregate_counters(all_audio_types) total_audio_type_local_ext_combos = aggregate_counters(all_audio_type_local_ext_combos) # Sort by frequency sorted_local = sorted(total_local_dirs.items(), key=lambda x: x[1], reverse=True) sorted_s3 = sorted(total_s3_patterns.items(), key=lambda x: x[1], reverse=True) return { "summary": { "total_records": total_records, "records_with_local_path": total_with_local, "records_with_s3_path": total_with_s3, "unique_local_dirs": len(total_local_dirs), "unique_s3_patterns": len(total_s3_patterns), "unique_s3_extensions": len(total_s3_extensions), "unique_local_extensions": len(total_local_extensions), "unique_sources": len(total_sources), "unique_audio_types": len(total_audio_types), "unique_audio_types_local_extensions": len(total_audio_type_local_ext_combos), "total_anomalies": len(all_anomalies), }, "local_directories": sorted_local[:50], "s3_patterns": sorted_s3[:50], "s3_buckets": dict(sorted(total_s3_buckets.items(), key=lambda x: x[1], reverse=True)), "s3_categories": dict( sorted(total_s3_categories.items(), key=lambda x: x[1], reverse=True)[:20] ), "s3_file_extensions": dict( sorted(total_s3_extensions.items(), key=lambda x: x[1], reverse=True) ), "local_file_extensions": dict( sorted(total_local_extensions.items(), key=lambda x: x[1], reverse=True) ), "sources": dict(sorted(total_sources.items(), key=lambda x: x[1], reverse=True)), "audio_types": dict(sorted(total_audio_types.items(), key=lambda x: x[1], reverse=True)), "audio_type_local_ext_combos": dict( sorted(total_audio_type_local_ext_combos.items(), key=lambda x: x[1], reverse=True) ), "anomaly_sample": all_anomalies[:100], }