PLOG: JSON to Human Readable Format converter

plog

#!/usr/bin/env python3
"""
* Save it as `/usr/local/bin/plog`
* `sudo chmod +x /usr/local/bin/plog`

and then in a new terminal:
`docker-compose logs -f <service> | plog`

you'll see your logs in humanized format.
"""

import json
import sys
from datetime import datetime
import re

def parse_timestamp(ts):
    """Parse ISO timestamp to readable format"""
    try:
        dt = datetime.fromisoformat(ts.replace('Z', '+00:00'))
        return dt.strftime('%H:%M:%S')
    except:
        return ts

def colorize(text, color_code):
    """Add ANSI color codes"""
    return f"\033[{color_code}m{text}\033[0m"

def get_level_color(level):
    """Get color based on log level"""
    colors = {
        'info': '36',    # cyan
        'warn': '33',    # yellow
        'warning': '33', # yellow
        'error': '31',   # red
        'debug': '35',   # magenta
    }
    return colors.get(level.lower(), '37')  # default white

def format_log_line(line):
    """Format a single log line"""
    try:
        # Extract service name from docker-compose output
        service_match = re.match(r'^([a-zA-Z0-9_-]+)\s+\|\s+(.*)$', line.strip())
        if service_match:
            service_name = service_match.group(1)
            json_part = service_match.group(2)
        else:
            service_name = "unknown"
            json_part = line.strip()
        
        # Parse JSON
        log_data = json.loads(json_part)
        
        # Extract key fields
        timestamp = parse_timestamp(log_data.get('ts', ''))
        level = log_data.get('level', 'info')
        msg = log_data.get('msg', '')
        func = log_data.get('func', '')
        caller = log_data.get('caller', '')
        
        # Color the level
        colored_level = colorize(level.upper().ljust(5), get_level_color(level))
        
        # Format function name
        func_part = f" [{func}]" if func else ""
        
        # Format caller (just filename:line)
        caller_part = ""
        if caller:
            caller_match = re.search(r'/([^/]+\.py:\d+)$', caller)
            if caller_match:
                caller_part = f" ({caller_match.group(1)})"
        
        # Build the formatted line
        formatted = f"{colorize(timestamp, '90')} {colored_level} {colorize(service_name, '94')}{func_part}{caller_part}: {msg}"
        
        # Add extra fields if they exist
        extra_fields = []
        for key, value in log_data.items():
            if key not in ['ts', 'level', 'msg', 'func', 'caller', 'stacktrace']:
                if key in ['run_id', 'org_id']:
                    # Truncate long IDs
                    value_str = str(value)[:8] + "..." if len(str(value)) > 8 else str(value)
                else:
                    value_str = str(value)
                extra_fields.append(f"{key}={value_str}")
        
        if extra_fields:
            formatted += f" {colorize('|', '90')} {colorize(' '.join(extra_fields), '90')}"
        
        return formatted
        
    except json.JSONDecodeError:
        # Not JSON, return as is
        return line.strip()
    except Exception as e:
        # Fallback for any other error
        return f"[PARSE ERROR] {line.strip()}"

def main():
    """Main function to process stdin"""
    try:
        for line in sys.stdin:
            if line.strip():
                formatted = format_log_line(line)
                print(formatted)
                sys.stdout.flush()
    except KeyboardInterrupt:
        pass
    except BrokenPipeError:
        pass

if __name__ == "__main__":
    main()

Save it as /usr/local/bin/plog and then in a new terminal:

docker-compose logs -f <service> | plog