import uuid from datetime import date, datetime, time, timedelta, timezone from typing import Sequence from sqlalchemy import and_, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.models.monitoring_visit_issue import MonitoringVisitIssue from app.schemas.monitoring_visit_issue import MonitoringVisitIssueCreate, MonitoringVisitIssueUpdate async def get_issue(db: AsyncSession, issue_id: uuid.UUID) -> MonitoringVisitIssue | None: result = await db.execute(select(MonitoringVisitIssue).where(MonitoringVisitIssue.id == issue_id)) return result.scalar_one_or_none() async def get_issue_by_issue_no( db: AsyncSession, study_id: uuid.UUID, issue_no: str, ) -> MonitoringVisitIssue | None: result = await db.execute( select(MonitoringVisitIssue).where( MonitoringVisitIssue.study_id == study_id, MonitoringVisitIssue.issue_no == issue_no, ) ) return result.scalar_one_or_none() async def generate_next_issue_no(db: AsyncSession, study_id: uuid.UUID) -> str: result = await db.execute(select(MonitoringVisitIssue.issue_no).where(MonitoringVisitIssue.study_id == study_id)) issue_nos = [str(item or "").strip() for item in result.scalars().all()] max_seq = 0 for issue_no in issue_nos: if "M" not in issue_no: continue seq_text = issue_no.split("M")[-1] if seq_text.isdigit(): max_seq = max(max_seq, int(seq_text)) next_seq = max_seq + 1 return f"001M{next_seq:05d}" async def list_issues( db: AsyncSession, study_id: uuid.UUID, *, site_id: uuid.UUID | None = None, category: str | None = None, severity: str | None = None, mark: str | None = None, visit_cycle: str | None = None, status: str | None = None, overdue: bool | None = None, rectification_completed: bool | None = None, due_from: date | None = None, due_to: date | None = None, created_from: date | None = None, created_to: date | None = None, keyword: str | None = None, skip: int = 0, limit: int = 500, ) -> Sequence[MonitoringVisitIssue]: stmt = select(MonitoringVisitIssue).where(MonitoringVisitIssue.study_id == study_id) if site_id: stmt = stmt.where(MonitoringVisitIssue.site_id == site_id) if category: stmt = stmt.where(MonitoringVisitIssue.category == category) if severity: stmt = stmt.where(MonitoringVisitIssue.severity == severity) if mark: stmt = stmt.where(MonitoringVisitIssue.mark.ilike(f"%{mark}%")) if visit_cycle: stmt = stmt.where(MonitoringVisitIssue.visit_cycle == visit_cycle) if status: stmt = stmt.where(MonitoringVisitIssue.status == status) if rectification_completed is not None: stmt = stmt.where(MonitoringVisitIssue.rectification_completed == rectification_completed) if due_from: stmt = stmt.where(MonitoringVisitIssue.due_at >= datetime.combine(due_from, time.min, tzinfo=timezone.utc)) if due_to: stmt = stmt.where(MonitoringVisitIssue.due_at < datetime.combine(due_to + timedelta(days=1), time.min, tzinfo=timezone.utc)) if created_from: stmt = stmt.where(MonitoringVisitIssue.created_at >= datetime.combine(created_from, time.min, tzinfo=timezone.utc)) if created_to: stmt = stmt.where(MonitoringVisitIssue.created_at < datetime.combine(created_to + timedelta(days=1), time.min, tzinfo=timezone.utc)) if keyword: term = f"%{keyword}%" stmt = stmt.where( or_( MonitoringVisitIssue.issue_no.ilike(term), MonitoringVisitIssue.category.ilike(term), MonitoringVisitIssue.severity.ilike(term), MonitoringVisitIssue.mark.ilike(term), MonitoringVisitIssue.visit_cycle.ilike(term), MonitoringVisitIssue.subject_code.ilike(term), MonitoringVisitIssue.subject_name.ilike(term), MonitoringVisitIssue.monitor_item.ilike(term), MonitoringVisitIssue.description.ilike(term), MonitoringVisitIssue.recommendation.ilike(term), MonitoringVisitIssue.action_taken.ilike(term), MonitoringVisitIssue.follow_up_progress.ilike(term), MonitoringVisitIssue.center_query.ilike(term), MonitoringVisitIssue.center_latest_reply.ilike(term), MonitoringVisitIssue.responsible_name.ilike(term), ) ) now = datetime.now(timezone.utc) overdue_expr = and_( MonitoringVisitIssue.status == "OPEN", MonitoringVisitIssue.due_at.is_not(None), MonitoringVisitIssue.due_at < now, ) if overdue is True: stmt = stmt.where(overdue_expr) elif overdue is False: stmt = stmt.where(or_(MonitoringVisitIssue.due_at.is_(None), MonitoringVisitIssue.due_at >= now, MonitoringVisitIssue.status == "CLOSED")) stmt = stmt.order_by(MonitoringVisitIssue.created_at.desc()).offset(skip).limit(limit) result = await db.execute(stmt) return result.scalars().all() async def create_issue( db: AsyncSession, study_id: uuid.UUID, issue_in: MonitoringVisitIssueCreate, *, created_by: uuid.UUID | None, created_at: datetime | None = None, open_duration_text: str | None = None, ) -> MonitoringVisitIssue: issue_no = issue_in.issue_no or await generate_next_issue_no(db, study_id) closed_at = issue_in.closed_at if issue_in.status == "CLOSED" else None if issue_in.status == "CLOSED" and closed_at is None: closed_at = datetime.now(timezone.utc) item = MonitoringVisitIssue( study_id=study_id, site_id=issue_in.site_id, issue_no=issue_no, category=issue_in.category, severity=issue_in.severity, mark=issue_in.mark, visit_cycle=issue_in.visit_cycle, subject_code=issue_in.subject_code, monitor_item=issue_in.monitor_item, monitor_type=issue_in.monitor_type, status=issue_in.status, source=issue_in.source, creator_name=issue_in.creator_name, recommendation=issue_in.recommendation, subject_name=issue_in.subject_name, description=issue_in.description, action_taken=issue_in.action_taken, follow_up_progress=issue_in.follow_up_progress, center_query=issue_in.center_query, center_latest_reply=issue_in.center_latest_reply, rectification_completed=issue_in.rectification_completed, found_date=issue_in.found_date, due_at=issue_in.due_at, actual_resolve_date=issue_in.actual_resolve_date, closed_at=closed_at, responsible_name=issue_in.responsible_name, created_by=created_by, open_duration_text=open_duration_text, ) if created_at is not None: item.created_at = created_at db.add(item) await db.commit() await db.refresh(item) return item async def update_issue( db: AsyncSession, issue: MonitoringVisitIssue, issue_in: MonitoringVisitIssueUpdate, *, created_by: uuid.UUID | None = None, created_at: datetime | None = None, open_duration_text: str | None = None, ) -> MonitoringVisitIssue: prev_status = issue.status update_data = issue_in.model_dump(exclude_unset=True) for key, value in update_data.items(): setattr(issue, key, value) next_status = update_data.get("status", issue.status) if created_by is not None: issue.created_by = created_by if created_at is not None: issue.created_at = created_at if open_duration_text is not None: issue.open_duration_text = open_duration_text if prev_status != "CLOSED" and next_status == "CLOSED" and "closed_at" not in update_data: issue.closed_at = datetime.now(timezone.utc) elif prev_status == "CLOSED" and next_status == "OPEN": issue.closed_at = None await db.commit() await db.refresh(issue) return issue async def delete_issue(db: AsyncSession, issue: MonitoringVisitIssue) -> None: await db.delete(issue) await db.commit()