diff --git a/app/features/core/utils.py b/app/features/core/utils.py index 6e2faa58..674b68bb 100644 --- a/app/features/core/utils.py +++ b/app/features/core/utils.py @@ -79,5 +79,20 @@ def format_validation_errors(exc: ValidationError) -> list[dict[str, Any]]: def gen_random(length: int = 16) -> str: import secrets + import string - return "".join(secrets.token_urlsafe(length)[:length]) + if length < 1: + msg = "length must be >= 1" + raise ValueError(msg) + + middle_alphabet = string.ascii_letters + string.digits + "-_" + edge_alphabet = string.ascii_letters + string.digits + + if 1 == length: + return secrets.choice(edge_alphabet) + + return ( + secrets.choice(edge_alphabet) + + "".join(secrets.choice(middle_alphabet) for _ in range(length - 2)) + + secrets.choice(edge_alphabet) + ) diff --git a/app/library/Utils.py b/app/library/Utils.py index b63b997b..baebbad6 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -33,8 +33,6 @@ FILES_TYPE: list = [ TAG_REGEX: re.Pattern[str] = re.compile(r"%{([^:}]+)(?::([^}]*))?}c") "Regex to find tags in templates." -DT_PATTERN: re.Pattern[str] = re.compile(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:[+-]\d{2}:\d{2}))\s?") -"Regex to match ISO 8601 datetime strings." class FileLogFormatter(logging.Formatter): @@ -969,116 +967,6 @@ def strip_newline(string: str) -> str: return res.strip() if res else "" -async def read_logfile(file: Path, offset: int = 0, limit: int = 50) -> dict: - """ - Read a log file and return a set of log lines along with pagination metadata. - - Args: - file (Path): The log file path. - offset (int): Number of lines to skip from the end (newer entries). - limit (int): Number of lines to return. - - Returns: - dict: A dictionary containing: - - logs: List of log entries. - - next_offset: Offset for the next page or None. - - end_is_reached: True if there are no older logs. - - """ - from hashlib import sha256 - - from anyio import open_file - - if not file.exists(): - return {"logs": [], "next_offset": None, "end_is_reached": True} - - result = [] - try: - async with await open_file(file, "rb") as f: - await f.seek(0, os.SEEK_END) - file_size: int = await f.tell() - - block_size = 1024 - block_end: int = file_size - buffer: bytes = b"" - lines: list = [] - - required_count: int = offset + limit + 1 - - while len(lines) < required_count and block_end > 0: - block_start: int = max(0, block_end - block_size) - await f.seek(block_start) - chunk: bytes = await f.read(block_end - block_start) - buffer: bytes = chunk + buffer # prepend the chunk - lines = buffer.splitlines() - block_end = block_start - - if len(lines) > offset + limit: - next_offset: int = offset + limit - end_is_reached = False - else: - next_offset = None - end_is_reached = True - - for line in lines[-(offset + limit) : -offset] if offset else lines[-limit:]: - line_bytes: bytes | str = line if isinstance(line, bytes) else line.encode() - msg: str = line.decode(errors="replace") - dt_match: re.Match[str] | None = DT_PATTERN.match(msg) - result.append( - { - "id": sha256(line_bytes).hexdigest(), - "line": msg[dt_match.end() :] if dt_match else msg, - "datetime": dt_match.group(1) if dt_match else None, - } - ) - - return {"logs": result, "next_offset": next_offset, "end_is_reached": end_is_reached} - except Exception: - return {"logs": [], "next_offset": None, "end_is_reached": True} - - -async def tail_log(file: Path, emitter: callable, sleep_time: float = 0.5): - """ - Continuously read a log file and emit new lines. - - Args: - file (str): The log file path. - emitter (callable): A callable to emit new lines. - sleep_time (float): The time to sleep between reads. - - """ - from asyncio import sleep as asyncio_sleep - from hashlib import sha256 - - from anyio import open_file - - if not file.exists(): - return - - try: - async with await open_file(file, "rb") as f: - await f.seek(0, os.SEEK_END) - while True: - line: bytes = await f.readline() - if not line: - await asyncio_sleep(sleep_time) - continue - - msg: str = line.decode(errors="replace") - dt_match: re.Match[str] | None = DT_PATTERN.match(msg) - - await emitter( - { - "id": sha256(line if isinstance(line, bytes) else line.encode()).hexdigest(), - "line": msg[dt_match.end() :] if dt_match else msg, - "datetime": dt_match.group(1) if dt_match else None, - } - ) - except Exception as e: - LOG.error(f"Error while tailing log file '{file!s}': {e!s}") - return - - def load_cookies(file: str | Path) -> tuple[bool, MozillaCookieJar]: """ Validate and load a cookie file. diff --git a/app/routes/api/logs.py b/app/routes/api/logs.py index 073d4fc4..d36ed350 100644 --- a/app/routes/api/logs.py +++ b/app/routes/api/logs.py @@ -1,5 +1,7 @@ import asyncio import logging +import os +import re from pathlib import Path from aiohttp import web @@ -8,10 +10,121 @@ from aiohttp.web import Request, Response from app.library.config import Config from app.library.encoder import Encoder from app.library.router import route -from app.library.Utils import read_logfile, tail_log LOG: logging.Logger = logging.getLogger(__name__) +DT_PATTERN: re.Pattern[str] = re.compile(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:[+-]\d{2}:\d{2}))\s?") +"Match ISO8601." + + +async def _read_logfile(file: Path, offset: int = 0, limit: int = 50) -> dict: + """ + Read a log file and return a set of log lines along with pagination metadata. + + Args: + file (Path): The log file path. + offset (int): Number of lines to skip from the end (newer entries). + limit (int): Number of lines to return. + + Returns: + dict: A dictionary containing: + - logs: List of log entries. + - next_offset: Offset for the next page or None. + - end_is_reached: True if there are no older logs. + + """ + from hashlib import sha256 + + from anyio import open_file + + if not file.exists(): + return {"logs": [], "next_offset": None, "end_is_reached": True} + + result = [] + try: + async with await open_file(file, "rb") as f: + await f.seek(0, os.SEEK_END) + file_size: int = await f.tell() + + block_size = 1024 + block_end: int = file_size + buffer: bytes = b"" + lines: list = [] + + required_count: int = offset + limit + 1 + + while len(lines) < required_count and block_end > 0: + block_start: int = max(0, block_end - block_size) + await f.seek(block_start) + chunk: bytes = await f.read(block_end - block_start) + buffer: bytes = chunk + buffer # prepend the chunk + lines = buffer.splitlines() + block_end = block_start + + if len(lines) > offset + limit: + next_offset: int = offset + limit + end_is_reached = False + else: + next_offset = None + end_is_reached = True + + for line in lines[-(offset + limit) : -offset] if offset else lines[-limit:]: + line_bytes: bytes | str = line if isinstance(line, bytes) else line.encode() + msg: str = line.decode(errors="replace") + dt_match: re.Match[str] | None = DT_PATTERN.match(msg) + result.append( + { + "id": sha256(line_bytes).hexdigest(), + "line": msg[dt_match.end() :] if dt_match else msg, + "datetime": dt_match.group(1) if dt_match else None, + } + ) + + return {"logs": result, "next_offset": next_offset, "end_is_reached": end_is_reached} + except Exception: + return {"logs": [], "next_offset": None, "end_is_reached": True} + + +async def _tail_log(file: Path, emitter: callable, sleep_time: float = 0.5): + """ + Live tail. + + Args: + file (str): The log file path. + emitter (callable): A callable to emit new lines. + sleep_time (float): The time to sleep between reads. + + """ + from hashlib import sha256 + + from anyio import open_file + + if not file.exists(): + return + + try: + async with await open_file(file, "rb") as f: + await f.seek(0, os.SEEK_END) + while True: + line: bytes = await f.readline() + if not line: + await asyncio.sleep(sleep_time) + continue + + msg: str = line.decode(errors="replace") + dt_match: re.Match[str] | None = DT_PATTERN.match(msg) + + await emitter( + { + "id": sha256(line if isinstance(line, bytes) else line.encode()).hexdigest(), + "line": msg[dt_match.end() :] if dt_match else msg, + "datetime": dt_match.group(1) if dt_match else None, + } + ) + except Exception as e: + LOG.error(f"Error while tailing log file '{file!s}': {e!s}") + return + @route("GET", "api/logs/", "logs") async def logs(request: Request, config: Config, encoder: Encoder) -> Response: @@ -35,7 +148,7 @@ async def logs(request: Request, config: Config, encoder: Encoder) -> Response: if limit < 1 or limit > 150: limit = 50 - logs_data = await read_logfile( + logs_data = await _read_logfile( file=Path(config.config_path) / "logs" / "app.log", offset=offset, limit=limit, @@ -85,12 +198,17 @@ async def stream_logs(request: Request, config: Config, encoder: Encoder) -> Res payload = f"event: log_lines\ndata: {encoder.encode(data)}\n\n" await response.write(payload.encode("utf-8")) - log_task: asyncio.Task[None] = asyncio.create_task(tail_log(file=log_file, emitter=emit_log), name="log_stream") + from app.features.core.utils import gen_random + + log_task: asyncio.Task[None] = asyncio.create_task( + _tail_log(file=log_file, emitter=emit_log), + name=f"log_stream_{gen_random(8)}", + ) try: LOG.debug("Log streaming connected.") while not log_task.done(): - await asyncio.sleep(0.5) + await asyncio.sleep(1.0) if request.transport is None or request.transport.is_closing(): log_task.cancel() break diff --git a/app/tests/test_utils.py b/app/tests/test_utils.py index 913eb8e9..39ce9674 100644 --- a/app/tests/test_utils.py +++ b/app/tests/test_utils.py @@ -34,15 +34,14 @@ from app.library.Utils import ( merge_dict, move_file, parse_tags, - read_logfile, rename_file, str_to_dt, strip_newline, - tail_log, timed_lru_cache, validate_url, validate_uuid, ) +from app.routes.api.logs import _read_logfile, _tail_log class TestTimedLruCache: @@ -1517,7 +1516,7 @@ class TestReadLogfile: """Test reading non-existent log file.""" async def test(): - result = await read_logfile(self.log_file) + result = await _read_logfile(self.log_file) assert isinstance(result, dict) assert "logs" in result @@ -1528,7 +1527,7 @@ class TestReadLogfile: self.log_file.write_text("line 1\nline 2\nline 3\n") async def test(): - result = await read_logfile(self.log_file, limit=2) + result = await _read_logfile(self.log_file, limit=2) assert isinstance(result, dict) assert "logs" in result @@ -1559,7 +1558,7 @@ class TestTailLog: async def test(): try: - await tail_log(self.log_file, emitter, sleep_time=0.1) + await _tail_log(self.log_file, emitter, sleep_time=0.1) except Exception: pass # Expected for non-existent file