Update processor.py

processor.py

@@ -1,11 +1,11 @@
 import json
 import logging
 import datasets
+import math
 from datasets import load_dataset, get_dataset_config_names, get_dataset_infos
 from huggingface_hub import HfApi, DatasetCard, DatasetCardData
 import re
 
-# Configure logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
@@ -14,42 +14,28 @@ class DatasetCommandCenter:
         self.token = token
         self.api = HfApi(token=token)
 
-    # ==========================================
-    # 1. METADATA & INSPECTION
-    # ==========================================
+    # --- 1. INSPECTION ---
 
     def get_dataset_metadata(self, dataset_id):
         configs = ['default']
         splits = ['train', 'test', 'validation']
         license_name = "unknown"
-
         try:
-            # 1. Fetch Configs
             try:
-                found_configs = get_dataset_config_names(dataset_id, token=self.token)
-                if found_configs: configs = found_configs
+                c = get_dataset_config_names(dataset_id, token=self.token)
+                if c: configs = c
             except: pass
 
-            # 2. Fetch Metadata
             try:
-                selected = configs[0]
                 infos = get_dataset_infos(dataset_id, token=self.token)
-                info = None
-                if selected in infos: info = infos[selected]
-                elif 'default' in infos: info = infos['default']
-                elif infos: info = list(infos.values())[0]
-
+                sel = configs[0]
+                info = infos.get(sel) or infos.get('default') or (list(infos.values())[0] if infos else None)
                 if info:
                     splits = list(info.splits.keys())
                     license_name = info.license or "unknown"
             except: pass
-
-            return {
-                "status": "success", 
-                "configs": configs, 
-                "splits": splits,
-                "license_detected": license_name
-            }
+            
+            return {"status": "success", "configs": configs, "splits": splits, "license_detected": license_name}
         except Exception as e:
             return {"status": "error", "message": str(e)}
 
@@ -61,8 +47,22 @@ class DatasetCommandCenter:
         except:
             return {"status": "success", "splits": ['train', 'test', 'validation']}
 
+    def _sanitize_for_json(self, obj):
+        """Recursively cleans data for JSON serialization (Fixes NaN crash)."""
+        if isinstance(obj, float):
+            if math.isnan(obj) or math.isinf(obj):
+                return None
+            return obj
+        elif isinstance(obj, dict):
+            return {k: self._sanitize_for_json(v) for k, v in obj.items()}
+        elif isinstance(obj, list):
+            return [self._sanitize_for_json(v) for v in obj]
+        elif isinstance(obj, (str, int, bool, type(None))):
+            return obj
+        else:
+            return str(obj)
+
     def _flatten_object(self, obj, parent_key='', sep='.'):
-        """Recursively finds keys for the UI dropdowns."""
         items = {}
         if isinstance(obj, str):
             s = obj.strip()
@@ -73,7 +73,7 @@ class DatasetCommandCenter:
         if isinstance(obj, dict):
             for k, v in obj.items():
                 new_key = f"{parent_key}{sep}{k}" if parent_key else k
-                items.update(self._flatten_object(v, new_key, sep=sep))
+                items.update(self._flatten_object(v, new_key, sep))
         elif isinstance(obj, list):
             items[parent_key or "list"] = "List"
         else:
@@ -92,7 +92,7 @@ class DatasetCommandCenter:
             for i, row in enumerate(ds_stream):
                 if i >= 10: break
                 
-                # Clean row for UI (No objects)
+                # Sanitize entire row to prevent JSON crash on UI
                 clean_row = self._sanitize_for_json(row)
                 sample_rows.append(clean_row)
 
@@ -100,7 +100,7 @@ class DatasetCommandCenter:
                 flattened = self._flatten_object(row)
                 available_paths.update(flattened.keys())
 
-                # List Mode Detection
+                # List Detection
                 for k, v in row.items():
                     if k not in schema_map: schema_map[k] = {"type": "Object"}
                     val = v
@@ -126,26 +126,21 @@ class DatasetCommandCenter:
         except Exception as e:
             return {"status": "error", "message": str(e)}
 
