ai-sessions.py

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///

"""解析 Codex 或 Claude 会话历史。"""

import argparse
import json
import os
import shutil
import sqlite3
import sys
import unicodedata
from dataclasses import dataclass, field
from datetime import date, datetime, time, timedelta
from pathlib import Path
from typing import Any, Iterable
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError


CODEX_DATABASE = Path.home() / '.codex' / 'thread_history_1.sqlite'
CLAUDE_DIRECTORY = Path.home() / '.claude'
DEFAULT_TIMEZONE = 'Asia/Shanghai'
CODEX_TOOL_TYPES = {
    'commandExecution': 'shell',
    'fileChange': 'fileChange',
    'imageView': 'imageView',
    'mcpToolCall': 'mcpToolCall',
    'webSearch': 'webSearch',
}
BLUE = '\033[34m'
FAINT = '\033[2m'
RESET = '\033[0m'


class HistoryError(RuntimeError):
    """表示会话历史无法解析。"""


@dataclass
class ConversationTurn:
    """表示一轮用户输入、工具调用和助手回答。"""

    question: str | None = None
    tools: list[str] = field(default_factory=list)
    answer: str | None = None

    def add_tool(self, name: str | None) -> None:
        if name and name not in self.tools:
            self.tools.append(name)


@dataclass
class SessionData:
    source: str
    session_id: str
    path: Path
    inputs: list[str] = field(default_factory=list)
    tools: list[str] = field(default_factory=list)
    final_output: str | None = None
    activity_dates: set[date] = field(default_factory=set)
    turns: list[ConversationTurn] = field(default_factory=list)
    started_at: datetime | None = None
    ended_at: datetime | None = None

    def refresh_summary(self) -> None:
        """根据对话轮次刷新会话级摘要字段。"""
        self.turns = [
            turn for turn in self.turns if turn.question or turn.tools or (turn.answer and turn.answer.strip())
        ]
        self.inputs = [turn.question for turn in self.turns if turn.question]
        self.tools = []
        for turn in self.turns:
            for tool_name in turn.tools:
                if tool_name not in self.tools:
                    self.tools.append(tool_name)
        self.final_output = next(
            (turn.answer for turn in reversed(self.turns) if turn.answer and turn.answer.strip()),
            None,
        )

    def matches_query(self, query: str | None) -> bool:
        """判断查询词是否出现在用户输入或最终输出中。"""
        if not query:
            return True

        normalized_query = query.casefold()
        answer_texts = [turn.answer or '' for turn in self.turns]
        searchable_texts = [*self.inputs, *answer_texts]
        return any(normalized_query in text.casefold() for text in searchable_texts)

    def add_activity_timestamp(self, timestamp: Any, display_timezone: ZoneInfo) -> None:
        """记录时间戳在显示时区对应的日期。"""
        activity_time = parse_timestamp(timestamp, display_timezone)
        if activity_time:
            self.activity_dates.add(activity_time.date())
            if self.started_at is None or activity_time < self.started_at:
                self.started_at = activity_time
            if self.ended_at is None or activity_time > self.ended_at:
                self.ended_at = activity_time


def get_display_timezone() -> ZoneInfo:
    """读取 TZ 环境变量，未设置时使用 Asia/Shanghai。"""
    timezone_name = os.environ.get('TZ') or DEFAULT_TIMEZONE
    try:
        return ZoneInfo(timezone_name)
    except ZoneInfoNotFoundError as error:
        raise HistoryError(f'无效的 TZ 时区：{timezone_name}') from error


def parse_timestamp(timestamp: Any, display_timezone: ZoneInfo) -> datetime | None:
    """将 Unix 毫秒或 ISO 时间戳转换为显示时区时间。"""
    if isinstance(timestamp, (int, float)):
        return datetime.fromtimestamp(timestamp / 1000, display_timezone)
    if not isinstance(timestamp, str) or not timestamp:
        return None

    try:
        parsed_timestamp = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
    except ValueError:
        return None
    if parsed_timestamp.tzinfo is None:
        parsed_timestamp = parsed_timestamp.replace(tzinfo=display_timezone)
    return parsed_timestamp.astimezone(display_timezone)


