#!/usr/bin/env python3
"""
Индексация сессий Claude — извлечение недоделок по проектам.
DEPRECATED: Используйте library.services.session.SessionManager
Использование:
python3 session_index.py index # Индексировать новые сессии
python3 session_index.py show # Показать недоделки
python3 session_index.py full # Полная переиндексация
python3 session_index.py stats # Статистика
Новый способ:
python3 -m library.services.session unfinished
"""
# Делегируем в SessionManager
try:
from library.services.session import SessionManager
USE_NEW_MANAGER = True
except ImportError:
USE_NEW_MANAGER = False
import json
import sys
from pathlib import Path
from datetime import datetime
from collections import defaultdict
CLAUDE_DIR = Path.home() / '.claude'
PROJECTS_DIR = CLAUDE_DIR / 'projects' / '-opt-claude-workspace'
INDEX_FILE = CLAUDE_DIR / 'tasks_index.json'
CHECKPOINT_FILE = CLAUDE_DIR / 'index_checkpoint'
def load_index() -> dict:
"""Загрузить существующий индекс"""
if INDEX_FILE.exists():
with open(INDEX_FILE, 'r') as f:
return json.load(f)
return {
"last_indexed": None,
"projects": {},
"total_pending": 0
}
def save_index(index: dict):
"""Сохранить индекс"""
with open(INDEX_FILE, 'w') as f:
json.dump(index, f, indent=2, ensure_ascii=False)
def get_checkpoint() -> datetime | None:
"""Получить точку последней индексации"""
if CHECKPOINT_FILE.exists():
ts = CHECKPOINT_FILE.read_text().strip()
try:
return datetime.fromisoformat(ts)
except ValueError:
pass # Invalid datetime format
return None
def save_checkpoint():
"""Сохранить точку индексации"""
CHECKPOINT_FILE.write_text(datetime.now().isoformat())
def classify_project(files: set, user_messages: list) -> tuple[str, str]:
"""Определить проект и подпроект по файлам"""
project = "other"
subproject = "general"
# По файлам
for f in files:
if 'pirotehnika' in f:
project = "pirotehnika"
if 'ozon' in f or 'marketplaces' in f:
subproject = "ozon"
elif 'crm' in f.lower() or 'client' in f.lower():
subproject = "crm"
elif 'mp1' in f or 'service' in f:
subproject = "mp1"
elif 'pim' in f:
subproject = "pim"
elif 'site' in f:
subproject = "site"
break
elif 'lideravto' in f:
project = "lideravto"
if 'site' in f:
subproject = "site"
elif 'crm' in f.lower():
subproject = "crm"
break
elif 'architect' in f:
project = "platform"
subproject = "architect"
break
elif 'system' in f:
project = "platform"
subproject = "system"
break
return project, subproject
def extract_session_data(session_file: Path) -> dict:
"""Извлечь данные из одной сессии"""
messages = []
with open(session_file, 'r') as f:
for line in f:
try:
msg = json.loads(line.strip())
if msg.get('type') in ('user', 'assistant'):
messages.append(msg)
except json.JSONDecodeError:
pass # Malformed JSON line
user_messages = []
files_changed = set()
todos_pending = []
todos_completed = []
for msg in messages:
msg_type = msg.get('type')
if msg_type == 'user':
content = msg.get('message', {}).get('content', '')
if isinstance(content, str) and len(content) > 3:
user_messages.append(content.strip()[:200])
elif msg_type == 'assistant':
content = msg.get('message', {}).get('content', [])
if isinstance(content, list):
for item in content:
if isinstance(item, dict):
tool_name = item.get('name', '')
tool_input = item.get('input', {})
if tool_name in ('Write', 'Edit'):
fp = tool_input.get('file_path', '')
if fp:
files_changed.add(fp)
elif tool_name == 'TodoWrite':
for t in tool_input.get('todos', []):
if isinstance(t, str):
todos_pending.append(t)
elif isinstance(t, dict):
content = t.get('content', '')
status = t.get('status', 'pending')
if content:
if status == 'completed':
todos_completed.append(content)
else:
todos_pending.append(content)
# Убираем выполненные из pending
pending_final = [t for t in todos_pending if t not in todos_completed]
# Убираем дубликаты
pending_final = list(dict.fromkeys(pending_final))
return {
'session_id': session_file.stem[:16],
'files_changed': files_changed,
'user_messages': user_messages,
'todos_pending': pending_final,
'todos_completed': list(set(todos_completed)),
'message_count': len(messages)
}
def determine_priority(task: str) -> str:
"""Определить приоритет задачи"""
task_lower = task.lower()
if any(w in task_lower for w in ['срочно', 'urgent', 'отменить', 'исправить ошибк']):
return 'urgent'
if any(w in task_lower for w in ['создать', 'добавить', 'реализовать', 'high']):
return 'high'
if any(w in task_lower for w in ['загрузить', 'синхрониз', 'обновить']):
return 'medium'
return 'low'
def index_sessions(full: bool = False):
"""Индексировать сессии"""
checkpoint = None if full else get_checkpoint()
index = {"last_indexed": None, "projects": {}, "total_pending": 0} if full else load_index()
# Получаем все файлы сессий
session_files = sorted(PROJECTS_DIR.glob("*.jsonl"), key=lambda x: x.stat().st_mtime)
if checkpoint:
# Фильтруем только новые
session_files = [f for f in session_files
if datetime.fromtimestamp(f.stat().st_mtime) > checkpoint]
if not session_files:
print(f"Новых сессий нет (после {checkpoint})")
return index
print(f"Индексирую {len(session_files)} сессий...")
for sf in session_files:
data = extract_session_data(sf)
if not data['todos_pending']:
continue
project, subproject = classify_project(data['files_changed'], data['user_messages'])
# Инициализируем структуру
if project not in index['projects']:
index['projects'][project] = {}
if subproject not in index['projects'][project]:
index['projects'][project][subproject] = []
# Добавляем задачи
existing_tasks = {t['task'] for t in index['projects'][project][subproject]}
for task in data['todos_pending']:
if task not in existing_tasks:
index['projects'][project][subproject].append({
'task': task,
'session': data['session_id'],
'priority': determine_priority(task)
})
# Считаем total
total = 0
for proj in index['projects'].values():
for tasks in proj.values():
total += len(tasks)
index['total_pending'] = total
index['last_indexed'] = datetime.now().isoformat()
save_index(index)
save_checkpoint()
print(f"Готово. Всего недоделок: {total}")
return index
def show_tasks(project_filter: str = None):
"""Показать недоделки"""
index = load_index()
if not index['projects']:
print("Индекс пуст. Запустите: python3 session_index.py index")
return
print(f"НЕДОДЕЛКИ ПО ПРОЕКТАМ (всего: {index['total_pending']})")
print(f"Последняя индексация: {index['last_indexed']}")
print()
idx = 1
task_map = {} # для выбора по номерам
# Сортировка по приоритету
priority_order = {'urgent': 0, 'high': 1, 'medium': 2, 'low': 3}
for project, subprojects in sorted(index['projects'].items()):
if project_filter and project != project_filter:
continue
for subproject, tasks in sorted(subprojects.items()):
if not tasks:
continue
# Сортируем по приоритету
sorted_tasks = sorted(tasks, key=lambda t: priority_order.get(t['priority'], 9))
print(f"{project}/{subproject} ({len(tasks)} задач):")
for t in sorted_tasks[:10]: # Показываем до 10
priority_mark = {'urgent': '!!!', 'high': '!', 'medium': '', 'low': ''}
mark = priority_mark.get(t['priority'], '')
print(f" [{idx}] {t['task'][:60]} {mark}")
task_map[idx] = t
idx += 1
if len(tasks) > 10:
print(f" ... и ещё {len(tasks) - 10}")
print()
print("---")
print("Укажи номера (1 3 5) или:")
print(" 'все' — взять все задачи")
print(" 'проект X' — только проект X")
print(" 'urgent' — только срочные")
return task_map
def show_stats():
"""Показать статистику"""
index = load_index()
print("СТАТИСТИКА ИНДЕКСА")
print(f"Последняя индексация: {index['last_indexed']}")
print(f"Всего недоделок: {index['total_pending']}")
print()
for project, subprojects in sorted(index['projects'].items()):
total = sum(len(tasks) for tasks in subprojects.values())
print(f"{project}: {total}")
for sub, tasks in sorted(subprojects.items()):
if tasks:
urgent = len([t for t in tasks if t['priority'] == 'urgent'])
high = len([t for t in tasks if t['priority'] == 'high'])
print(f" {sub}: {len(tasks)} (urgent: {urgent}, high: {high})")
def main():
if len(sys.argv) < 2:
print("Использование:")
print(" session_index.py index — индексировать новые сессии")
print(" session_index.py show — показать недоделки")
print(" session_index.py full — полная переиндексация")
print(" session_index.py stats — статистика")
return
cmd = sys.argv[1]
if cmd == 'index':
index_sessions(full=False)
elif cmd == 'full':
index_sessions(full=True)
elif cmd == 'show':
project = sys.argv[2] if len(sys.argv) > 2 else None
show_tasks(project)
elif cmd == 'stats':
show_stats()
else:
print(f"Неизвестная команда: {cmd}")
if __name__ == '__main__':
main()