Spaces:
Sleeping
Sleeping
File size: 6,690 Bytes
63da3a1 96c1a4b 63da3a1 96c1a4b 63da3a1 96c1a4b 63da3a1 | 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 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | from pydantic import BaseModel, Field
from typing import Dict, List, Optional, Any
from datetime import datetime
from bson import ObjectId
from app.models.user import PyObjectId
import logging
logger = logging.getLogger(__name__)
class CanvasInsight(BaseModel):
"""Individual insight extracted from chat messages"""
content: str
source_persona: str
source_message_id: Optional[str] = None
source_chat_session: Optional[str] = None
confidence_score: float = Field(ge=0.0, le=1.0, default=0.8)
extracted_at: datetime = Field(default_factory=datetime.utcnow)
keywords: List[str] = Field(default_factory=list)
class CanvasSection(BaseModel):
"""A themed section of the PhD Canvas with related insights"""
title: str
description: str
insights: List[CanvasInsight] = Field(default_factory=list)
priority: int = Field(default=1, ge=1, le=5) # 1=highest priority
updated_at: datetime = Field(default_factory=datetime.utcnow)
class PhdCanvas(BaseModel):
"""Main PhD Canvas model storing all user insights organized by sections"""
id: PyObjectId = Field(default_factory=PyObjectId, alias="_id")
user_id: PyObjectId
# Canvas sections organized by theme
sections: Dict[str, CanvasSection] = Field(default_factory=dict)
# Metadata
created_at: datetime = Field(default_factory=datetime.utcnow)
last_updated: datetime = Field(default_factory=datetime.utcnow)
last_chat_processed: Optional[datetime] = None
total_insights: int = Field(default=0)
# Settings
auto_update: bool = Field(default=True)
print_optimized: bool = Field(default=True)
class Config:
allow_population_by_field_name = True
arbitrary_types_allowed = True
json_encoders = {ObjectId: str}
def update_section(self, section_key: str, insights: List[CanvasInsight]):
"""Update a specific canvas section with new insights"""
if section_key not in self.sections:
self.sections[section_key] = CanvasSection(
title=self._get_section_title(section_key),
description=self._get_section_description(section_key)
)
existing_insights_map = {
insight.content.strip().lower(): insight
for insight in self.sections[section_key].insights
}
# Also track existing chat session + message combinations
existing_sources = {
(insight.source_chat_session, insight.source_message_id)
for insight in self.sections[section_key].insights
if insight.source_chat_session and insight.source_message_id
}
new_insights = []
for insight in insights:
# Normalize content for comparison
normalized_content = insight.content.strip().lower()
# Check if this exact content already exists
if normalized_content in existing_insights_map:
logger.debug(f"Skipping duplicate insight: {insight.content[:50]}...")
continue
# Check if this source was already processed
if insight.source_chat_session and insight.source_message_id:
source_key = (insight.source_chat_session, insight.source_message_id)
if source_key in existing_sources:
logger.debug(f"Skipping already processed source: {source_key}")
continue
# This is genuinely new
new_insights.append(insight)
if new_insights:
logger.info(f"Adding {len(new_insights)} new insights to section '{section_key}'")
self.sections[section_key].insights.extend(new_insights)
self.sections[section_key].updated_at = datetime.utcnow()
self.last_updated = datetime.utcnow()
else:
logger.info(f"No new insights to add to section '{section_key}' (all {len(insights)} were duplicates)")
# Update total insights count
self.total_insights = sum(len(section.insights) for section in self.sections.values())
def _get_section_title(self, section_key: str) -> str:
"""Get human-readable title for section"""
titles = {
"research_progress": "Research Progress & Milestones",
"methodology": "Research Methods & Approach",
"theoretical_framework": "Theoretical Foundations",
"challenges_obstacles": "Challenges & Solutions",
"next_steps": "Action Items & Next Steps",
"writing_communication": "Writing & Communication",
"career_development": "Academic Career Planning",
"literature_review": "Literature & Sources",
"data_analysis": "Data & Analysis",
"motivation_mindset": "Motivation & Mindset"
}
return titles.get(section_key, section_key.replace("_", " ").title())
def _get_section_description(self, section_key: str) -> str:
"""Get description for each section"""
descriptions = {
"research_progress": "Key milestones, accomplishments, and timeline updates",
"methodology": "Research design decisions and methodological insights",
"theoretical_framework": "Theoretical perspectives and conceptual foundations",
"challenges_obstacles": "Challenges faced and strategies for overcoming them",
"next_steps": "Immediate action items and upcoming priorities",
"writing_communication": "Writing strategies and communication insights",
"career_development": "Professional development and career planning",
"literature_review": "Literature gaps, sources, and review strategies",
"data_analysis": "Data collection and analysis approaches",
"motivation_mindset": "Motivational insights and mental health considerations"
}
return descriptions.get(section_key, "General insights and guidance")
class CanvasResponse(BaseModel):
"""Response model for canvas API endpoints"""
id: str
user_id: str
sections: Dict[str, CanvasSection]
created_at: datetime
last_updated: datetime
last_chat_processed: Optional[datetime]
total_insights: int
auto_update: bool
print_optimized: bool
class UpdateCanvasRequest(BaseModel):
"""Request model for updating canvas"""
force_full_update: bool = Field(default=False)
include_chat_sessions: Optional[List[str]] = None # Specific sessions to include
exclude_sections: Optional[List[str]] = None # Sections to skip updating |