""" Git operation tracking API endpoints. """ from typing import List, Optional from fastapi import APIRouter, Depends, HTTPException, status, Query from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func from sqlalchemy.orm import selectinload from app.database.connection import get_db from app.models.git_operation import GitOperation from app.models.session import Session from app.api.schemas import GitOperationRequest, GitOperationResponse router = APIRouter() @router.post("/git", response_model=GitOperationResponse, status_code=status.HTTP_201_CREATED) async def record_git_operation( request: GitOperationRequest, db: AsyncSession = Depends(get_db) ): """ Record a git operation performed during development. Called by hooks when git commands are executed via Bash tool. """ try: # Verify session exists result = await db.execute( select(Session).where(Session.id == request.session_id) ) session = result.scalars().first() if not session: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Session {request.session_id} not found" ) # Create git operation record git_operation = GitOperation( session_id=request.session_id, timestamp=request.timestamp, operation=request.operation, command=request.command, result=request.result, success=request.success, files_changed=request.files_changed, lines_added=request.lines_added, lines_removed=request.lines_removed, commit_hash=request.commit_hash, branch_from=request.branch_from, branch_to=request.branch_to ) db.add(git_operation) await db.commit() await db.refresh(git_operation) return GitOperationResponse( id=git_operation.id, session_id=git_operation.session_id, operation=git_operation.operation, timestamp=git_operation.timestamp, success=git_operation.success, commit_hash=git_operation.commit_hash ) except HTTPException: raise except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to record git operation: {str(e)}" ) @router.get("/git/operations") async def get_git_operations( session_id: Optional[int] = Query(None, description="Filter by session ID"), project_id: Optional[int] = Query(None, description="Filter by project ID"), operation: Optional[str] = Query(None, description="Filter by operation type"), limit: int = Query(50, description="Maximum number of results"), offset: int = Query(0, description="Number of results to skip"), db: AsyncSession = Depends(get_db) ): """Get git operations with optional filtering.""" try: query = select(GitOperation).options( selectinload(GitOperation.session).selectinload(Session.project) ) # Apply filters if session_id: query = query.where(GitOperation.session_id == session_id) elif project_id: query = query.join(Session).where(Session.project_id == project_id) if operation: query = query.where(GitOperation.operation == operation) # Order by timestamp descending query = query.order_by(GitOperation.timestamp.desc()).offset(offset).limit(limit) result = await db.execute(query) git_operations = result.scalars().all() return [ { "id": op.id, "session_id": op.session_id, "project_name": op.session.project.name, "timestamp": op.timestamp, "operation": op.operation, "command": op.command, "result": op.result, "success": op.success, "files_changed": op.files_changed, "files_count": op.files_count, "lines_added": op.lines_added, "lines_removed": op.lines_removed, "total_lines_changed": op.total_lines_changed, "net_lines_changed": op.net_lines_changed, "commit_hash": op.commit_hash, "commit_message": op.get_commit_message(), "commit_category": op.get_commit_category(), "change_size_category": op.get_change_size_category() } for op in git_operations ] except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to get git operations: {str(e)}" ) @router.get("/git/operations/{operation_id}") async def get_git_operation( operation_id: int, db: AsyncSession = Depends(get_db) ): """Get detailed information about a specific git operation.""" try: result = await db.execute( select(GitOperation) .options(selectinload(GitOperation.session).selectinload(Session.project)) .where(GitOperation.id == operation_id) ) git_operation = result.scalars().first() if not git_operation: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Git operation {operation_id} not found" ) return { "id": git_operation.id, "session_id": git_operation.session_id, "project_name": git_operation.session.project.name, "timestamp": git_operation.timestamp, "operation": git_operation.operation, "command": git_operation.command, "result": git_operation.result, "success": git_operation.success, "files_changed": git_operation.files_changed, "files_count": git_operation.files_count, "lines_added": git_operation.lines_added, "lines_removed": git_operation.lines_removed, "total_lines_changed": git_operation.total_lines_changed, "net_lines_changed": git_operation.net_lines_changed, "commit_hash": git_operation.commit_hash, "branch_from": git_operation.branch_from, "branch_to": git_operation.branch_to, "commit_message": git_operation.get_commit_message(), "branch_name": git_operation.get_branch_name(), "is_commit": git_operation.is_commit, "is_push": git_operation.is_push, "is_pull": git_operation.is_pull, "is_branch_operation": git_operation.is_branch_operation, "is_merge_commit": git_operation.is_merge_commit(), "is_feature_commit": git_operation.is_feature_commit(), "is_bugfix_commit": git_operation.is_bugfix_commit(), "is_refactor_commit": git_operation.is_refactor_commit(), "commit_category": git_operation.get_commit_category(), "change_size_category": git_operation.get_change_size_category() } except HTTPException: raise except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to get git operation: {str(e)}" ) @router.get("/git/stats/commits") async def get_commit_stats( project_id: Optional[int] = Query(None, description="Filter by project ID"), days: int = Query(30, description="Number of days to include"), db: AsyncSession = Depends(get_db) ): """Get commit statistics and patterns.""" try: # Base query for commits query = select(GitOperation).where(GitOperation.operation == "commit") # Apply filters if project_id: query = query.join(Session).where(Session.project_id == project_id) if days > 0: from datetime import datetime, timedelta start_date = datetime.utcnow() - timedelta(days=days) query = query.where(GitOperation.timestamp >= start_date) result = await db.execute(query) commits = result.scalars().all() if not commits: return { "total_commits": 0, "commit_categories": {}, "change_size_distribution": {}, "average_lines_per_commit": 0.0, "commit_frequency": [] } # Calculate statistics total_commits = len(commits) # Categorize commits categories = {} for commit in commits: category = commit.get_commit_category() categories[category] = categories.get(category, 0) + 1 # Change size distribution size_distribution = {} for commit in commits: size_category = commit.get_change_size_category() size_distribution[size_category] = size_distribution.get(size_category, 0) + 1 # Average lines per commit total_lines = sum(commit.total_lines_changed for commit in commits) avg_lines_per_commit = total_lines / total_commits if total_commits > 0 else 0 # Daily commit frequency daily_commits = {} for commit in commits: date_key = commit.timestamp.date().isoformat() daily_commits[date_key] = daily_commits.get(date_key, 0) + 1 commit_frequency = [ {"date": date, "commits": count} for date, count in sorted(daily_commits.items()) ] return { "total_commits": total_commits, "commit_categories": categories, "change_size_distribution": size_distribution, "average_lines_per_commit": round(avg_lines_per_commit, 1), "commit_frequency": commit_frequency, "top_commit_messages": [ { "message": commit.get_commit_message(), "lines_changed": commit.total_lines_changed, "timestamp": commit.timestamp } for commit in sorted(commits, key=lambda c: c.total_lines_changed, reverse=True)[:5] if commit.get_commit_message() ] } except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to get commit stats: {str(e)}" ) @router.get("/git/stats/activity") async def get_git_activity_stats( project_id: Optional[int] = Query(None, description="Filter by project ID"), days: int = Query(30, description="Number of days to include"), db: AsyncSession = Depends(get_db) ): """Get overall git activity statistics.""" try: query = select(GitOperation) # Apply filters if project_id: query = query.join(Session).where(Session.project_id == project_id) if days > 0: from datetime import datetime, timedelta start_date = datetime.utcnow() - timedelta(days=days) query = query.where(GitOperation.timestamp >= start_date) result = await db.execute(query) operations = result.scalars().all() if not operations: return { "total_operations": 0, "operations_by_type": {}, "success_rate": 0.0, "most_active_days": [] } # Operations by type operations_by_type = {} successful_operations = 0 for op in operations: operations_by_type[op.operation] = operations_by_type.get(op.operation, 0) + 1 if op.success: successful_operations += 1 success_rate = successful_operations / len(operations) * 100 if operations else 0 # Daily activity daily_activity = {} for op in operations: date_key = op.timestamp.date().isoformat() if date_key not in daily_activity: daily_activity[date_key] = 0 daily_activity[date_key] += 1 most_active_days = [ {"date": date, "operations": count} for date, count in sorted(daily_activity.items(), key=lambda x: x[1], reverse=True)[:7] ] return { "total_operations": len(operations), "operations_by_type": operations_by_type, "success_rate": round(success_rate, 1), "most_active_days": most_active_days, "timeline": [ { "date": date, "operations": count } for date, count in sorted(daily_activity.items()) ] } except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to get git activity stats: {str(e)}" )