Add tail-style log reader and -n flag

README.md

@@ -77,14 +77,17 @@ Output:
 pmo start   [all | service-name | service-id]
 pmo stop    [all | service-name | service-id]
 pmo restart [all | service-name | service-id]
-pmo logs    [all | service-name | service-id]
+pmo logs    [-f] [-n NUM] [all | service-name | service-id]
 pmo flush   [all | service-name | service-id]
 pmo dry-run [all | service-name | service-id]
 pmo ls
 pmo status  [all | service-name | service-id]
 
 ```
 
+- `-n NUM` shows the last `NUM` log lines.
+- `-f` follows the log output in real time.
+
 ## Configuration
 
 The `pmo.yml` file supports two formats:

pmo/cli.py

@@ -72,9 +72,9 @@ def setup_arg_parser() -> argparse.ArgumentParser:
     log_parser = subparsers.add_parser('logs', aliases=['log'], help=f'{Emojis.LOG} View service logs')
     log_parser.add_argument('service', nargs='*', default=['all'],
                           help='Service names or IDs (multiple allowed) or "all" to view all logs')
-    log_parser.add_argument('--no-follow', '-n', action='store_true',
-                          help='Do not follow logs in real-time')
-    log_parser.add_argument('--lines', '-l', type=int, 
+    log_parser.add_argument('-f', '--follow', action='store_true',
+                          help='Follow logs in real-time')
+    log_parser.add_argument('-n', '--lines', '-l', type=int, dest='lines',
                           help='Number of lines to show initially (default: 15 for all services, 30 for specific services)')
 
     # Flush command
@@ -237,7 +237,7 @@ def handle_restart(manager: ServiceManager, service_specs: List[str]) -> bool:
 def handle_log(manager: ServiceManager, log_manager: LogManager, args) -> bool:
     """Handle log command with support for multiple services and remote hostnames"""
     service_specs = args.service
-    follow = not args.no_follow
+    follow = args.follow
 
     # Set default line values based on whether specific services are specified
     if args.lines is None:

pmo/logs.py

@@ -7,7 +7,7 @@
 import time
 import re
 from pathlib import Path
-from typing import List, Dict, Optional, Tuple
+from typing import List, Dict, Optional, Tuple, IO
 
 from rich.console import Console
 from rich.theme import Theme
@@ -338,35 +338,45 @@ def tail_logs(self, service_names: List[str], follow: bool = True, lines: Option
             self._follow_logs(log_files, hostname)
         else:
             self._display_recent_logs(log_files, lines, hostname)
-
+
+    def _read_last_lines(self, file: IO[bytes], lines: int) -> List[str]:
+        """Read last N lines from a binary file efficiently."""
+        block_size = 1024
+        file.seek(0, os.SEEK_END)
+        pointer = file.tell()
+        buffer = bytearray()
+
+        while pointer > 0 and buffer.count(b"\n") <= lines:
+            read_size = min(block_size, pointer)
+            pointer -= read_size
+            file.seek(pointer)
+            buffer[:0] = file.read(read_size)
+
+        text = buffer.decode(errors="replace")
+        return text.splitlines()[-lines:]
+
     def _display_recent_logs(self, log_files: List[Tuple[str, str, Path, str]], lines: int, hostname: Optional[str] = None):
         """Display recent log lines"""
         for service, log_type, log_path, service_id in log_files:
-            # PM2-style title
             console.print(f"\n[dim]{log_path} last {lines} lines:[/]")
-            
+
             try:
-                # Read last N lines
-                with open(log_path, 'r') as f:
-                    content = f.readlines()
-                    last_lines = content[-lines:] if len(content) >= lines else content
-
-                    # Print each line with service ID, PM2 format
-                    for line in last_lines:
-                        timestamp, message = self._parse_log_line(line)
-                        # 根据日志类型选择样式
-                        if log_type == "merged":
-                            style = "stdout_service"  # 合并日志使用stdout样式
-                        else:
-                            style = "stderr_service" if log_type == "stderr" else "stdout_service"
-                        # Use Text object to avoid Rich markup parsing in message content
-                        text = Text()
-                        text.append(f"{service_id} | ")
-                        if hostname:
-                            text.append(f"{hostname}:")
-                        text.append(service, style=style)
-                        text.append(f" | {timestamp}: {message}")
-                        console.print(text)
+                with open(log_path, "rb") as f:
+                    last_lines = self._read_last_lines(f, lines)
+
+                for line in last_lines:
+                    timestamp, message = self._parse_log_line(line)
+                    if log_type == "merged":
+                        style = "stdout_service"
+                    else:
+                        style = "stderr_service" if log_type == "stderr" else "stdout_service"
+                    text = Text()
+                    text.append(f"{service_id} | ")
+                    if hostname:
+                        text.append(f"{hostname}:")
+                    text.append(service, style=style)
+                    text.append(f" | {timestamp}: {message}")
+                    console.print(text)
             except Exception as e:
                 print_error(f"Error reading log file: {str(e)}")
 
@@ -425,4 +435,4 @@ def _follow_logs(self, log_files: List[Tuple[str, str, Path, str]], hostname: Op
         finally:
             # Close all files
             for f in file_handlers.values():
-                f.close()
+                f.close()

tests/test_cli.py

@@ -294,7 +294,7 @@ def test_log_command(self, temp_dir, basic_config_file):
             # 创建模拟参数
             args = MagicMock()
             args.service = ["service1"]
-            args.no_follow = False
+            args.follow = True
             args.lines = 20
 
             # 设置服务解析行为
@@ -318,7 +318,7 @@ def test_log_command(self, temp_dir, basic_config_file):
                 assert result is True
 
             # 测试不跟随模式
-            args.no_follow = True
+            args.follow = False
             mock_log_manager.tail_logs.reset_mock()
             with patch('pmo.cli.resolve_remote_service_spec',
                        return_value=(None, ["service1"])) as mock_resolve:

tests/test_integration.py

@@ -224,7 +224,7 @@ def test_log_integration(self, cli_test_config):
         # 模拟参数对象
         class Args:
             service = ['test-service']
-            no_follow = True
+            follow = False
             lines = 5
 
         args = Args()