app/services/learner_workflow.py - master - artyuhovs/AIFeedBackTrainingBot

app/services/learner_workflow.py
191 строка7 KB
Codex
Завершить архитектурное оздоровление и исправить итоги
15 июл 2026, 11:05
15 июл 2026, 11:05c65f9a6
from __future__ import annotations

from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime
from enum import StrEnum

from app.db.read_models import (
    LearnerAssignmentRecord,
    PeerReviewTaskRecord,
    ReceivedReviewRecord,
    WrittenReviewFeedbackRecord,
)


class LearnerPhase(StrEnum):
    UPCOMING = "upcoming"
    SUBMISSION = "submission"
    PEER_REVIEW = "peer_review"
    WAITING_FOR_PEER_REVIEW = "waiting_for_peer_review"
    WAITING_FOR_RESULTS = "waiting_for_results"
    REVIEW_FEEDBACK = "review_feedback"
    WAITING_FOR_REVIEW_FEEDBACK = "waiting_for_review_feedback"
    REVIEWER_FEEDBACK = "reviewer_feedback"
    IMPROVEMENT = "improvement"
    WAITING_FOR_IMPROVEMENT = "waiting_for_improvement"
    COMPLETE_WITHOUT_REVIEWS = "complete_without_reviews"
    MISSED_SUBMISSION = "missed_submission"
    MISSED_IMPROVEMENT = "missed_improvement"
    COMPLETE = "complete"


@dataclass(frozen=True)
class LearnerWorkflowItem:
    assignment: LearnerAssignmentRecord
    phase: LearnerPhase
    reviews_to_write: tuple[PeerReviewTaskRecord, ...] = ()
    reviews_received: tuple[ReceivedReviewRecord, ...] = ()
    written_review_feedback: tuple[WrittenReviewFeedbackRecord, ...] = ()

    @property
    def stream_publication_id(self) -> int:
        return int(self.assignment["stream_publication_id"])

    @property
    def is_current(self) -> bool:
        return self.phase not in {
            LearnerPhase.UPCOMING,
            LearnerPhase.COMPLETE,
            LearnerPhase.COMPLETE_WITHOUT_REVIEWS,
            LearnerPhase.MISSED_SUBMISSION,
            LearnerPhase.MISSED_IMPROVEMENT,
        }


TERMINAL_LEARNER_PHASES = {
    LearnerPhase.COMPLETE,
    LearnerPhase.COMPLETE_WITHOUT_REVIEWS,
    LearnerPhase.MISSED_SUBMISSION,
    LearnerPhase.MISSED_IMPROVEMENT,
}


def learning_is_complete(items: Sequence[LearnerWorkflowItem]) -> bool:
    return bool(items) and all(item.phase in TERMINAL_LEARNER_PHASES for item in items)


def build_learner_workflow(
    assignments: Sequence[LearnerAssignmentRecord],
    review_tasks: Sequence[PeerReviewTaskRecord],
    received_reviews: Mapping[int, Sequence[ReceivedReviewRecord]],
    at: str,
    written_review_feedback: Mapping[
        int,
        Sequence[WrittenReviewFeedbackRecord],
    ]
    | None = None,
) -> list[LearnerWorkflowItem]:
    """Derive every learner-facing stage from one timeline and one set of rules."""
    moment = datetime.fromisoformat(at)
    reviews_by_publication: dict[int, list[PeerReviewTaskRecord]] = {}
    for review in review_tasks:
        if _get(review, "status") not in {"assigned", "completed"}:
            continue
        publication_id = int(review["stream_publication_id"])
        reviews_by_publication.setdefault(publication_id, []).append(review)

    items: list[LearnerWorkflowItem] = []
    feedback_by_publication = written_review_feedback or {}
    for assignment in assignments:
        assignment_id = int(assignment["id"])
        publication_id = int(assignment["stream_publication_id"])
        reviews_to_write = tuple(reviews_by_publication.get(publication_id, ()))
        reviews_for_work = tuple(received_reviews.get(assignment_id, ()))
        feedback_for_written_reviews = tuple(feedback_by_publication.get(publication_id, ()))
        phase = _phase_for_assignment(
            assignment,
            reviews_to_write,
            reviews_for_work,
            feedback_for_written_reviews,
            moment,
        )
        items.append(
            LearnerWorkflowItem(
                assignment=assignment,
                phase=phase,
                reviews_to_write=reviews_to_write,
                reviews_received=reviews_for_work,
                written_review_feedback=feedback_for_written_reviews,
            )
        )
    return items