-    # ==========================================
-    # 2. CORE LOGIC
-    # ==========================================
+    # --- 2. LOGIC ---
 
     def _get_value_by_path(self, obj, path):
         if not path: return obj
         keys = path.split('.')
         current = obj
-        
         for key in keys:
             if isinstance(current, str):
                 s = current.strip()
                 if (s.startswith('{') and s.endswith('}')) or (s.startswith('[') and s.endswith(']')):
                     try: current = json.loads(s)
-                    except: pass 
-
+                    except: pass
             if isinstance(current, dict) and key in current:
                 current = current[key]
-            else:
-                return None
+            else: return None
         return current
 
     def _extract_from_list_logic(self, row, source_col, filter_key, filter_val, target_path):
@@ -155,118 +150,84 @@ class DatasetCommandCenter:
             except: return None
         if not isinstance(data, list): return None
 
-        matched_item = None
+        matched = None
         for item in data:
             if str(item.get(filter_key, '')) == str(filter_val):
-                matched_item = item
+                matched = item
                 break
-        
-        if matched_item:
-            return self._get_value_by_path(matched_item, target_path)
+        if matched: return self._get_value_by_path(matched, target_path)
         return None
 
     def _apply_projection(self, row, recipe):
         new_row = {}
+        # Context
         eval_context = row.copy()
         eval_context['row'] = row
         eval_context['json'] = json
         eval_context['re'] = re
         
-        for col_def in recipe['columns']:
-            t_type = col_def.get('type', 'simple')
-            target_col = col_def['name']
-            
+        for col in recipe['columns']:
             try:
-                if t_type == 'simple':
-                    new_row[target_col] = self._get_value_by_path(row, col_def['source'])
-                elif t_type == 'list_search':
-                    new_row[target_col] = self._extract_from_list_logic(
-                        row, col_def['source'], col_def['filter_key'], col_def['filter_val'], col_def['target_key']
-                    )
-                elif t_type == 'python':
-                    val = eval(col_def['expression'], {}, eval_context)
-                    new_row[target_col] = val
+                c_type = col.get('type', 'simple')
+                name = col['name']
+                if c_type == 'simple':
+                    new_row[name] = self._get_value_by_path(row, col['source'])
+                elif c_type == 'list_search':
+                    new_row[name] = self._extract_from_list_logic(row, col['source'], col['filter_key'], col['filter_val'], col['target_key'])
+                elif c_type == 'python':
+                    new_row[name] = eval(col['expression'], {}, eval_context)
             except Exception as e:
-                raise ValueError(f"Column '{target_col}' failed: {str(e)}")
-
+                raise ValueError(f"Column '{col['name']}' error: {e}")
         return new_row
 
-    def _sanitize_for_json(self, obj):
-        """Helper to ensure objects are JSON serializable (fixes Preview crash)."""
-        if isinstance(obj, dict):
-            return {k: self._sanitize_for_json(v) for k, v in obj.items()}
-        elif isinstance(obj, list):
-            return [self._sanitize_for_json(v) for v in obj]
-        elif isinstance(obj, (str, int, float, bool, type(None))):
-            return obj
-        else:
-            return str(obj) # Convert Timestamps, Images, etc to string
-
-    # ==========================================
-    # 3. PREVIEW & EXECUTE
-    # ==========================================
+    # --- 3. PREVIEW & PUSH ---
 
     def preview_transform(self, dataset_id, config, split, recipe):
         conf = config if config != 'default' else None
-        
         try:
-            ds_stream = load_dataset(dataset_id, name=conf, split=split, streaming=True, token=self.token)
-            processed = []
-            
-            for i, row in enumerate(ds_stream):
-                if len(processed) >= 5: break
+            ds = load_dataset(dataset_id, name=conf, split=split, streaming=True, token=self.token)
+            out = []
+            for i, row in enumerate(ds):
+                if len(out) >= 5: break
                 
                 # Filter
-                passed = True
                 if recipe.get('filter_rule'):
                     try:
                         ctx = row.copy()
                         ctx['row'] = row
                         ctx['json'] = json
                         ctx['re'] = re
