""" Analysis result and summary models """ from datetime import datetime from bson import ObjectId from typing import Optional, Dict, Any from app.models.database import get_database class AnalysisResult: """Video analysis result model""" COLLECTION_NAME = 'analysis_results' def __init__(self, video_id: ObjectId, content: str, fps: int, analysis_type: str = 'content', created_at: Optional[datetime] = None, _id: Optional[ObjectId] = None): """ Initialize AnalysisResult model Args: video_id: Video document ID content: Analysis content fps: Frames per second used analysis_type: Type of analysis (content, summary) created_at: Creation timestamp _id: Document ID """ self._id = _id or ObjectId() self.video_id = video_id self.content = content self.fps = fps self.analysis_type = analysis_type self.created_at = created_at or datetime.utcnow() def to_dict(self) -> Dict[str, Any]: """Convert to dictionary""" return { '_id': self._id, 'video_id': self.video_id, 'content': self.content, 'fps': self.fps, 'analysis_type': self.analysis_type, 'created_at': self.created_at } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'AnalysisResult': """Create AnalysisResult from dictionary""" return cls( _id=data.get('_id'), video_id=data['video_id'], content=data['content'], fps=data['fps'], analysis_type=data.get('analysis_type', 'content'), created_at=data.get('created_at') ) def save(self) -> ObjectId: """Save analysis result to database""" db = get_database() result = db[self.COLLECTION_NAME].insert_one(self.to_dict()) self._id = result.inserted_id return self._id @classmethod def find_by_video_id(cls, video_id: str, analysis_type: Optional[str] = None) -> Optional['AnalysisResult']: """Find analysis result by video ID""" db = get_database() query = {'video_id': ObjectId(video_id)} if analysis_type: query['analysis_type'] = analysis_type doc = db[cls.COLLECTION_NAME].find_one(query, sort=[('created_at', -1)]) if doc: return cls.from_dict(doc) return None class Summary: """Video summary model""" COLLECTION_NAME = 'summaries' def __init__(self, video_id: ObjectId, summary_text: str, created_at: Optional[datetime] = None, _id: Optional[ObjectId] = None): """ Initialize Summary model Args: video_id: Video document ID summary_text: Summary text created_at: Creation timestamp _id: Document ID """ self._id = _id or ObjectId() self.video_id = video_id self.summary_text = summary_text self.created_at = created_at or datetime.utcnow() def to_dict(self) -> Dict[str, Any]: """Convert to dictionary""" return { '_id': self._id, 'video_id': self.video_id, 'summary_text': self.summary_text, 'created_at': self.created_at } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'Summary': """Create Summary from dictionary""" return cls( _id=data.get('_id'), video_id=data['video_id'], summary_text=data['summary_text'], created_at=data.get('created_at') ) def save(self) -> ObjectId: """Save summary to database""" db = get_database() result = db[self.COLLECTION_NAME].insert_one(self.to_dict()) self._id = result.inserted_id return self._id @classmethod def find_by_video_id(cls, video_id: str) -> Optional['Summary']: """Find summary by video ID""" db = get_database() doc = db[cls.COLLECTION_NAME].find_one( {'video_id': ObjectId(video_id)}, sort=[('created_at', -1)] ) if doc: return cls.from_dict(doc) return None