telemetry.py

"""
telemetry.py - StoryWeaver 结构化交互日志工具

职责：
1. 为每个游戏会话分配稳定的 session_id
2. 以 JSONL 形式落盘每回合交互记录
3. 为评估脚本和案例分析提供统一的日志格式
"""

from __future__ import annotations

import json
import os
import uuid
from datetime import datetime
from pathlib import Path
from typing import Any


PROJECT_ROOT = Path(__file__).resolve().parent
DEFAULT_LOG_DIR = PROJECT_ROOT / "logs" / "interactions"


def _resolve_log_dir() -> Path:
    custom_dir = os.getenv("STORYWEAVER_LOG_DIR", "").strip()
    if custom_dir:
        return Path(custom_dir).expanduser()
    return DEFAULT_LOG_DIR


def create_session_metadata(session_id: str | None = None) -> dict[str, Any]:
    """
    创建新的会话元数据。

    每个会话对应一个单独的 JSONL 文件，便于回放和分析。
    """
    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    new_session_id = session_id or f"sw-{timestamp}-{uuid.uuid4().hex[:8]}"
    log_dir = _resolve_log_dir()
    log_path = log_dir / f"{new_session_id}.jsonl"
    return {
        "session_id": new_session_id,
        "turn_index": 0,
        "interaction_log_path": str(log_path),
    }


def ensure_session_metadata(game_session: dict[str, Any]) -> dict[str, Any]:
    """确保游戏会话中带有日志所需的元数据。"""
    if "session_id" not in game_session or "interaction_log_path" not in game_session:
        game_session.update(create_session_metadata())
    if "turn_index" not in game_session:
        game_session["turn_index"] = 0
    return game_session


def append_turn_log(game_session: dict[str, Any], record: dict[str, Any]) -> str:
    """
    追加一条结构化交互日志。

    Returns:
        日志文件路径，便于调试和脚本复用。
    """
    ensure_session_metadata(game_session)

    game_session["turn_index"] += 1
    log_path = Path(game_session["interaction_log_path"])
    log_path.parent.mkdir(parents=True, exist_ok=True)

    payload = {
        "timestamp": datetime.now().isoformat(timespec="seconds"),
        "session_id": game_session["session_id"],
        "turn_index": game_session["turn_index"],
        **record,
    }

    with log_path.open("a", encoding="utf-8") as fh:
        json.dump(payload, fh, ensure_ascii=False)
        fh.write("\n")

    return str(log_path)