🔍 Code Extractor

class ReviewCycle_v1

Maturity: 28

Model representing a review cycle for a document version.

File:
/tf/active/vicechatdev/CDocs single class/models/review.py
Lines:
286 - 1291
Complexity:
moderate

Purpose

Model representing a review cycle for a document version.

Source Code

class ReviewCycle(WorkflowCycleBase, DocumentVersionRelationMixin, BaseModel):
    """Model representing a review cycle for a document version."""
    
    def __init__(self, data: Optional[Dict[str, Any]] = None, uid: Optional[str] = None):
        """
        Initialize a review cycle.
        
        Args:
            data: Dictionary of review cycle properties
            uid: Review cycle UID to load from database (if data not provided)
        """
        if data is None and uid is not None:
            # Fetch review cycle data from database
            data = db.get_node_by_uid(uid)
            
        BaseModel.__init__(self, data or {})
        WorkflowCycleBase.__init__(self, data or {})
        DocumentVersionRelationMixin.__init__(self)
        
        # Set default workflow type
        if 'workflow_type' not in self._data:
            self._data['workflow_type'] = 'REVIEW'
            
        # Ensure backward compatibility - map old fields to new fields
        if 'startDate' in self._data and 'started_at' not in self._data:
            self._data['started_at'] = self._data['startDate']
        if 'dueDate' in self._data and 'due_date' not in self._data:
            self._data['due_date'] = self._data['dueDate']
        if 'completionDate' in self._data and 'completed_at' not in self._data:
            self._data['completed_at'] = self._data['completionDate']
            
        self._comments_cache = None
        self._reviewers_cache = None
    
    @classmethod
    def create(cls, document_version_uid: str, 
              reviewers: List[Union[DocUser, str]],
              due_date: Optional[datetime] = None,
              instructions: str = '',
              properties: Optional[Dict[str, Any]] = None) -> Optional['ReviewCycle']:
        """
        Create a new review cycle.
        
        Args:
            document_version_uid: UID of the document version to review
            reviewers: List of users or UIDs to assign as reviewers
            due_date: Date when review should be completed
            instructions: Instructions for reviewers
            properties: Additional properties for the review cycle
            
        Returns:
            New ReviewCycle instance or None if creation failed
        """
        try:
            # Generate due date if not provided
            if not due_date:
                due_date = datetime.now() + timedelta(days=settings.DEFAULT_REVIEW_PERIOD_DAYS)
                
            # Prepare properties
            props = properties or {}
            props.update({
                'status': WorkflowStatus.PENDING,
                'started_at': datetime.now(),
                'startDate': datetime.now(),  # For backward compatibility
                'due_date': due_date,
                'dueDate': due_date,  # For backward compatibility
                'instructions': instructions,
                'workflow_type': 'REVIEW',
                'document_version_uid': document_version_uid  # Store for relationship creation
            })
            
            # Create node in database
            review_data = db.create_node_with_relationship(
                NodeLabels.REVIEW_CYCLE,
                props,
                document_version_uid,
                RelTypes.FOR_REVIEW
            )
            
            if not review_data:
                logger.error(f"Failed to create review cycle for document version {document_version_uid}")
                return None
                
            # Create the review cycle instance
            review_cycle = cls(review_data)
            
            # Add reviewers
            for reviewer in reviewers:
                review_cycle.add_reviewer(reviewer)
            
            # Update document version status
            from CDocs.models.document import DocumentVersion
            version = DocumentVersion(uid=document_version_uid)
            if version:
                # Update version status
                version.status = 'IN_REVIEW'
                
                # Also update document status
                document = version.document
                if document:
                    document.status = 'IN_REVIEW'
            
            return review_cycle
            
        except Exception as e:
            logger.error(f"Error creating review cycle: {e}")
            return None
    
    @classmethod
    def get_reviews_for_document(cls, document_uid: str) -> List['ReviewCycle']:
        """
        Get all review cycles for a document.
        
        Args:
            document_uid: UID of the document
            
        Returns:
            List of ReviewCycle instances
        """
        try:
            # Query for review cycles related to the document
            result = db.run_query(
                """
                MATCH (d:ControlledDocument {UID: $doc_uid})-[:HAS_VERSION]->(v:DocumentVersion)
                MATCH (v)<-[:FOR_REVIEW]-(r:ReviewCycle)
                RETURN r.UID as uid
                ORDER BY r.startDate DESC
                """,
                {"doc_uid": document_uid}
            )
            
            # Create ReviewCycle instances
            reviews = [cls(uid=record['uid']) for record in result if 'uid' in record]
            return reviews
            
        except Exception as e:
            logger.error(f"Error getting reviews for document {document_uid}: {e}")
            return []

    @property
    def status(self) -> str:
        """Get review cycle status."""
        return self._data.get('status', WorkflowStatus.PENDING)
    
    @status.setter
    def status(self, value: str) -> None:
        """Set review cycle status."""
        if value not in settings.REVIEW_STATUSES:
            logger.warning(f"Invalid review status: {value}")
            return
            
        old_status = self.status
        
        self._data['status'] = value
        update_data = {'status': value}
        
        # Set completion date if we're transitioning to a terminal status
        if value in [WorkflowStatus.COMPLETED, WorkflowStatus.REJECTED] and old_status not in [WorkflowStatus.COMPLETED, WorkflowStatus.REJECTED]:
            self._data['completionDate'] = datetime.now()
            self._data['completed_at'] = datetime.now().isoformat()
            update_data['completionDate'] = self._data['completionDate']
            update_data['completed_at'] = self._data['completed_at']
            
        db.update_node(self.uid, update_data)

    @property
    def initiated_by_uid(self) -> Optional[str]:
        """Get the UID of the user who initiated this review cycle."""
        return self._data.get('initiated_by_uid')
        
    @initiated_by_uid.setter
    def initiated_by_uid(self, uid: str) -> None:
        """Set the UID of the user who initiated this review cycle."""
        if uid:
            self._data['initiated_by_uid'] = uid
            db.update_node(self.uid, {'initiated_by_uid': uid})

    @property
    def required_approval_percentage(self) -> int:
        """Get the required approval percentage for this review cycle."""
        return self._data.get('required_approval_percentage', 100)
    
    @required_approval_percentage.setter
    def required_approval_percentage(self, value: int) -> None:
        """Set the required approval percentage."""
        if not isinstance(value, int) or value < 1 or value > 100:
            logger.warning(f"Invalid approval percentage: {value}. Must be an integer between 1 and 100.")
            return
            
        self._data['required_approval_percentage'] = value
        db.update_node(self.uid, {'required_approval_percentage': value})

    def save(self) -> bool:
        """Save changes to database."""
        try:
            # If node doesn't exist, create it
            if not db.node_exists(self.uid):
                created = db.create_node_with_uid(
                    NodeLabels.REVIEW_CYCLE,
                    self._data,
                    self.uid
                )
                
                if created:
                    # If document version relationship needs to be established
                    document_version_uid = self._data.get('document_version_uid')
                    if document_version_uid:
                        # Create relationship between review cycle and document version
                        db.create_relationship(
                            document_version_uid,
                            self.uid,
                            RelTypes.FOR_REVIEW
                        )
                    
                    # Add relationships to reviewers if needed
                    reviewer_uids = self._data.get('reviewer_uids', [])
                    for reviewer_uid in reviewer_uids:
                        db.create_relationship(
                            self.uid, 
                            reviewer_uid,
                            RelTypes.REVIEWED_BY
                        )
                        
                return created
            
            # Otherwise update existing node
            return db.update_node(self.uid, self._data)
        except Exception as e:
            logger.error(f"Error saving review cycle: {e}")
            import traceback
            logger.error(traceback.format_exc())
            return False
    
    @property
    def start_date(self) -> Optional[datetime]:
        """Get when review cycle started."""
        return self._data.get('startDate') or (
            datetime.fromisoformat(self._data['started_at']) 
            if 'started_at' in self._data and self._data['started_at'] else None
        )
    
    @property
    def due_date(self) -> Optional[datetime]:
        """Get when review is due."""
        return self._data.get('dueDate') or (
            datetime.fromisoformat(self._data['due_date']) 
            if 'due_date' in self._data and self._data['due_date'] else None
        )
    
    @due_date.setter
    def due_date(self, date: datetime) -> None:
        """Set due date."""
        self._data['dueDate'] = date  # For backward compatibility
        self._data['due_date'] = date.isoformat() if date else None
        db.update_node(self.uid, {'dueDate': date, 'due_date': self._data['due_date']})
    
    @property
    def completion_date(self) -> Optional[datetime]:
        """Get when review was completed."""
        return self._data.get('completionDate') or (
            datetime.fromisoformat(self._data['completed_at']) 
            if 'completed_at' in self._data and self._data['completed_at'] else None
        )
    
    @completion_date.setter
    def completion_date(self, date: datetime) -> None:
        """Set completion date."""
        self._data['completionDate'] = date  # For backward compatibility
        self._data['completed_at'] = date.isoformat() if date else None
        db.update_node(self.uid, {'completionDate': date, 'completed_at': self._data['completed_at']})
    
    @property
    def instructions(self) -> str:
        """Get instructions for reviewers."""
        return self._data.get('instructions', '')
    
    @property
    def is_completed(self) -> bool:
        """Whether review cycle is completed."""
        return self.status in [WorkflowStatus.COMPLETED, WorkflowStatus.REJECTED]
    
    @property
    def is_overdue(self) -> bool:
        """Whether review is overdue."""
        return (
            self.due_date is not None 
            and datetime.now() > self.due_date 
            and not self.is_completed
        )
    
    @property
    def document_version_uid(self) -> Optional[str]:
        """Get the UID of the document version being reviewed."""
        # First check if we have it cached
        if self._data.get('document_version_uid'):
            return self._data.get('document_version_uid')
            
        # Otherwise query the database
        result = db.run_query(
            """
            MATCH (r:ReviewCycle {UID: $uid})-[:FOR_REVIEW]->(v:DocumentVersion)
            RETURN v.UID as version_uid
            """,
            {"uid": self.uid}
        )
        
        if result and 'version_uid' in result[0]:
            # Cache it for future use
            self._data['document_version_uid'] = result[0]['version_uid']
            return result[0]['version_uid']
            
        return None
    
    @property
    def document_uid(self) -> Optional[str]:
        """Get the UID of the controlled document being reviewed."""
        result = db.run_query(
            """
            MATCH (r:ReviewCycle {UID: $uid})-[:FOR_REVIEW]->(v:DocumentVersion)<-[:HAS_VERSION]-(d:ControlledDocument)
            RETURN d.UID as document_uid
            """,
            {"uid": self.uid}
        )
        
        if result and 'document_uid' in result[0]:
            return result[0]['document_uid']
            
        return None
    
    @property
    def document_version_number(self) -> Optional[str]:
        """Get the version number being reviewed."""
        result = db.run_query(
            """
            MATCH (r:ReviewCycle {UID: $uid})-[:FOR_REVIEW]->(v:DocumentVersion)
            RETURN v.version_number as document_version_number
            """,
            {"uid": self.uid}
        )
        
        if result and 'document_version_number' in result[0]:
            return result[0]['document_version_number']
            
        return None
    
    @property
    def document_version(self) -> Optional[Any]:
        """Get the document version being reviewed."""
        from CDocs.models.document import DocumentVersion
        
        version_uid = self.document_version_uid
        if version_uid:
            return DocumentVersion(uid=version_uid)
            
        return None
    
    @property
    def document(self) -> Optional[Any]:
        """Get the controlled document being reviewed."""
        from CDocs.models.document import ControlledDocument
        
        document_uid = self.document_uid
        if document_uid:
            return ControlledDocument(uid=document_uid)
            
        return None
    
    @property
    def reviewers(self) -> List[DocUser]:
        """Get users assigned as reviewers."""
        if self._reviewers_cache is not None:
            return self._reviewers_cache
            
        result = db.run_query(
            """
            MATCH (r:ReviewCycle {UID: $uid})-[:REVIEWED_BY]->(u:User)
            RETURN u
            """,
            {"uid": self.uid}
        )
        
        reviewers = [DocUser(record['u']) for record in result if 'u' in record]
        self._reviewers_cache = reviewers
        return reviewers
    
    @property
    def reviewer_uids(self) -> List[str]:
        """Get UIDs of users assigned as reviewers."""
        return [reviewer.uid for reviewer in self.reviewers]
    
    def add_reviewer(self, reviewer: Union[DocUser, str]) -> bool:
        """
        Add a reviewer to the review cycle.
        
        Args:
            reviewer: User or UID to add as reviewer
            
        Returns:
            Boolean indicating success
        """
        try:
            reviewer_uid = reviewer.uid if isinstance(reviewer, DocUser) else reviewer
            
            # Check if already a reviewer
            if reviewer_uid in self.reviewer_uids:
                return True
                
            # Add reviewer relationship
            success = db.create_relationship(
                self.uid,
                reviewer_uid,
                RelTypes.REVIEWED_BY
            )
            
            # Create a ReviewerAssignment node to track this reviewer's progress
            if success:
                # Get user info 
                user_info = db.run_query(
                    """
                    MATCH (u:User {UID: $uid})
                    RETURN u.Name as name
                    """,
                    {"uid": reviewer_uid}
                )
                
                # Create assignment data
                assignment_uid = str(uuid.uuid4())
                assignment_data = {
                    'UID': assignment_uid,
                    'reviewer_uid': reviewer_uid,
                    'user_uid': reviewer_uid,  # For workflow_base compatibility
                    'reviewer_name': user_info[0]['name'] if user_info else "Unknown",
                    'user_name': user_info[0]['name'] if user_info else "Unknown",  # For workflow_base compatibility
                    'review_cycle_uid': self.uid,
                    'status': AssignmentStatus.PENDING,
                    'assigned_at': datetime.now().isoformat(),
                    'assigned_date': datetime.now()  # For backward compatibility
                }
                
                # Create the node
                db.create_node(NodeLabels.REVIEWER_ASSIGNMENT, assignment_data)
                
                # Create the ASSIGNMENT relationship between ReviewCycle and ReviewerAssignment
                db.create_relationship(
                    self.uid, 
                    assignment_uid,
                    RelTypes.ASSIGNMENT
                )
            
            # Clear cache
            self._reviewers_cache = None
            
            return success
                
        except Exception as e:
            logger.error(f"Error adding reviewer: {e}")
            return False
    
    def remove_reviewer(self, reviewer: Union[DocUser, str]) -> bool:
        """
        Remove a reviewer from the review cycle.
        
        Args:
            reviewer: User or UID to remove
            
        Returns:
            Boolean indicating success
        """
        try:
            reviewer_uid = reviewer.uid if isinstance(reviewer, DocUser) else reviewer
            
            # Remove relationship
            rel_result = db.run_query(
                """
                MATCH (r:ReviewCycle {UID: $review_uid})-[rel:REVIEWED_BY]->(u:User {UID: $user_uid})
                DELETE rel
                RETURN count(rel) AS deleted
                """,
                {"review_uid": self.uid, "user_uid": reviewer_uid}
            )
            
            success = rel_result and rel_result[0].get('deleted', 0) > 0
            
            # Also remove any associated ReviewerAssignment
            if success:
                db.run_query(
                    """
                    MATCH (a:ReviewerAssignment)
                    WHERE a.review_cycle_uid = $review_uid AND a.reviewer_uid = $reviewer_uid
                    DELETE a
                    """,
                    {"review_uid": self.uid, "reviewer_uid": reviewer_uid}
                )
            
            # Clear cache
            self._reviewers_cache = None
            
            return success
                
        except Exception as e:
            logger.error(f"Error removing reviewer: {e}")
            return False
    
    @property
    def comments(self) -> List[ReviewComment]:
        """Get all comments for this review cycle."""
        if self._comments_cache is not None:
            return self._comments_cache
            
        result = db.run_query(
            """
            MATCH (c:ReviewComment)-[:COMMENTED_ON]->(r:ReviewCycle {UID: $uid})
            RETURN c
            ORDER BY c.timestamp DESC
            """,
            {"uid": self.uid}
        )
        
        comments = [ReviewComment(record['c']) for record in result if 'c' in record]
        self._comments_cache = comments
        return comments
    
    def add_comment(self, commenter: Union[DocUser, str], 
                   text: str,
                   requires_resolution: bool = False) -> Optional[ReviewComment]:
        """
        Add a comment to the review cycle.
        
        Args:
            commenter: User making the comment or their UID
            text: Comment text
            requires_resolution: Whether this comment requires resolution
            
        Returns:
            New ReviewComment instance or None if creation failed
        """
        comment = ReviewComment.create(
            self.uid, 
            commenter, 
            text, 
            requires_resolution
        )
        
        if comment:
            # Update status if needed
            if self.status == WorkflowStatus.PENDING:
                self.status = WorkflowStatus.IN_PROGRESS
                
            # Clear cache
            self._comments_cache = None
            
        return comment
    
    def get_unresolved_comments(self) -> List[ReviewComment]:
        """
        Get all unresolved comments that require resolution.
        
        Returns:
            List of unresolved ReviewComment instances
        """
        # First get all comments using the updated method which handles relationships correctly
        all_comments = self.comments
        
        # Then filter for unresolved ones
        return [
            comment for comment in all_comments 
            if comment.requires_resolution and not comment.is_resolved
        ]
    
    def complete_review(self, approved: bool = True) -> bool:
        """
        Complete the review cycle.
        
        Args:
            approved: Whether the document was approved in the review
            
        Returns:
            Boolean indicating success
        """
        try:
            # Check if we have unresolved comments
            unresolved = self.get_unresolved_comments()
            if unresolved and approved:
                logger.warning(f"Cannot complete review with {len(unresolved)} unresolved comments")
                return False
                
            # Update status
            self.status = WorkflowStatus.COMPLETED if approved else WorkflowStatus.REJECTED
            
            # Update document version status if needed
            document_version = self.document_version
            if document_version:
                if approved:
                    # Move to approval if document type requires it
                    doc_type = document_version.document.doc_type
                    doc_type_details = settings.get_document_type(doc_type)
                    
                    if doc_type_details and doc_type_details.get('required_approvals'):
                        document_version.status = 'IN_APPROVAL'
                        if document_version.document:
                            document_version.document.status = 'IN_APPROVAL'
                    else:
                        # No approval needed, mark as approved
                        document_version.status = 'APPROVED'
                        if document_version.document:
                            document_version.document.status = 'APPROVED'
                else:
                    # Review rejected, revert to draft
                    document_version.status = 'DRAFT'
                    if document_version.document:
                        document_version.document.status = 'DRAFT'
            
            return True
            
        except Exception as e:
            logger.error(f"Error completing review: {e}")
            return False
    
    def get_review_status(self, reviewer: Union[DocUser, str]) -> Dict[str, Any]:
        """
        Get review status for a specific reviewer.
        
        Args:
            reviewer: Reviewer to check
            
        Returns:
            Dictionary with review status information
        """
        reviewer_uid = reviewer.uid if isinstance(reviewer, DocUser) else reviewer
        
        # Check if user is a reviewer
        if reviewer_uid not in self.reviewer_uids:
            return {'status': 'NOT_ASSIGNED'}
            
        # Check if reviewer has commented
        result = db.run_query(
            """
            MATCH (r:ReviewCycle {UID: $review_uid})-[:COMMENTED_ON]->(c:ReviewComment)
            WHERE c.commenterUID = $reviewer_uid
            RETURN count(c) as comment_count
            """,
            {"review_uid": self.uid, "reviewer_uid": reviewer_uid}
        )
        
        comment_count = result[0]['comment_count'] if result else 0
        
        if comment_count > 0:
            return {
                'status': 'COMMENTED',
                'comment_count': comment_count
            }
        else:
            return {'status': 'PENDING'}
    
    def get_reviewer_assignments(self) -> List['ReviewerAssignment']:
        """Get all reviewer assignments for this review cycle."""
        try:
            # First try to get assignments using the ASSIGNMENT relationship
            result = db.run_query(
                """
                MATCH (r:ReviewCycle {UID: $review_uid})-[:ASSIGNMENT]->(a:ReviewerAssignment)
                RETURN a
                """,
                {"review_uid": self.uid}
            )
            
            assignments = []
            
            # If we found assignments via relationships, use those
            if result and len(result) > 0:
                assignments = [ReviewerAssignment(record['a']) for record in result if 'a' in record]
                
                # If we found some assignments but not for all reviewers, we need to check for legacy ones
                if len(assignments) < len(self.reviewer_uids):
                    # Get all reviewer UIDs that already have an assignment
                    assigned_reviewer_uids = [a.reviewer_uid for a in assignments]
                    
                    # Look for reviewers without assignments
                    for reviewer_uid in self.reviewer_uids:
                        if reviewer_uid not in assigned_reviewer_uids:
                            # Create assignment for this reviewer
                            assignment = self.get_reviewer_assignment(reviewer_uid)
                            if assignment:
                                assignments.append(assignment)
            else:
                # Try the old method using REVIEWED_BY + property lookup
                result = db.run_query(
                    """
                    MATCH (r:ReviewCycle {UID: $review_uid})-[:REVIEWED_BY]->(u:User)
                    OPTIONAL MATCH (a:ReviewerAssignment)
                    WHERE a.review_cycle_uid = $review_uid AND a.reviewer_uid = u.UID
                    RETURN u, a
                    """,
                    {"review_uid": self.uid}
                )
                
                for record in result:
                    user = record.get('u', {})
                    assignment_data = record.get('a', {})
                    
                    # If no existing assignment record, create one with default values
                    if not assignment_data:
                        # Create a new ReviewerAssignment
                        assignment_uid = str(uuid.uuid4())
                        assignment_data = {
                            'UID': assignment_uid,
                            'reviewer_uid': user.get('UID'),
                            'user_uid': user.get('UID'),  # For workflow_base compatibility
                            'reviewer_name': user.get('Name'),
                            'user_name': user.get('Name'),  # For workflow_base compatibility
                            'review_cycle_uid': self.uid,
                            'status': 'PENDING',
                            'assigned_date': datetime.now(),
                            'assigned_at': datetime.now().isoformat()  # For workflow_base compatibility
                        }
                        
                        # Create the node in the database
                        from CDocs.db.schema_manager import NodeLabels
                        db.create_node(NodeLabels.REVIEWER_ASSIGNMENT, assignment_data)
                        
                        # Create the relationship
                        db.create_relationship(
                            self.uid,
                            assignment_uid,
                            RelTypes.ASSIGNMENT
                        )
                    else:
                        # Create relationship if it doesn't exist already (backward compatibility)
                        db.create_relationship(
                            self.uid,
                            assignment_data.get('UID'),
                            RelTypes.ASSIGNMENT
                        )
                    
                    # Create and return the assignment object
                    assignments.append(ReviewerAssignment(assignment_data))
            
            return assignments
                
        except Exception as e:
            logger.error(f"Error getting reviewer assignments: {e}")
            return []

    def get_reviewer_assignment(self, reviewer_uid: str) -> Optional['ReviewerAssignment']:
        """
        Get assignment for a specific reviewer.
        
        Args:
            reviewer_uid: UID of the reviewer
            
        Returns:
            ReviewerAssignment instance or None if not found
        """
        try:
            # First check if this user is a reviewer (using the REVIEWED_BY relationship)
            reviewer_check = db.run_query(
                """
                MATCH (r:ReviewCycle {UID: $review_uid})-[:REVIEWED_BY]->(u:User {UID: $reviewer_uid})
                RETURN count(u) > 0 as is_reviewer
                """,
                {"review_uid": self.uid, "reviewer_uid": reviewer_uid}
            )
            
            # Exit early if not a reviewer
            if not reviewer_check or not reviewer_check[0].get('is_reviewer', False):
                logger.warning(f"User {reviewer_uid} is not a reviewer for cycle {self.uid}")
                return None
                
            # Try to find existing assignment node using ASSIGNMENT relationship
            result = db.run_query(
                """
                MATCH (r:ReviewCycle {UID: $review_uid})-[:ASSIGNMENT]->(a:ReviewerAssignment)
                WHERE a.reviewer_uid = $reviewer_uid OR a.user_uid = $reviewer_uid
                RETURN a
                """,
                {"review_uid": self.uid, "reviewer_uid": reviewer_uid}
            )
            
            # If assignment exists via relationship, return it
            if result and 'a' in result[0]:
                return ReviewerAssignment(result[0]['a'])
            
            # Also check by property for backward compatibility
            result = db.run_query(
                """
                MATCH (a:ReviewerAssignment)
                WHERE a.review_cycle_uid = $review_uid
                AND (a.reviewer_uid = $reviewer_uid OR a.user_uid = $reviewer_uid)
                RETURN a
                """,
                {"review_uid": self.uid, "reviewer_uid": reviewer_uid}
            )
            
            # If assignment exists via property, return it
            if result and 'a' in result[0]:
                # Create the relationship that was missing
                db.create_relationship(
                    self.uid,
                    result[0]['a']['UID'],
                    RelTypes.ASSIGNMENT
                )
                return ReviewerAssignment(result[0]['a'])
            
            # Otherwise, create a new assignment to represent this reviewer    
            # Get user info
            user_info = db.run_query(
                """
                MATCH (u:User {UID: $uid})
                RETURN u.Name as name
                """,
                {"uid": reviewer_uid}
            )
            
            # Create new assignment
            assignment_uid = str(uuid.uuid4())
            assignment_data = {
                'UID': assignment_uid,
                'reviewer_uid': reviewer_uid,
                'user_uid': reviewer_uid,  # For workflow_base compatibility
                'reviewer_name': user_info[0]['name'] if user_info else "Unknown",
                'user_name': user_info[0]['name'] if user_info else "Unknown",  # For workflow_base compatibility
                'review_cycle_uid': self.uid,
                'status': AssignmentStatus.PENDING,
                'assigned_date': datetime.now(),
                'assigned_at': datetime.now().isoformat()  # For workflow_base compatibility
            }
            
            # Create the node in the database
            from CDocs.db.schema_manager import NodeLabels
            db.create_node(NodeLabels.REVIEWER_ASSIGNMENT, assignment_data)
            
            # Create the ASSIGNMENT relationship
            db.create_relationship(
                self.uid,
                assignment_uid,
                RelTypes.ASSIGNMENT
            )
            
            # Return the new assignment
            return ReviewerAssignment(assignment_data)
                
        except Exception as e:
            logger.error(f"Error getting reviewer assignment: {e}")
            return None

    def is_reviewer(self, reviewer_uid: str) -> bool:
        """
        Check if a user is a reviewer for this cycle.
        
        Args:
            reviewer_uid: UID of the user to check
            
        Returns:
            Boolean indicating if user is a reviewer
        """
        try:
            # Use the REVIEWED_BY relationship directly
            result = db.run_query(
                """
                MATCH (r:ReviewCycle {UID: $review_uid})-[:REVIEWED_BY]->(u:User {UID: $reviewer_uid})
                RETURN count(u) > 0 as is_reviewer
                """,
                {"review_uid": self.uid, "reviewer_uid": reviewer_uid}
            )
            
            return result and result[0].get('is_reviewer', False)
                
        except Exception as e:
            logger.error(f"Error checking if user is reviewer: {e}")
            return False

    def get_next_reviewer(self, current_sequence: int) -> Optional['ReviewerAssignment']:
        """
        Get the next reviewer in sequence for sequential reviews.
        
        Args:
            current_sequence: Current sequence number
            
        Returns:
            Next ReviewerAssignment in sequence or None
        """
        try:
            if not self.sequential:
                return None
                
            result = db.run_query(
                """
                MATCH (a:ReviewerAssignment)
                WHERE a.review_cycle_uid = $review_uid
                AND a.sequence_order > $sequence
                AND a.status = 'PENDING'
                RETURN a
                ORDER BY a.sequence_order
                LIMIT 1
                """,
                {"review_uid": self.uid, "sequence": current_sequence}
            )
            
            if result and 'a' in result[0]:
                return ReviewerAssignment(result[0]['a'])
                
            return None
            
        except Exception as e:
            logger.error(f"Error getting next reviewer: {e}")
            return None

    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary representation."""
        result = BaseModel.to_dict(self)
        
        # Add status details
        status_details = settings.REVIEW_STATUSES.get(self.status, {})
        if status_details:
            result['statusName'] = status_details.get('name', self.status)
            result['statusColor'] = status_details.get('color', '#000000')
            result['statusIcon'] = status_details.get('icon', 'clock')
            
        # Add reviewer information
        result['reviewers'] = [
            {
                'UID': reviewer.uid,
                'name': reviewer.name,
                'email': reviewer.email
            }
            for reviewer in self.reviewers
        ]
        
        # Add comment counts
        all_comments = self.comments
        result['commentCount'] = len(all_comments)
        result['unresolvedCount'] = len(self.get_unresolved_comments())
        
        # Add document version info
        version = self.document_version
        if version:
            result['documentVersion'] = {
                'UID': version.uid,
                'versionNumber': version.version_number,
                'status': version.status
            }
            document = version.document
            if document:
                result['document'] = {
                    'UID': document.uid,
                    'docNumber': document.doc_number,
                    'title': document.title
                }
        
        return result

    def update_status(self) -> None:
        """Update the cycle status based on assignment statuses."""
        assignments = self.get_reviewer_assignments()
        
        if not assignments:
            return
        
        # Count completion status
        total = len(assignments)
        completed = sum(1 for a in assignments if a.status == AssignmentStatus.COMPLETED)
        rejected = sum(1 for a in assignments if a.status == AssignmentStatus.REJECTED)
        
        # Calculate completion percentage
        completion_percentage = (completed / total) * 100 if total > 0 else 0
        
        # If any rejections, mark as rejected
        if rejected > 0:
            self.status = WorkflowStatus.REJECTED
            return
            
        # If all completed, mark as completed
        if completed == total:
            self.status = WorkflowStatus.COMPLETED
            return
            
        # If above required percentage, mark as completed
        required_percentage = self.required_approval_percentage
        if completion_percentage >= required_percentage:
            self.status = WorkflowStatus.COMPLETED
            return
            
        # Otherwise mark as in progress if at least one assignment is active
        if self.status == WorkflowStatus.PENDING and any(a.status == AssignmentStatus.IN_PROGRESS for a in assignments):
            self.status = WorkflowStatus.IN_PROGRESS
            
    def get_progress(self) -> Dict[str, int]:
        """Get progress statistics for the cycle."""
        assignments = self.get_reviewer_assignments()
        
        # Count by status
        total = len(assignments)
        pending = sum(1 for a in assignments if a.status == AssignmentStatus.PENDING)
        in_progress = sum(1 for a in assignments if a.status == AssignmentStatus.IN_PROGRESS)
        completed = sum(1 for a in assignments if a.status == AssignmentStatus.COMPLETED)
        rejected = sum(1 for a in assignments if a.status == AssignmentStatus.REJECTED)
        skipped = sum(1 for a in assignments if a.status == AssignmentStatus.SKIPPED)
        
        return {
            "total": total,
            "pending": pending,
            "in_progress": in_progress,
            "completed": completed,
            "rejected": rejected,
            "skipped": skipped,
        }

Parameters

Name Type Default Kind
bases WorkflowCycleBase, DocumentVersionRelationMixin, BaseModel -

Parameter Details

bases: Parameter of type WorkflowCycleBase, DocumentVersionRelationMixin, BaseModel

Return Value

Returns unspecified type

Class Interface

Methods

__init__(self, data, uid)

Purpose: Initialize a review cycle. Args: data: Dictionary of review cycle properties uid: Review cycle UID to load from database (if data not provided)

Parameters:

  • data: Type: Optional[Dict[str, Any]]
  • uid: Type: Optional[str]

Returns: None

create(cls, document_version_uid, reviewers, due_date, instructions, properties) -> Optional['ReviewCycle']

Purpose: Create a new review cycle. Args: document_version_uid: UID of the document version to review reviewers: List of users or UIDs to assign as reviewers due_date: Date when review should be completed instructions: Instructions for reviewers properties: Additional properties for the review cycle Returns: New ReviewCycle instance or None if creation failed

Parameters:

  • cls: Parameter
  • document_version_uid: Type: str
  • reviewers: Type: List[Union[DocUser, str]]
  • due_date: Type: Optional[datetime]
  • instructions: Type: str
  • properties: Type: Optional[Dict[str, Any]]

Returns: Returns Optional['ReviewCycle']

get_reviews_for_document(cls, document_uid) -> List['ReviewCycle']

Purpose: Get all review cycles for a document. Args: document_uid: UID of the document Returns: List of ReviewCycle instances

Parameters:

  • cls: Parameter
  • document_uid: Type: str

Returns: Returns List['ReviewCycle']

status(self) -> str property

Purpose: Get review cycle status.

Returns: Returns str

status(self, value) -> None

Purpose: Set review cycle status.

Parameters:

  • value: Type: str

Returns: Returns None

initiated_by_uid(self) -> Optional[str] property

Purpose: Get the UID of the user who initiated this review cycle.

Returns: Returns Optional[str]

initiated_by_uid(self, uid) -> None

Purpose: Set the UID of the user who initiated this review cycle.

Parameters:

  • uid: Type: str

Returns: Returns None

required_approval_percentage(self) -> int property

Purpose: Get the required approval percentage for this review cycle.

Returns: Returns int

required_approval_percentage(self, value) -> None

Purpose: Set the required approval percentage.

Parameters:

  • value: Type: int

Returns: Returns None

save(self) -> bool

Purpose: Save changes to database.

Returns: Returns bool

start_date(self) -> Optional[datetime] property

Purpose: Get when review cycle started.

Returns: Returns Optional[datetime]

due_date(self) -> Optional[datetime] property

Purpose: Get when review is due.

Returns: Returns Optional[datetime]

due_date(self, date) -> None

Purpose: Set due date.

Parameters:

  • date: Type: datetime

Returns: Returns None

completion_date(self) -> Optional[datetime] property

Purpose: Get when review was completed.

Returns: Returns Optional[datetime]

completion_date(self, date) -> None

Purpose: Set completion date.

Parameters:

  • date: Type: datetime

Returns: Returns None

instructions(self) -> str property

Purpose: Get instructions for reviewers.

Returns: Returns str

is_completed(self) -> bool property

Purpose: Whether review cycle is completed.

Returns: Returns bool

is_overdue(self) -> bool property

Purpose: Whether review is overdue.

Returns: Returns bool

document_version_uid(self) -> Optional[str] property

Purpose: Get the UID of the document version being reviewed.

Returns: Returns Optional[str]

document_uid(self) -> Optional[str] property

Purpose: Get the UID of the controlled document being reviewed.

Returns: Returns Optional[str]

document_version_number(self) -> Optional[str] property

Purpose: Get the version number being reviewed.

Returns: Returns Optional[str]

document_version(self) -> Optional[Any] property

Purpose: Get the document version being reviewed.

Returns: Returns Optional[Any]

document(self) -> Optional[Any] property

Purpose: Get the controlled document being reviewed.

Returns: Returns Optional[Any]

reviewers(self) -> List[DocUser] property

Purpose: Get users assigned as reviewers.

Returns: Returns List[DocUser]

reviewer_uids(self) -> List[str] property

Purpose: Get UIDs of users assigned as reviewers.

Returns: Returns List[str]

add_reviewer(self, reviewer) -> bool

Purpose: Add a reviewer to the review cycle. Args: reviewer: User or UID to add as reviewer Returns: Boolean indicating success

Parameters:

  • reviewer: Type: Union[DocUser, str]

Returns: Returns bool

remove_reviewer(self, reviewer) -> bool

Purpose: Remove a reviewer from the review cycle. Args: reviewer: User or UID to remove Returns: Boolean indicating success

Parameters:

  • reviewer: Type: Union[DocUser, str]

Returns: Returns bool

comments(self) -> List[ReviewComment] property

Purpose: Get all comments for this review cycle.

Returns: Returns List[ReviewComment]

add_comment(self, commenter, text, requires_resolution) -> Optional[ReviewComment]

Purpose: Add a comment to the review cycle. Args: commenter: User making the comment or their UID text: Comment text requires_resolution: Whether this comment requires resolution Returns: New ReviewComment instance or None if creation failed

Parameters:

  • commenter: Type: Union[DocUser, str]
  • text: Type: str
  • requires_resolution: Type: bool

Returns: Returns Optional[ReviewComment]

get_unresolved_comments(self) -> List[ReviewComment]

Purpose: Get all unresolved comments that require resolution. Returns: List of unresolved ReviewComment instances

Returns: Returns List[ReviewComment]

complete_review(self, approved) -> bool

Purpose: Complete the review cycle. Args: approved: Whether the document was approved in the review Returns: Boolean indicating success

Parameters:

  • approved: Type: bool

Returns: Returns bool

get_review_status(self, reviewer) -> Dict[str, Any]

Purpose: Get review status for a specific reviewer. Args: reviewer: Reviewer to check Returns: Dictionary with review status information

Parameters:

  • reviewer: Type: Union[DocUser, str]

Returns: Returns Dict[str, Any]

get_reviewer_assignments(self) -> List['ReviewerAssignment']

Purpose: Get all reviewer assignments for this review cycle.

Returns: Returns List['ReviewerAssignment']

get_reviewer_assignment(self, reviewer_uid) -> Optional['ReviewerAssignment']

Purpose: Get assignment for a specific reviewer. Args: reviewer_uid: UID of the reviewer Returns: ReviewerAssignment instance or None if not found

Parameters:

  • reviewer_uid: Type: str

Returns: Returns Optional['ReviewerAssignment']

is_reviewer(self, reviewer_uid) -> bool

Purpose: Check if a user is a reviewer for this cycle. Args: reviewer_uid: UID of the user to check Returns: Boolean indicating if user is a reviewer

Parameters:

  • reviewer_uid: Type: str

Returns: Returns bool

get_next_reviewer(self, current_sequence) -> Optional['ReviewerAssignment']

Purpose: Get the next reviewer in sequence for sequential reviews. Args: current_sequence: Current sequence number Returns: Next ReviewerAssignment in sequence or None

Parameters:

  • current_sequence: Type: int

Returns: Returns Optional['ReviewerAssignment']

to_dict(self) -> Dict[str, Any]

Purpose: Convert to dictionary representation.

Returns: Returns Dict[str, Any]

update_status(self) -> None

Purpose: Update the cycle status based on assignment statuses.

Returns: Returns None

get_progress(self) -> Dict[str, int]

Purpose: Get progress statistics for the cycle.

Returns: Returns Dict[str, int]

Required Imports

import logging
import uuid
from typing import Dict
from typing import List
from typing import Any

Usage Example

# Example usage:
# result = ReviewCycle(bases)

Tags

class reviewcycle

Similar Components

AI-powered semantic similarity - components with related functionality:

  • class ReviewCycle 97.6% similar

    Model representing a review cycle for a document version.

    From: /tf/active/vicechatdev/CDocs/models/review.py

  • class ApprovalCycle_v1 80.4% similar

    Model representing a approval cycle for a document version.

    From: /tf/active/vicechatdev/CDocs/models/approval.py

  • class ApprovalCycle_v2 78.9% similar

    Model representing an approval cycle for a document version.

    From: /tf/active/vicechatdev/CDocs single class/models/approval.py

  • class ApprovalCycle 78.2% similar

    Model representing an approval cycle for a document version.

    From: /tf/active/vicechatdev/CDocs/models/approval_bis.py

  • function create_review_cycle_v1 72.8% similar

    Creates a review cycle for a specific document version, assigning reviewers with configurable approval requirements and workflow settings.

    From: /tf/active/vicechatdev/CDocs single class/controllers/review_controller.py
← Back to Browse