Open file and optionally read last N lines
(self)
| 52 | self.inode = None |
| 53 | |
| 54 | def open_file(self): |
| 55 | """Open file and optionally read last N lines""" |
| 56 | try: |
| 57 | self.file = open(self.filepath, "r", encoding="utf-8", errors="ignore") |
| 58 | if self.search_pattern: |
| 59 | print( |
| 60 | f"{Colors.CYAN}[INFO] Searching for pattern '{self.search_pattern}' in {self.filepath}{Colors.RESET}" |
| 61 | ) |
| 62 | self.file.seek(0) |
| 63 | lines = self.file.readlines() |
| 64 | match_indices = [i for i, line in enumerate(lines) if self.search_pattern in line] |
| 65 | for idx in match_indices: |
| 66 | start = max(0, idx - 5) |
| 67 | end = min(len(lines), idx + 6) |
| 68 | print(f"{Colors.MAGENTA}[{self.filepath}:{idx + 1}]{Colors.RESET}") |
| 69 | for i in range(start, end): |
| 70 | prefix = f"{Colors.MAGENTA}>> {Colors.RESET}" if i == idx else " " |
| 71 | print(prefix + self.format_output(lines[i].rstrip("\n"))) |
| 72 | print( |
| 73 | f"{Colors.CYAN}[INFO] Finished searching in {self.filepath}, now monitoring for new lines...{Colors.RESET}" |
| 74 | ) |
| 75 | stat = os.stat(self.filepath) |
| 76 | self.inode = stat.st_ino |
| 77 | if self.last_n_lines > 0: |
| 78 | # Read last N lines |
| 79 | self.file.seek(0, 2) |
| 80 | file_size = self.file.tell() |
| 81 | block_size = 4096 |
| 82 | blocks = [] |
| 83 | lines_found = 0 |
| 84 | pos = file_size |
| 85 | while pos > 0 and lines_found < self.last_n_lines: |
| 86 | read_size = min(block_size, pos) |
| 87 | pos -= read_size |
| 88 | self.file.seek(pos) |
| 89 | block = self.file.read(read_size) |
| 90 | blocks.insert(0, block) |
| 91 | lines_found = sum(b.count("\n") for b in blocks) |
| 92 | all_data = "".join(blocks) |
| 93 | last_lines = all_data.splitlines()[-self.last_n_lines :] |
| 94 | for line in last_lines: |
| 95 | print(self.format_output(line)) |
| 96 | self.file.seek(0, 2) |
| 97 | else: |
| 98 | self.file.seek(0, 2) |
| 99 | self.file_size = self.file.tell() |
| 100 | return True |
| 101 | except Exception as e: |
| 102 | print(f"{Colors.RED}[ERROR] Failed to open {self.filepath}: {e}{Colors.RESET}") |
| 103 | return False |
| 104 | |
| 105 | def check_rotation(self): |
| 106 | """Check if file has been rotated""" |
no test coverage detected