Fix_2.0

Author: anilkulkarni87

src/digdaggraph/cli.py

@@ -9,15 +9,42 @@
 logger = get_logger(__name__)
 
 
+def _label_for_schedule(schedule_obj) -> str:
+    """
+    Build a robust label for the schedule table row.
+    Prefer cron humanization when possible, but never fail the row.
+    """
+    try:
+        import yaml
+        label_core = yaml.safe_dump(schedule_obj).strip()
+    except Exception:
+        import json
+
+        label_core = f"{schedule_obj!r}" if isinstance(schedule_obj, str) else json.dumps(schedule_obj)
+
+    # Try to humanize cron, but never make this fatal
+    try:
+        if isinstance(schedule_obj, dict) and "cron>" in schedule_obj:
+            from cron_descriptor import get_description
+
+            return f"{label_core}\n{get_description(schedule_obj['cron>'])}"
+    except Exception as e:
+        logger.warning(f"cron description failed: {e}")
+
+    return f"schedule\n{label_core}"
+
+
 def main() -> None:
     start_time = time.time()
     count = 0
     Path(GRAPHS_DIR).mkdir(exist_ok=True)
     schedule_entries: list[ScheduleEntry] = []
 
-    for path in Path(os.getcwd()).rglob("*.dig"):
-        if GRAPHS_DIR in str(path):
-            continue
+    # Discover .dig files
+    dig_files = [p for p in Path(os.getcwd()).rglob("*.dig") if GRAPHS_DIR not in str(p)]
+    logger.info(f"Found {len(dig_files)} .dig files")
+
+    for path in dig_files:
         input_file_path = path
         out_dir = Path(os.getcwd()) / GRAPHS_DIR / path.parent.name
         out_dir.mkdir(parents=True, exist_ok=True)
@@ -26,41 +53,35 @@ def main() -> None:
         print(f"Generating graph for {input_file_path} → {output_dot_file}")
         try:
             generate_graph(input_filepath=str(input_file_path), output_dot_file=output_dot_file)
-            # collect schedule entries (best-effort lightweight parse)
-            try:
-                import yaml
-                import json
-                from cron_descriptor import get_description
-                from .yaml_includes import DigLoader, resolve_includes
-
-                with open(input_file_path, encoding="utf-8") as f:
-                    data_raw = yaml.load(f, Loader=DigLoader)
-                    data = resolve_includes(data_raw) or {}
-                    if "schedule" in data:
-                        if isinstance(data["schedule"], dict) and "cron>" in data["schedule"]:
-                            label = (
-                                f"{yaml.safe_dump(data['schedule']).strip()}"
-                                f"\n{get_description(data['schedule']['cron>'])}"
-                            )
-                        else:
-                            label = f"schedule\n{json.dumps(data['schedule'])}"
-                        schedule_entries.append(
-                            ScheduleEntry(
-                                project=path.parent.name,
-                                workflow=path.name,
-                                schedule_text=label,
-                                href=f"./{GRAPHS_DIR}/{path.parent.name}/{path.name.replace('.dig','.html')}",
-                            )
-                        )
-            except Exception:
-                # don't let schedule parsing failures break the whole run
-                pass
             count += 1
             logger.info(f"COMPLETE generating graph for {input_file_path}")
         except Exception as e:
             logger.error(f"FAILED generating graph for {input_file_path}: {e}", exc_info=True)
+            # continue on other files
+            continue
+
+        # Collect schedule entry (robust, never fatal)
+        try:
+            import yaml
+            from .yaml_includes import DigLoader, resolve_includes
+
+            with open(input_file_path, encoding="utf-8") as f:
+                data_raw = yaml.load(f, Loader=DigLoader)
+                data = resolve_includes(data_raw) or {}
+            if "schedule" in data:
+                label = _label_for_schedule(data["schedule"])
+                schedule_entries.append(
+                    ScheduleEntry(
+                        project=path.parent.name,
+                        workflow=path.name,
+                        schedule_text=label,
+                        href=f"./{GRAPHS_DIR}/{path.parent.name}/{path.name.replace('.dig','.html')}",
+                    )
+                )
+        except Exception as e:
+            logger.warning(f"Schedule collection failed for {input_file_path}: {e}")
 
-    # Always write the index, even if there were zero .dig files
+    # Always write the index
     write_scheduled_workflows(schedule_entries, out_path=SCHEDULE_INDEX_FILE)
 
     elapsed = time.time() - start_time