feat: refine subject visits and project workflows

Add early termination visit workflow with ordering, non-applicable visit handling, visit window display, and medication adherence support. Extend monitoring visit issue template fields, site scoping, setup draft project info handling, login security UI, attachment behavior, and related tests/migrations.
2026-05-09 17:10:34 +08:00
parent 74feca4467
commit 917ab7ccf1
41 changed files with 3463 additions and 701 deletions
@@ -14,6 +14,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.core.deps import get_current_user, get_db_session, require_study_member, require_study_not_locked, require_study_roles
 from app.crud import audit as audit_crud
 from app.crud import monitoring_visit_issue as issue_crud
+from app.crud import site as site_crud
 from app.crud import study as study_crud
 from app.schemas.monitoring_visit_issue import (
    MonitoringVisitIssueCreate,
@@ -25,9 +26,13 @@ from app.schemas.monitoring_visit_issue import (
 router = APIRouter()

 _HEADER_ALIASES: dict[str, list[str]] = {
+    "site_name": ["项目/中心", "中心", "中心名称", "site_name", "site"],
    "issue_no": ["问题编号", "问题编码", "issue_no", "IssueNo"],
    "open_duration_text": ["开放时长", "open_duration"],
    "category": ["问题分类", "category"],
+    "severity": ["严重程度", "severity"],
+    "mark": ["标记", "mark"],
+    "visit_cycle": ["访视周期", "访视", "visit_cycle"],
    "subject_code": ["受试者", "受试者编号", "subject_code"],
    "subject_name": ["受试者姓名", "subject_name"],
    "monitor_item": ["监查项", "monitor_item"],
@@ -40,6 +45,9 @@ _HEADER_ALIASES: dict[str, list[str]] = {
    "description": ["问题描述", "description"],
    "action_taken": ["采取措施", "action_taken"],
    "follow_up_progress": ["跟进计划及进展", "follow_up_progress"],
+    "center_query": ["中心质疑", "center_query"],
+    "center_latest_reply": ["中心最新回复", "中心回复", "center_latest_reply"],
+    "rectification_completed": ["是否完成整改", "完成整改", "rectification_completed"],
    "found_date": ["发现时间", "found_date"],
    "due_at": ["截止时间", "截止日期", "超期截止时间", "due_at"],
    "actual_resolve_date": ["实际解决日期", "actual_resolve_date"],
@@ -48,17 +56,24 @@ _HEADER_ALIASES: dict[str, list[str]] = {
 }

 _AUDIT_FIELDS: tuple[str, ...] = (
+    "site_id",
    "issue_no",
    "source",
    "monitor_type",
    "monitor_item",
    "category",
+    "severity",
+    "mark",
+    "visit_cycle",
    "recommendation",
    "subject_name",
    "subject_code",
    "description",
    "action_taken",
    "follow_up_progress",
+    "center_query",
+    "center_latest_reply",
+    "rectification_completed",
    "found_date",
    "due_at",
    "actual_resolve_date",
@@ -75,6 +90,14 @@ async def _ensure_study_exists(db: AsyncSession, study_id: uuid.UUID):
    return study


+async def _ensure_site_belongs_to_study(db: AsyncSession, study_id: uuid.UUID, site_id: uuid.UUID | None) -> None:
+    if site_id is None:
+        return
+    site = await site_crud.get_site(db, site_id)
+    if not site or site.study_id != study_id:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="中心不存在或不属于当前项目")
+
+
 def _normalize_status(value: str | None) -> str:
    text = (value or "").strip().upper()
    zh = {
@@ -91,6 +114,21 @@ def _normalize_status(value: str | None) -> str:
    return "OPEN"


+def _parse_bool(value: object | None) -> bool | None:
+    if value is None:
+        return None
+    if isinstance(value, bool):
+        return value
+    text = str(value).strip().lower()
+    if not text:
+        return None
+    if text in {"1", "true", "yes", "y", "是", "已完成", "完成"}:
+        return True
+    if text in {"0", "false", "no", "n", "否", "未完成"}:
+        return False
+    return None
+
+
 def _pick(row: dict[str, object], key: str) -> str | None:
    for alias in _HEADER_ALIASES[key]:
        if alias not in row:
@@ -167,9 +205,13 @@ def _to_read(item) -> MonitoringVisitIssueRead:
    return MonitoringVisitIssueRead(
        id=item.id,
        study_id=item.study_id,
+        site_id=item.site_id,
        issue_no=item.issue_no,
        open_duration=_calc_open_duration(item.created_at, item.status, item.closed_at, item.open_duration_text),
        category=item.category,
+        severity=item.severity,
+        mark=item.mark,
+        visit_cycle=item.visit_cycle,
        subject_code=item.subject_code,
        monitor_item=item.monitor_item,
        monitor_type=item.monitor_type,
@@ -183,6 +225,9 @@ def _to_read(item) -> MonitoringVisitIssueRead:
        description=item.description,
        action_taken=item.action_taken,
        follow_up_progress=item.follow_up_progress,
+        center_query=item.center_query,
+        center_latest_reply=item.center_latest_reply,
+        rectification_completed=bool(item.rectification_completed),
        found_date=item.found_date,
        overdue=overdue,
        due_at=item.due_at,
@@ -206,6 +251,8 @@ def _format_datetime(value: datetime | None) -> str:


 def _to_audit_value(value: object | None):
+    if isinstance(value, uuid.UUID):
+        return str(value)
    if isinstance(value, datetime):
        parsed = value if value.tzinfo else value.replace(tzinfo=timezone.utc)
        return parsed.isoformat()
@@ -289,22 +336,41 @@ def _normalize_file_rows(filename: str, content: bytes) -> list[dict[str, object
 )
 async def list_monitoring_visit_issues(
    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_value: str | None = Query(default=None, alias="status"),
    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,
    db: AsyncSession = Depends(get_db_session),
 ) -> list[MonitoringVisitIssueRead]:
    await _ensure_study_exists(db, study_id)
+    await _ensure_site_belongs_to_study(db, study_id, site_id)
    normalized_status = _normalize_status(status_value) if status_value else None
    items = await issue_crud.list_issues(
        db,
        study_id,
+        site_id=site_id,
        category=(category or None),
+        severity=(severity or None),
+        mark=(mark or None),
+        visit_cycle=(visit_cycle or None),
        status=normalized_status,
        overdue=overdue,
+        rectification_completed=rectification_completed,
+        due_from=due_from,
+        due_to=due_to,
+        created_from=created_from,
+        created_to=created_to,
        keyword=(keyword or None),
        skip=skip,
        limit=min(limit, 2000),
@@ -325,6 +391,7 @@ async def create_monitoring_visit_issue(
    current_user=Depends(get_current_user),
 ) -> MonitoringVisitIssueRead:
    await _ensure_study_exists(db, study_id)
+    await _ensure_site_belongs_to_study(db, study_id, payload.site_id)

    if payload.issue_no:
        existing = await issue_crud.get_issue_by_issue_no(db, study_id, payload.issue_no)
@@ -358,20 +425,39 @@ async def create_monitoring_visit_issue(
 )
 async def export_monitoring_visit_issues(
    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_value: str | None = Query(default=None, alias="status"),
    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,
    db: AsyncSession = Depends(get_db_session),
 ) -> StreamingResponse:
    await _ensure_study_exists(db, study_id)
+    await _ensure_site_belongs_to_study(db, study_id, site_id)
    normalized_status = _normalize_status(status_value) if status_value else None
    items = await issue_crud.list_issues(
        db,
        study_id,
+        site_id=site_id,
        category=(category or None),
+        severity=(severity or None),
+        mark=(mark or None),
+        visit_cycle=(visit_cycle or None),
        status=normalized_status,
        overdue=overdue,
+        rectification_completed=rectification_completed,
+        due_from=due_from,
+        due_to=due_to,
+        created_from=created_from,
+        created_to=created_to,
        keyword=(keyword or None),
        skip=0,
        limit=50000,
@@ -381,15 +467,23 @@ async def export_monitoring_visit_issues(
    ws = wb.active
    ws.title = "监查访视问题"
    headers = [
+        "项目/中心",
        "问题编号",
+        "受试者筛选号",
+        "访视周期",
+        "问题分类",
+        "严重程度",
+        "建议措施",
+        "问题描述",
+        "中心质疑",
+        "中心最新回复",
+        "是否完成整改",
+        "标记",
+        "状态",
+        "是否超期",
        "问题来源",
        "监查类型",
        "监查项",
-        "问题分类",
-        "建议措施",
-        "受试者",
-        "受试者缩写号",
-        "问题描述",
        "采取措施",
        "跟进计划及进展",
        "发现时间",
@@ -403,20 +497,31 @@ async def export_monitoring_visit_issues(
        "创建时间",
    ]
    ws.append(headers)
+    site_ids = {item.site_id for item in items if item.site_id}
+    site_map = await site_crud.get_sites_by_ids(db, site_ids)

    for item in items:
        view = _to_read(item)
+        site_name = site_map.get(view.site_id).name if view.site_id in site_map else ""
        ws.append(
            [
+                site_name,
                view.issue_no or "",
+                view.subject_code or "",
+                view.visit_cycle or "",
+                view.category or "",
+                view.severity or "",
+                view.recommendation or "",
+                view.description or "",
+                view.center_query or "",
+                view.center_latest_reply or "",
+                "是" if view.rectification_completed else "否",
+                view.mark or "",
+                view.progress or "",
+                "是" if view.overdue else "否",
                view.source or "",
                view.monitor_type or "",
                view.monitor_item or "",
-                view.category or "",
-                view.recommendation or "",
-                view.subject_name or "",
-                view.subject_code or "",
-                view.description or "",
                view.action_taken or "",
                view.follow_up_progress or "",
                _format_date(view.found_date),
@@ -486,6 +591,8 @@ async def update_monitoring_visit_issue(

    if not payload.model_fields_set:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="至少提供一个更新字段")
+    if "site_id" in payload.model_fields_set:
+        await _ensure_site_belongs_to_study(db, study_id, payload.site_id)

    before_snapshot = _issue_audit_snapshot(item)
    try:
@@ -568,6 +675,8 @@ async def import_monitoring_visit_issues(
    created_count = 0
    updated_count = 0
    skipped_rows: list[str] = []
+    sites = await site_crud.list_by_study(db, study_id, limit=5000, include_inactive=True)
+    site_by_name = {site.name.strip().lower(): site for site in sites if site.name}

    for idx, row in enumerate(rows, start=2):
        issue_no = _pick(row, "issue_no")
@@ -576,6 +685,15 @@ async def import_monitoring_visit_issues(
            skipped_rows.append(f"第{idx}行：问题编号或问题分类为空")
            continue

+        site_id = None
+        site_name = _pick(row, "site_name")
+        if site_name:
+            site = site_by_name.get(site_name.strip().lower())
+            if not site:
+                skipped_rows.append(f"第{idx}行：中心「{site_name}」不存在")
+                continue
+            site_id = site.id
+
        status_text = _normalize_status(_pick(row, "status"))
        creator_name = _pick(row, "creator_name") or getattr(current_user, "full_name", None)
        created_at = _parse_datetime(row.get("创建时间") or row.get("created_at") or _pick(row, "created_at"))
@@ -586,7 +704,11 @@ async def import_monitoring_visit_issues(
        try:
            payload = MonitoringVisitIssueCreate(
                issue_no=issue_no,
+                site_id=site_id,
                category=category,
+                severity=_pick(row, "severity"),
+                mark=_pick(row, "mark"),
+                visit_cycle=_pick(row, "visit_cycle"),
                subject_code=_pick(row, "subject_code"),
                subject_name=_pick(row, "subject_name"),
                monitor_item=_pick(row, "monitor_item"),
@@ -598,6 +720,9 @@ async def import_monitoring_visit_issues(
                description=_pick(row, "description"),
                action_taken=_pick(row, "action_taken"),
                follow_up_progress=_pick(row, "follow_up_progress"),
+                center_query=_pick(row, "center_query"),
+                center_latest_reply=_pick(row, "center_latest_reply"),
+                rectification_completed=_parse_bool(_pick(row, "rectification_completed")) or False,
                found_date=_parse_date(row.get("发现时间") or row.get("found_date") or _pick(row, "found_date")),
                due_at=due_at,
                actual_resolve_date=_parse_date(
@@ -263,6 +263,13 @@ def _to_date_text(value: date | None) -> str:
    return value.isoformat()


+def _project_snapshot_has_draft_values(snapshot: ProjectPublishSnapshot | None) -> bool:
+    if not snapshot:
+        return False
+    data = snapshot.model_dump(mode="json")
+    return any(value not in ("", None, []) for value in data.values())
+
+
 def _build_project_publish_snapshot(study) -> ProjectPublishSnapshot:
    return ProjectPublishSnapshot(
        code=getattr(study, "code", None) or "",
@@ -288,6 +295,61 @@ def _build_project_publish_snapshot(study) -> ProjectPublishSnapshot:
    )


+def _resolve_setup_project_snapshot(setup_data: StudySetupConfigData, study) -> ProjectPublishSnapshot:
+    if _project_snapshot_has_draft_values(setup_data.projectInfo):
+        return setup_data.projectInfo
+    return _build_project_publish_snapshot(study)
+
+
+def _parse_optional_snapshot_date(value: str) -> date | None:
+    text = (value or "").strip()
+    if not text:
+        return None
+    return date.fromisoformat(text)
+
+
+def _normalize_optional_snapshot_text(value: str | None) -> str | None:
+    if value is None:
+        return None
+    text = value.strip()
+    return text or None
+
+
+def _snapshot_visit_schedule(snapshot: ProjectPublishSnapshot) -> list[dict[str, int | str]]:
+    return [item.model_dump(mode="json") for item in snapshot.visit_schedule]
+
+
+def _apply_project_publish_snapshot_to_study(study, snapshot: ProjectPublishSnapshot) -> bool:
+    next_values = {
+        "code": snapshot.code.strip(),
+        "name": snapshot.name.strip(),
+        "project_full_name": _normalize_optional_snapshot_text(snapshot.project_full_name),
+        "sponsor": _normalize_optional_snapshot_text(snapshot.sponsor),
+        "protocol_no": _normalize_optional_snapshot_text(snapshot.protocol_no),
+        "lead_unit": _normalize_optional_snapshot_text(snapshot.lead_unit),
+        "principal_investigator": _normalize_optional_snapshot_text(snapshot.principal_investigator),
+        "main_pm": _normalize_optional_snapshot_text(snapshot.main_pm),
+        "research_analysis": _normalize_optional_snapshot_text(snapshot.research_analysis),
+        "research_product": _normalize_optional_snapshot_text(snapshot.research_product),
+        "control_product": _normalize_optional_snapshot_text(snapshot.control_product),
+        "indication": _normalize_optional_snapshot_text(snapshot.indication),
+        "research_population": _normalize_optional_snapshot_text(snapshot.research_population),
+        "research_design": _normalize_optional_snapshot_text(snapshot.research_design),
+        "plan_start_date": _parse_optional_snapshot_date(snapshot.plan_start_date),
+        "plan_end_date": _parse_optional_snapshot_date(snapshot.plan_end_date),
+        "planned_site_count": snapshot.planned_site_count,
+        "planned_enrollment_count": snapshot.planned_enrollment_count,
+        "status": snapshot.status or "DRAFT",
+        "visit_schedule": _snapshot_visit_schedule(snapshot),
+    }
+    changed = False
+    for field, value in next_values.items():
+        if getattr(study, field, None) != value:
+            setattr(study, field, value)
+            changed = True
+    return changed
+
+
 def _to_setup_config_read(
    record,
    *,
@@ -360,6 +422,63 @@ def _ensure_study_timeline_valid(
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="项目计划结束日期不能早于开始日期")


+async def _validate_setup_project_snapshot(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    snapshot: ProjectPublishSnapshot,
+    strict_required: bool = True,
+) -> None:
+    errors: list[dict[str, str]] = []
+    code = snapshot.code.strip()
+    name = snapshot.name.strip()
+    if strict_required and not code:
+        errors.append({"field": "projectInfo.code", "message": "项目编号不能为空"})
+    if strict_required and not name:
+        errors.append({"field": "projectInfo.name", "message": "项目名称不能为空"})
+    try:
+        plan_start = _parse_optional_snapshot_date(snapshot.plan_start_date)
+        plan_end = _parse_optional_snapshot_date(snapshot.plan_end_date)
+    except ValueError:
+        errors.append({"field": "projectInfo.plan_start_date", "message": "项目计划日期格式应为 YYYY-MM-DD"})
+        plan_start = None
+        plan_end = None
+    if plan_start and plan_end and plan_start > plan_end:
+        errors.append({"field": "projectInfo.plan_end_date", "message": "项目计划结束日期不能早于开始日期"})
+
+    seen_visit_codes: set[str] = set()
+    for index, item in enumerate(snapshot.visit_schedule):
+        row_prefix = f"projectInfo.visit_schedule[{index}]"
+        visit_code = item.visit_code.strip()
+        has_visit_values = bool(visit_code) or any(
+            value != 0
+            for value in (item.baseline_offset_days, item.window_before_days, item.window_after_days)
+        )
+        if strict_required and not visit_code:
+            errors.append({"field": f"{row_prefix}.visit_code", "message": "访视编号不能为空"})
+        if not has_visit_values:
+            continue
+        if visit_code:
+            if len(visit_code) > 50:
+                errors.append({"field": f"{row_prefix}.visit_code", "message": "访视编号不能超过50个字符"})
+            if visit_code in seen_visit_codes:
+                errors.append({"field": f"{row_prefix}.visit_code", "message": f"访视编号重复：{visit_code}"})
+            seen_visit_codes.add(visit_code)
+        if item.baseline_offset_days < 0 or item.baseline_offset_days > 3650:
+            errors.append({"field": f"{row_prefix}.baseline_offset_days", "message": "基线偏移天数应在0到3650之间"})
+        if item.window_before_days < 0 or item.window_before_days > 365:
+            errors.append({"field": f"{row_prefix}.window_before_days", "message": "窗口前天数应在0到365之间"})
+        if item.window_after_days < 0 or item.window_after_days > 365:
+            errors.append({"field": f"{row_prefix}.window_after_days", "message": "窗口后天数应在0到365之间"})
+
+    if errors:
+        _raise_validation_error(errors)
+    if code:
+        existing = await study_crud.get_by_code(db, code)
+        if existing and existing.id != study_id:
+            raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="项目编号已存在")
+
+
 def _resolve_version_label(record: Any) -> str:
    branch_name = (getattr(record, "branch_name", None) or "main").strip() or "main"
    display_version = int(getattr(record, "display_version", 0) or 0)
@@ -897,18 +1016,23 @@ async def upsert_study_setup_config(

    sites = await site_crud.list_by_study(db, study_id, skip=0, limit=1000, include_inactive=True)
    site_lookup = {str(site.id): site.name or "" for site in sites}
+    setup_project_snapshot = _resolve_setup_project_snapshot(payload.data, study)
+    await _validate_setup_project_snapshot(db, study_id=study_id, snapshot=setup_project_snapshot, strict_required=False)
+    setup_plan_start = _parse_optional_snapshot_date(setup_project_snapshot.plan_start_date)
+    setup_plan_end = _parse_optional_snapshot_date(setup_project_snapshot.plan_end_date)
    _validate_setup_data(
        payload.data,
        site_lookup,
-        project_plan_start=getattr(study, "plan_start_date", None),
-        project_plan_end=getattr(study, "plan_end_date", None),
+        project_plan_start=setup_plan_start,
+        project_plan_end=setup_plan_end,
+        strict_required=False,
    )

    existing = await study_setup_config_crud.get_by_study(db, study_id)
    old_config = dict(existing.config or {}) if existing else {}
    force_draft = bool(payload.force_draft)
    if existing and existing.publish_status == "PUBLISHED" and existing.published_project_snapshot:
-        current_project_snapshot = _build_project_publish_snapshot(study).model_dump(mode="json")
+        current_project_snapshot = setup_project_snapshot.model_dump(mode="json")
        if current_project_snapshot != dict(existing.published_project_snapshot or {}):
            force_draft = True
    record, conflict = await study_setup_config_crud.upsert(
@@ -971,7 +1095,12 @@ async def publish_study_setup_config(
    record = await study_setup_config_crud.get_by_study(db, study_id)
    if not record:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="请先保存立项配置草稿")
-    current_project_snapshot_for_publish = _build_project_publish_snapshot(study).model_dump(mode="json")
+    setup_data = StudySetupConfigData.model_validate(record.config or {})
+    project_snapshot_for_publish = _resolve_setup_project_snapshot(setup_data, study)
+    await _validate_setup_project_snapshot(db, study_id=study_id, snapshot=project_snapshot_for_publish)
+    setup_plan_start = _parse_optional_snapshot_date(project_snapshot_for_publish.plan_start_date)
+    setup_plan_end = _parse_optional_snapshot_date(project_snapshot_for_publish.plan_end_date)
+    current_project_snapshot_for_publish = project_snapshot_for_publish.model_dump(mode="json")
    force_create_snapshot = bool(
        (record.published_project_snapshot or {}) != current_project_snapshot_for_publish
    )
@@ -979,10 +1108,10 @@ async def publish_study_setup_config(
    sites = await site_crud.list_by_study(db, study_id, skip=0, limit=1000, include_inactive=True)
    site_lookup = {str(site.id): site.name or "" for site in sites}
    _validate_setup_data(
-        StudySetupConfigData.model_validate(record.config or {}),
+        setup_data,
        site_lookup,
-        project_plan_start=getattr(study, "plan_start_date", None),
-        project_plan_end=getattr(study, "plan_end_date", None),
+        project_plan_start=setup_plan_start,
+        project_plan_end=setup_plan_end,
    )

    published, conflict = await study_setup_config_crud.publish(
@@ -1001,6 +1130,7 @@ async def publish_study_setup_config(
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="配置发布失败")

    try:
+        _apply_project_publish_snapshot_to_study(study, project_snapshot_for_publish)
        projection = await apply_setup_projection_on_publish(
            db,
            study_id=study_id,
@@ -1017,16 +1147,14 @@ async def publish_study_setup_config(
                for item in projection.skipped_items
            ],
        )
-        study_after_publish = await study_crud.get(db, study_id)
-        if study_after_publish:
-            project_snapshot = _build_project_publish_snapshot(study_after_publish).model_dump(mode="json")
-            published.published_project_snapshot = project_snapshot
-            await study_setup_config_crud.set_version_project_snapshot(
-                db,
-                study_id,
-                version=published.version,
-                project_snapshot=project_snapshot,
-            )
+        project_snapshot = project_snapshot_for_publish.model_dump(mode="json")
+        published.published_project_snapshot = project_snapshot
+        await study_setup_config_crud.set_version_project_snapshot(
+            db,
+            study_id,
+            version=published.version,
+            project_snapshot=project_snapshot,
+        )
        await audit_crud.log_action(
            db,
            study_id=study_id,
@@ -1362,10 +1490,13 @@ async def merge_study_setup_config_to_main(
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="合并主分支失败")

    try:
+        merged_setup_data = StudySetupConfigData.model_validate(merged.published_config or {})
+        merged_project_snapshot = _resolve_setup_project_snapshot(merged_setup_data, study)
+        _apply_project_publish_snapshot_to_study(study, merged_project_snapshot)
        projection = await apply_setup_projection_on_publish(
            db,
            study_id=study_id,
-            setup_data=StudySetupConfigData.model_validate(merged.published_config or {}),
+            setup_data=merged_setup_data,
            operator_id=current_user.id,
        )
        projection_summary = SetupProjectionSummary(
@@ -1375,16 +1506,14 @@ async def merge_study_setup_config_to_main(
            warnings=projection.warnings,
            skipped_items=[{"site_id": item.site_id, "reason": item.reason} for item in projection.skipped_items],
        )
-        study_after_publish = await study_crud.get(db, study_id)
-        if study_after_publish:
-            project_snapshot = _build_project_publish_snapshot(study_after_publish).model_dump(mode="json")
-            merged.published_project_snapshot = project_snapshot
-            await study_setup_config_crud.set_version_project_snapshot(
-                db,
-                study_id,
-                version=merged.version,
-                project_snapshot=project_snapshot,
-            )
+        project_snapshot = merged_project_snapshot.model_dump(mode="json")
+        merged.published_project_snapshot = project_snapshot
+        await study_setup_config_crud.set_version_project_snapshot(
+            db,
+            study_id,
+            version=merged.version,
+            project_snapshot=project_snapshot,
+        )
        await audit_crud.log_action(
            db,
            study_id=study_id,
@@ -9,7 +9,8 @@ from app.crud import site as site_crud
 from app.crud import subject as subject_crud
 from app.crud import study as study_crud
 from app.crud import visit as visit_crud
-from app.schemas.visit import VisitCreate, VisitRead, VisitUpdate
+from app.schemas.subject import SubjectUpdate
+from app.schemas.visit import EarlyTerminationCreate, VisitCreate, VisitRead, VisitUpdate

 router = APIRouter()

@@ -110,6 +111,60 @@ async def create_visit(
    return visit


+@router.post(
+    "/early-termination",
+    response_model=VisitRead,
+    status_code=status.HTTP_201_CREATED,
+    dependencies=[Depends(require_study_roles(["PM", "CRA"])), Depends(require_study_not_locked())],
+)
+async def create_early_termination(
+    study_id: uuid.UUID,
+    subject_id: uuid.UUID,
+    termination_in: EarlyTerminationCreate,
+    db: AsyncSession = Depends(get_db_session),
+    current_user=Depends(get_current_user),
+) -> VisitRead:
+    subject = await _ensure_subject(db, study_id, subject_id)
+    await _ensure_subject_active(db, subject)
+    reason = termination_in.reason.strip()
+    if not reason:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="提前终止原因不能为空")
+    if subject.baseline_date and termination_in.termination_date < subject.baseline_date:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="提前终止日期不能早于基线/治疗日期")
+
+    try:
+        visit = await visit_crud.create_early_termination_visit(
+            db,
+            study_id=study_id,
+            subject=subject,
+            termination_date=termination_in.termination_date,
+            reason=reason,
+            notes=termination_in.notes,
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
+    await subject_crud.update_subject(
+        db,
+        subject,
+        SubjectUpdate(
+            status="DROPPED",
+            completion_date=termination_in.termination_date,
+            drop_reason=reason,
+        ),
+    )
+    await audit_crud.log_action(
+        db,
+        study_id=study_id,
+        entity_type="visit",
+        entity_id=visit.id,
+        action="CREATE_EARLY_TERMINATION",
+        detail=f"参与者 {subject.subject_no} 已提前终止",
+        operator_id=current_user.id,
+        operator_role=current_user.role,
+    )
+    return visit
+
+
@router.patch(
    "/{visit_id}",
    response_model=VisitRead,