Cheng Zhou
151 lines
5.3 KiB
Python
151 lines
5.3 KiB
Python
"""
|
|
项目概览示例数据注入脚本
|
|
|
|
运行方式:
|
|
docker compose run --rm backend python scripts/seed_overview_data.py
|
|
"""
|
|
import asyncio
|
|
import sys
|
|
import uuid
|
|
from datetime import date, timedelta
|
|
from pathlib import Path
|
|
|
|
# 添加app目录到Python路径
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from sqlalchemy import select
|
|
from app.db.session import SessionLocal
|
|
from app.models.site import Site
|
|
from app.models.startup_feasibility import StartupFeasibility
|
|
from app.models.startup_ethics import StartupEthics
|
|
from app.models.finance_contract import FinanceContract
|
|
from app.models.kickoff_meeting import KickoffMeeting
|
|
from app.models.subject import Subject
|
|
|
|
|
|
async def seed_data():
|
|
"""注入示例数据"""
|
|
async with SessionLocal() as db:
|
|
# 获取第一个项目用于测试
|
|
stmt = select(Site.study_id).limit(1)
|
|
result = await db.execute(stmt)
|
|
study_id_row = result.first()
|
|
|
|
if not study_id_row:
|
|
print("❌ 数据库中没有项目,请先创建项目和中心")
|
|
return
|
|
|
|
study_id = study_id_row[0]
|
|
print(f"✓ 使用项目 ID: {study_id}")
|
|
|
|
# 获取该项目的所有中心
|
|
stmt = select(Site).where(Site.study_id == study_id)
|
|
result = await db.execute(stmt)
|
|
sites = result.scalars().all()
|
|
|
|
if not sites:
|
|
print("❌ 项目下没有中心,请先创建中心")
|
|
return
|
|
|
|
print(f"✓ 找到 {len(sites)} 个中心")
|
|
|
|
# 为每个中心设置入组目标
|
|
targets = [80, 60, 70, 50, 40]
|
|
for idx, site in enumerate(sites[:5]):
|
|
site.enrollment_target = targets[idx] if idx < len(targets) else 50
|
|
print(f" • {site.name}: 目标入组 {site.enrollment_target} 人")
|
|
|
|
await db.commit()
|
|
print("✓ 入组目标已更新")
|
|
|
|
# 为中心添加启动流程数据
|
|
base_date = date(2024, 12, 1)
|
|
|
|
for idx, site in enumerate(sites[:5]):
|
|
# 机构立项
|
|
feasibility = StartupFeasibility(
|
|
id=uuid.uuid4(),
|
|
study_id=study_id,
|
|
site_id=site.id,
|
|
submit_date=base_date + timedelta(days=idx * 10),
|
|
accept_date=base_date + timedelta(days=idx * 10 + 5),
|
|
approved_date=base_date + timedelta(days=idx * 10 + 15) if idx < 4 else None,
|
|
project_no=f"PROJ-{idx+1:03d}"
|
|
)
|
|
db.add(feasibility)
|
|
|
|
# 伦理审批
|
|
ethics = StartupEthics(
|
|
id=uuid.uuid4(),
|
|
study_id=study_id,
|
|
site_id=site.id,
|
|
submit_date=base_date + timedelta(days=idx * 10 + 20),
|
|
accept_date=base_date + timedelta(days=idx * 10 + 25),
|
|
meeting_date=base_date + timedelta(days=idx * 10 + 35) if idx < 4 else None,
|
|
approved_date=base_date + timedelta(days=idx * 10 + 40) if idx < 4 else None,
|
|
approval_no=f"EC-{idx+1:03d}" if idx < 4 else None
|
|
)
|
|
db.add(ethics)
|
|
|
|
# 合同签署 (前3个中心)
|
|
if idx < 3:
|
|
contract = FinanceContract(
|
|
id=uuid.uuid4(),
|
|
study_id=study_id,
|
|
site_name=site.name,
|
|
contract_no=f"CT-{idx+1:03d}",
|
|
signed_date=base_date + timedelta(days=idx * 10 + 50),
|
|
amount=500000.0,
|
|
currency="CNY"
|
|
)
|
|
db.add(contract)
|
|
|
|
# 启动会 (前3个中心)
|
|
if idx < 3:
|
|
kickoff = KickoffMeeting(
|
|
id=uuid.uuid4(),
|
|
study_id=study_id,
|
|
site_id=site.id,
|
|
kickoff_date=base_date + timedelta(days=idx * 10 + 60),
|
|
attendees=["研究者", "CRA", "PM"]
|
|
)
|
|
db.add(kickoff)
|
|
|
|
await db.commit()
|
|
print("✓ 启动流程数据已添加")
|
|
|
|
# 为前3个中心添加入组受试者
|
|
enrollment_counts = [52, 18, 70]
|
|
start_enrollment_date = date(2025, 1, 1)
|
|
|
|
for idx, site in enumerate(sites[:3]):
|
|
count = enrollment_counts[idx] if idx < len(enrollment_counts) else 0
|
|
|
|
for i in range(count):
|
|
# 分散在6个月内入组
|
|
enrollment_date = start_enrollment_date + timedelta(days=i * 4)
|
|
|
|
subject = Subject(
|
|
id=uuid.uuid4(),
|
|
study_id=study_id,
|
|
site_id=site.id,
|
|
subject_no=f"S{idx+1:02d}-{i+1:03d}",
|
|
status="ENROLLED",
|
|
screening_date=enrollment_date - timedelta(days=7),
|
|
consent_date=enrollment_date - timedelta(days=3),
|
|
enrollment_date=enrollment_date
|
|
)
|
|
db.add(subject)
|
|
|
|
print(f" • {site.name}: 已入组 {count} 人")
|
|
|
|
await db.commit()
|
|
print("✓ 受试者入组数据已添加")
|
|
|
|
print("\n✅ 示例数据注入完成!")
|
|
print("请刷新项目概览页面查看效果")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(seed_data())
|