def _phase_for_assignment(
    assignment: Mapping[str, object],
    reviews_to_write: Sequence[Mapping[str, object]],
    reviews_received: Sequence[Mapping[str, object]],
    written_review_feedback: Sequence[Mapping[str, object]],
    moment: datetime,
) -> LearnerPhase:
    scheduled_at = _datetime(assignment, "scheduled_at")
    deadline_at = _datetime(assignment, "deadline_at")
    review_deadline_at = _datetime(assignment, "review_deadline_at")
    review_quality_deadline_at = _datetime(assignment, "review_quality_deadline_at")

    if scheduled_at is not None and moment < scheduled_at:
        return LearnerPhase.UPCOMING
    if deadline_at is not None and moment < deadline_at and not _get(assignment, "review_closed_at"):
        return LearnerPhase.SUBMISSION
    if review_deadline_at is None:
        return (
            LearnerPhase.COMPLETE
            if _get(assignment, "status") in {"submitted", "completed", "excused"}
            else LearnerPhase.MISSED_SUBMISSION
        )
    if not _get(assignment, "review_closed_at") and moment < review_deadline_at:
        return LearnerPhase.PEER_REVIEW if reviews_to_write else LearnerPhase.WAITING_FOR_PEER_REVIEW

    if not _get(assignment, "review_closed_at"):
        return LearnerPhase.WAITING_FOR_RESULTS
    has_feedback_context = bool(reviews_received or written_review_feedback)
    if _get(assignment, "status") == "excused":
        return LearnerPhase.COMPLETE
    if _get(assignment, "status") not in {"submitted", "completed"}:
        if (
            _get(assignment, "review_quality_closed_at")
            and written_review_feedback
            and not _get(assignment, "reviewer_feedback_seen_at")
        ):
            return LearnerPhase.REVIEWER_FEEDBACK
        return LearnerPhase.MISSED_SUBMISSION
    if not has_feedback_context:
        return LearnerPhase.COMPLETE_WITHOUT_REVIEWS
    if not _get(assignment, "review_quality_closed_at"):
        if review_quality_deadline_at is None or moment < review_quality_deadline_at:
            if reviews_received and any(not review_quality_completed(review) for review in reviews_received):
                return LearnerPhase.REVIEW_FEEDBACK
            return LearnerPhase.WAITING_FOR_REVIEW_FEEDBACK
        return LearnerPhase.WAITING_FOR_IMPROVEMENT
    if _get(assignment, "correction_status") in {"completed", "excused"}:
        return LearnerPhase.COMPLETE
    correction_opened_at = _datetime(assignment, "correction_opened_at")
    correction_deadline_at = _datetime(assignment, "correction_deadline_at")
    if correction_opened_at is None or correction_deadline_at is None:
        return LearnerPhase.WAITING_FOR_IMPROVEMENT
    if moment < correction_opened_at:
        return LearnerPhase.WAITING_FOR_IMPROVEMENT
    if correction_opened_at <= moment < correction_deadline_at:
        return LearnerPhase.IMPROVEMENT
    return (
        LearnerPhase.COMPLETE
        if _get(assignment, "correction_status") in {"submitted", "completed", "excused"}
        else LearnerPhase.MISSED_IMPROVEMENT
    )


def _datetime(row: Mapping[str, object], key: str) -> datetime | None:
    value = _get(row, key)
    return datetime.fromisoformat(str(value)) if value else None


def _get(row: Mapping[str, object], key: str, default: object = None) -> object:
    try:
        return row[key]
    except (KeyError, IndexError):
        return default


def review_quality_completed(review: Mapping[str, object]) -> bool:
    return bool(_get(review, "quality_score") is not None and str(_get(review, "quality_comment") or "").strip())