basicmachines-co-basic-memory/src/basic_memory/schemas/project_info.py

"""Schema for project info response."""

import os
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Any

from pydantic import Field, BaseModel

from basic_memory.utils import generate_permalink


class ProjectStatistics(BaseModel):
    """Statistics about the current project."""

    # Basic counts
    total_entities: int = Field(description="Total number of entities in the knowledge base")
    total_observations: int = Field(description="Total number of observations across all entities")
    total_relations: int = Field(description="Total number of relations between entities")
    total_unresolved_relations: int = Field(
        description="Number of relations with unresolved targets"
    )

    # Entity counts by type
    entity_types: Dict[str, int] = Field(
        description="Count of entities by type (e.g., note, conversation)"
    )

    # Observation counts by category
    observation_categories: Dict[str, int] = Field(
        description="Count of observations by category (e.g., tech, decision)"
    )

    # Relation counts by type
    relation_types: Dict[str, int] = Field(
        description="Count of relations by type (e.g., implements, relates_to)"
    )

    # Graph metrics
    most_connected_entities: List[Dict[str, Any]] = Field(
        description="Entities with the most relations, including their titles and permalinks"
    )
    isolated_entities: int = Field(description="Number of entities with no relations")


class ActivityMetrics(BaseModel):
    """Activity metrics for the current project."""

    # Recent activity
    recently_created: List[Dict[str, Any]] = Field(
        description="Recently created entities with timestamps"
    )
    recently_updated: List[Dict[str, Any]] = Field(
        description="Recently updated entities with timestamps"
    )

    # Growth over time (last 6 months)
    monthly_growth: Dict[str, Dict[str, int]] = Field(
        description="Monthly growth statistics for entities, observations, and relations"
    )


class SystemStatus(BaseModel):
    """System status information."""

    # Version information
    version: str = Field(description="Basic Memory version")

    # Database status
    database_path: str = Field(description="Path to the SQLite database")
    database_size: str = Field(description="Size of the database in human-readable format")

    # Watch service status
    watch_status: Optional[Dict[str, Any]] = Field(
        default=None, description="Watch service status information (if running)"
    )

    # System information
    timestamp: datetime = Field(description="Timestamp when the information was collected")


class ProjectInfoResponse(BaseModel):
    """Response for the project_info tool."""

    # Project configuration
    project_name: str = Field(description="Name of the current project")
    project_path: str = Field(description="Path to the current project files")
    available_projects: Dict[str, Dict[str, Any]] = Field(
        description="Map of configured project names to detailed project information"
    )
    default_project: str = Field(description="Name of the default project")

    # Statistics
    statistics: ProjectStatistics = Field(description="Statistics about the knowledge base")

    # Activity metrics
    activity: ActivityMetrics = Field(description="Activity and growth metrics")

    # System status
    system: SystemStatus = Field(description="System and service status information")


class ProjectInfoRequest(BaseModel):
    """Request model for switching projects."""

    name: str = Field(..., description="Name of the project to switch to")
    path: str = Field(..., description="Path to the project directory")
    set_default: bool = Field(..., description="Set the project as the default")


class WatchEvent(BaseModel):
    timestamp: datetime
    path: str
    action: str  # new, delete, etc
    status: str  # success, error
    checksum: Optional[str]
    error: Optional[str] = None


class WatchServiceState(BaseModel):
    # Service status
    running: bool = False
    start_time: datetime = datetime.now()  # Use directly with Pydantic model
    pid: int = os.getpid()  # Use directly with Pydantic model

    # Stats
    error_count: int = 0
    last_error: Optional[datetime] = None
    last_scan: Optional[datetime] = None

    # File counts
    synced_files: int = 0

    # Recent activity
    recent_events: List[WatchEvent] = []  # Use directly with Pydantic model

    def add_event(
        self,
        path: str,
        action: str,
        status: str,
        checksum: Optional[str] = None,
        error: Optional[str] = None,
    ) -> WatchEvent:  # pragma: no cover
        event = WatchEvent(
            timestamp=datetime.now(),
            path=path,
            action=action,
            status=status,
            checksum=checksum,
            error=error,
        )
        self.recent_events.insert(0, event)
        self.recent_events = self.recent_events[:100]  # Keep last 100
        return event

    def record_error(self, error: str):  # pragma: no cover
        self.error_count += 1
        self.add_event(path="", action="sync", status="error", error=error)
        self.last_error = datetime.now()


class ProjectWatchStatus(BaseModel):
    """Project with its watch status."""

    name: str = Field(..., description="Name of the project")
    path: str = Field(..., description="Path to the project")
    watch_status: Optional[WatchServiceState] = Field(
        None, description="Watch status information for the project"
    )


class ProjectItem(BaseModel):
    """Simple representation of a project."""

    name: str
    path: str
    is_default: bool = False

    @property
    def permalink(self) -> str:  # pragma: no cover
        return generate_permalink(self.name)

    @property
    def home(self) -> Path:  # pragma: no cover
        return Path(self.name)

    @property
    def project_url(self) -> str:  # pragma: no cover
        return f"/{generate_permalink(self.name)}"


class ProjectList(BaseModel):
    """Response model for listing projects."""

    projects: List[ProjectItem]
    default_project: str


class ProjectStatusResponse(BaseModel):
    """Response model for switching projects."""

    message: str = Field(..., description="Status message about the project switch")
    status: str = Field(..., description="Status of the switch (success or error)")
    default: bool = Field(..., description="True if the project was set as the default")
    old_project: Optional[ProjectItem] = Field(
        None, description="Information about the project being switched from"
    )
    new_project: Optional[ProjectItem] = Field(
        None, description="Information about the project being switched to"
    )