ctms/backend/app/services/project_reminder_service.py

from __future__ import annotations

import uuid
from datetime import date, datetime, time, timedelta, timezone
from zoneinfo import ZoneInfo

from sqlalchemy import and_, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.deps import get_cra_site_scope, is_system_admin
from app.core.project_permissions import role_has_api_permission
from app.crud import member as member_crud
from app.crud import site as site_crud
from app.models.acknowledgement import Acknowledgement, AcknowledgementType
from app.models.ae import AdverseEvent
from app.models.collaboration import CollaborationEditRequest, CollaborationFile, CollaborationMember
from app.models.distribution import Distribution, DistributionStatus, DistributionTargetType
from app.models.document import Document, DocumentStatus
from app.models.document_version import DocumentVersion
from app.models.milestone import Milestone
from app.models.monitoring_visit_issue import MonitoringVisitIssue
from app.models.site import Site
from app.models.subject import Subject
from app.models.user import User
from app.models.visit import Visit
from app.services import notification_service

BUSINESS_TIMEZONE = ZoneInfo("Asia/Shanghai")
RISK_DUE_SOON_DAYS = 3
MILESTONE_DUE_SOON_DAYS = 7
VISIT_WINDOW_DUE_SOON_DAYS = 3


def _date_due_at(value: date | None) -> datetime | None:
    if value is None:
        return None
    return datetime.combine(value, time.max, tzinfo=BUSINESS_TIMEZONE).astimezone(timezone.utc)


def _as_utc(value: datetime | None) -> datetime | None:
    if value is None:
        return None
    if value.tzinfo is None:
        return value.replace(tzinfo=timezone.utc)
    return value.astimezone(timezone.utc)


async def _count_and_earliest(db: AsyncSession, model, *filters):
    return (await db.execute(
        select(func.count(model.id), func.min(model.report_due_date if model is AdverseEvent else model.due_at))
        .where(*filters)
    )).one()


async def _sync_risk_reminders(
    db: AsyncSession,
    *,
    study_id: uuid.UUID,
    user,
    role: str,
    current_time: datetime,
) -> None:
    can_read_aes = is_system_admin(user) or await role_has_api_permission(
        db, study_id, role, "subject_aes:read"
    )
    can_read_monitoring = is_system_admin(user) or await role_has_api_permission(
        db, study_id, role, "monitoring_issues:read"
    )
    cra_scope = await get_cra_site_scope(db, study_id, user)
    site_ids = cra_scope[0] if cra_scope else None
    today = current_time.astimezone(BUSINESS_TIMEZONE).date()
    due_soon_date = today + timedelta(days=RISK_DUE_SOON_DAYS)

    ae_scope = [AdverseEvent.study_id == study_id, AdverseEvent.status != "CLOSED"]
    if site_ids is not None:
        ae_scope.append(AdverseEvent.site_id.in_(site_ids))
    overdue_aes = 0
    overdue_ae_due = None
    due_soon_aes = 0
    due_soon_ae_due = None
    if can_read_aes:
        overdue_aes, overdue_ae_due = await _count_and_earliest(
            db,
            AdverseEvent,
            *ae_scope,
            AdverseEvent.report_due_date.is_not(None),
            AdverseEvent.report_due_date < today,
        )
        due_soon_aes, due_soon_ae_due = await _count_and_earliest(
            db,
            AdverseEvent,
            *ae_scope,
            AdverseEvent.report_due_date >= today,
            AdverseEvent.report_due_date <= due_soon_date,
        )
    await notification_service.sync_aggregate_notification(
        db,
        study_id=study_id,
        recipient_id=user.id,
        category="RISK_OVERDUE_AE",
        priority="URGENT",
        title="逾期 AE 待处理",
        message=f"当前有 {overdue_aes} 条逾期 AE 需要跟进",
        action_path="/risk-issues/sae",
        source_type="RISK_OVERDUE_AE",
        source_id=str(study_id),
        count=int(overdue_aes) if can_read_aes else 0,
        due_at=_date_due_at(overdue_ae_due),
    )
    await notification_service.sync_aggregate_notification(
        db,
        study_id=study_id,
        recipient_id=user.id,
        category="RISK_AE_DUE_SOON",
        priority="HIGH",
        title="AE 上报时限临近",
        message=f"未来 {RISK_DUE_SOON_DAYS} 天内有 {due_soon_aes} 条 AE 到达上报时限",
        action_path="/risk-issues/sae",
        source_type="RISK_AE_DUE_SOON",
        source_id=str(study_id),
        count=int(due_soon_aes) if can_read_aes else 0,
        due_at=_date_due_at(due_soon_ae_due),
    )

    monitoring_scope = [
        MonitoringVisitIssue.study_id == study_id,
        MonitoringVisitIssue.status == "OPEN",
        MonitoringVisitIssue.due_at.is_not(None),
    ]
    if site_ids is not None:
        monitoring_scope.append(MonitoringVisitIssue.site_id.in_(site_ids))
    overdue_monitoring = 0
    overdue_monitoring_due = None
    due_soon_monitoring = 0
    due_soon_monitoring_due = None
    if can_read_monitoring:
        overdue_monitoring, overdue_monitoring_due = await _count_and_earliest(
            db,
            MonitoringVisitIssue,
            *monitoring_scope,
            MonitoringVisitIssue.due_at < current_time,
        )
        due_soon_monitoring, due_soon_monitoring_due = await _count_and_earliest(
            db,
            MonitoringVisitIssue,
            *monitoring_scope,
            MonitoringVisitIssue.due_at >= current_time,
            MonitoringVisitIssue.due_at <= current_time + timedelta(days=RISK_DUE_SOON_DAYS),
        )
    await notification_service.sync_aggregate_notification(
        db,
        study_id=study_id,
        recipient_id=user.id,
        category="RISK_OVERDUE_MONITORING",
        priority="HIGH",
        title="监查问题已逾期",
        message=f"当前有 {overdue_monitoring} 条监查问题已超过计划解决日期",
        action_path="/risk-issues/monitoring-visits",
        source_type="RISK_OVERDUE_MONITORING",
        source_id=str(study_id),
        count=int(overdue_monitoring) if can_read_monitoring else 0,
        due_at=_as_utc(overdue_monitoring_due),
    )
    await notification_service.sync_aggregate_notification(
        db,
        study_id=study_id,
        recipient_id=user.id,
        category="RISK_MONITORING_DUE_SOON",
        priority="HIGH",
        title="监查问题整改时限临近",
        message=f"未来 {RISK_DUE_SOON_DAYS} 天内有 {due_soon_monitoring} 条监查问题到期",
        action_path="/risk-issues/monitoring-visits",
        source_type="RISK_MONITORING_DUE_SOON",
        source_id=str(study_id),
        count=int(due_soon_monitoring) if can_read_monitoring else 0,
        due_at=_as_utc(due_soon_monitoring_due),
    )