def get_today(display_timezone: ZoneInfo) -> date:
    """获取显示时区中的今天。"""
    return datetime.now(display_timezone).date()


def get_date_timestamp_bounds(target_date: date, display_timezone: ZoneInfo) -> tuple[int, int]:
    """获取显示时区指定日期的 Unix 毫秒边界。"""
    next_date = target_date + timedelta(days=1)
    start_timestamp = datetime.combine(target_date, time.min, tzinfo=display_timezone)
    end_timestamp = datetime.combine(next_date, time.min, tzinfo=display_timezone)
    return int(start_timestamp.timestamp() * 1000), int(end_timestamp.timestamp() * 1000)


def get_text(value: Any) -> str | None:
    """从字符串或消息内容块中提取可展示文本。"""
    if isinstance(value, str):
        return value
    if not isinstance(value, list):
        return None

    text_parts: list[str] = []
    for block in value:
        if isinstance(block, str):
            text_parts.append(block)
        elif isinstance(block, dict) and block.get('type') == 'text':
            text = block.get('text')
            if isinstance(text, str):
                text_parts.append(text)

    return '\n'.join(text_parts) or None


def get_tool_names(value: Any) -> Iterable[str]:
    """递归提取消息内容中的工具名称。"""
    if isinstance(value, dict):
        if value.get('type') == 'tool_use':
            name = value.get('name') or value.get('tool_name')
            if isinstance(name, str):
                yield name
        for nested_value in value.values():
            yield from get_tool_names(nested_value)
    elif isinstance(value, list):
        for nested_value in value:
            yield from get_tool_names(nested_value)


def connect_readonly(database_path: Path) -> sqlite3.Connection:
    """以只读方式打开 SQLite 历史数据库。"""
    if not database_path.is_file():
        raise HistoryError(f'找不到 Codex 历史数据库：{database_path}')

    try:
        return sqlite3.connect(f'file:{database_path}?mode=ro', uri=True)
    except sqlite3.Error as error:
        raise HistoryError(f'打开 Codex 历史数据库失败：{error}') from error


def resolve_codex_session(session_id: str, database_path: Path) -> tuple[str, Path]:
    """解析 Codex 会话 ID，并允许使用唯一前缀。"""
    with connect_readonly(database_path) as connection:
        rows = connection.execute(
            """
            SELECT DISTINCT thread_id
            FROM thread_items
            WHERE thread_id = ? OR thread_id LIKE ?
            ORDER BY thread_id
            """,
            (session_id, f'{session_id}%'),
        ).fetchall()

    if not rows:
        raise HistoryError(f'找不到 Codex 会话：{session_id}')
    if len(rows) > 1:
        matches = ', '.join(row[0] for row in rows)
        raise HistoryError(f'Codex 会话前缀不唯一：{matches}')
    return rows[0][0], database_path


def codex_session_exists(session_id: str, database_path: Path) -> bool:
    """判断 Codex 会话或唯一前缀是否存在。"""
    try:
        resolve_codex_session(session_id, database_path)
    except HistoryError:
        return False
    return True