-                        if not eval(recipe['filter_rule'], {}, ctx): passed = False
-                    except: passed = False # Skip crashing rows in preview
+                        if not eval(recipe['filter_rule'], {}, ctx): continue
+                    except: continue # Skip crashing filters in preview
                 
-                if passed:
-                    try:
-                        projected = self._apply_projection(row, recipe)
-                        # SANITIZE OUTPUT so Flask doesn't crash on Timestamps/Images
-                        clean_projected = self._sanitize_for_json(projected)
-                        processed.append(clean_projected)
-                    except Exception as e:
-                        processed.append({"_preview_error": f"Error: {str(e)}"})
-                        
-            return processed
+                try:
+                    # Apply & Sanitize
+                    proj = self._apply_projection(row, recipe)
+                    out.append(self._sanitize_for_json(proj))
+                except Exception as e:
+                    out.append({"_preview_error": str(e)})
+            return out
         except Exception as e:
             raise e
 
     def _generate_card(self, source_id, target_id, recipe, license_name):
-        card_data = DatasetCardData(
-            language="en",
-            license=license_name,
-            tags=["dataset-command-center", "etl"],
-            base_model=source_id,
-        )
-        content = f"""
-# {target_id.split('/')[-1]}
-This dataset is a transformation of [{source_id}](https://huggingface.co/datasets/{source_id}).
-## Recipe
-"""
-        for col in recipe['columns']:
-            content += f"- **{col['name']}**: {col.get('type')} ({col.get('source') or 'expr'})\n"
+        content = f"# {target_id}\nDerived from [{source_id}](https://huggingface.co/datasets/{source_id}).\n\n## Recipe\n"
+        for c in recipe['columns']:
+            content += f"- **{c['name']}**: {c.get('type')} ({c.get('source') or c.get('expression')})\n"
         content += f"\n**License:** {license_name}"
-        return DatasetCard.from_template(card_data, content=content)
+        return DatasetCard.from_template(DatasetCardData(license=license_name, tags=["etl"]), content=content)
 
     def process_and_push(self, source_id, config, split, target_id, recipe, max_rows=None, new_license=None):
-        logger.info(f"Job: {source_id} -> {target_id}")
+        logger.info(f"Pushing {source_id} -> {target_id}")
         conf = config if config != 'default' else None
         
         def gen():
-            ds_stream = load_dataset(source_id, name=conf, split=split, streaming=True, token=self.token)
+            ds = load_dataset(source_id, name=conf, split=split, streaming=True, token=self.token)
             count = 0
-            for i, row in enumerate(ds_stream):
+            for i, row in enumerate(ds):
                 if max_rows and count >= int(max_rows): break
-
+                
+                # Filter
                 if recipe.get('filter_rule'):
                     try:
                         ctx = row.copy()
@@ -274,23 +235,21 @@ This dataset is a transformation of [{source_id}](https://huggingface.co/dataset
                         ctx['json'] = json
                         ctx['re'] = re
                         if not eval(recipe['filter_rule'], {}, ctx): continue
-                    except Exception as e:
-                        raise ValueError(f"Filter error row {i}: {e}")
+                    except Exception as e: raise ValueError(f"Filter error row {i}: {e}")
 
+                # Project
                 try:
                     yield self._apply_projection(row, recipe)
                     count += 1
-                except ValueError as ve: raise ve
-                except Exception as e: raise ValueError(f"Error row {i}: {e}")
+                except Exception as e: raise ValueError(f"Row {i} error: {e}")
 
         try:
-            new_dataset = datasets.Dataset.from_generator(gen)
-            new_dataset.push_to_hub(target_id, token=self.token)
+            new_ds = datasets.Dataset.from_generator(gen)
+            new_ds.push_to_hub(target_id, token=self.token)
             try:
                 card = self._generate_card(source_id, target_id, recipe, new_license or "unknown")
                 card.push_to_hub(target_id, token=self.token)
             except: pass
-            return {"status": "success", "rows_processed": len(new_dataset)}
+            return {"status": "success", "rows_processed": len(new_ds)}
         except Exception as e:
-            logger.error(f"Job Failed: {e}")
             return {"status": "failed", "error": str(e)}