automated progress bar

utils/user_progress.py

@@ -0,0 +1,113 @@
+# utils/user_progress.py
+
+import datetime
+from sqlalchemy import and_
+from utils.database import get_db
+from data.models import UserProgress, Annotation, Validation, Annotator, TTSData
+from utils.logger import Logger
+
+log = Logger()
+
+
+def get_user_progress(user_id: int, target_annotator_id: int) -> UserProgress:
+    """Get or create user progress record for a specific target annotator"""
+    with get_db() as db:
+        progress = db.query(UserProgress).filter(
+            and_(
+                UserProgress.user_id == user_id,
+                UserProgress.target_annotator_id == target_annotator_id
+            )
+        ).first()
+        
+        if not progress:
+            # Create new progress record
+            progress = UserProgress(
+                user_id=user_id,
+                target_annotator_id=target_annotator_id,
+                last_reviewed_annotation_id=None,
+                last_position=0,
+                updated_at=datetime.datetime.now()
+            )
+            db.add(progress)
+            db.commit()
+            log.info(f"Created new progress record for user {user_id} on target annotator {target_annotator_id}")
+        
+        return progress
+
+
+def update_user_progress(user_id: int, target_annotator_id: int, annotation_id: int, position: int):
+    """Update user progress with the latest reviewed annotation"""
+    with get_db() as db:
+        progress = get_user_progress(user_id, target_annotator_id)
+        progress.last_reviewed_annotation_id = annotation_id
+        progress.last_position = position
+        progress.updated_at = datetime.datetime.now()
+        db.commit()
+        log.info(f"Updated progress for user {user_id}: annotation {annotation_id}, position {position}")
+
+
+def get_next_unreviewed_annotation(user_id: int, target_annotator_id: int) -> tuple[int, int]:
+    """
+    Get the next unreviewed annotation ID and its position in the review list.
+    Returns (annotation_id, position) or (None, 0) if no unreviewed items found.
+    """
+    with get_db() as db:
+        # Get all annotations for the target annotator, ordered by ID
+        annotations = db.query(Annotation).filter(
+            Annotation.annotator_id == target_annotator_id
+        ).order_by(Annotation.id).all()
+        
+        if not annotations:
+            return None, 0
+        
+        # Find the first annotation that hasn't been reviewed by this user
+        for position, annotation in enumerate(annotations):
+            # Check if this annotation has been validated by the current user
+            validation = db.query(Validation).filter(
+                and_(
+                    Validation.annotation_id == annotation.id,
+                    Validation.validator_id == user_id
+                )
+            ).first()
+            
+            if not validation:
+                # This annotation hasn't been reviewed yet
+                return annotation.id, position
+        
+        # All annotations have been reviewed
+        return None, len(annotations)
+
+
+def get_last_reviewed_position(user_id: int, target_annotator_id: int) -> int:
+    """Get the position of the last reviewed annotation by the user"""
+    with get_db() as db:
+        progress = db.query(UserProgress).filter(
+            and_(
+                UserProgress.user_id == user_id,
+                UserProgress.target_annotator_id == target_annotator_id
+            )
+        ).first()
+        
+        if progress and progress.last_position is not None:
+            return progress.last_position
+        return 0
+
+
+def get_annotations_from_position(target_annotator_id: int, start_position: int, batch_size: int = 10) -> list:
+    """
+    Get annotations starting from a specific position with pagination.
+    Returns list of (annotation, filename, sentence) tuples.
+    """
+    with get_db() as db:
+        # Get annotations ordered by ID, starting from the specified position
+        annotations = db.query(
+            Annotation,
+            TTSData.filename,
+            TTSData.sentence
+        ).join(
+            TTSData, Annotation.tts_data_id == TTSData.id
+        ).filter(
+            Annotation.annotator_id == target_annotator_id
+        ).order_by(Annotation.id).offset(start_position).limit(batch_size).all()
+        
+        return annotations