async def _sync_document_distribution_reminders(
    db: AsyncSession,
    *,
    study_id: uuid.UUID,
    user,
    role: str,
    current_time: datetime,
) -> None:
    can_read_documents = is_system_admin(user) or await role_has_api_permission(
        db, study_id, role, "documents:read"
    )
    if not can_read_documents:
        await notification_service.resolve_missing_recipient_sources(
            db,
            study_id=study_id,
            recipient_id=user.id,
            source_type="DOCUMENT_DISTRIBUTION",
            active_source_ids=set(),
        )
        return
    site_ids = await site_crud.list_ids_by_contact_user(db, study_id, user.id)
    target_filters = [
        and_(
            Distribution.target_type == DistributionTargetType.USER,
            Distribution.target_id == str(user.id),
        ),
        and_(
            Distribution.target_type == DistributionTargetType.ROLE,
            Distribution.target_id == role,
        ),
    ]
    if site_ids:
        target_filters.append(and_(
            Distribution.target_type == DistributionTargetType.SITE,
            Distribution.target_id.in_({str(site_id) for site_id in site_ids}),
        ))
    rows = (await db.execute(
        select(Distribution, Document, DocumentVersion)
        .join(Document, Distribution.document_id == Document.id)
        .join(DocumentVersion, Distribution.version_id == DocumentVersion.id)
        .outerjoin(
            Acknowledgement,
            and_(
                Acknowledgement.distribution_id == Distribution.id,
                Acknowledgement.user_id == user.id,
                Acknowledgement.ack_type == AcknowledgementType.RECEIVED,
            ),
        )
        .where(
            Document.trial_id == study_id,
            Document.status == DocumentStatus.ACTIVE,
            Distribution.status == DistributionStatus.ACTIVE,
            Acknowledgement.id.is_(None),
            or_(*target_filters),
        )
    )).all()

    active_source_ids: set[str] = set()
    for distribution, document, version in rows:
        source_id = str(distribution.id)
        active_source_ids.add(source_id)
        distribution_due_at = _as_utc(distribution.due_at)
        if distribution_due_at and distribution_due_at < current_time:
            stage = "OVERDUE"
            category = "DOCUMENT_ACK_OVERDUE"
            priority = "URGENT"
            title = "文件回执已逾期"
        elif distribution_due_at and distribution_due_at <= current_time + timedelta(days=RISK_DUE_SOON_DAYS):
            stage = "DUE_SOON"
            category = "DOCUMENT_ACK_DUE_SOON"
            priority = "HIGH"
            title = "文件回执即将到期"
        else:
            stage = "PENDING"
            category = "DOCUMENT_DISTRIBUTION"
            priority = "NORMAL"
            title = "有新的文件版本待接收"
        due_version = distribution_due_at.isoformat() if distribution_due_at else "none"
        await notification_service.sync_state_notification(
            db,
            study_id=study_id,
            recipient_id=user.id,
            category=category,
            priority=priority,
            title=title,
            message=f"“{document.title}” {version.version_no} 已分发，请完成接收回执",
            action_path=f"/documents/{document.id}",
            source_type="DOCUMENT_DISTRIBUTION",
            source_id=source_id,
            source_version=f"{stage}:{due_version}",
            active=True,
            due_at=distribution_due_at,
        )
    await notification_service.resolve_missing_recipient_sources(
        db,
        study_id=study_id,
        recipient_id=user.id,
        source_type="DOCUMENT_DISTRIBUTION",
        active_source_ids=active_source_ids,
    )


