update

Author: Tom Softreck

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "dialogchain"
-version = "0.1.15"
+version = "0.1.16"
 description = "DialogChain - A flexible and extensible dialog processing framework"
 authors = ["Tom Sapletta <info@softreck.dev>"]
 readme = "README.md"

src/dialogchain/cli.py

@@ -55,28 +55,46 @@ def run(config, env_file, route, dry_run, verbose):
     # Run routes
     try:
         if route:
-            # Find the specific route by name
             route_config = next((r for r in config_data.get('routes', []) if r.get('name') == route), None)
             if not route_config:
                 click.echo(f"❌ Route '{route}' not found in configuration", err=True)
                 return
 
+            if verbose:
+                click.echo(f"🚀 Starting route: {route}")
+                
             # Create source and destination
-            source = engine.create_source(route_config['from'])
-            destination = engine.create_destination(route_config['to'])
+            source = engine.create_source(route_config.get('from'))
+            destination = engine.create_destination(route_config.get('to'))
+            
+            if verbose:
+                click.echo(f"🔌 Source: {type(source).__name__}")
+                click.echo(f"🎯 Destination: {type(destination).__name__}")
 
             # Run the specific route
-            asyncio.run(engine.run_route(route_config, source, destination))
+            asyncio.run(engine.run_route_config(route_config, source, destination))
+            
+            if verbose:
+                click.echo(f"✅ Completed route: {route}")
+                
         else:
+            if verbose:
+                click.echo("🚀 Starting all routes...")
             # Run all routes
             asyncio.run(engine.run_all_routes())
+            if verbose:
+                click.echo("✅ All routes completed")
+                
     except KeyboardInterrupt:
-        click.echo("\n🛑 Shutting down...")
+        click.echo("\n🛑 Operation cancelled by user")
     except Exception as e:
-        click.echo(f"❌ Error: {e}", err=True)
+        click.echo(f"❌ Error running routes: {e}", err=True)
         if verbose:
             import traceback
             traceback.print_exc()
+        return 1
+        
+    return 0
 
 
 def update_env_file(env_path, required_vars):

src/dialogchain/connectors.py

@@ -102,14 +102,30 @@ def __init__(self, interval: str):
         self.interval = self._parse_interval(interval)
 
     async def receive(self) -> AsyncIterator[Dict[str, Any]]:
-        """Yield timer events"""
-        while True:
-            yield {
-                "type": "timer_event",
-                "timestamp": datetime.now().isoformat(),
-                "interval": self.interval,
-            }
-            await asyncio.sleep(self.interval)
+        """Yield timer events
+        
+        Yields:
+            Dictionary with timer event data
+        """
+        logger = setup_logger(__name__)
+        logger.info(f"⏱️  Timer source started with interval: {self.interval}s")
+        
+        try:
+            while True:
+                event = {
+                    "type": "timer_event",
+                    "timestamp": datetime.now().isoformat(),
+                    "interval": self.interval,
+                }
+                logger.debug(f"⏱️  Timer event: {event}")
+                yield event
+                await asyncio.sleep(self.interval)
+        except asyncio.CancelledError:
+            logger.info("⏱️  Timer source cancelled")
+            raise
+        except Exception as e:
+            logger.error(f"⏱️  Timer source error: {e}")
+            raise
 
     def _parse_interval(self, interval_str: str) -> float:
         """Parse interval string to seconds
@@ -749,52 +765,56 @@ async def send(self, message: Any) -> None:
 
 
 class LogDestination(Destination):
-    """Log destination for both console and file logging"""
+    """Log destination for both console and file logging
+    
+    URI format: log:info or log:debug or log:error
+    """
 
     def __init__(self, uri: str):
         parsed = urlparse(uri)
-        # Handle different URI formats:
-        # - log:relative/path
-        # - log:///absolute/path
-        # - log://hostname/absolute/path
-        # - log://relative/path
-        if parsed.scheme == 'log' and not parsed.netloc and not parsed.path.startswith('//'):
-            # Case: log:relative/path
-            self.log_file = parsed.path
-        elif parsed.scheme == 'log' and parsed.netloc and not parsed.path:
-            # Case: log://hostname (without path)
-            self.log_file = parsed.netloc
-        elif parsed.scheme == 'log' and (parsed.netloc or parsed.path):
-            # Case: log:///path or log://hostname/path
-            self.log_file = parsed.path.lstrip('/') or parsed.netloc
-        else:
-            self.log_file = None
+        # Extract log level from URI (e.g., 'log:info' -> 'info')
+        self.log_level = parsed.path.upper() if parsed.path else 'INFO'
+        if self.log_level not in ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']:
+            self.log_level = 'INFO'
+            
+        self.logger = setup_logger('dialogchain.destination.log')
+        self.logger.info(f" Log destination initialized with level: {self.log_level}")
 
     async def send(self, message: Any) -> None:
-        """Log message to console and optionally to a file"""
-        log_msg = f"📝 {datetime.now().isoformat()}: {message}"
-        
-        # Always print to console
-        print(log_msg)
+        """Log message with the specified log level
         
