Files
7000pctAUTO 86160392f3
Some checks failed
CI / test (push) Has been cancelled
CI / lint (push) Has been cancelled
CI / build (push) Has been cancelled
feat: add models, context, and search modules
2026-01-31 10:23:33 +00:00

185 lines
6.2 KiB
Python

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional, List, Dict, Any
from enum import Enum
class CommandType(Enum):
GIT = "git"
DOCKER = "docker"
BUILD = "build"
TEST = "test"
DEPLOY = "deploy"
FILE_OP = "file_op"
SYSTEM = "system"
OTHER = "other"
@dataclass
class Project:
id: Optional[int] = None
name: str = ""
path: str = ""
git_remote: Optional[str] = None
tech_stack: List[str] = field(default_factory=list)
created_at: datetime = field(default_factory=datetime.utcnow)
updated_at: datetime = field(default_factory=datetime.utcnow)
def to_dict(self) -> Dict[str, Any]:
return {
"id": self.id,
"name": self.name,
"path": self.path,
"git_remote": self.git_remote,
"tech_stack": self.tech_stack,
"created_at": self.created_at.isoformat(),
"updated_at": self.updated_at.isoformat(),
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "Project":
return cls(
id=data.get("id"),
name=data.get("name", ""),
path=data.get("path", ""),
git_remote=data.get("git_remote"),
tech_stack=data.get("tech_stack", []),
created_at=datetime.fromisoformat(data["created_at"]) if data.get("created_at") else datetime.utcnow(),
updated_at=datetime.fromisoformat(data["updated_at"]) if data.get("updated_at") else datetime.utcnow(),
)
@dataclass
class Command:
id: Optional[int] = None
workflow_id: Optional[int] = None
project_id: Optional[int] = None
command: str = ""
command_type: CommandType = CommandType.OTHER
exit_code: Optional[int] = None
duration_ms: Optional[int] = None
working_directory: str = ""
timestamp: datetime = field(default_factory=datetime.utcnow)
tags: List[str] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
return {
"id": self.id,
"workflow_id": self.workflow_id,
"project_id": self.project_id,
"command": self.command,
"command_type": self.command_type.value,
"exit_code": self.exit_code,
"duration_ms": self.duration_ms,
"working_directory": self.working_directory,
"timestamp": self.timestamp.isoformat(),
"tags": self.tags,
"metadata": self.metadata,
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "Command":
return cls(
id=data.get("id"),
workflow_id=data.get("workflow_id"),
project_id=data.get("project_id"),
command=data.get("command", ""),
command_type=CommandType(data.get("command_type", "other")),
exit_code=data.get("exit_code"),
duration_ms=data.get("duration_ms"),
working_directory=data.get("working_directory", ""),
timestamp=datetime.fromisoformat(data["timestamp"]) if data.get("timestamp") else datetime.utcnow(),
tags=data.get("tags", []),
metadata=data.get("metadata", {}),
)
@dataclass
class Workflow:
id: Optional[int] = None
project_id: Optional[int] = None
name: str = ""
description: str = ""
commands: List[Command] = field(default_factory=list)
created_at: datetime = field(default_factory=datetime.utcnow)
updated_at: datetime = field(default_factory=datetime.utcnow)
is_automated: bool = False
pattern_confidence: float = 0.0
usage_count: int = 0
def to_dict(self) -> Dict[str, Any]:
return {
"id": self.id,
"project_id": self.project_id,
"name": self.name,
"description": self.description,
"commands": [c.to_dict() for c in self.commands],
"created_at": self.created_at.isoformat(),
"updated_at": self.updated_at.isoformat(),
"is_automated": self.is_automated,
"pattern_confidence": self.pattern_confidence,
"usage_count": self.usage_count,
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "Workflow":
return cls(
id=data.get("id"),
project_id=data.get("project_id"),
name=data.get("name", ""),
description=data.get("description", ""),
commands=[Command.from_dict(c) for c in data.get("commands", [])],
created_at=datetime.fromisoformat(data["created_at"]) if data.get("created_at") else datetime.utcnow(),
updated_at=datetime.fromisoformat(data["updated_at"]) if data.get("updated_at") else datetime.utcnow(),
is_automated=data.get("is_automated", False),
pattern_confidence=data.get("pattern_confidence", 0.0),
usage_count=data.get("usage_count", 0),
)
@dataclass
class Suggestion:
id: Optional[int] = None
project_id: Optional[int] = None
command: str = ""
context: str = ""
confidence: float = 0.0
frequency: int = 0
last_used: Optional[datetime] = None
pattern_id: Optional[int] = None
def to_dict(self) -> Dict[str, Any]:
return {
"id": self.id,
"project_id": self.project_id,
"command": self.command,
"context": self.context,
"confidence": self.confidence,
"frequency": self.frequency,
"last_used": self.last_used.isoformat() if self.last_used else None,
"pattern_id": self.pattern_id,
}
@dataclass
class Pattern:
id: Optional[int] = None
project_id: Optional[int] = None
name: str = ""
command_sequence: List[str] = field(default_factory=list)
occurrences: int = 0
confidence: float = 0.0
created_at: datetime = field(default_factory=datetime.utcnow)
def to_dict(self) -> Dict[str, Any]:
return {
"id": self.id,
"project_id": self.project_id,
"name": self.name,
"command_sequence": self.command_sequence,
"occurrences": self.occurrences,
"confidence": self.confidence,
"created_at": self.created_at.isoformat(),
}