def parse_codex(
    session_id: str,
    database_path: Path,
    display_timezone: ZoneInfo,
) -> SessionData:
    """解析 Codex SQLite 会话。"""
    resolved_session_id, resolved_path = resolve_codex_session(session_id, database_path)
    session = SessionData('codex', resolved_session_id, resolved_path)
    turns_by_id: dict[str, ConversationTurn] = {}

    with connect_readonly(database_path) as connection:
        rows = connection.execute(
            """
            SELECT item_type, item_json, created_at_ms, turn_id
            FROM thread_items
            WHERE thread_id = ?
            ORDER BY rollout_ordinal
            """,
            (resolved_session_id,),
        ).fetchall()

    for item_type, item_json, created_at_ms, turn_id in rows:
        session.add_activity_timestamp(created_at_ms, display_timezone)
        turn_key = str(turn_id)
        turn = turns_by_id.setdefault(turn_key, ConversationTurn())
        try:
            item = json.loads(item_json)
        except json.JSONDecodeError:
            continue
        if not isinstance(item, dict):
            continue

        if item_type == 'userMessage':
            content = get_text(item.get('content'))
            if content:
                turn.question = f'{turn.question}\n{content.strip()}' if turn.question else content.strip()
            continue

        tool_name = CODEX_TOOL_TYPES.get(item_type)
        if tool_name:
            if item_type == 'mcpToolCall' and isinstance(item.get('tool'), str):
                tool_name = item['tool']
            turn.add_tool(tool_name)

        if item_type != 'agentMessage':
            continue

        output = item.get('text')
        if not isinstance(output, str) or not output.strip():
            continue
        output = output.strip()
        if item.get('phase') == 'final_answer' or turn.answer is None:
            turn.answer = output

    session.turns = list(turns_by_id.values())
    session.refresh_summary()
    return session


def list_codex_session_ids(
    database_path: Path,
    display_timezone: ZoneInfo,
    target_date: date | None,
) -> list[str]:
    """列出 Codex 历史中的会话 ID。"""
    query = """
        SELECT thread_id
        FROM thread_items
    """
    query_parameters: tuple[int, ...] = ()
    if target_date:
        start_timestamp, end_timestamp = get_date_timestamp_bounds(target_date, display_timezone)
        query += 'WHERE created_at_ms >= ? AND created_at_ms < ?\n'
        query_parameters = (start_timestamp, end_timestamp)
    query += 'GROUP BY thread_id ORDER BY MAX(created_at_ms) DESC'

    with connect_readonly(database_path) as connection:
        rows = connection.execute(query, query_parameters).fetchall()
    return [row[0] for row in rows]


def claude_session_candidates(session_id: str, claude_directory: Path) -> list[Path]:
    """查找 Claude 会话文件。"""
    session_path = Path(session_id).expanduser()
    if session_path.is_file():
        return [session_path]

    candidates: list[Path] = []
    for search_directory in (
        claude_directory / 'projects',
        claude_directory / 'transcripts',
    ):
        if not search_directory.is_dir():
            continue
        candidates.extend(
            path
            for path in search_directory.rglob(f'{session_id}.jsonl')
            if path.is_file() and not path.name.endswith('.wakatime')
        )
    return candidates


def resolve_claude_session(session_id: str, claude_directory: Path) -> Path:
    """解析 Claude 会话文件路径。"""
    candidates = claude_session_candidates(session_id, claude_directory)
    if not candidates:
        raise HistoryError(f'找不到 Claude 会话：{session_id}')
    if len(candidates) > 1:
        matches = ', '.join(str(path) for path in candidates)
        raise HistoryError(f'Claude 会话对应多个文件：{matches}')
    return candidates[0]


def list_claude_session_paths(claude_directory: Path) -> list[Path]:
    """列出 Claude 项目和 transcript 中的会话文件。"""
    session_paths: set[Path] = set()
    for search_directory in (
        claude_directory / 'projects',
        claude_directory / 'transcripts',
    ):
        if not search_directory.is_dir():
            continue
        session_paths.update(
            path for path in search_directory.rglob('*.jsonl') if path.is_file() and not path.name.endswith('.wakatime')
        )
    return sorted(session_paths, key=lambda path: path.stat().st_mtime, reverse=True)


def is_real_user_input(item: dict[str, Any], content: str | None) -> bool:
    """过滤 Claude 的内部命令和工具返回消息。"""
    if not content or item.get('isMeta') is True:
        return False
    if isinstance(item.get('toolUseResult'), (dict, list, str)):
        return False
    return '<local-command-caveat>' not in content and '<command-name>' not in content


