未知（继上次中断）

backend/app/services/document_service.py

@@ -8,10 +8,11 @@ from typing import Iterable
 import aiofiles
 from fastapi import HTTPException, UploadFile, status
 from fastapi.responses import FileResponse
-from sqlalchemy import update as sa_update
+from sqlalchemy import delete as sa_delete, or_, select, update as sa_update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core import rbac
+from app.core.deps import get_cra_site_scope
 from app.crud import acknowledgement as acknowledgement_crud
 from app.crud import distribution as distribution_crud
 from app.crud import document as document_crud
@@ -20,20 +21,16 @@ from app.crud import member as member_crud
 from app.crud import site as site_crud
 from app.crud import study as study_crud
 from app.crud import user as user_crud
-from app.crud import version_workflow as workflow_crud
-from app.crud import workflow_template as template_crud
 from app.models.acknowledgement import Acknowledgement, AcknowledgementType
 from app.models.audit_log import AuditLog
-from app.models.distribution import Distribution, DistributionTargetType
+from app.models.distribution import Distribution, DistributionStatus, DistributionTargetType
 from app.models.document import Document, DocumentScopeType, DocumentStatus
 from app.models.document_version import DocumentVersion, DocumentVersionStatus
-from app.models.version_workflow import VersionWorkflow, WorkflowStatus
-from app.models.workflow_action import WorkflowAction, WorkflowActionType
-from app.models.workflow_template import WorkflowTemplate
 from app.schemas.acknowledgement import AcknowledgementCreate
 from app.schemas.distribution import DistributionCreate, DistributionRead, DistributionStats
 from app.schemas.document import DocumentCreate, DocumentDetail, DocumentSummary
 from app.schemas.document_version import DocumentVersionRead, DocumentVersionSummary
+from app.schemas.notification import NotificationItem
 from app.schemas.user import UserDisplay
 
 UPLOAD_ROOT = Path(__file__).resolve().parent.parent / "uploads" / "documents"
