Cheng Zhou
301 lines
10 KiB
Python
301 lines
10 KiB
Python
"""概览数据聚合CRUD操作"""
|
|
import uuid
|
|
from datetime import datetime
|
|
from typing import Dict, List, Tuple
|
|
|
|
from sqlalchemy import Date, cast, func, select, and_, or_
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.site import Site
|
|
from app.models.startup_feasibility import StartupFeasibility
|
|
from app.models.startup_ethics import StartupEthics
|
|
from app.models.contract_fee import ContractFee
|
|
from app.models.kickoff_meeting import KickoffMeeting
|
|
from app.models.subject import Subject
|
|
|
|
|
|
async def get_project_overview(db: AsyncSession, study_id: uuid.UUID) -> Dict:
|
|
"""
|
|
获取项目概览数据，聚合各中心的进度信息
|
|
|
|
返回格式:
|
|
{
|
|
"study_id": "...",
|
|
"updated_at": "2026-01-16T16:10:00",
|
|
"centers": [...],
|
|
"enrollment_by_month": [...]
|
|
}
|
|
"""
|
|
# 获取所有中心
|
|
stmt = select(Site).where(Site.study_id == study_id).order_by(Site.created_at)
|
|
result = await db.execute(stmt)
|
|
sites = result.scalars().all()
|
|
|
|
if not sites:
|
|
return {
|
|
"study_id": str(study_id),
|
|
"updated_at": datetime.now().isoformat(),
|
|
"centers": [],
|
|
"enrollment_by_month": []
|
|
}
|
|
|
|
site_ids = [site.id for site in sites]
|
|
|
|
# 并行获取各类数据
|
|
feasibility_data = await _get_feasibility_data(db, study_id, site_ids)
|
|
ethics_data = await _get_ethics_data(db, study_id, site_ids)
|
|
contract_data = await _get_contract_data(db, study_id, site_ids)
|
|
kickoff_data = await _get_kickoff_data(db, study_id, site_ids)
|
|
enrollment_data = await _get_enrollment_data(db, study_id, site_ids)
|
|
monthly_enrollment = await _get_monthly_enrollment(db, study_id)
|
|
|
|
# 组装中心数据
|
|
centers = []
|
|
for site in sites:
|
|
site_id_str = str(site.id)
|
|
|
|
# 机构立项状态
|
|
feas = feasibility_data.get(site_id_str, {})
|
|
institution_status = "COMPLETED" if feas.get("approved_date") else "IN_PROGRESS" if feas.get("submit_date") else "NOT_STARTED"
|
|
institution_completed = feas.get("approved_date")
|
|
|
|
# 伦理记录状态 - 依赖机构立项完成
|
|
eth = ethics_data.get(site_id_str, {})
|
|
if institution_status == "COMPLETED":
|
|
ethics_status = "COMPLETED" if eth.get("approved_date") else "IN_PROGRESS" if eth.get("submit_date") else "NOT_STARTED"
|
|
else:
|
|
# 机构立项未完成,伦理强制为未开始
|
|
ethics_status = "NOT_STARTED"
|
|
ethics_completed = eth.get("approved_date") if institution_status == "COMPLETED" else None
|
|
|
|
# 合同签署状态 - 依赖伦理记录完成
|
|
contract = contract_data.get(site_id_str, {})
|
|
if ethics_status == "COMPLETED":
|
|
contract_status = "COMPLETED" if contract.get("signed_date") else "NOT_STARTED"
|
|
else:
|
|
# 伦理未完成,合同强制为未开始
|
|
contract_status = "NOT_STARTED"
|
|
contract_completed = contract.get("signed_date") if ethics_status == "COMPLETED" else None
|
|
|
|
# 启动会状态 - 依赖合同签署完成
|
|
kickoff = kickoff_data.get(site_id_str, {})
|
|
if contract_status == "COMPLETED":
|
|
startup_status = "COMPLETED" if kickoff.get("kickoff_date") else "NOT_STARTED"
|
|
else:
|
|
# 合同未完成,启动会强制为未开始
|
|
startup_status = "NOT_STARTED"
|
|
startup_completed = kickoff.get("kickoff_date") if contract_status == "COMPLETED" else None
|
|
|
|
# 入组状态
|
|
enrollment = enrollment_data.get(site_id_str, {"actual": 0})
|
|
enrollment_actual = enrollment["actual"]
|
|
enrollment_target = site.enrollment_target or 0
|
|
|
|
if enrollment_actual >= enrollment_target and enrollment_target > 0:
|
|
enrollment_status = "COMPLETED"
|
|
elif enrollment_actual > 0:
|
|
enrollment_status = "IN_PROGRESS"
|
|
elif startup_status == "COMPLETED":
|
|
enrollment_status = "NOT_STARTED"
|
|
else:
|
|
enrollment_status = "NOT_STARTED"
|
|
|
|
# 检查状态 - 依赖入组完成
|
|
# 暂未实现数据源,固定为NOT_STARTED
|
|
if enrollment_status == "COMPLETED":
|
|
inspection_status = "NOT_STARTED" # 未来从inspection表获取
|
|
else:
|
|
inspection_status = "NOT_STARTED"
|
|
|
|
# 关中心状态 - 依赖检查完成
|
|
# 暂未实现数据源,固定为NOT_STARTED
|
|
if inspection_status == "COMPLETED":
|
|
closeout_status = "NOT_STARTED" # 未来从closeout表获取
|
|
else:
|
|
closeout_status = "NOT_STARTED"
|
|
|
|
|
|
centers.append({
|
|
"center_id": str(site.id),
|
|
"center_name": site.name,
|
|
"institution_initiation_status": institution_status,
|
|
"ethics_status": ethics_status,
|
|
"contract_sign_status": contract_status,
|
|
"startup_status": startup_status,
|
|
"enrollment_status": enrollment_status,
|
|
"inspection_status": inspection_status,
|
|
"closeout_status": closeout_status,
|
|
"institution_initiation_completed_at": institution_completed,
|
|
"ethics_completed_at": ethics_completed,
|
|
"contract_sign_completed_at": contract_completed,
|
|
"startup_completed_at": startup_completed,
|
|
"enrollment_completed_at": None, # 暂未实现
|
|
"inspection_completed_at": None,
|
|
"closeout_completed_at": None,
|
|
"enrollment_target": enrollment_target,
|
|
"enrollment_actual": enrollment_actual,
|
|
})
|
|
|
|
return {
|
|
"study_id": str(study_id),
|
|
"updated_at": datetime.now().isoformat(),
|
|
"centers": centers,
|
|
"enrollment_by_month": monthly_enrollment
|
|
}
|
|
|
|
|
|
async def _get_feasibility_data(db: AsyncSession, study_id: uuid.UUID, site_ids: List[uuid.UUID]) -> Dict:
|
|
"""获取各中心的立项数据"""
|
|
stmt = select(
|
|
StartupFeasibility.site_id,
|
|
StartupFeasibility.submit_date,
|
|
StartupFeasibility.approved_date
|
|
).where(
|
|
and_(
|
|
StartupFeasibility.study_id == study_id,
|
|
StartupFeasibility.site_id.in_(site_ids)
|
|
)
|
|
).order_by(StartupFeasibility.created_at.desc())
|
|
|
|
result = await db.execute(stmt)
|
|
rows = result.all()
|
|
|
|
data = {}
|
|
for row in rows:
|
|
site_id_str = str(row.site_id)
|
|
if site_id_str not in data:
|
|
data[site_id_str] = {
|
|
"submit_date": row.submit_date.isoformat() if row.submit_date else None,
|
|
"approved_date": row.approved_date.isoformat() if row.approved_date else None,
|
|
}
|
|
|
|
return data
|
|
|
|
|
|
async def _get_ethics_data(db: AsyncSession, study_id: uuid.UUID, site_ids: List[uuid.UUID]) -> Dict:
|
|
"""获取各中心的伦理数据"""
|
|
stmt = select(
|
|
StartupEthics.site_id,
|
|
StartupEthics.submit_date,
|
|
StartupEthics.approved_date
|
|
).where(
|
|
and_(
|
|
StartupEthics.study_id == study_id,
|
|
StartupEthics.site_id.in_(site_ids)
|
|
)
|
|
).order_by(StartupEthics.created_at.desc())
|
|
|
|
result = await db.execute(stmt)
|
|
rows = result.all()
|
|
|
|
data = {}
|
|
for row in rows:
|
|
site_id_str = str(row.site_id)
|
|
if site_id_str not in data:
|
|
data[site_id_str] = {
|
|
"submit_date": row.submit_date.isoformat() if row.submit_date else None,
|
|
"approved_date": row.approved_date.isoformat() if row.approved_date else None,
|
|
}
|
|
|
|
return data
|
|
|
|
|
|
async def _get_contract_data(db: AsyncSession, study_id: uuid.UUID, site_ids: List[uuid.UUID]) -> Dict:
|
|
"""获取各中心的合同签署数据"""
|
|
stmt = select(
|
|
ContractFee.center_id,
|
|
func.min(ContractFee.signed_date).label("signed_date")
|
|
).where(
|
|
ContractFee.project_id == study_id,
|
|
ContractFee.center_id.in_(site_ids),
|
|
).group_by(ContractFee.center_id)
|
|
|
|
result = await db.execute(stmt)
|
|
rows = result.all()
|
|
|
|
data = {}
|
|
for row in rows:
|
|
data[str(row.center_id)] = {
|
|
"signed_date": row.signed_date.isoformat() if row.signed_date else None,
|
|
}
|
|
|
|
return data
|
|
|
|
|
|
async def _get_kickoff_data(db: AsyncSession, study_id: uuid.UUID, site_ids: List[uuid.UUID]) -> Dict:
|
|
"""获取各中心的启动会数据"""
|
|
stmt = select(
|
|
KickoffMeeting.site_id,
|
|
KickoffMeeting.kickoff_date
|
|
).where(
|
|
and_(
|
|
KickoffMeeting.study_id == study_id,
|
|
KickoffMeeting.site_id.in_(site_ids)
|
|
)
|
|
).order_by(KickoffMeeting.created_at.desc())
|
|
|
|
result = await db.execute(stmt)
|
|
rows = result.all()
|
|
|
|
data = {}
|
|
for row in rows:
|
|
site_id_str = str(row.site_id)
|
|
if site_id_str not in data:
|
|
data[site_id_str] = {
|
|
"kickoff_date": row.kickoff_date.isoformat() if row.kickoff_date else None,
|
|
}
|
|
|
|
return data
|
|
|
|
|
|
async def _get_enrollment_data(db: AsyncSession, study_id: uuid.UUID, site_ids: List[uuid.UUID]) -> Dict:
|
|
"""获取各中心的入组数据"""
|
|
stmt = select(
|
|
Subject.site_id,
|
|
func.count(Subject.id).label("count")
|
|
).where(
|
|
and_(
|
|
Subject.study_id == study_id,
|
|
Subject.site_id.in_(site_ids),
|
|
or_(
|
|
Subject.enrollment_date.isnot(None),
|
|
Subject.status == "ENROLLED"
|
|
)
|
|
)
|
|
).group_by(Subject.site_id)
|
|
|
|
result = await db.execute(stmt)
|
|
rows = result.all()
|
|
|
|
data = {}
|
|
for row in rows:
|
|
data[str(row.site_id)] = {"actual": row.count}
|
|
|
|
return data
|
|
|
|
|
|
async def _get_monthly_enrollment(db: AsyncSession, study_id: uuid.UUID) -> List[Dict]:
|
|
"""获取月度入组统计"""
|
|
month_col = func.to_char(
|
|
func.coalesce(Subject.enrollment_date, func.cast(Subject.created_at, Date)),
|
|
'YYYY-MM'
|
|
).label("month")
|
|
|
|
stmt = select(
|
|
month_col,
|
|
func.count(Subject.id).label("count")
|
|
).where(
|
|
and_(
|
|
Subject.study_id == study_id,
|
|
or_(
|
|
Subject.enrollment_date.isnot(None),
|
|
Subject.status == "ENROLLED"
|
|
)
|
|
)
|
|
).group_by(month_col).order_by(month_col)
|
|
|
|
result = await db.execute(stmt)
|
|
rows = result.all()
|
|
|
|
return [{"month": row.month, "count": row.count} for row in rows]
|