def parse_claude(
    session_id: str,
    claude_directory: Path,
    display_timezone: ZoneInfo,
    allow_empty: bool = False,
) -> SessionData:
    """解析 Claude JSONL 会话。"""
    session_path = resolve_claude_session(session_id, claude_directory)
    resolved_session_id = session_path.stem
    session = SessionData('claude', resolved_session_id, session_path)
    turns: list[ConversationTurn] = []
    current_turn: ConversationTurn | None = None

    try:
        lines = session_path.open(encoding='utf-8')
    except OSError as error:
        raise HistoryError(f'读取 Claude 会话失败：{session_path}：{error}') from error

    line_number = 0
    with lines:
        for line_number, line in enumerate(lines, start=1):
            try:
                item = json.loads(line)
            except json.JSONDecodeError:
                continue
            if not isinstance(item, dict):
                continue

            session.add_activity_timestamp(item.get('timestamp'), display_timezone)

            item_type = item.get('type')
            message = item.get('message')
            message_role = message.get('role') if isinstance(message, dict) else None
            content_value = message.get('content') if isinstance(message, dict) else item.get('content')
            content = get_text(content_value)

            if item_type == 'user' and message_role != 'assistant':
                if is_real_user_input(item, content):
                    if current_turn:
                        turns.append(current_turn)
                    current_turn = ConversationTurn(question=content.strip())
                    continue

            if current_turn is None:
                current_turn = ConversationTurn()
            for tool_name in get_tool_names(item):
                current_turn.add_tool(tool_name)
            if item_type == 'tool_use':
                current_turn.add_tool(item.get('tool_name'))
            if item_type == 'assistant' or message_role == 'assistant':
                if content and content.strip():
                    current_turn.answer = content.strip()

    if current_turn:
        turns.append(current_turn)
    session.turns = turns
    session.refresh_summary()
    if not allow_empty and not session.inputs and not session.tools and not session.final_output:
        raise HistoryError(f'Claude 会话没有可解析内容：{session_path}（最后检查行：{line_number}）')

    return session


def detect_source(session_id: str, claude_directory: Path) -> str:
    """根据会话 ID 和本地历史文件判断来源。"""
    if session_id.startswith('ses_') or session_id.endswith('.jsonl'):
        return 'claude'
    if Path(session_id).expanduser().is_file() or claude_session_candidates(session_id, claude_directory):
        return 'claude'
    return 'codex'


def print_labeled_text(
    label: str,
    content: str | None,
    indent_continuation: bool = True,
    separator: str = '>',
) -> None:
    """以紧凑标签格式输出多行文本。"""
    if not content or not content.strip():
        print(f'{label}{separator} （无）')
        return

    lines = content.strip().splitlines()
    print(f'{label}{separator} {lines[0]}')
    for line in lines[1:]:
        print(f'   {line}' if indent_continuation else line)


def use_terminal_color() -> bool:
    """判断当前输出是否适合使用 ANSI 颜色。"""
    return sys.stdout.isatty() and 'NO_COLOR' not in os.environ


def colorize(text: str, color: str, enabled: bool) -> str:
    """为文本添加 ANSI 颜色。"""
    if not enabled:
        return text
    return f'{color}{text}{RESET}'


def print_session_separator(use_color: bool, with_blank_lines: bool = True) -> None:
    """输出终端宽度的淡色会话分隔线。"""
    width = max(shutil.get_terminal_size((80, 20)).columns, 1)
    separator = '─' * width
    if with_blank_lines:
        print()
    print(colorize(separator, FAINT, use_color))
    if with_blank_lines:
        print()


def print_session_time(session: SessionData, display_timezone: ZoneInfo) -> None:
    """输出会话起止时间和显示时区。"""
    if session.started_at and session.ended_at:
        start_time = session.started_at.strftime('%Y-%m-%d %H:%M')
        end_time = session.ended_at.strftime('%Y-%m-%d %H:%M')
        print(f'Time: {start_time} ~ {end_time} (TZ={display_timezone.key})')
    else:
        print(f'Time: （无） (TZ={display_timezone.key})')


