File size: 3,000 Bytes
bd91486 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | import sqlite3
from typing import Dict, List, Optional
from dataclasses import dataclass
from datetime import datetime
import json
@dataclass
class ProjectState:
project_id: str
name: str
language: str
framework: str
status: str
completion_percentage: float
last_modified: datetime
code_snippets: Dict[str, str]
dependencies: List[str]
class ProjectTracker:
def __init__(self, db_path: str = "cogenbai_projects.db"):
self.conn = sqlite3.connect(db_path)
self._init_db()
def _init_db(self):
cursor = self.conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS projects (
project_id TEXT PRIMARY KEY,
name TEXT,
language TEXT,
framework TEXT,
status TEXT,
completion_percentage REAL,
last_modified TIMESTAMP,
code_snippets TEXT,
dependencies TEXT
)
''')
self.conn.commit()
def create_project(self, project: ProjectState) -> bool:
try:
cursor = self.conn.cursor()
cursor.execute('''
INSERT INTO projects VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (
project.project_id,
project.name,
project.language,
project.framework,
project.status,
project.completion_percentage,
project.last_modified.isoformat(),
json.dumps(project.code_snippets),
json.dumps(project.dependencies)
))
self.conn.commit()
return True
except Exception as e:
print(f"Error creating project: {e}")
return False
def update_project(self, project_id: str, updates: Dict) -> bool:
try:
cursor = self.conn.cursor()
set_clause = ", ".join([f"{k} = ?" for k in updates.keys()])
query = f"UPDATE projects SET {set_clause} WHERE project_id = ?"
cursor.execute(query, list(updates.values()) + [project_id])
self.conn.commit()
return True
except Exception as e:
print(f"Error updating project: {e}")
return False
def get_project(self, project_id: str) -> Optional[ProjectState]:
cursor = self.conn.cursor()
cursor.execute("SELECT * FROM projects WHERE project_id = ?", (project_id,))
row = cursor.fetchone()
if row:
return ProjectState(
project_id=row[0],
name=row[1],
language=row[2],
framework=row[3],
status=row[4],
completion_percentage=row[5],
last_modified=datetime.fromisoformat(row[6]),
code_snippets=json.loads(row[7]),
dependencies=json.loads(row[8])
)
return None
|