-        # Also log using the logger
-        logger.info(log_msg)
-
-        if self.log_file:
-            try:
-                # Ensure directory exists
-                log_dir = os.path.dirname(self.log_file)
-                if log_dir and not os.path.exists(log_dir):
-                    os.makedirs(log_dir, exist_ok=True)
-
-                # Make sure to use absolute path
-                abs_log_file = os.path.abspath(self.log_file)
-                with open(abs_log_file, "a", encoding="utf-8") as f:
-                    f.write(log_msg + "\n")
-            except Exception as e:
-                error_msg = f"❌ Log file error: {e}"
-                print(error_msg)
-                logger.error(error_msg)
+        Args:
+            message: The message to log. Can be a string or any JSON-serializable object.
+            
+        Raises:
+            Exception: If there's an error processing the message
+        """
+        try:
+            # Convert message to string if it's not already
+            if not isinstance(message, (str, bytes)):
+                try:
+                    message = json.dumps(message, indent=2, default=str)
+                except (TypeError, ValueError) as e:
+                    self.logger.warning(f"Could not convert message to JSON: {e}")
+                    message = str(message)
+
+            # Log with the appropriate level
+            if self.log_level == 'DEBUG':
+                self.logger.debug(message)
+            elif self.log_level == 'INFO':
+                self.logger.info(message)
+            elif self.log_level == 'WARNING':
+                self.logger.warning(message)
+            elif self.log_level == 'ERROR':
+                self.logger.error(message)
+            elif self.log_level == 'CRITICAL':
+                self.logger.critical(message)
+            else:
+                self.logger.info(message)  # Default to info
+                
+        except Exception as e:
+            self.logger.error(f"Error in LogDestination.send: {e}")
+            raise
 
 
 class GRPCDestination(Destination):

src/dialogchain/engine.py

@@ -62,21 +62,33 @@ def parse_uri(uri: str) -> Tuple[str, str]:
 
 class DialogChainEngine:
     def __init__(self, config: Dict[str, Any], verbose: bool = False):
+        # Initialize logger first
+        self.logger = setup_logger(__name__)
+        self.logger.info("🔧 Initializing DialogChainEngine")
+        
         self.config = config
         self.verbose = verbose
         self.routes = config.get("routes", [])
         self.running_processes = {}
         self._is_running = False
         self._tasks = []
+        
+        self.logger.info(f"🔧 Loaded {len(self.routes)} routes from config")
+        if self.verbose:
+            self.logger.info(f"🔧 Verbose mode enabled")
 
         # Validate the configuration on initialization
+        self.logger.info("🔧 Validating configuration...")
         errors = self.validate_config()
         if errors:
             error_msg = "Invalid configuration:\n" + "\n".join(
                 f"- {error}" for error in errors
             )
+            self.logger.error(f"❌ Configuration validation failed: {error_msg}")
             raise ValueError(error_msg)
 
+        self.logger.info("✅ Engine initialized successfully")
+            
     @property
     def is_running(self) -> bool:
         """Return whether the engine is currently running."""
@@ -109,7 +121,7 @@ async def start(self):
                     await destination.connect()
 
                 # Create and store task
-                task = asyncio.create_task(self.run_route(route, source, destination))
+                task = asyncio.create_task(self.run_route_config(route, source, destination))
                 self._tasks.append(task)
 
             self.log("Engine started successfully")
@@ -176,7 +188,7 @@ async def run_all_routes(self):
 
                 # Create and start task for this route
                 task = asyncio.create_task(
-                    self.run_route(route_config, source, destination)
+                    self.run_route_config(route_config, source, destination)
                 )
                 tasks.append(task)
 
@@ -194,7 +206,7 @@ async def run_all_routes(self):
         self.log(f"Starting {len(tasks)} routes...")
         await asyncio.gather(*tasks, return_exceptions=True)
 
-    async def run_route(self, route: Dict[str, Any], source: Source, destination: Destination):
+    async def run_route_config(self, route: Dict[str, Any], source: Source, destination: Destination):
         """Run a specific route with the given source and destination
         
         Args:
@@ -206,8 +218,20 @@ async def run_route(self, route: Dict[str, Any], source: Source, destination: De
         It runs the route configuration with the provided source and destination.
         """
         route_name = route.get("name", "unnamed")
+        logger.info(f"🔧 Starting route: {route_name}")
+        logger.info(f"🔧 Route config: {route}")
+        logger.info(f"🔧 Source type: {type(source).__name__}")
+        logger.info(f"🔧 Destination type: {type(destination).__name__}")
         self.log(f"Starting route: {route_name}")
 
+        if not source:
+            logger.error(f"❌ No source provided for route: {route_name}")
+            return
+            
+        if not destination:
+            logger.error(f"❌ No destination provided for route: {route_name}")
+            return
+        
         try:
             # Connect to source if not already connected
             if hasattr(source, 'connect') and callable(source.connect) and not getattr(source, 'is_connected', False):
@@ -447,6 +471,49 @@ def create_destination(self, uri: str) -> Destination:
             return HTTPDestination(uri, **config)
         elif scheme == "file":
             return FileDestination(path, **config)
+        elif scheme == "log":
+            # Handle log destination (e.g., log:info, log:error)
+            log_level = path.upper() if path else "INFO"
+            
+            # Create a simple destination that logs messages
+            class LogDestination:
+                def __init__(self, level):
+                    self.level = level
+                    self.sent_messages = []
+                    
+                async def send(self, message):
+                    log_message = f"[Log Destination] {message}"
+                    if self.level == "DEBUG":
+                        self.logger.debug(log_message)
+                    elif self.level == "INFO":
+                        self.logger.info(log_message)
+                    elif self.level == "WARNING":
+                        self.logger.warning(log_message)
+                    elif self.level == "ERROR":
+                        self.logger.error(log_message)
+                    elif self.level == "CRITICAL":
+                        self.logger.critical(log_message)
+                    else:
+                        self.logger.info(log_message)  # Default to info
+                    
+                    # Store the message for testing/verification
+                    self.sent_messages.append(message)
+                    return True
+                    
+                async def connect(self):
+                    pass
+                    
+                async def disconnect(self):
+                    pass
+                    
+                async def __aenter__(self):
+                    return self
+                    
+                async def __aexit__(self, exc_type, exc_val, exc_tb):
+                    pass
+            
+            return LogDestination(log_level)
+            
         elif scheme == "mock":
             # For testing purposes
             from unittest.mock import AsyncMock