#!/usr/bin/env python3
"""
Session Recovery — восстановление оборванных сессий Claude Code
Показывает последние сессии с контекстом для продолжения работы.
Использование:
python3 session_recovery.py # Последние 5 сессий
python3 session_recovery.py --days 3 # За последние 3 дня
python3 session_recovery.py --session UUID # Детали конкретной сессии
python3 session_recovery.py --project piro # Фильтр по проекту
"""
import json
import os
import re
import sys
from pathlib import Path
from datetime import datetime, timedelta
from collections import defaultdict
from typing import Dict, List, Any, Optional
CLAUDE_DIR = Path.home() / '.claude'
PROJECTS_DIR = CLAUDE_DIR / 'projects' / '-opt-claude-workspace'
HISTORY_FILE = CLAUDE_DIR / 'history.jsonl'
def ts_to_dt(ts: int) -> datetime:
"""Timestamp (ms) -> datetime"""
return datetime.fromtimestamp(ts / 1000)
def classify_project(messages: List[Dict]) -> str:
"""Определить проект по содержимому сообщений"""
text = ' '.join([m.get('display', '') for m in messages]).lower()
if any(x in text for x in ['pirotehnika', 'пиротехник', 'ozon', 'озон', 'фейерверк', 'pim', '1c', '1с']):
return 'pirotehnika'
elif any(x in text for x in ['lider', 'лидер', 'cs-cart', 'авто', 'запчаст']):
return 'lider'
elif any(x in text for x in ['marketplace', 'маркетплейс', 'streamlit', 'mvp']):
return 'marketplace'
elif any(x in text for x in ['platform', 'платформ', 'architect', 'архитектор', 'claude.md', 'agent', 'агент']):
return 'platform'
elif any(x in text for x in ['infra', 'сервер', 'backup', 'nginx', 'docker', 'deploy']):
return 'infra'
else:
return 'unknown'
def classify_agent(messages: List[Dict]) -> str:
"""Определить агента по содержимому"""
text = ' '.join([m.get('display', '') for m in messages]).lower()
if any(x in text for x in ['архитектор', 'architect', 'стандарт', 'паттерн', 'теория']):
return 'Архитектор'
elif any(x in text for x in ['проектор', 'projector', 'задач', 'план']):
return 'Проектор'
elif any(x in text for x in ['кодер', 'coder', 'код', 'функци', 'класс', 'def ', 'async']):
return 'Кодер'
elif any(x in text for x in ['инфра', 'infra', 'сервер', 'deploy', 'nginx', 'docker']):
return 'Инфра'
else:
return 'Свободная работа'
def load_sessions() -> Dict[str, Dict]:
"""Загрузить все сессии из history.jsonl"""
sessions = {}
if not HISTORY_FILE.exists():
return sessions
with open(HISTORY_FILE, 'r') as f:
for line in f:
try:
entry = json.loads(line.strip())
sid = entry.get('sessionId', 'unknown')
if sid not in sessions:
sessions[sid] = {
'id': sid,
'messages': [],
'first_ts': entry.get('timestamp'),
'last_ts': entry.get('timestamp'),
}
sessions[sid]['messages'].append(entry)
sessions[sid]['last_ts'] = entry.get('timestamp')
except json.JSONDecodeError:
pass # Malformed JSON line
# Обогатить классификацией
for sid, data in sessions.items():
data['project'] = classify_project(data['messages'])
data['agent'] = classify_agent(data['messages'])
data['msg_count'] = len(data['messages'])
data['first_dt'] = ts_to_dt(data['first_ts'])
data['last_dt'] = ts_to_dt(data['last_ts'])
data['duration_h'] = (data['last_ts'] - data['first_ts']) / 3600000
# Первые сообщения для контекста
data['preview'] = [m.get('display', '')[:100] for m in data['messages'][:5]]
return sessions
def get_session_details(session_id: str) -> Optional[Dict]:
"""Загрузить полные детали сессии"""
session_file = PROJECTS_DIR / f"{session_id}.jsonl"
if not session_file.exists():
return None
messages = []
with open(session_file, 'r') as f:
for line in f:
try:
messages.append(json.loads(line.strip()))
except json.JSONDecodeError:
pass # Malformed JSON line
# Извлечь ключевую информацию
result = {
'id': session_id,
'total_messages': len(messages),
'human_messages': [],
'files_mentioned': set(),
'commands_run': [],
'last_context': [],
}
for msg in messages:
msg_type = msg.get('type')
if msg_type == 'human':
content = msg.get('message', {}).get('content', '')
if isinstance(content, str) and content.strip():
result['human_messages'].append(content[:200])
elif msg_type == 'assistant':
content = msg.get('message', {}).get('content', [])
if isinstance(content, list):
for item in content:
if isinstance(item, dict):
# Файлы
if item.get('name') in ['Read', 'Write', 'Edit']:
fp = item.get('input', {}).get('file_path', '')
if fp:
result['files_mentioned'].add(fp)
# Команды
elif item.get('name') == 'Bash':
cmd = item.get('input', {}).get('command', '')
if cmd:
result['commands_run'].append(cmd[:100])
result['files_mentioned'] = list(result['files_mentioned'])
result['last_context'] = result['human_messages'][-10:]
return result
def print_session_card(session: Dict, detailed: bool = False):
"""Напечатать карточку сессии"""
print(f"\n{'='*70}")
print(f"🔹 Сессия: {session['id'][:8]}...")
print(f" Проект: {session['project']} | Агент: {session['agent']}")
print(f" Время: {session['first_dt'].strftime('%Y-%m-%d %H:%M')} — {session['last_dt'].strftime('%H:%M')}")
print(f" Сообщений: {session['msg_count']} | Длительность: {session['duration_h']:.1f}ч")
print(f"\n 📝 Контекст:")
for i, preview in enumerate(session['preview'][:3]):
clean = preview.replace('\n', ' ').strip()
if clean:
print(f" {i+1}. {clean[:80]}{'...' if len(clean) > 80 else ''}")
if detailed:
details = get_session_details(session['id'])
if details:
print(f"\n 📁 Файлы ({len(details['files_mentioned'])}):")
for f in details['files_mentioned'][:5]:
print(f" - {f}")
print(f"\n 💬 Последние запросы:")
for msg in details['last_context'][-5:]:
clean = msg.replace('\n', ' ').strip()
print(f" • {clean[:100]}...")
def print_recovery_dashboard(sessions: Dict[str, Dict], days: int = 7, project_filter: str = None):
"""Напечатать дашборд для восстановления"""
cutoff = datetime.now() - timedelta(days=days)
# Фильтрация
recent = []
for sid, data in sessions.items():
if data['last_dt'] < cutoff:
continue
if project_filter and project_filter.lower() not in data['project'].lower():
continue
recent.append(data)
# Сортировка по времени (последние сверху)
recent.sort(key=lambda x: x['last_ts'], reverse=True)
print("\n" + "="*70)
print(" ВОССТАНОВЛЕНИЕ СЕССИЙ CLAUDE CODE")
print("="*70)
print(f"\nПериод: последние {days} дней | Найдено: {len(recent)} сессий")
if project_filter:
print(f"Фильтр: {project_filter}")
# Группировка по проектам
by_project = defaultdict(list)
for s in recent:
by_project[s['project']].append(s)
print("\n📊 По проектам:")
for proj, sess in sorted(by_project.items(), key=lambda x: -len(x[1])):
total_h = sum(s['duration_h'] for s in sess)
print(f" {proj}: {len(sess)} сессий, {total_h:.1f}ч")
# Последние 5 сессий
print("\n" + "-"*70)
print("📋 ПОСЛЕДНИЕ СЕССИИ (для продолжения):")
for i, session in enumerate(recent[:5]):
print_session_card(session, detailed=(i == 0))
# Команда для продолжения
if recent:
last = recent[0]
print("\n" + "-"*70)
print("💡 ДЛЯ ПРОДОЛЖЕНИЯ ПОСЛЕДНЕЙ СЕССИИ:")
print(f"\n Проект: {last['project']}")
print(f" Агент: {last['agent']}")
print(f" Контекст: {last['preview'][0][:60] if last['preview'] else 'нет'}...")
print(f"\n Скажите: 'продолжить {last['project']}'")
print(f" Или: 'что делали вчера?'")
def main():
import argparse
parser = argparse.ArgumentParser(description='Session Recovery')
parser.add_argument('--days', type=int, default=7, help='Период в днях')
parser.add_argument('--session', type=str, help='UUID конкретной сессии')
parser.add_argument('--project', type=str, help='Фильтр по проекту')
parser.add_argument('--json', action='store_true', help='Вывод в JSON')
args = parser.parse_args()
sessions = load_sessions()
if args.session:
# Детали конкретной сессии
if args.session in sessions:
print_session_card(sessions[args.session], detailed=True)
else:
# Поиск по частичному ID
matches = [s for sid, s in sessions.items() if args.session in sid]
if matches:
for m in matches:
print_session_card(m, detailed=True)
else:
print(f"❌ Сессия {args.session} не найдена")
elif args.json:
# JSON вывод для интеграции
cutoff = datetime.now() - timedelta(days=args.days)
recent = [
{
'id': s['id'],
'project': s['project'],
'agent': s['agent'],
'last': s['last_dt'].isoformat(),
'messages': s['msg_count'],
'preview': s['preview'][:2],
}
for s in sessions.values()
if s['last_dt'] >= cutoff
]
recent.sort(key=lambda x: x['last'], reverse=True)
print(json.dumps(recent[:10], indent=2, ensure_ascii=False))
else:
# Дашборд
print_recovery_dashboard(sessions, args.days, args.project)
if __name__ == '__main__':
main()