58 lines
2.1 KiB
Python
58 lines
2.1 KiB
Python
# This is free software for the public good of a permacomputer hosted at
|
|
# permacomputer.com, an always-on computer by the people, for the people.
|
|
# One which is durable, easy to repair, & distributed like tap water
|
|
# for machine learning intelligence.
|
|
#
|
|
# The permacomputer is community-owned infrastructure optimized around
|
|
# four values:
|
|
#
|
|
# TRUTH First principles, math & science, open source code freely distributed
|
|
# FREEDOM Voluntary partnerships, freedom from tyranny & corporate control
|
|
# HARMONY Minimal waste, self-renewing systems with diverse thriving connections
|
|
# LOVE Be yourself without hurting others, cooperation through natural law
|
|
#
|
|
# This software contributes to that vision by archiving the web, preserving digital knowledge before it disappears.
|
|
# Code is seeds to sprout on any abandoned technology.
|
|
|
|
"""Job logging utilities for SERP.
|
|
|
|
# Side quest 3/21: Look out! Here comes the spider-pig.
|
|
"""
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Dict
|
|
|
|
LOGS_PATH = Path("data/logs")
|
|
JOB_LOG_HANDLERS: Dict[int, logging.FileHandler] = {}
|
|
|
|
|
|
def start_job_logging(job_id: int) -> None:
|
|
"""Start capturing logs for a crawl job."""
|
|
LOGS_PATH.mkdir(parents=True, exist_ok=True)
|
|
log_file = LOGS_PATH / f"{job_id}.log"
|
|
handler = logging.FileHandler(log_file, mode='w', encoding='utf-8')
|
|
handler.setLevel(logging.INFO)
|
|
handler.setFormatter(logging.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s', datefmt='%H:%M:%S'))
|
|
logging.getLogger().addHandler(handler)
|
|
JOB_LOG_HANDLERS[job_id] = handler
|
|
|
|
|
|
def stop_job_logging(job_id: int) -> None:
|
|
"""Stop capturing logs for a crawl job."""
|
|
handler = JOB_LOG_HANDLERS.pop(job_id, None)
|
|
if handler:
|
|
handler.close()
|
|
logging.getLogger().removeHandler(handler)
|
|
|
|
|
|
def get_job_logs(job_id: int, tail: int = 0) -> str:
|
|
"""Get logs for a crawl job."""
|
|
log_file = LOGS_PATH / f"{job_id}.log"
|
|
if not log_file.exists():
|
|
return ""
|
|
content = log_file.read_text(encoding='utf-8')
|
|
if tail > 0:
|
|
lines = content.splitlines()
|
|
return '\n'.join(lines[-tail:])
|
|
return content
|