Refactor: move log related inside the logs module
This commit is contained in:
arabcoders
parent
b1a05b1d79
commit
f88422e92b
4 changed files with 142 additions and 122 deletions
|
|
@ -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)
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue