NotesAgentic/backend/app/operation_logs.py

"""有界的异步操作诊断，与业务/Trace 数据分开。仅存储明确允许的元数据。切勿在此诊断通道中存储提示、工具参数、提供程序响应正文或原始异常消息。"""
from __future__ import annotations

import json
import logging
import math
import queue
import re
import sqlite3
import threading
import traceback
from contextvars import ContextVar
from contextlib import closing
from datetime import datetime, timezone
from pathlib import Path

from app.config import get_settings

request_id: ContextVar[str] = ContextVar('log_request_id', default='')
agent_run_id: ContextVar[str] = ContextVar('log_agent_run_id', default='')
_allowed = {'run_id', 'task_id', 'note_id', 'job_id', 'provider_id', 'model',
            'device', 'error_code', 'error_type', 'status', 'duration_ms', 'count',
            'step', 'sequence', 'tool', 'method', 'route', 'request_id', 'fallback',
            'frames', 'source', 'changed_fields'}
_safe = re.compile(r'[^\w .:/@{}\[\],()=+\-]', re.UNICODE)


def metadata(values: dict) -> dict:
    result = {}
    for key, value in values.items():
        if key not in _allowed or value is None:
            continue
        if isinstance(value, (int, float, bool)):
            if not isinstance(value, float) or math.isfinite(value):
                result[key] = value
        else:
            text = str(value)
            text = re.sub(r'(?i)(?:bearer\s+\S+|sk-[\w-]+)', '[REDACTED]', text)
            result[key] = _safe.sub('', text)[:500]
    return result


class LogStore:
    def __init__(self, path: Path, *, retain: int = 20_000):
        self.path = path
        self.retain = retain
        self.queue: queue.Queue = queue.Queue(maxsize=4096)
        self.dropped = 0
        self.failed = 0
        self.closed = False
        self.state_lock = threading.Lock()
        self.thread = threading.Thread(target=self._write, name='operation-logs', daemon=True)
        path.parent.mkdir(parents=True, exist_ok=True)
        with closing(self._connect()) as conn, conn:
            conn.execute('CREATE TABLE IF NOT EXISTS logs (id INTEGER PRIMARY KEY, timestamp TEXT NOT NULL, level TEXT NOT NULL, source TEXT NOT NULL, event TEXT NOT NULL, details TEXT NOT NULL)')
            conn.execute('CREATE INDEX IF NOT EXISTS logs_level_id ON logs(level, id)')
            conn.execute('CREATE INDEX IF NOT EXISTS logs_source_id ON logs(source, id)')
        self.thread.start()

    def _connect(self):
        conn = sqlite3.connect(self.path, timeout=5)
        conn.row_factory = sqlite3.Row
        return conn

    def emit(self, level: str, source: str, event: str, details: dict):
        row = (datetime.now(timezone.utc).isoformat(), level, source[:100], event[:160], json.dumps(metadata(details), ensure_ascii=False))
        with self.state_lock:
            if self.closed:
                return
            try:
                self.queue.put_nowait(row)
            except queue.Full:
                self.dropped += 1

    def _write(self):
        while True:
            first = self.queue.get()
            batch = [first]
            while len(batch) < 128:
                try:
                    batch.append(self.queue.get_nowait())
                except queue.Empty:
                    break
            stop = None in batch
            rows = [row for row in batch if row is not None]
            try:
                if rows:
                    with closing(self._connect()) as conn, conn:
                        conn.executemany('INSERT INTO logs(timestamp,level,source,event,details) VALUES(?,?,?,?,?)', rows)
                        conn.execute('DELETE FROM logs WHERE id <= (SELECT id FROM logs ORDER BY id DESC LIMIT 1 OFFSET ?)', (self.retain,))
            except Exception:
                self.failed += len(rows)
            finally:
                for _ in batch:
                    self.queue.task_done()
            if stop:
                return

    def query(self, *, limit=50, before=None, level='', source='', q=''):
        clauses, args = [], []
        for column, value in [('level', level), ('source', source)]:
            if value:
                clauses.append(f'{column} = ?')
                args.append(value)
        if before is not None:
            clauses.append('id < ?')
            args.append(before)
        if q:
            clauses.append('(instr(event, ?) > 0 OR instr(details, ?) > 0)')
            args += [q, q]
        where = ' WHERE ' + ' AND '.join(clauses) if clauses else ''
        with closing(self._connect()) as conn, conn:
            rows = conn.execute('SELECT * FROM logs' + where + ' ORDER BY id DESC LIMIT ?', (*args, limit + 1)).fetchall()
            sources = [row[0] for row in conn.execute('SELECT DISTINCT source FROM logs ORDER BY source')]
        items = [{**dict(row), 'details': json.loads(row['details'])} for row in rows[:limit]]
        return {'items': items, 'next_cursor': items[-1]['id'] if len(rows) > limit else None,
                'sources': sources, 'pending': self.queue.qsize(), 'dropped': self.dropped,
                'write_failures': self.failed, 'retention': self.retain}

    def close(self):
        with self.state_lock:
            if self.closed:
                return
            self.closed = True
        self.queue.put(None)
        self.thread.join(timeout=15)


_store: LogStore | None = None
_lock = threading.Lock()


def get_store() -> LogStore:
    global _store
    path = get_settings().data_dir / 'logs' / 'operations.sqlite3'
    with _lock:
        if _store is None or _store.path != path or _store.closed:
            if _store is not None and not _store.closed:
                _store.close()
            _store = LogStore(path)
        return _store


def log_event(module: str, event: str, *, level='INFO', error: BaseException | None = None, **details):
    if request_id.get():
        details.setdefault('request_id', request_id.get())
    if agent_run_id.get():
        details.setdefault('run_id', agent_run_id.get())
    if error:
        details['error_type'] = type(error).__name__
        details.setdefault('error_code', getattr(error, 'code', None))
        details['frames'] = '; '.join(f'{Path(f.filename).name}:{f.lineno}:{f.name}' for f in traceback.extract_tb(error.__traceback__)[-8:])
    try:
        get_store().emit(level, module, event, details)
    except Exception:
        # 日志记录不得将成功的保存/运行变成业务失败。
        logging.getLogger('operation_log_storage').error('Operational log storage unavailable')


class ApplicationLogHandler(logging.Handler):
    def emit(self, record):
        if record.name == 'operation_log_storage' or getattr(record, '_notes_operation_logged', False):
            return
        record._notes_operation_logged = True
        # 旧日志消息可能包含笔记文本或凭据，f-string 也不例外。保留源码位置与错误类型；结构化调用点负责携带 ID。
        log_event(record.name, 'application.warning' if record.levelno < 40 else 'application.error',
                  level=record.levelname, error=record.exc_info[1] if record.exc_info else None,
                  frames=f'{Path(record.pathname).name}:{record.lineno}:{record.funcName}')


def install_logging():
    # Uvicorn 的默认记录器在根记录器之前停止传播。
    for name in ('', 'uvicorn'):
        logger = logging.getLogger(name)
        if not any(isinstance(h, ApplicationLogHandler) for h in logger.handlers):
            logger.addHandler(ApplicationLogHandler(level=logging.WARNING))


def shutdown_logging():
    if _store is not None and not _store.closed:
        _store.close()