@@ -83,6 +80,18 @@ async def _ensure_study_access(db: AsyncSession, trial_id: uuid.UUID, current_us
     return membership
 
 
+async def _ensure_study_member(db: AsyncSession, trial_id: uuid.UUID, current_user):
+    study = await study_crud.get(db, trial_id)
+    if not study:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="项目不存在")
+    if current_user.role == "ADMIN":
+        return None
+    membership = await member_crud.get_member(db, trial_id, current_user.id)
+    if not membership or not membership.is_active:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="不是项目成员")
+    return membership
+
+
 async def create_document(
     db: AsyncSession,
     doc_in: DocumentCreate,
@@ -140,6 +149,8 @@ async def list_documents(
     current_user,
 ) -> list[DocumentSummary]:
     await _ensure_study_access(db, trial_id, current_user, action="view")
+    cra_scope = await get_cra_site_scope(db, trial_id, current_user)
+    cra_site_ids = cra_scope[0] if cra_scope else None
     docs = await document_crud.list_documents(
         db,
         trial_id=trial_id,
@@ -147,6 +158,7 @@ async def list_documents(
         doc_type=doc_type,
         status=status,
         scope_type=scope_type,
+        cra_site_ids=cra_site_ids,
         skip=skip,
         limit=limit,
     )
@@ -185,6 +197,9 @@ async def get_document_detail(
     if not doc:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
     await _ensure_study_access(db, doc.trial_id, current_user, action="view")
+    cra_scope = await get_cra_site_scope(db, doc.trial_id, current_user)
+    if cra_scope and doc.scope_type == DocumentScopeType.SITE and doc.site_id not in cra_scope[0]:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="权限不足")
     versions = await version_crud.list_by_document(db, document_id)
 
     current_version = None
@@ -223,6 +238,7 @@ async def create_version(
     document_id: uuid.UUID,
     *,
     version_no: str,
+    version_date: str,
     change_summary: str | None,
     parent_version_id: uuid.UUID | None,
     file: UploadFile,
@@ -233,6 +249,8 @@ async def create_version(
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
     await _ensure_study_access(db, doc.trial_id, current_user, action="create_version")
 
+    previous_effective_id = doc.current_effective_version_id
+
     existing = await version_crud.get_by_document_version_no(db, document_id, version_no)
     if existing:
         raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="版本号已存在")
@@ -251,11 +269,21 @@ async def create_version(
     async with aiofiles.open(dest_path, "wb") as out_file:
         await out_file.write(content)
 
+    now = datetime.now(timezone.utc)
+    effective_at = now
+    try:
+        parsed = datetime.fromisoformat(version_date)
+        if parsed.tzinfo is None:
+            parsed = parsed.replace(tzinfo=timezone.utc)
+        effective_at = parsed
+    except ValueError:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="版本日期格式不正确")
     version = DocumentVersion(
         document_id=document_id,
         version_no=version_no,
         parent_version_id=parent_version_id,
-        status=DocumentVersionStatus.DRAFT,
+        status=DocumentVersionStatus.EFFECTIVE,
+        effective_at=effective_at,
         file_uri=str(dest_path),
         file_hash=file_hash,
         file_size=len(content),
@@ -264,6 +292,21 @@ async def create_version(
         created_by=current_user.id,
     )
     db.add(version)
+    await db.flush()
+    await db.execute(
+        sa_update(DocumentVersion)
+        .where(
+            DocumentVersion.document_id == document_id,
+            DocumentVersion.status == DocumentVersionStatus.EFFECTIVE,
+            DocumentVersion.id != version.id,
+        )
+        .values(status=DocumentVersionStatus.SUPERSEDED, superseded_at=now)
+    )
+    await db.execute(
+        sa_update(Document)
+        .where(Document.id == document_id)
+        .values(current_effective_version_id=version.id)
+    )
     db.add(
         AuditLog(
             study_id=doc.trial_id,
@@ -275,227 +318,98 @@ async def create_version(
             operator_role=_role_value(current_user),
         )
     )
+
+    if previous_effective_id:
+        previous_distributions = await distribution_crud.list_by_version(db, previous_effective_id)
+        if previous_distributions:
+            seen: set[tuple[str, str]] = set()
+            for dist in previous_distributions:
+                key = (dist.target_type, dist.target_id)
+                if key in seen:
+                    continue
+                seen.add(key)
+                db.add(
+                    Distribution(
+                        document_id=document_id,
+                        version_id=version.id,
+                        target_type=dist.target_type,
+                        target_id=dist.target_id,
+                        created_by=current_user.id,
+                    )
+                )
     await db.commit()
     await db.refresh(version)
     return version
 
 
-async def submit_version(
+async def delete_version(
     db: AsyncSession,
     version_id: uuid.UUID,
-    template_id: uuid.UUID,
-    comment: str | None,
     current_user,
-) -> VersionWorkflow:
+) -> None:
     version = await version_crud.get(db, version_id)
     if not version:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="版本不存在")
     doc = await document_crud.get(db, version.document_id)
     if not doc:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
-    await _ensure_study_access(db, doc.trial_id, current_user, action="submit")
+    await _ensure_study_access(db, doc.trial_id, current_user, action="delete_document")
 
-    if version.status != DocumentVersionStatus.DRAFT:
-        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="版本状态不允许提交")
-    pending = await version_crud.list_pending_versions(db, version.document_id)
-    if any(v.id != version.id for v in pending):
-        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="存在待审批版本")
-
-    template = await template_crud.get(db, template_id)
-    if not template:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="审批模板不存在")
+    distribution_ids = (
+        await db.execute(
+            select(Distribution.id).where(Distribution.version_id == version.id)
+        )
+    ).scalars().all()
+    if distribution_ids:
+        await db.execute(sa_delete(Acknowledgement).where(Acknowledgement.distribution_id.in_(distribution_ids)))
+        await db.execute(sa_delete(Distribution).where(Distribution.id.in_(distribution_ids)))
 
     before = _version_snapshot(version)
-    version.status = DocumentVersionStatus.SUBMITTED
-    version.submitted_at = datetime.now(timezone.utc)
 
-    workflow = VersionWorkflow(
-        document_id=version.document_id,
-        version_id=version.id,
-        template_id=template_id,
-        status=WorkflowStatus.PENDING,
-        current_node=1,
-        submitted_by=current_user.id,
-        submitted_at=version.submitted_at,
-    )
-    action = WorkflowAction(
-        workflow=workflow,
-        node_order=workflow.current_node,
-        action=WorkflowActionType.SUBMIT,
-        actor_id=current_user.id,
-        comment=comment,
-    )
-    db.add(version)
-    db.add(workflow)
-    db.add(action)
+    if doc.current_effective_version_id == version.id:
+        candidates = await version_crud.list_by_document(db, version.document_id)
+        candidate = next((item for item in candidates if item.id != version.id), None)
+        if candidate:
+            update_values: dict[str, object] = {"status": DocumentVersionStatus.EFFECTIVE, "superseded_at": None}
+            if not candidate.effective_at:
+                update_values["effective_at"] = datetime.now(timezone.utc)
+            await db.execute(
+                sa_update(DocumentVersion)
+                .where(DocumentVersion.id == candidate.id)
+                .values(**update_values)
+            )
+            await db.execute(
+                sa_update(Document)
+                .where(Document.id == doc.id)
+                .values(current_effective_version_id=candidate.id)
+            )
+        else:
+            await db.execute(
+                sa_update(Document)
+                .where(Document.id == doc.id)
+                .values(current_effective_version_id=None)
+            )
+
+    await db.execute(sa_delete(DocumentVersion).where(DocumentVersion.id == version.id))
     db.add(
         AuditLog(
             study_id=doc.trial_id,
             entity_type="DOCUMENT_VERSION",
             entity_id=version.id,
-            action="VERSION_SUBMITTED",
-            detail=_audit_detail(before, _version_snapshot(version)),
+            action="VERSION_DELETED",
+            detail=_audit_detail(before, None),
             operator_id=current_user.id,
             operator_role=_role_value(current_user),
         )
     )
     await db.commit()
-    await db.refresh(workflow)
-    return workflow
 
-
-async def approve_version(
-    db: AsyncSession,
-    version_id: uuid.UUID,
-    comment: str | None,
-    current_user,
-) -> VersionWorkflow:
-    version = await version_crud.get(db, version_id)
-    if not version:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="版本不存在")
-    doc = await document_crud.get(db, version.document_id)
-    if not doc:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
-    await _ensure_study_access(db, doc.trial_id, current_user, action="approve")
-
-    workflow = await workflow_crud.get_by_version(db, version_id)
-    if not workflow or workflow.status != WorkflowStatus.PENDING:
-        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="审批流程不可用")
-
-    before = _version_snapshot(version)
-    version.status = DocumentVersionStatus.APPROVED
-    version.approved_at = datetime.now(timezone.utc)
-    workflow.status = WorkflowStatus.APPROVED
-    workflow.completed_at = version.approved_at
-
-    db.add(version)
-    db.add(workflow)
-    db.add(
-        WorkflowAction(
-            workflow_id=workflow.id,
-            node_order=workflow.current_node,
-            action=WorkflowActionType.APPROVE,
-            actor_id=current_user.id,
-            comment=comment,
-        )
-    )
-    db.add(
-        AuditLog(
-            study_id=doc.trial_id,
-            entity_type="DOCUMENT_VERSION",
-            entity_id=version.id,
-            action="VERSION_APPROVED",
-            detail=_audit_detail(before, _version_snapshot(version)),
-            operator_id=current_user.id,
-            operator_role=_role_value(current_user),
-        )
-    )
-    await db.commit()
-    await db.refresh(workflow)
-    return workflow
-
-
-async def reject_version(
-    db: AsyncSession,
-    version_id: uuid.UUID,
-    comment: str | None,
-    current_user,
-) -> VersionWorkflow:
-    version = await version_crud.get(db, version_id)
-    if not version:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="版本不存在")
-    doc = await document_crud.get(db, version.document_id)
-    if not doc:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
-    await _ensure_study_access(db, doc.trial_id, current_user, action="approve")
-
-    workflow = await workflow_crud.get_by_version(db, version_id)
-    if not workflow or workflow.status != WorkflowStatus.PENDING:
-        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="审批流程不可用")
-
-    before = _version_snapshot(version)
-    version.status = DocumentVersionStatus.WITHDRAWN
-    version.withdrawn_at = datetime.now(timezone.utc)
-    workflow.status = WorkflowStatus.REJECTED
-    workflow.completed_at = version.withdrawn_at
-
-    db.add(version)
-    db.add(workflow)
-    db.add(
-        WorkflowAction(
-            workflow_id=workflow.id,
-            node_order=workflow.current_node,
-            action=WorkflowActionType.REJECT,
-            actor_id=current_user.id,
-            comment=comment,
-        )
-    )
-    db.add(
-        AuditLog(
-            study_id=doc.trial_id,
-            entity_type="DOCUMENT_VERSION",
-            entity_id=version.id,
-            action="VERSION_REJECTED",
-            detail=_audit_detail(before, _version_snapshot(version)),
-            operator_id=current_user.id,
-            operator_role=_role_value(current_user),
-        )
-    )
-    await db.commit()
-    await db.refresh(workflow)
-    return workflow
-
-
-async def make_version_effective(
-    db: AsyncSession,
-    version_id: uuid.UUID,
-    current_user,
-) -> DocumentVersion:
-    version = await version_crud.get(db, version_id)
-    if not version:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="版本不存在")
-    doc = await document_crud.get(db, version.document_id)
-    if not doc:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
-    await _ensure_study_access(db, doc.trial_id, current_user, action="approve")
-
-    if version.status != DocumentVersionStatus.APPROVED:
-        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="版本未审批通过")
-
-    now = datetime.now(timezone.utc)
-    before = _version_snapshot(version)
-    version.status = DocumentVersionStatus.EFFECTIVE
-    version.effective_at = now
-
-    await db.execute(
-        sa_update(DocumentVersion)
-        .where(
-            DocumentVersion.document_id == version.document_id,
-            DocumentVersion.status == DocumentVersionStatus.EFFECTIVE,
-            DocumentVersion.id != version.id,
-        )
-        .values(status=DocumentVersionStatus.SUPERSEDED, superseded_at=now)
-    )
-    await db.execute(
-        sa_update(Document)
-        .where(Document.id == version.document_id)
-        .values(current_effective_version_id=version.id)
-    )
-    db.add(version)
-    db.add(
-        AuditLog(
-            study_id=doc.trial_id,
-            entity_type="DOCUMENT_VERSION",
-            entity_id=version.id,
-            action="VERSION_EFFECTIVE",
-            detail=_audit_detail(before, _version_snapshot(version)),
-            operator_id=current_user.id,
-            operator_role=_role_value(current_user),
-        )
-    )
-    await db.commit()
-    await db.refresh(version)
-    return version
+    file_path = Path(version.file_uri)
+    if file_path.exists():
+        try:
+            file_path.unlink()
+        except OSError:
+            pass
 
 
 async def delete_document(
@@ -555,19 +469,6 @@ async def get_version_download_response(
     )
 
 
-async def list_workflow_templates(
-    db: AsyncSession,
-    trial_id: uuid.UUID | None,
-    current_user,
-) -> list[WorkflowTemplate]:
-    if trial_id:
-        await _ensure_study_access(db, trial_id, current_user, action="view")
-    elif current_user.role != "ADMIN":
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="权限不足")
-    templates = await template_crud.list_active(db, trial_id)
-    return list(templates)
-
-
 async def create_distributions(
     db: AsyncSession,
     version_id: uuid.UUID,
@@ -580,7 +481,7 @@ async def create_distributions(
     doc = await document_crud.get(db, version.document_id)
     if not doc:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
-    await _ensure_study_access(db, doc.trial_id, current_user, action="distribute")
+    await _ensure_study_member(db, doc.trial_id, current_user)
 
     if version.status != DocumentVersionStatus.EFFECTIVE:
         raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="只有生效版本可分发")
@@ -625,7 +526,7 @@ async def list_distributions(
     doc = await document_crud.get(db, version.document_id)
     if not doc:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
-    await _ensure_study_access(db, doc.trial_id, current_user, action="view")
+    await _ensure_study_member(db, doc.trial_id, current_user)
 
     distributions = await distribution_crud.list_by_version(db, version_id)
     result: list[DistributionRead] = []
@@ -752,3 +653,69 @@ def _aggregate_ack_stats(
         if due_at < now and not acks:
             stats.overdue = target_count
     return stats
+
+
+async def list_distribution_notifications(
+    db: AsyncSession,
+    study_id: uuid.UUID,
+    current_user,
+    *,
+    skip: int = 0,
+    limit: int = 20,
+) -> list[NotificationItem]:
+    role_value = _role_value(current_user)
+    membership = None
+    if role_value != "ADMIN":
+        membership = await member_crud.get_member(db, study_id, current_user.id)
+    role_in_study = membership.role_in_study if membership else None
+
+    target_filters = [
+        (Distribution.target_type == DistributionTargetType.USER)
+        & (Distribution.target_id == str(current_user.id))
+    ]
+    if role_in_study:
+        target_filters.append(
+            (Distribution.target_type == DistributionTargetType.ROLE)
+            & (Distribution.target_id == role_in_study)
+        )
+
+    stmt = (
+        select(
+            Distribution.id,
+            Distribution.document_id,
+            Distribution.version_id,
+            Distribution.created_at,
+            Document.title,
+            Document.doc_no,
+            DocumentVersion.version_no,
+            DocumentVersion.change_summary,
+            DocumentVersion.effective_at,
+        )
+        .join(DocumentVersion, Distribution.version_id == DocumentVersion.id)
+        .join(Document, Distribution.document_id == Document.id)
+        .where(
+            Document.trial_id == study_id,
+            Distribution.status == DistributionStatus.ACTIVE,
+            or_(*target_filters),
+        )
+        .order_by(Distribution.created_at.desc())
+        .offset(skip)
+        .limit(limit)
+    )
+    result = await db.execute(stmt)
+    items: list[NotificationItem] = []
+    for row in result.fetchall():
+        items.append(
+            NotificationItem(
+                id=row.id,
+                document_id=row.document_id,
+                version_id=row.version_id,
+                document_title=row.title,
+                document_no=row.doc_no,
+                version_no=row.version_no,
+                change_summary=row.change_summary,
+                effective_at=row.effective_at,
+                created_at=row.created_at,
+            )
+        )
+    return items

backend/app/services/visit_scheduler.py

@@ -0,0 +1,24 @@
+import asyncio
+from datetime import datetime, timedelta
+from zoneinfo import ZoneInfo
+
+from app.crud import visit as visit_crud
+from app.db.session import SessionLocal
+
+
+async def _run_mark_lost_once() -> None:
+    async with SessionLocal() as session:
+        await visit_crud.mark_overdue_as_lost_global(session)
+
+
+async def run_daily_lost_visit_job(stop_event: asyncio.Event) -> None:
+    tz = ZoneInfo("Asia/Shanghai")
+    while not stop_event.is_set():
+        now = datetime.now(tz)
+        next_midnight = (now + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
+        sleep_seconds = max(1, (next_midnight - now).total_seconds())
+        try:
+            await asyncio.wait_for(stop_event.wait(), timeout=sleep_seconds)
+            break
+        except asyncio.TimeoutError:
+            await _run_mark_lost_once()