interaminense · April 3, 2026 02:31
diff --git a/catalize.py b/catalize.py
 #!/usr/bin/env python3
 """Catalize — structured, colorized log pipe for Tomcat / Liferay DXP."""

 import re
 import sys
 import time
 import threading

 # ── ANSI ─────────────────────────────────────────────────────────────────────
 R   = '\033[0m'
 B   = '\033[1m'
 DIM = '\033[2m'
 IT  = '\033[3m'
 UL  = '\033[4m'

 RE  = '\033[31m';  BRE = '\033[91m'
 GR  = '\033[32m';  BGR = '\033[92m'
 YE  = '\033[33m';  BYE = '\033[93m'
 BL  = '\033[34m';  BBL = '\033[94m'
 MA  = '\033[35m';  BMA = '\033[95m'
 CY  = '\033[36m';  BCY = '\033[96m'
 WH  = '\033[37m';  BWH = '\033[97m'

 BG_RE = '\033[41m'

 CLEAR_LINE = '\033[2K\r'

 # ── Column widths ─────────────────────────────────────────────────────────────
 W_TIME   = 12
 W_LEVEL  = 7
 W_THREAD = 22
 W_CLASS  = 28

 # ── Level config ──────────────────────────────────────────────────────────────
 LEVELS = {
    'SEVERE':  ('☠', f'{BG_RE}{B}{BWH}', BRE, BRE),
    'FATAL':   ('☠', f'{BG_RE}{B}{BWH}', BRE, BRE),
    'ERROR':   ('✖', f'{BRE}{B}',        BRE, BRE),
    'WARNING': ('⚠', f'{BYE}{B}',        YE,  YE),
    'WARN':    ('⚠', f'{BYE}{B}',        YE,  YE),
    'INFO':    ('●', f'{BBL}',           '',  BBL),
    'CONFIG':  ('◆', f'{MA}',            '',  MA),
    'DEBUG':   ('○', f'{DIM}{CY}',       DIM, DIM),
    'FINE':    ('·', f'{DIM}',           DIM, DIM),
    'FINER':   ('·', f'{DIM}',           DIM, DIM),
    'FINEST':  ('·', f'{DIM}',           DIM, DIM),
 }
 DEFAULT_LEVEL = ('·', '', '', DIM)

 # ── Startup phase detection ───────────────────────────────────────────────────
 PHASES = [
    (re.compile(r'Server startup in',                  re.I), None),           # done
    (re.compile(r'Deploying web application',          re.I), 'Deploying web application...'),
    (re.compile(r'Starting ProtocolHandler',           re.I), 'Starting protocol handlers...'),
    (re.compile(r'OSGi framework started|Starting modules framework', re.I),
                                                              'OSGi framework starting...'),
    (re.compile(r'com\.liferay\.portal\.events',       re.I), 'Running Liferay startup events...'),
    (re.compile(r'com\.liferay',                       re.I), 'Loading Liferay bundles...'),
    (re.compile(r'Initializing Spring',                re.I), 'Initializing Spring context...'),
    (re.compile(r'Connector \[',                       re.I), 'Configuring connectors...'),
    (re.compile(r'Catalina',                           re.I), 'Initializing Catalina...'),
 ]

 # ── Log line regex ────────────────────────────────────────────────────────────
 LOG_RE = re.compile(
    r'^(\d{2}-\w{3}-\d{4})\s+'
    r'(\d{2}:\d{2}:\d{2}\.\d+)\s+'
    r'(\w+)\s+'
    r'(\[\S+?\])\s+'
    r'([\w.$]+(?:\.[\w$<>]+)*)\s*'
    r'(.*)'
 )

 SUCCESS_KW = ('Server startup in', 'Successfully deployed', 'Deployment of web application')
 LIFERAY_KW = ('Welcome to Liferay', 'Liferay DXP', 'OSGi framework started', 'Starting modules framework')

 # ── Inline highlights ─────────────────────────────────────────────────────────
 INLINE = [
    (re.compile(r'(\d[\d.,]*\s*(?:ms|millis(?:econds?)?|seconds?|mins?|minutes?|hours?)\b)', re.I), BGR),
    (re.compile(r'(\d[\d.,]*\s*(?:KB|MB|GB|TB|bytes?)\b)',                                   re.I), BMA),
    (re.compile(r'\b(\d{1,3}(?:\.\d{1,3}){3})\b'),                                                 BCY),
    (re.compile(r'(:\d{2,5})(?=/|\s|$)'),                                                          BYE),
    (re.compile(r'(https?://\S+)'),                                                        UL + BCY),
    (re.compile(r'(?<![:\w])(/[\w./\-]+)'),                                                UL + DIM),
    (re.compile(r'("(?:[^"\\]|\\.)*")'),                                                           BWH),
    (re.compile(r'(?<!\w)(\d+)(?!\w)'),                                                            YE),
 ]

 def colorize_msg(msg: str) -> str:
    for pattern, color in INLINE:
        msg = pattern.sub(lambda m, c=color: f'{c}{m.group(1)}{R}', msg)
    return msg

 def abbrev_thread(thread: str) -> str:
    inner = thread.strip('[]')
    if len(inner) <= W_THREAD - 2:
        return thread
    parts = inner.rsplit('-', 2)
    tail = '-'.join(parts[-2:]) if len(parts) >= 2 else inner[-(W_THREAD - 4):]
    return f'[…{tail}]'

 def abbrev_class(cls: str) -> str:
    if len(cls) <= W_CLASS:
        return cls
    parts = cls.split('.')
    short = '.'.join(parts[-3:]) if len(parts) >= 3 else cls
    return f'…{short}' if len(short) < len(cls) else cls[-W_CLASS:]

 def fit(s: str, width: int) -> str:
    visible = re.sub(r'\033\[[0-9;]*m', '', s)
    diff = width - len(visible)
    return s + ' ' * diff if diff >= 0 else visible[:width]

 def box(msg: str, color: str) -> str:
    inner = f'  {msg}  '
    w = len(re.sub(r'\033\[[0-9;]*m', '', inner)) + 2
    return (
        f'\n{color}╔{"═" * w}╗{R}\n'
        f'{color}║{R}{B}{inner}{R}{color}║{R}\n'
        f'{color}╚{"═" * w}╝{R}\n'
    )

 # ── Spinner ───────────────────────────────────────────────────────────────────
 class Spinner:
    FRAMES  = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠣', '⠏']
    BAR_W   = 28
    BAR_CHR = ('━', '╺')

    def __init__(self):
        self.phase    = 'Initializing...'
        self.done     = False
        self.lock     = threading.Lock()
        self._start   = time.time()
        self._frame   = 0
        self._t       = threading.Thread(target=self._run, daemon=True)
        self._t.start()

    def _elapsed(self) -> str:
        s = int(time.time() - self._start)
        return f'{s // 60:02d}:{s % 60:02d}'

    def _bar(self) -> str:
        """Animated indeterminate bar that bounces."""
        w   = self.BAR_W
        pos = self._frame % (w * 2)
        if pos >= w:
            pos = w * 2 - pos
        filled = min(pos + 6, w)
        empty  = w - filled
        bar    = self.BAR_CHR[0] * filled + self.BAR_CHR[1] + self.BAR_CHR[1] * empty
        return f'{BCY}{bar[:w]}{R}'

    def _run(self):
        while not self.done:
            with self.lock:
                f   = self.FRAMES[self._frame % len(self.FRAMES)]
                el  = self._elapsed()
                ph  = self.phase[:45]
                bar = self._bar()
                line = (
                    f'{CLEAR_LINE}'
                    f'{BCY}{B}{f}{R} '
                    f'{B}Starting up{R}  '
                    f'{bar}  '
                    f'{DIM}{ph:<45}{R}  '
                    f'{DIM}[{el}]{R}'
                )
                sys.stdout.write(line)
                sys.stdout.flush()
                self._frame += 1
            time.sleep(0.08)

    def update(self, phase: str):
        with self.lock:
            self.phase = phase

    def clear_line(self):
        sys.stdout.write(CLEAR_LINE)
        sys.stdout.flush()

    def print_above(self, formatted: str):
        """Interrupt spinner to print a line, then resume."""
        with self.lock:
            sys.stdout.write(CLEAR_LINE)
            print(formatted)
            sys.stdout.flush()

    def stop(self):
        self.done = True
        self._t.join(timeout=0.5)
        sys.stdout.write(CLEAR_LINE)
        sys.stdout.flush()


 # ── Log formatter ─────────────────────────────────────────────────────────────
 _last_was_error = False

 def format_line(line: str) -> str:
    global _last_was_error

    if re.match(r'^\s+at ', line) or re.match(r'^\s+\.\.\. \d+ more', line):
        _last_was_error = False
        return f'  {DIM}{RE}│{R}  {DIM}{RE}{line.strip()}{R}'

    if re.match(r'^\s*Caused by:', line) or re.match(r'^[\w.$]+(?:Exception|Error)', line):
        _last_was_error = False
        return f'  {BRE}│{R}  {BRE}{B}{line.strip()}{R}'

    m = LOG_RE.match(line)
    if m:
        _, time_s, level, thread, cls, msg = m.groups()

        icon, badge, msg_color, bar_color = LEVELS.get(level, DEFAULT_LEVEL)
        is_error   = level in ('SEVERE', 'FATAL', 'ERROR')
        is_warn    = level in ('WARN', 'WARNING')
        is_success = any(k in msg for k in SUCCESS_KW)
        is_liferay = any(k in line for k in LIFERAY_KW)

        prefix = '\n' if is_error and not _last_was_error else ''
        _last_was_error = is_error

        if is_success:
            m2 = re.search(r'Server startup in \[?(\d+)\]?\s*(\w+)', msg)
            duration = f'{m2.group(1)} {m2.group(2)}' if m2 else msg.strip()
            return box(f'🚀  Server started in {duration}', BGR)

        if is_liferay and 'Welcome to Liferay' in line:
            return box(f'🌐  {msg.strip()}', BMA)

        ts_s     = f'{DIM}{time_s:<{W_TIME}}{R}'
        icon_s   = f'{badge}{icon}{R}'
        level_s  = f'{badge}{level:<{W_LEVEL}}{R}'
        thread_s = f'{BCY}{fit(abbrev_thread(thread), W_THREAD)}{R}'
        cls_s    = f'{DIM}{IT}{fit(abbrev_class(cls), W_CLASS)}{R}'
        bar      = f'{bar_color}▌{R}'

        if is_error:
            msg_s = f'{BRE}{msg}{R}'
        elif is_warn:
            msg_s = f'{YE}{msg}{R}'
        elif is_liferay:
            msg_s = f'{BMA}{B}{msg}{R}'
        elif msg_color:
            msg_s = f'{msg_color}{msg}{R}'
        else:
            msg_s = colorize_msg(msg)

        return f'{prefix}{bar} {ts_s}  {icon_s} {level_s}  {thread_s}  {cls_s}  {msg_s}'

    _last_was_error = False
    if any(k in line for k in LIFERAY_KW):
        return f'{BMA}{B}{line}{R}'
    if any(k in line for k in SUCCESS_KW):
        return f'{BGR}{B}{line}{R}'
    if 'SEVERE' in line or 'FATAL' in line:
        return f'\n{BRE}{B}{line}{R}'
    if 'ERROR' in line:
        return f'{BRE}{line}{R}'
    if 'WARN' in line:
        return f'{BYE}{line}{R}'
    return f'{DIM}{line}{R}'


 # ── Main ──────────────────────────────────────────────────────────────────────
 def main():
    spinner = Spinner()
    startup_done = False

    try:
        for raw in sys.stdin:
            line = raw.rstrip('\n')
            formatted = format_line(line)

            if not startup_done:
                # Detect startup completion
                if re.search(r'Server startup in', line, re.I):
                    spinner.stop()
                    startup_done = True
                    print(formatted)
                    sys.stdout.flush()
                    continue

                # Update spinner phase
                for pattern, phase in PHASES[1:]:
                    if pattern.search(line):
                        spinner.update(phase)
                        break

                # Print all lines above the spinner
                spinner.print_above(formatted)
                continue

            # Post-startup: full colorized streaming
            print(formatted)
            sys.stdout.flush()

    except KeyboardInterrupt:
        if not startup_done:
            spinner.stop()
        print(f'\n{DIM}Catalize: interrupted.{R}')


 if __name__ == '__main__':
    main()
	#!/usr/bin/env python3
	"""Catalize — structured, colorized log pipe for Tomcat / Liferay DXP."""

	import re
	import sys
	import time
	import threading

	# ── ANSI ─────────────────────────────────────────────────────────────────────
	R = '\033[0m'
	B = '\033[1m'
	DIM = '\033[2m'
	IT = '\033[3m'
	UL = '\033[4m'

	RE = '\033[31m'; BRE = '\033[91m'
	GR = '\033[32m'; BGR = '\033[92m'
	YE = '\033[33m'; BYE = '\033[93m'
	BL = '\033[34m'; BBL = '\033[94m'
	MA = '\033[35m'; BMA = '\033[95m'
	CY = '\033[36m'; BCY = '\033[96m'
	WH = '\033[37m'; BWH = '\033[97m'

	BG_RE = '\033[41m'

	CLEAR_LINE = '\033[2K\r'

	# ── Column widths ─────────────────────────────────────────────────────────────
	W_TIME = 12
	W_LEVEL = 7
	W_THREAD = 22
	W_CLASS = 28

	# ── Level config ──────────────────────────────────────────────────────────────
	LEVELS = {
	'SEVERE': ('☠', f'{BG_RE}{B}{BWH}', BRE, BRE),
	'FATAL': ('☠', f'{BG_RE}{B}{BWH}', BRE, BRE),
	'ERROR': ('✖', f'{BRE}{B}', BRE, BRE),
	'WARNING': ('⚠', f'{BYE}{B}', YE, YE),
	'WARN': ('⚠', f'{BYE}{B}', YE, YE),
	'INFO': ('●', f'{BBL}', '', BBL),
	'CONFIG': ('◆', f'{MA}', '', MA),
	'DEBUG': ('○', f'{DIM}{CY}', DIM, DIM),
	'FINE': ('·', f'{DIM}', DIM, DIM),
	'FINER': ('·', f'{DIM}', DIM, DIM),
	'FINEST': ('·', f'{DIM}', DIM, DIM),
	}
	DEFAULT_LEVEL = ('·', '', '', DIM)

	# ── Startup phase detection ───────────────────────────────────────────────────
	PHASES = [
	(re.compile(r'Server startup in', re.I), None), # done
	(re.compile(r'Deploying web application', re.I), 'Deploying web application...'),
	(re.compile(r'Starting ProtocolHandler', re.I), 'Starting protocol handlers...'),
	(re.compile(r'OSGi framework started\|Starting modules framework', re.I),
	'OSGi framework starting...'),
	(re.compile(r'com\.liferay\.portal\.events', re.I), 'Running Liferay startup events...'),
	(re.compile(r'com\.liferay', re.I), 'Loading Liferay bundles...'),
	(re.compile(r'Initializing Spring', re.I), 'Initializing Spring context...'),
	(re.compile(r'Connector \[', re.I), 'Configuring connectors...'),
	(re.compile(r'Catalina', re.I), 'Initializing Catalina...'),
	]

	# ── Log line regex ────────────────────────────────────────────────────────────
	LOG_RE = re.compile(
	r'^(\d{2}-\w{3}-\d{4})\s+'
	r'(\d{2}:\d{2}:\d{2}\.\d+)\s+'
	r'(\w+)\s+'
	r'(\[\S+?\])\s+'
	r'([\w.$]+(?:\.[\w$<>]+))\s'
	r'(.*)'
	)

	SUCCESS_KW = ('Server startup in', 'Successfully deployed', 'Deployment of web application')
	LIFERAY_KW = ('Welcome to Liferay', 'Liferay DXP', 'OSGi framework started', 'Starting modules framework')

	# ── Inline highlights ─────────────────────────────────────────────────────────
	INLINE = [
	(re.compile(r'(\d[\d.,]\s(?:ms\|millis(?:econds?)?\|seconds?\|mins?\|minutes?\|hours?)\b)', re.I), BGR),
	(re.compile(r'(\d[\d.,]\s(?:KB\|MB\|GB\|TB\|bytes?)\b)', re.I), BMA),
	(re.compile(r'\b(\d{1,3}(?:\.\d{1,3}){3})\b'), BCY),
	(re.compile(r'(:\d{2,5})(?=/\|\s\|$)'), BYE),
	(re.compile(r'(https?://\S+)'), UL + BCY),
	(re.compile(r'(?<![:\w])(/[\w./\-]+)'), UL + DIM),
	(re.compile(r'("(?:[^"\\]\|\\.)*")'), BWH),
	(re.compile(r'(?<!\w)(\d+)(?!\w)'), YE),
	]

	def colorize_msg(msg: str) -> str:
	for pattern, color in INLINE:
	msg = pattern.sub(lambda m, c=color: f'{c}{m.group(1)}{R}', msg)
	return msg

	def abbrev_thread(thread: str) -> str:
	inner = thread.strip('[]')
	if len(inner) <= W_THREAD - 2:
	return thread
	parts = inner.rsplit('-', 2)
	tail = '-'.join(parts[-2:]) if len(parts) >= 2 else inner[-(W_THREAD - 4):]
	return f'[…{tail}]'

	def abbrev_class(cls: str) -> str:
	if len(cls) <= W_CLASS:
	return cls
	parts = cls.split('.')
	short = '.'.join(parts[-3:]) if len(parts) >= 3 else cls
	return f'…{short}' if len(short) < len(cls) else cls[-W_CLASS:]

	def fit(s: str, width: int) -> str:
	visible = re.sub(r'\033\[[0-9;]*m', '', s)
	diff = width - len(visible)
	return s + ' ' * diff if diff >= 0 else visible[:width]

	def box(msg: str, color: str) -> str:
	inner = f' {msg} '
	w = len(re.sub(r'\033\[[0-9;]*m', '', inner)) + 2
	return (
	f'\n{color}╔{"═" * w}╗{R}\n'
	f'{color}║{R}{B}{inner}{R}{color}║{R}\n'
	f'{color}╚{"═" * w}╝{R}\n'
	)

	# ── Spinner ───────────────────────────────────────────────────────────────────
	class Spinner:
	FRAMES = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠣', '⠏']
	BAR_W = 28
	BAR_CHR = ('━', '╺')

	def __init__(self):
	self.phase = 'Initializing...'
	self.done = False
	self.lock = threading.Lock()
	self._start = time.time()
	self._frame = 0
	self._t = threading.Thread(target=self._run, daemon=True)
	self._t.start()

	def _elapsed(self) -> str:
	s = int(time.time() - self._start)
	return f'{s // 60:02d}:{s % 60:02d}'

	def _bar(self) -> str:
	"""Animated indeterminate bar that bounces."""
	w = self.BAR_W
	pos = self._frame % (w * 2)
	if pos >= w:
	pos = w * 2 - pos
	filled = min(pos + 6, w)
	empty = w - filled
	bar = self.BAR_CHR[0] * filled + self.BAR_CHR[1] + self.BAR_CHR[1] * empty
	return f'{BCY}{bar[:w]}{R}'

	def _run(self):
	while not self.done:
	with self.lock:
	f = self.FRAMES[self._frame % len(self.FRAMES)]
	el = self._elapsed()
	ph = self.phase[:45]
	bar = self._bar()
	line = (
	f'{CLEAR_LINE}'
	f'{BCY}{B}{f}{R} '
	f'{B}Starting up{R} '
	f'{bar} '
	f'{DIM}{ph:<45}{R} '
	f'{DIM}[{el}]{R}'
	)
	sys.stdout.write(line)
	sys.stdout.flush()
	self._frame += 1
	time.sleep(0.08)

	def update(self, phase: str):
	with self.lock:
	self.phase = phase

	def clear_line(self):
	sys.stdout.write(CLEAR_LINE)
	sys.stdout.flush()

	def print_above(self, formatted: str):
	"""Interrupt spinner to print a line, then resume."""
	with self.lock:
	sys.stdout.write(CLEAR_LINE)
	print(formatted)
	sys.stdout.flush()

	def stop(self):
	self.done = True
	self._t.join(timeout=0.5)
	sys.stdout.write(CLEAR_LINE)
	sys.stdout.flush()


	# ── Log formatter ─────────────────────────────────────────────────────────────
	_last_was_error = False

	def format_line(line: str) -> str:
	global _last_was_error

	if re.match(r'^\s+at ', line) or re.match(r'^\s+\.\.\. \d+ more', line):
	_last_was_error = False
	return f' {DIM}{RE}│{R} {DIM}{RE}{line.strip()}{R}'

	if re.match(r'^\s*Caused by:', line) or re.match(r'^[\w.$]+(?:Exception\|Error)', line):
	_last_was_error = False
	return f' {BRE}│{R} {BRE}{B}{line.strip()}{R}'

	m = LOG_RE.match(line)
	if m:
	_, time_s, level, thread, cls, msg = m.groups()

	icon, badge, msg_color, bar_color = LEVELS.get(level, DEFAULT_LEVEL)
	is_error = level in ('SEVERE', 'FATAL', 'ERROR')
	is_warn = level in ('WARN', 'WARNING')
	is_success = any(k in msg for k in SUCCESS_KW)
	is_liferay = any(k in line for k in LIFERAY_KW)

	prefix = '\n' if is_error and not _last_was_error else ''
	_last_was_error = is_error

	if is_success:
	m2 = re.search(r'Server startup in \[?(\d+)\]?\s*(\w+)', msg)
	duration = f'{m2.group(1)} {m2.group(2)}' if m2 else msg.strip()
	return box(f'🚀 Server started in {duration}', BGR)

	if is_liferay and 'Welcome to Liferay' in line:
	return box(f'🌐 {msg.strip()}', BMA)

	ts_s = f'{DIM}{time_s:<{W_TIME}}{R}'
	icon_s = f'{badge}{icon}{R}'
	level_s = f'{badge}{level:<{W_LEVEL}}{R}'
	thread_s = f'{BCY}{fit(abbrev_thread(thread), W_THREAD)}{R}'
	cls_s = f'{DIM}{IT}{fit(abbrev_class(cls), W_CLASS)}{R}'
	bar = f'{bar_color}▌{R}'

	if is_error:
	msg_s = f'{BRE}{msg}{R}'
	elif is_warn:
	msg_s = f'{YE}{msg}{R}'
	elif is_liferay:
	msg_s = f'{BMA}{B}{msg}{R}'
	elif msg_color:
	msg_s = f'{msg_color}{msg}{R}'
	else:
	msg_s = colorize_msg(msg)

	return f'{prefix}{bar} {ts_s} {icon_s} {level_s} {thread_s} {cls_s} {msg_s}'

	_last_was_error = False
	if any(k in line for k in LIFERAY_KW):
	return f'{BMA}{B}{line}{R}'
	if any(k in line for k in SUCCESS_KW):
	return f'{BGR}{B}{line}{R}'
	if 'SEVERE' in line or 'FATAL' in line:
	return f'\n{BRE}{B}{line}{R}'
	if 'ERROR' in line:
	return f'{BRE}{line}{R}'
	if 'WARN' in line:
	return f'{BYE}{line}{R}'
	return f'{DIM}{line}{R}'


	# ── Main ──────────────────────────────────────────────────────────────────────
	def main():
	spinner = Spinner()
	startup_done = False

	try:
	for raw in sys.stdin:
	line = raw.rstrip('\n')
	formatted = format_line(line)

	if not startup_done:
	# Detect startup completion
	if re.search(r'Server startup in', line, re.I):
	spinner.stop()
	startup_done = True
	print(formatted)
	sys.stdout.flush()
	continue

	# Update spinner phase
	for pattern, phase in PHASES[1:]:
	if pattern.search(line):
	spinner.update(phase)
	break

	# Print all lines above the spinner
	spinner.print_above(formatted)
	continue

	# Post-startup: full colorized streaming
	print(formatted)
	sys.stdout.flush()

	except KeyboardInterrupt:
	if not startup_done:
	spinner.stop()
	print(f'\n{DIM}Catalize: interrupted.{R}')


	if __name__ == '__main__':
	main()
No results found