def print_session(
    session: SessionData,
    display_timezone: ZoneInfo,
    use_color: bool | None = None,
) -> None:
    """以紧凑格式输出会话正文。"""
    if use_color is None:
        use_color = use_terminal_color()
    colored_session_id = colorize(session.session_id, BLUE, use_color)
    print(f'[{session.source}] UUID {colored_session_id}')
    print_session_time(session, display_timezone)
    print()
    if not session.turns:
        print(' Q: （无）')
        print_labeled_text(' A:', session.final_output, indent_continuation=False, separator='')
        return

    label_width = len(f'Q{len(session.turns)}:') + 1
    for index, turn in enumerate(session.turns, start=1):
        question_label = f'Q{index}:'.rjust(label_width)
        print_labeled_text(question_label, turn.question, separator='')
        if turn.tools:
            tool_text = ', '.join(turn.tools)
            tool_label = f'T{index}:'.rjust(label_width)
            print(f'{tool_label} {tool_text}')

    answer_index = get_last_answer_index(session)
    answer_label = f'A{answer_index}:'.rjust(label_width)
    answer_prefix = f'{answer_label} '
    answer_text = shorten_to_terminal_width(session.final_output, answer_prefix)
    print_labeled_text(
        answer_label,
        answer_text,
        indent_continuation=False,
        separator='',
    )


def load_all_sessions(
    source: str,
    codex_database: Path,
    claude_directory: Path,
    display_timezone: ZoneInfo,
    target_date: date | None,
) -> list[SessionData]:
    """加载指定来源的全部可解析会话。"""
    sessions: list[SessionData] = []

    if source in ('auto', 'codex'):
        session_ids = list_codex_session_ids(codex_database, display_timezone, target_date)
        for session_id in session_ids:
            sessions.append(parse_codex(session_id, codex_database, display_timezone))

    if source in ('auto', 'claude'):
        for session_path in list_claude_session_paths(claude_directory):
            try:
                session = parse_claude(
                    str(session_path),
                    claude_directory,
                    display_timezone,
                    allow_empty=True,
                )
                if (target_date is None or target_date in session.activity_dates) and (
                    session.inputs or session.tools or session.final_output
                ):
                    sessions.append(session)
            except HistoryError as error:
                print(f'警告：跳过 Claude 会话：{error}', file=sys.stderr)

    if not sessions:
        raise HistoryError('没有找到可解析的会话')
    return sessions


def get_display_width(text: str) -> int:
    """计算文本在终端中的显示列数。"""
    display_width = 0
    for character in text:
        if unicodedata.combining(character):
            continue
        display_width += 2 if unicodedata.east_asian_width(character) in ('W', 'F') else 1
    return display_width


def get_last_answer_index(session: SessionData) -> int:
    """获取最后一个助手回答所在的轮次编号。"""
    return next(
        (index for index, turn in reversed(list(enumerate(session.turns, start=1))) if turn.answer),
        max(len(session.turns), 1),
    )


def shorten_to_terminal_width(content: str | None, prefix: str) -> str:
    """按终端宽度将文本压成不会换行的单行。"""
    single_line = ' '.join(content.split()) if content else '（无）'
    terminal_width = max(shutil.get_terminal_size((80, 20)).columns, 1)
    available_width = max(terminal_width - get_display_width(prefix) - 1, 1)
    if get_display_width(single_line) <= available_width:
        return single_line

    truncated_characters: list[str] = []
    current_width = 0
    target_width = max(available_width - 1, 0)
    for character in single_line:
        character_width = get_display_width(character)
        if current_width + character_width > target_width:
            break
        truncated_characters.append(character)
        current_width += character_width
    return ''.join(truncated_characters) + '…'