async def _sync_milestone_reminders(
    db: AsyncSession,
    *,
    study_id: uuid.UUID,
    user,
    role: str,
    current_time: datetime,
) -> None:
    can_read = is_system_admin(user) or await role_has_api_permission(
        db, study_id, role, "project_milestones:read"
    )
    active_source_ids: set[str] = set()
    if can_read:
        today = current_time.astimezone(BUSINESS_TIMEZONE).date()
        due_soon_date = today + timedelta(days=MILESTONE_DUE_SOON_DAYS)
        milestones = (await db.scalars(
            select(Milestone).where(
                Milestone.study_id == study_id,
                Milestone.owner_id == user.id,
                Milestone.status != "DONE",
            )
        )).all()
        for milestone in milestones:
            due_date = milestone.adjusted_end_date or milestone.planned_date
            if due_date is None or due_date > due_soon_date:
                continue
            source_id = str(milestone.id)
            active_source_ids.add(source_id)
            overdue = due_date < today
            await notification_service.sync_state_notification(
                db,
                study_id=study_id,
                recipient_id=user.id,
                category="MILESTONE_OVERDUE" if overdue else "MILESTONE_DUE_SOON",
                priority="HIGH" if overdue else "NORMAL",
                title="项目里程碑已逾期" if overdue else "项目里程碑即将到期",
                message=f"“{milestone.name}”计划日期为 {due_date.isoformat()}",
                action_path="/project/milestones",
                source_type="PROJECT_MILESTONE",
                source_id=source_id,
                source_version=f"{'OVERDUE' if overdue else 'DUE_SOON'}:{due_date.isoformat()}:{milestone.status}",
                active=True,
                due_at=_date_due_at(due_date),
            )
    await notification_service.resolve_missing_recipient_sources(
        db,
        study_id=study_id,
        recipient_id=user.id,
        source_type="PROJECT_MILESTONE",
        active_source_ids=active_source_ids,
    )


async def _sync_visit_window_reminders(
    db: AsyncSession,
    *,
    study_id: uuid.UUID,
    user,
    role: str,
    current_time: datetime,
) -> None:
    """Project active visit windows to the members who can actually maintain them."""
    can_manage_visits = is_system_admin(user) or await role_has_api_permission(
        db, study_id, role, "visits:update"
    )
    active_source_ids: set[str] = set()
    if can_manage_visits:
        cra_scope = await get_cra_site_scope(db, study_id, user)
        site_ids = cra_scope[0] if cra_scope else None
        filters = [
            Visit.study_id == study_id,
            Visit.actual_date.is_(None),
            Visit.status.not_in(["DONE", "CANCELLED"]),
            Subject.status.not_in(["COMPLETED", "DROPPED"]),
            Site.is_active.is_(True),
        ]
        if site_ids is not None:
            filters.append(Subject.site_id.in_(site_ids))
        rows = (await db.execute(
            select(Visit, Subject)
            .join(Subject, Subject.id == Visit.subject_id)
            .join(Site, Site.id == Subject.site_id)
            .where(*filters)
        )).all()

        today = current_time.astimezone(BUSINESS_TIMEZONE).date()
        due_soon_date = today + timedelta(days=VISIT_WINDOW_DUE_SOON_DAYS)
        for visit, subject in rows:
            window_start = visit.window_start or visit.planned_date
            window_end = visit.window_end or visit.planned_date
            if window_start is None or window_end is None or window_start > window_end:
                continue

            if window_end < today:
                category = "VISIT_WINDOW_MISSED"
                priority = "HIGH"
                title = "受试者访视已错过窗口"
                message = "一个受试者访视已错过计划窗口，请进入详情核对并处理"
                stage = "MISSED"
            elif window_start <= due_soon_date:
                category = "VISIT_WINDOW_DUE_SOON"
                priority = "NORMAL"
                title = "受试者访视窗口临近"
                message = "一个受试者访视已进入近期窗口，请及时核对访视安排"
                stage = "DUE_SOON"
            else:
                continue

            source_id = str(visit.id)
            active_source_ids.add(source_id)
            await notification_service.sync_state_notification(
                db,
                study_id=study_id,
                recipient_id=user.id,
                category=category,
                priority=priority,
                title=title,
                message=message,
                action_path=f"/subjects/{subject.id}",
                source_type="SUBJECT_VISIT_WINDOW",
                source_id=source_id,
                source_version=f"{stage}:{window_start.isoformat()}:{window_end.isoformat()}",
                active=True,
                due_at=_date_due_at(window_end),
            )
    await notification_service.resolve_missing_recipient_sources(
        db,
        study_id=study_id,
        recipient_id=user.id,
        source_type="SUBJECT_VISIT_WINDOW",
        active_source_ids=active_source_ids,
    )


