diff --git a/archive/scanner.py b/archive/scanner.py index ca91511..4d2f8a0 100644 --- a/archive/scanner.py +++ b/archive/scanner.py @@ -25,7 +25,7 @@ def __init__(self): } def scan_directory(self, root_path): - inventory = { + inventory: dict = { "meta": {"root": str(root_path), "total_files": 0, "algorithms_found": 0}, "files": [], } @@ -59,15 +59,13 @@ def analyze_file(self, file_path): # print(f"Error parsing {file_path}: {e}") # Reduce noise return None - file_data = {"path": str(file_path), "functions": []} + file_data: dict = {"path": str(file_path), "functions": []} # Find all function definitions (simplified traversal) # Note: Ideally this should be language-agnostic node type checking # For now relying on tree-sitter standard naming common in supported langs - cursor = tree.walk() - def traverse(node): - if node.type == "function_definition" or node.type == "method_declaration": + if node.type in {"function_definition", "method_declaration"}: self._analyze_function_node(node, code_bytes, file_data) for child in node.children: @@ -80,7 +78,7 @@ def traverse(node): def _analyze_function_node(self, node, code_bytes, file_data): try: func_name = node.child_by_field_name("name").text.decode("utf-8") - except: + except Exception: func_name = "anonymous" function_entry = {"name": func_name, "line": node.start_point[0] + 1, "algorithms": []} diff --git a/dissect/cli.py b/dissect/cli.py index cf132d0..d1dea4a 100644 --- a/dissect/cli.py +++ b/dissect/cli.py @@ -14,6 +14,8 @@ from .exporters.mermaid import save_mermaid from .trace_receiver import parse_trace_file +TRACE_FILE_HELP = "Path to trace file (JSON)" + def explain_command(args): """Generate AI summary of the trace.""" @@ -138,12 +140,12 @@ def main(): # Trace command trace_parser = subparsers.add_parser("trace", help="Parse and inspect a trace file") - trace_parser.add_argument("--file", "-f", required=True, help="Path to trace file (JSON)") + trace_parser.add_argument("--file", "-f", required=True, help=TRACE_FILE_HELP) trace_parser.set_defaults(func=trace_command) # Visualize command viz_parser = subparsers.add_parser("visualize", help="Generate visualization from trace") - viz_parser.add_argument("--file", "-f", required=True, help="Path to trace file (JSON)") + viz_parser.add_argument("--file", "-f", required=True, help=TRACE_FILE_HELP) viz_parser.add_argument( "--format", "-t", @@ -160,7 +162,7 @@ def main(): explain_parser = subparsers.add_parser( "explain", help="Generate AI-powered insights from trace" ) - explain_parser.add_argument("--file", "-f", required=True, help="Path to trace file (JSON)") + explain_parser.add_argument("--file", "-f", required=True, help=TRACE_FILE_HELP) explain_parser.add_argument( "--provider", "-p", diff --git a/dissect/graph.py b/dissect/graph.py index ea2539c..28174aa 100644 --- a/dissect/graph.py +++ b/dissect/graph.py @@ -127,10 +127,11 @@ def dfs(node_id: str, visited: set) -> tuple: return (node_duration + max_child_duration, [node] + max_child_path) - roots = self.get_root_nodes() - if not roots: - return [] + return self._best_root_path(self.get_root_nodes(), dfs) + @staticmethod + def _best_root_path(roots, dfs): + """Racine au plus long chemin (extrait pour complexite).""" max_duration = 0 critical_path = [] for root in roots: @@ -138,7 +139,6 @@ def dfs(node_id: str, visited: set) -> tuple: if duration > max_duration: max_duration = duration critical_path = path - return critical_path def to_dict(self) -> Dict[str, Any]: diff --git a/dissect/trace_receiver.py b/dissect/trace_receiver.py index b696726..f3bbd7e 100644 --- a/dissect/trace_receiver.py +++ b/dissect/trace_receiver.py @@ -6,10 +6,18 @@ """ import json +from datetime import datetime from typing import Any, Dict, List, Optional from .graph import Edge, Node, NodeType, OrchestrationGraph +UTC_SUFFIX = "+00:00" + + +def _parse_ts(ts): + """ISO-8601 tolerant Z (meme regle aux 3 sites d appel).""" + return datetime.fromisoformat(ts.replace("Z", UTC_SUFFIX)) + class TraceParser: """ @@ -210,11 +218,8 @@ def _parse_timestamp(self, ts: Any) -> Optional[float]: return ts if isinstance(ts, str): # Try ISO format - from datetime import datetime - try: - dt = datetime.fromisoformat(ts.replace("Z", "+00:00")) - return dt.timestamp() + return _parse_ts(ts).timestamp() except Exception: return None return None @@ -352,11 +357,8 @@ def _parse_timestamp(self, ts: Any) -> Optional[float]: return ts / 1000 return ts if isinstance(ts, str): - from datetime import datetime - try: - dt = datetime.fromisoformat(ts.replace("Z", "+00:00")) - return dt.timestamp() + return _parse_ts(ts).timestamp() except Exception: return None return None @@ -454,11 +456,8 @@ def _parse_timestamp(self, ts: Any) -> Optional[float]: return ts / 1000 return ts if isinstance(ts, str): - from datetime import datetime - try: - dt = datetime.fromisoformat(ts.replace("Z", "+00:00")) - return dt.timestamp() + return _parse_ts(ts).timestamp() except Exception: return None return None