def print_session_index(
    sessions: list[SessionData],
    display_timezone: ZoneInfo,
) -> None:
    """输出会话索引，不展开全部正文。"""
    use_color = use_terminal_color()
    sorted_sessions = sorted(
        sessions,
        key=lambda session: session.ended_at.timestamp() if session.ended_at else float('-inf'),
    )
    for index, session in enumerate(sorted_sessions):
        if index:
            print_session_separator(use_color, with_blank_lines=False)
        first_input = session.inputs[0] if session.inputs else None
        colored_session_id = colorize(session.session_id, BLUE, use_color)
        question_index = next(
            (turn_index for turn_index, turn in enumerate(session.turns, start=1) if turn.question),
            1,
        )
        answer_index = get_last_answer_index(session)
        label_width = len(f'Q{max(len(session.turns), 1)}:') + 1
        question_label = f'Q{question_index}:'.rjust(label_width)
        answer_label = f'A{answer_index}:'.rjust(label_width)
        question_prefix = f'{question_label} '
        answer_prefix = f'{answer_label} '
        print(f'[{session.source}] {colored_session_id}')
        print_session_time(session, display_timezone)
        print(f'{question_prefix}{shorten_to_terminal_width(first_input, question_prefix)}')
        print(f'{answer_prefix}{shorten_to_terminal_width(session.final_output, answer_prefix)}')


def print_matching_sessions(
    sessions: list[SessionData],
    query: str,
    display_timezone: ZoneInfo,
    target_date: date | None,
) -> None:
    """输出匹配查询词的完整会话。"""
    matching_sessions = [session for session in sessions if session.matches_query(query)]
    if not matching_sessions:
        raise HistoryError(f'没有找到请求或最终响应包含“{query}”的会话')

    matching_sessions.sort(key=lambda session: session.ended_at.timestamp() if session.ended_at else float('-inf'))
    use_color = use_terminal_color()
    scope = str(target_date) if target_date else '全部日期'
    print(f'范围={scope} TZ={display_timezone.key} | 匹配={len(matching_sessions)}')
    for index, session in enumerate(matching_sessions):
        if index:
            print_session_separator(use_color)
        print_session(session, display_timezone, use_color)


def build_parser() -> argparse.ArgumentParser:
    """创建命令行参数解析器。"""
    parser = argparse.ArgumentParser(
        description='解析 Codex 或 Claude 会话历史，输出输入、工具调用和最终输出。',
        epilog='示例：ai-sessions tantivy；ai-sessions 019... tantivy；ai-sessions --source claude',
    )
    parser.add_argument(
        'session_or_query',
        nargs='?',
        metavar='SESSION|QUERY',
        help='会话 UUID 或查询词；普通文本会自动当作查询词',
    )
    parser.add_argument(
        'query_text',
        nargs='?',
        metavar='QUERY',
        help='指定会话后的查询词',
    )
    parser.add_argument(
        '-q',
        '--query',
        dest='query',
        help='按请求或最终响应文本过滤会话，不区分大小写',
    )
    parser.add_argument(
        '--source',
        choices=('auto', 'codex', 'claude'),
        default='auto',
        help='会话来源，默认按 ID 自动判断',
    )
    parser.add_argument(
        '-a',
        '--all',
        action='store_true',
        help='扫描全部日期的会话，默认只扫描 TZ 对应的今天',
    )
    parser.add_argument(
        '-d',
        '--date',
        dest='date_text',
        metavar='YYYY-MM-DD',
        help='按 TZ 时区的指定日期过滤，默认今天',
    )
    parser.add_argument(
        '--codex-db',
        type=Path,
        default=CODEX_DATABASE,
        help=f'Codex 历史数据库（默认：{CODEX_DATABASE}）',
    )
    parser.add_argument(
        '--claude-dir',
        type=Path,
        default=CLAUDE_DIRECTORY,
        help=f'Claude 数据目录（默认：{CLAUDE_DIRECTORY}）',
    )
    return parser


