Add comprehensive development intelligence system that tracks: - Development sessions with automatic start/stop - Full conversation history with semantic search - Tool usage and file operation analytics - Think time and engagement analysis - Git activity correlation - Learning pattern recognition - Productivity insights and metrics Features: - FastAPI backend with SQLite database - Modern web dashboard with interactive charts - Claude Code hook integration for automatic tracking - Comprehensive test suite with 100+ tests - Complete API documentation (OpenAPI/Swagger) - Privacy-first design with local data storage 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
202 lines
7.1 KiB
Python
202 lines
7.1 KiB
Python
"""
|
|
Git operation model for tracking repository changes.
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from typing import Optional, List
|
|
from sqlalchemy import String, Text, Integer, DateTime, JSON, ForeignKey, Boolean
|
|
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
|
from sqlalchemy.sql import func
|
|
|
|
from .base import Base, TimestampMixin
|
|
|
|
|
|
class GitOperation(Base, TimestampMixin):
|
|
"""
|
|
Represents a git operation performed during a development session.
|
|
|
|
Tracks commits, pushes, pulls, branch operations, and other git commands
|
|
to provide insight into version control workflow.
|
|
"""
|
|
|
|
__tablename__ = "git_operations"
|
|
|
|
# Primary key
|
|
id: Mapped[int] = mapped_column(primary_key=True)
|
|
|
|
# Foreign key to session
|
|
session_id: Mapped[int] = mapped_column(ForeignKey("sessions.id"), nullable=False, index=True)
|
|
|
|
# Operation timing
|
|
timestamp: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
nullable=False,
|
|
default=func.now(),
|
|
index=True
|
|
)
|
|
|
|
# Operation details
|
|
operation: Mapped[str] = mapped_column(String(50), nullable=False, index=True) # commit, push, pull, branch, etc.
|
|
command: Mapped[str] = mapped_column(Text, nullable=False)
|
|
result: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
|
|
success: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
|
|
|
|
# File and change tracking
|
|
files_changed: Mapped[Optional[List[str]]] = mapped_column(JSON, nullable=True)
|
|
lines_added: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
|
|
lines_removed: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
|
|
|
|
# Commit-specific fields
|
|
commit_hash: Mapped[Optional[str]] = mapped_column(String(40), nullable=True, index=True)
|
|
|
|
# Branch operation fields
|
|
branch_from: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
|
|
branch_to: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
|
|
|
|
# Relationships
|
|
session: Mapped["Session"] = relationship("Session", back_populates="git_operations")
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<GitOperation(id={self.id}, operation='{self.operation}', success={self.success})>"
|
|
|
|
@property
|
|
def is_commit(self) -> bool:
|
|
"""Check if this is a commit operation."""
|
|
return self.operation == "commit"
|
|
|
|
@property
|
|
def is_push(self) -> bool:
|
|
"""Check if this is a push operation."""
|
|
return self.operation == "push"
|
|
|
|
@property
|
|
def is_pull(self) -> bool:
|
|
"""Check if this is a pull operation."""
|
|
return self.operation == "pull"
|
|
|
|
@property
|
|
def is_branch_operation(self) -> bool:
|
|
"""Check if this is a branch-related operation."""
|
|
return self.operation in {"branch", "checkout", "merge", "rebase"}
|
|
|
|
@property
|
|
def total_lines_changed(self) -> int:
|
|
"""Get total lines changed (added + removed)."""
|
|
added = self.lines_added or 0
|
|
removed = self.lines_removed or 0
|
|
return added + removed
|
|
|
|
@property
|
|
def net_lines_changed(self) -> int:
|
|
"""Get net lines changed (added - removed)."""
|
|
added = self.lines_added or 0
|
|
removed = self.lines_removed or 0
|
|
return added - removed
|
|
|
|
@property
|
|
def files_count(self) -> int:
|
|
"""Get number of files changed."""
|
|
return len(self.files_changed) if self.files_changed else 0
|
|
|
|
def get_commit_message(self) -> Optional[str]:
|
|
"""Extract commit message from the command."""
|
|
if not self.is_commit:
|
|
return None
|
|
|
|
command = self.command
|
|
if "-m" in command:
|
|
# Extract message between quotes after -m
|
|
parts = command.split("-m")
|
|
if len(parts) > 1:
|
|
message_part = parts[1].strip()
|
|
# Remove quotes
|
|
if message_part.startswith('"') and message_part.endswith('"'):
|
|
return message_part[1:-1]
|
|
elif message_part.startswith("'") and message_part.endswith("'"):
|
|
return message_part[1:-1]
|
|
else:
|
|
# Find first quoted string
|
|
import re
|
|
match = re.search(r'["\']([^"\']*)["\']', message_part)
|
|
if match:
|
|
return match.group(1)
|
|
|
|
return None
|
|
|
|
def get_branch_name(self) -> Optional[str]:
|
|
"""Get branch name for branch operations."""
|
|
if self.branch_to:
|
|
return self.branch_to
|
|
elif self.branch_from:
|
|
return self.branch_from
|
|
|
|
# Try to extract from command
|
|
if "checkout" in self.command:
|
|
parts = self.command.split()
|
|
if len(parts) > 2:
|
|
return parts[-1] # Last argument is usually the branch
|
|
|
|
return None
|
|
|
|
def is_merge_commit(self) -> bool:
|
|
"""Check if this is a merge commit."""
|
|
commit_msg = self.get_commit_message()
|
|
return commit_msg is not None and "merge" in commit_msg.lower()
|
|
|
|
def is_feature_commit(self) -> bool:
|
|
"""Check if this appears to be a feature commit."""
|
|
commit_msg = self.get_commit_message()
|
|
if not commit_msg:
|
|
return False
|
|
|
|
feature_keywords = ["add", "implement", "create", "new", "feature"]
|
|
return any(keyword in commit_msg.lower() for keyword in feature_keywords)
|
|
|
|
def is_bugfix_commit(self) -> bool:
|
|
"""Check if this appears to be a bugfix commit."""
|
|
commit_msg = self.get_commit_message()
|
|
if not commit_msg:
|
|
return False
|
|
|
|
bugfix_keywords = ["fix", "bug", "resolve", "correct", "patch"]
|
|
return any(keyword in commit_msg.lower() for keyword in bugfix_keywords)
|
|
|
|
def is_refactor_commit(self) -> bool:
|
|
"""Check if this appears to be a refactoring commit."""
|
|
commit_msg = self.get_commit_message()
|
|
if not commit_msg:
|
|
return False
|
|
|
|
refactor_keywords = ["refactor", "cleanup", "improve", "optimize", "reorganize"]
|
|
return any(keyword in commit_msg.lower() for keyword in refactor_keywords)
|
|
|
|
def get_commit_category(self) -> str:
|
|
"""Categorize the commit based on its message."""
|
|
if not self.is_commit:
|
|
return "non-commit"
|
|
|
|
if self.is_merge_commit():
|
|
return "merge"
|
|
elif self.is_feature_commit():
|
|
return "feature"
|
|
elif self.is_bugfix_commit():
|
|
return "bugfix"
|
|
elif self.is_refactor_commit():
|
|
return "refactor"
|
|
else:
|
|
return "other"
|
|
|
|
def get_change_size_category(self) -> str:
|
|
"""Categorize the size of changes in this operation."""
|
|
total_changes = self.total_lines_changed
|
|
|
|
if total_changes == 0:
|
|
return "no-changes"
|
|
elif total_changes < 10:
|
|
return "small"
|
|
elif total_changes < 50:
|
|
return "medium"
|
|
elif total_changes < 200:
|
|
return "large"
|
|
else:
|
|
return "very-large" |