#!/usr/bin/env python3
"""
Анализатор сессий Claude Code
Парсит history.jsonl и классифицирует сессии по проектам/задачам
"""
import json
import os
from datetime import datetime
from collections import defaultdict
from pathlib import Path
CLAUDE_DIR = Path.home() / '.claude'
HISTORY_FILE = CLAUDE_DIR / "history.jsonl"
PROJECTS_DIR = CLAUDE_DIR / "projects/-opt-claude-workspace"
OUTPUT_DIR = CLAUDE_DIR / "sessions/reconstructed"
# Ключевые слова для определения проекта
PROJECT_KEYWORDS = {
"pirotehnika": [
"pirotehnika", "пиротехник", "ozon", "озон", "nocodb",
"sync", "товар", "заказ", "транзакц", "аналитик",
"1с", "1c", "odata", "pim", "маркетплейс", "marketplace"
],
"lider": [
"lider", "лидер", "bazon", "базон", "cscart", "cs-cart",
"авто", "auto", "запчаст", "совместимост", "oem",
"агрегат", "узел", "деталь", "каталог"
],
"platform": [
"platform", "платформ", "architect", "архитектор",
"проектор", "projector", "агент", "agent", "терминал",
"методолог", "стандарт", "тикет", "ticket", "сессия"
],
"marketplace": [
"marketplace mvp", "streamlit", "доставк", "delivery",
"массов", "bulk", "экспорт"
],
"content-factory": [
"контент завод", "content factory", "ролик", "видео"
]
}
# Ключевые слова для определения агента
AGENT_KEYWORDS = {
"Архитектор": ["архитектор", "architect", "методолог", "стандарт", "теория"],
"Проектор": ["проектор", "projector", "проект", "задач"],
"Кодер": ["код", "code", "функци", "скрипт", "парс", "api"],
"Инфра": ["сервер", "nginx", "docker", "deploy", "ssh"],
"Аналитик": ["аналит", "отчёт", "данны", "статистик"]
}
def load_history():
"""Загрузка history.jsonl"""
sessions = defaultdict(list)
with open(HISTORY_FILE, 'r') as f:
for line in f:
try:
entry = json.loads(line.strip())
session_id = entry.get('sessionId')
if session_id and session_id != 'null':
sessions[session_id].append(entry)
except json.JSONDecodeError:
continue
return sessions
def classify_text(text, keywords_dict):
"""Классифицировать текст по ключевым словам"""
text_lower = text.lower()
scores = defaultdict(int)
for category, keywords in keywords_dict.items():
for keyword in keywords:
if keyword in text_lower:
scores[category] += 1
if scores:
return max(scores, key=scores.get), max(scores.values())
return None, 0
def analyze_session(session_id, messages):
"""Анализ одной сессии"""
all_text = " ".join([m.get('display', '') for m in messages])
# Определяем проект
project, proj_score = classify_text(all_text, PROJECT_KEYWORDS)
# Определяем агента
agent, agent_score = classify_text(all_text, AGENT_KEYWORDS)
# Временные рамки
timestamps = [m.get('timestamp', 0) for m in messages]
start_ts = min(timestamps) / 1000 if timestamps else 0
end_ts = max(timestamps) / 1000 if timestamps else 0
start_dt = datetime.fromtimestamp(start_ts) if start_ts else None
end_dt = datetime.fromtimestamp(end_ts) if end_ts else None
# Первые сообщения (для понимания контекста)
first_messages = [m.get('display', '')[:100] for m in messages[:5]]
return {
'session_id': session_id,
'project': project or 'unknown',
'project_confidence': proj_score,
'agent': agent or 'unknown',
'agent_confidence': agent_score,
'message_count': len(messages),
'start': start_dt.isoformat() if start_dt else None,
'end': end_dt.isoformat() if end_dt else None,
'duration_hours': round((end_ts - start_ts) / 3600, 2) if start_ts and end_ts else 0,
'first_messages': first_messages
}
def generate_report(sessions_data):
"""Генерация отчёта"""
# Группировка по проектам
by_project = defaultdict(list)
for s in sessions_data:
by_project[s['project']].append(s)
report = []
report.append("# Реконструкция сессий Claude Code\n")
report.append(f"**Дата анализа:** {datetime.now().isoformat()}\n")
report.append(f"**Всего сессий:** {len(sessions_data)}\n")
report.append(f"**Период:** {min(s['start'] for s in sessions_data if s['start'])} — {max(s['end'] for s in sessions_data if s['end'])}\n")
report.append("\n---\n")
# Сводка по проектам
report.append("## Сводка по проектам\n")
report.append("| Проект | Сессий | Сообщений | Часов |\n")
report.append("|--------|--------|-----------|-------|\n")
for project in sorted(by_project.keys()):
sessions = by_project[project]
total_msgs = sum(s['message_count'] for s in sessions)
total_hours = sum(s['duration_hours'] for s in sessions)
report.append(f"| {project} | {len(sessions)} | {total_msgs} | {total_hours:.1f} |\n")
report.append("\n---\n")
# Детали по каждому проекту
for project in sorted(by_project.keys()):
report.append(f"## Проект: {project}\n\n")
sessions = sorted(by_project[project], key=lambda x: x['start'] or '', reverse=True)
for s in sessions[:10]: # Топ 10 сессий
report.append(f"### Сессия {s['session_id'][:8]}\n")
report.append(f"- **Период:** {s['start']} — {s['end']}\n")
report.append(f"- **Агент:** {s['agent']} (уверенность: {s['agent_confidence']})\n")
report.append(f"- **Сообщений:** {s['message_count']}\n")
report.append(f"- **Длительность:** {s['duration_hours']} ч\n")
report.append(f"- **Первые сообщения:**\n")
for msg in s['first_messages'][:3]:
report.append(f" - {msg[:80]}...\n")
report.append("\n")
return "".join(report)
def save_session_files(sessions_data):
"""Сохранение отдельных файлов сессий"""
os.makedirs(OUTPUT_DIR, exist_ok=True)
# Группировка по дате
by_date = defaultdict(list)
for s in sessions_data:
if s['start']:
date = s['start'][:10]
by_date[date].append(s)
for date, sessions in by_date.items():
filename = f"{OUTPUT_DIR}/{date}.md"
content = [f"# Сессии за {date}\n\n"]
for s in sorted(sessions, key=lambda x: x['start'] or ''):
content.append(f"## {s['start'][11:16]} | {s['project']} | {s['agent']}\n")
content.append(f"**UUID:** `{s['session_id'][:8]}`\n")
content.append(f"**Сообщений:** {s['message_count']}\n\n")
content.append("**Начало диалога:**\n")
for msg in s['first_messages'][:3]:
content.append(f"> {msg[:100]}\n")
content.append("\n---\n\n")
with open(filename, 'w') as f:
f.write("".join(content))
print(f"Сохранено {len(by_date)} файлов в {OUTPUT_DIR}")
def main():
print("Загрузка history.jsonl...")
sessions = load_history()
print(f"Найдено {len(sessions)} сессий")
print("Анализ сессий...")
sessions_data = []
for session_id, messages in sessions.items():
analysis = analyze_session(session_id, messages)
sessions_data.append(analysis)
print("Генерация отчёта...")
report = generate_report(sessions_data)
# Сохранение общего отчёта
report_file = CLAUDE_DIR / "sessions/SESSIONS_REPORT.md"
with open(report_file, 'w') as f:
f.write(report)
print(f"Отчёт сохранён: {report_file}")
# Сохранение JSON для дальнейшей обработки
json_file = CLAUDE_DIR / "sessions/sessions_data.json"
with open(json_file, 'w') as f:
json.dump(sessions_data, f, indent=2, ensure_ascii=False, default=str)
print(f"JSON сохранён: {json_file}")
# Сохранение файлов по датам
save_session_files(sessions_data)
# Вывод сводки
print("\n=== СВОДКА ===")
by_project = defaultdict(int)
for s in sessions_data:
by_project[s['project']] += 1
for project, count in sorted(by_project.items(), key=lambda x: -x[1]):
print(f" {project}: {count} сессий")
if __name__ == "__main__":
main()