def get_target_date(arguments: argparse.Namespace, display_timezone: ZoneInfo) -> date | None:
    """解析日期参数，返回按 TZ 过滤的目标日期。"""
    if arguments.all and arguments.date_text:
        raise HistoryError('不能同时使用 --all 和 --date')
    if arguments.all:
        return None
    if not arguments.date_text:
        return get_today(display_timezone)
    try:
        return date.fromisoformat(arguments.date_text)
    except ValueError as error:
        raise HistoryError(f'日期格式必须是 YYYY-MM-DD：{arguments.date_text}') from error


def normalize_arguments(arguments: argparse.Namespace) -> tuple[str | None, str | None]:
    """将简短的位置参数形式转换为会话 ID 和查询词。"""
    first_value = arguments.session_or_query
    second_value = arguments.query_text
    option_query = arguments.query

    if second_value and option_query:
        raise HistoryError('不能同时使用位置查询词和 --query')
    if not first_value:
        return None, option_query

    if second_value:
        return first_value, second_value

    source = arguments.source
    is_claude_session = bool(claude_session_candidates(first_value, arguments.claude_dir))
    if source == 'claude' or first_value.startswith('ses_') or first_value.endswith('.jsonl'):
        is_session = is_claude_session or first_value.startswith('ses_') or first_value.endswith('.jsonl')
    elif source == 'codex':
        is_session = codex_session_exists(first_value, arguments.codex_db)
    else:
        is_session = is_claude_session or codex_session_exists(first_value, arguments.codex_db)

    if is_session:
        return first_value, option_query
    if option_query:
        raise HistoryError('第一个位置参数不是会话 ID，不能同时作为查询词使用 --query')
    return None, first_value


def main() -> int:
    """解析参数并输出会话。"""
    arguments = build_parser().parse_args()

    try:
        display_timezone = get_display_timezone()
        target_date = get_target_date(arguments, display_timezone)
        session_id, query = normalize_arguments(arguments)
        if session_id:
            source = arguments.source if arguments.source != 'auto' else detect_source(session_id, arguments.claude_dir)
            if source == 'codex':
                session = parse_codex(session_id, arguments.codex_db, display_timezone)
            else:
                session = parse_claude(session_id, arguments.claude_dir, display_timezone)
            if not session.matches_query(query):
                raise HistoryError(f'会话不匹配查询：{query}')
            print_session(session, display_timezone)
            return 0

        sessions = load_all_sessions(
            arguments.source,
            arguments.codex_db,
            arguments.claude_dir,
            display_timezone,
            target_date,
        )
        if query:
            print_matching_sessions(
                sessions,
                query,
                display_timezone,
                target_date,
            )
        else:
            print_session_index(
                sessions,
                display_timezone,
            )
    except HistoryError as error:
        print(f'错误：{error}', file=sys.stderr)
        return 1

    return 0


if __name__ == '__main__':
    raise SystemExit(main())

ai-sessions -h
usage: ai-sessions [-h] [-q QUERY] [--source {auto,codex,claude}] [-a] [-d YYYY-MM-DD] [--codex-db CODEX_DB]
                   [--claude-dir CLAUDE_DIR]
                   [SESSION|QUERY] [QUERY]

解析 Codex 或 Claude 会话历史，输出输入、工具调用和最终输出。

positional arguments:
  SESSION|QUERY         会话 UUID 或查询词；普通文本会自动当作查询词
  QUERY                 指定会话后的查询词

options:
  -h, --help            show this help message and exit
  -q QUERY, --query QUERY
                        按请求或最终响应文本过滤会话，不区分大小写
  --source {auto,codex,claude}
                        会话来源，默认按 ID 自动判断
  -a, --all             扫描全部日期的会话，默认只扫描 TZ 对应的今天
  -d YYYY-MM-DD, --date YYYY-MM-DD
                        按 TZ 时区的指定日期过滤，默认今天
  --codex-db CODEX_DB   Codex 历史数据库（默认：/Users/garden/.codex/thread_history_1.sqlite）
  --claude-dir CLAUDE_DIR
                        Claude 数据目录（默认：/Users/garden/.claude）

示例：ai-sessions tantivy；ai-sessions 019... tantivy；ai-sessions --source claude