async def _sync_collaboration_edit_request_reminders(
    db: AsyncSession,
    *,
    study_id: uuid.UUID,
    user,
    role: str,
) -> None:
    can_read = is_system_admin(user) or await role_has_api_permission(
        db, study_id, role, "collaboration:read"
    )
    if not can_read:
        await notification_service.resolve_missing_recipient_sources(
            db,
            study_id=study_id,
            recipient_id=user.id,
            source_type="COLLABORATION_EDIT_REQUEST",
            active_source_ids=set(),
        )
        return
    manages_file = or_(
        CollaborationFile.owner_id == user.id,
        select(CollaborationMember.id).where(
            CollaborationMember.file_id == CollaborationFile.id,
            CollaborationMember.user_id == user.id,
            CollaborationMember.role == "MANAGER",
        ).exists(),
    )
    rows = (await db.execute(
        select(CollaborationEditRequest, CollaborationFile, User)
        .join(CollaborationFile, CollaborationFile.id == CollaborationEditRequest.file_id)
        .join(User, User.id == CollaborationEditRequest.requester_id)
        .where(
            CollaborationFile.study_id == study_id,
            CollaborationFile.status == "ACTIVE",
            CollaborationFile.deleted_at.is_(None),
            CollaborationEditRequest.status == "PENDING",
            manages_file,
        )
    )).all()
    active_source_ids: set[str] = set()
    for request, item, requester in rows:
        source_id = str(request.id)
        active_source_ids.add(source_id)
        requester_name = str(requester.full_name or requester.email or "项目成员")
        await notification_service.sync_state_notification(
            db,
            study_id=study_id,
            recipient_id=user.id,
            category="COLLABORATION_EDIT_REQUEST",
            priority="NORMAL",
            title="新的编辑权限申请",
            message=f"{requester_name} 申请编辑“{item.title}”",
            action_path=f"/knowledge/collaboration?editRequestFile={item.id}",
            source_type="COLLABORATION_EDIT_REQUEST",
            source_id=source_id,
            source_version="PENDING",
            active=True,
            dedupe_key=f"collaboration-edit-request:{request.id}",
        )
    await notification_service.resolve_missing_recipient_sources(
        db,
        study_id=study_id,
        recipient_id=user.id,
        source_type="COLLABORATION_EDIT_REQUEST",
        active_source_ids=active_source_ids,
    )


async def sync_project_reminders(
    db: AsyncSession,
    study_id: uuid.UUID,
    user,
    *,
    now: datetime | None = None,
) -> None:
    membership = await member_crud.get_member(db, study_id, user.id)
    if not membership or not membership.is_active:
        await notification_service.resolve_recipient_study_notifications(
            db,
            study_id=study_id,
            recipient_id=user.id,
        )
        await db.commit()
        return
    role = membership.role_in_study
    current_time = now or datetime.now(timezone.utc)
    if current_time.tzinfo is None:
        current_time = current_time.replace(tzinfo=timezone.utc)
    else:
        current_time = current_time.astimezone(timezone.utc)

    await _sync_risk_reminders(
        db,
        study_id=study_id,
        user=user,
        role=role,
        current_time=current_time,
    )
    await _sync_document_distribution_reminders(
        db,
        study_id=study_id,
        user=user,
        role=role,
        current_time=current_time,
    )
    await _sync_milestone_reminders(
        db,
        study_id=study_id,
        user=user,
        role=role,
        current_time=current_time,
    )
    await _sync_visit_window_reminders(
        db,
        study_id=study_id,
        user=user,
        role=role,
        current_time=current_time,
    )
    await _sync_collaboration_edit_request_reminders(
        db,
        study_id=study_id,
        user=user,
        role=role,
    )
    await db.commit()