Cheng Zhou
3e77127687
增加通用提醒状态、数据库迁移和定时同步，覆盖风险时效、文件回执、项目里程碑、访视窗口与协作申请。 新增网页端和桌面端提醒中心、真实投递诊断与固定隐私通知正文，并补齐登录来源聚合、测试和说明文档。
872 lines
33 KiB
Python
872 lines
33 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Iterable
|
|
from urllib.parse import quote
|
|
|
|
import aiofiles
|
|
from fastapi import HTTPException, UploadFile, status
|
|
from fastapi.responses import FileResponse
|
|
from sqlalchemy import String, and_, cast, delete as sa_delete, or_, select, update as sa_update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.deps import get_cra_site_scope
|
|
from app.core.deps import get_operator_role_label, is_system_admin
|
|
from app.core.project_permissions import role_has_api_permission
|
|
from app.crud import acknowledgement as acknowledgement_crud
|
|
from app.crud import distribution as distribution_crud
|
|
from app.crud import document as document_crud
|
|
from app.crud import etmf as etmf_crud
|
|
from app.crud import document_version as version_crud
|
|
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.models.acknowledgement import Acknowledgement, AcknowledgementType
|
|
from app.models.audit_log import AuditLog
|
|
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.desktop_notification import DesktopNotificationDelivery
|
|
from app.models.notification import Notification
|
|
from app.models.study import Study
|
|
from app.schemas.acknowledgement import AcknowledgementCreate
|
|
from app.schemas.distribution import DistributionCreate, DistributionRead, DistributionStats
|
|
from app.schemas.document import DocumentCreate, DocumentDetail, DocumentSummary, DocumentUpdate
|
|
from app.schemas.document_version import DocumentVersionRead, DocumentVersionSummary
|
|
from app.schemas.notification import NotificationItem
|
|
from app.schemas.user import UserDisplay
|
|
from app.services import notification_service
|
|
|
|
UPLOAD_ROOT = Path(__file__).resolve().parent.parent / "uploads" / "documents"
|
|
|
|
DOCUMENT_ACTION_PERMISSIONS = {
|
|
"view": "documents:read",
|
|
"ack": "documents:read",
|
|
"create_document": "documents:create",
|
|
"update_document": "documents:update",
|
|
"create_version": "documents:update",
|
|
"distribute": "documents:update",
|
|
"delete_document": "documents:delete",
|
|
}
|
|
|
|
|
|
def _safe_original_filename(value: str | None) -> str:
|
|
filename = (value or "").replace("\\", "/").rsplit("/", 1)[-1].strip()
|
|
filename = "".join(character for character in filename if ord(character) >= 32 and ord(character) != 127)
|
|
return filename[:255] or "document"
|
|
|
|
|
|
def _content_disposition(filename: str, disposition: str = "attachment") -> str:
|
|
fallback = "".join(character if 32 <= ord(character) < 127 and character not in {'"', "\\"} else "_" for character in filename)
|
|
fallback = fallback or "download"
|
|
encoded = quote(filename, safe="")
|
|
return f'{disposition}; filename="{fallback}"; filename*=UTF-8\'\'{encoded}'
|
|
|
|
|
|
def _legacy_download_filename(version: DocumentVersion, document: Document) -> str:
|
|
"""Return a readable fallback for rows created before original_filename existed."""
|
|
stored_name = Path(version.file_uri).name
|
|
suffix = Path(stored_name).suffix
|
|
title = _safe_original_filename(document.title)
|
|
if suffix and title.lower().endswith(suffix.lower()):
|
|
return title
|
|
return f"{title}{suffix}"
|
|
|
|
|
|
def _audit_detail(before: dict | None, after: dict | None) -> str:
|
|
payload = {"before": before, "after": after}
|
|
return json.dumps(payload, ensure_ascii=True)
|
|
|
|
|
|
def _doc_snapshot(doc: Document) -> dict:
|
|
status = doc.status.value if hasattr(doc.status, "value") else str(doc.status)
|
|
return {
|
|
"id": str(doc.id),
|
|
"trial_id": str(doc.trial_id),
|
|
"site_id": str(doc.site_id) if doc.site_id else None,
|
|
"etmf_node_id": str(doc.etmf_node_id) if doc.etmf_node_id else None,
|
|
"doc_no": doc.doc_no,
|
|
"status": status,
|
|
"current_effective_version_id": str(doc.current_effective_version_id) if doc.current_effective_version_id else None,
|
|
}
|
|
|
|
|
|
def _version_snapshot(version: DocumentVersion) -> dict:
|
|
status = version.status.value if hasattr(version.status, "value") else str(version.status)
|
|
return {
|
|
"id": str(version.id),
|
|
"document_id": str(version.document_id),
|
|
"version_no": version.version_no,
|
|
"status": status,
|
|
"file_hash": version.file_hash,
|
|
}
|
|
|
|
|
|
def _version_read(version: DocumentVersion, users_by_id: dict[uuid.UUID, object] | None = None) -> DocumentVersionRead:
|
|
creator = users_by_id.get(version.created_by) if users_by_id and version.created_by else None
|
|
return DocumentVersionRead.model_validate(version).model_copy(
|
|
update={"created_by_user": UserDisplay.model_validate(creator) if creator else None}
|
|
)
|
|
|
|
|
|
async def _ensure_study_access(db: AsyncSession, trial_id: uuid.UUID, current_user, action: str):
|
|
study = await study_crud.get(db, trial_id)
|
|
if not study:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="项目不存在")
|
|
if is_system_admin(current_user):
|
|
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="不是项目成员")
|
|
endpoint_key = DOCUMENT_ACTION_PERMISSIONS.get(action, "documents:update")
|
|
allowed = await role_has_api_permission(db, trial_id, membership.role_in_study, endpoint_key, check_prerequisites=False)
|
|
if not allowed:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="权限不足")
|
|
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 is_system_admin(current_user):
|
|
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,
|
|
current_user,
|
|
) -> Document:
|
|
await _ensure_study_access(db, doc_in.trial_id, current_user, action="create_document")
|
|
etmf_node = None
|
|
if doc_in.etmf_node_id:
|
|
etmf_node = await etmf_crud.get_node(db, doc_in.etmf_node_id)
|
|
if not etmf_node:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="eTMF目录不存在")
|
|
if etmf_node.study_id != doc_in.trial_id:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="eTMF目录不属于当前项目")
|
|
if etmf_node.scope_type == DocumentScopeType.SITE and not doc_in.site_id:
|
|
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="中心级eTMF目录需指定中心")
|
|
|
|
scope_type = etmf_node.scope_type if etmf_node else doc_in.scope_type
|
|
if scope_type == DocumentScopeType.SITE and not doc_in.site_id:
|
|
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="分中心文件需指定分中心")
|
|
site_id = doc_in.site_id if scope_type == DocumentScopeType.SITE else None
|
|
if site_id:
|
|
site = await site_crud.get_site(db, site_id)
|
|
if not site or not site.is_active:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="中心已停用")
|
|
existing = await document_crud.get_by_trial_doc_no(db, doc_in.trial_id, doc_in.doc_no)
|
|
if existing:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="文档编号已存在")
|
|
|
|
doc = Document(
|
|
trial_id=doc_in.trial_id,
|
|
site_id=site_id,
|
|
etmf_node_id=doc_in.etmf_node_id,
|
|
doc_no=doc_in.doc_no,
|
|
doc_type=doc_in.doc_type,
|
|
title=doc_in.title,
|
|
scope_type=scope_type,
|
|
owner_id=doc_in.owner_id,
|
|
description=doc_in.description,
|
|
)
|
|
db.add(doc)
|
|
db.add(
|
|
AuditLog(
|
|
study_id=doc_in.trial_id,
|
|
entity_type="DOCUMENT",
|
|
entity_id=doc.id,
|
|
action="DOCUMENT_CREATED",
|
|
detail=_audit_detail(None, _doc_snapshot(doc)),
|
|
operator_id=current_user.id,
|
|
operator_role=await get_operator_role_label(db, doc.trial_id, current_user),
|
|
)
|
|
)
|
|
await db.commit()
|
|
await db.refresh(doc)
|
|
return doc
|
|
|
|
|
|
async def list_documents(
|
|
db: AsyncSession,
|
|
*,
|
|
trial_id: uuid.UUID,
|
|
site_id: uuid.UUID | None,
|
|
doc_type: str | None,
|
|
status: str | None,
|
|
scope_type: str | None,
|
|
etmf_node_id: uuid.UUID | None = None,
|
|
skip: int,
|
|
limit: int,
|
|
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,
|
|
site_id=site_id,
|
|
doc_type=doc_type,
|
|
status=status,
|
|
scope_type=scope_type,
|
|
etmf_node_id=etmf_node_id,
|
|
cra_site_ids=cra_site_ids,
|
|
skip=skip,
|
|
limit=limit,
|
|
)
|
|
|
|
summaries: list[DocumentSummary] = []
|
|
for doc in docs:
|
|
current_version = None
|
|
if doc.current_effective_version_id:
|
|
version = await version_crud.get(db, doc.current_effective_version_id)
|
|
if version:
|
|
current_version = DocumentVersionSummary.model_validate(version)
|
|
summaries.append(
|
|
DocumentSummary(
|
|
id=doc.id,
|
|
trial_id=doc.trial_id,
|
|
site_id=doc.site_id,
|
|
etmf_node_id=doc.etmf_node_id,
|
|
doc_type=doc.doc_type,
|
|
title=doc.title,
|
|
scope_type=doc.scope_type,
|
|
owner_id=doc.owner_id,
|
|
current_effective_version_id=doc.current_effective_version_id,
|
|
current_effective_version=current_version,
|
|
created_at=doc.created_at,
|
|
updated_at=doc.updated_at,
|
|
)
|
|
)
|
|
return summaries
|
|
|
|
|
|
async def get_document_detail(
|
|
db: AsyncSession,
|
|
document_id: uuid.UUID,
|
|
current_user,
|
|
) -> DocumentDetail:
|
|
doc = await document_crud.get(db, 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")
|
|
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
|
|
if doc.current_effective_version_id:
|
|
version = await version_crud.get(db, doc.current_effective_version_id)
|
|
if version:
|
|
current_version = DocumentVersionSummary.model_validate(version)
|
|
|
|
distribution_stats = await _aggregate_distribution_stats(db, doc.current_effective_version_id)
|
|
owner = None
|
|
if doc.owner_id:
|
|
owner_obj = await user_crud.get_by_id(db, doc.owner_id)
|
|
if owner_obj:
|
|
owner = UserDisplay.model_validate(owner_obj)
|
|
creator_ids = {v.created_by for v in versions if v.created_by}
|
|
version_creators = await user_crud.get_users_by_ids(db, creator_ids)
|
|
return DocumentDetail(
|
|
id=doc.id,
|
|
trial_id=doc.trial_id,
|
|
site_id=doc.site_id,
|
|
etmf_node_id=doc.etmf_node_id,
|
|
doc_type=doc.doc_type,
|
|
title=doc.title,
|
|
scope_type=doc.scope_type,
|
|
owner_id=doc.owner_id,
|
|
current_effective_version_id=doc.current_effective_version_id,
|
|
current_effective_version=current_version,
|
|
description=doc.description,
|
|
created_at=doc.created_at,
|
|
updated_at=doc.updated_at,
|
|
owner=owner,
|
|
version_timeline=[_version_read(v, version_creators) for v in versions],
|
|
distribution_stats=distribution_stats,
|
|
)
|
|
|
|
|
|
async def update_document(
|
|
db: AsyncSession,
|
|
document_id: uuid.UUID,
|
|
doc_in: DocumentUpdate,
|
|
current_user,
|
|
) -> Document:
|
|
doc = await document_crud.get(db, 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="update_document")
|
|
|
|
values = doc_in.model_dump(exclude_unset=True)
|
|
if not values:
|
|
return doc
|
|
|
|
next_scope = values.get("scope_type", doc.scope_type)
|
|
next_scope_value = getattr(next_scope, "value", next_scope)
|
|
next_site_id = values.get("site_id", doc.site_id)
|
|
if next_scope_value == DocumentScopeType.SITE.value:
|
|
if not next_site_id:
|
|
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="分中心文件需指定分中心")
|
|
site = await site_crud.get_site(db, next_site_id)
|
|
if not site or site.study_id != doc.trial_id:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="分中心不属于当前项目")
|
|
if not site.is_active:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="中心已停用")
|
|
else:
|
|
values["site_id"] = None
|
|
|
|
before = _doc_snapshot(doc)
|
|
await document_crud.update(db, document_id, values, commit=False)
|
|
updated = await document_crud.get(db, document_id)
|
|
if not updated:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
|
|
db.add(
|
|
AuditLog(
|
|
study_id=updated.trial_id,
|
|
entity_type="DOCUMENT",
|
|
entity_id=updated.id,
|
|
action="DOCUMENT_UPDATED",
|
|
detail=_audit_detail(before, _doc_snapshot(updated)),
|
|
operator_id=current_user.id,
|
|
operator_role=await get_operator_role_label(db, updated.trial_id, current_user),
|
|
)
|
|
)
|
|
await db.commit()
|
|
await db.refresh(updated)
|
|
return updated
|
|
|
|
|
|
async def create_version(
|
|
db: AsyncSession,
|
|
document_id: uuid.UUID,
|
|
*,
|
|
version_no: str,
|
|
version_date: str,
|
|
change_summary: str | None,
|
|
parent_version_id: uuid.UUID | None,
|
|
file: UploadFile,
|
|
current_user,
|
|
) -> DocumentVersion:
|
|
doc = await document_crud.get(db, 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="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="版本号已存在")
|
|
|
|
if parent_version_id:
|
|
parent = await version_crud.get(db, parent_version_id)
|
|
if not parent or parent.document_id != document_id:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="父版本不存在或不属于该文档")
|
|
|
|
content = await file.read()
|
|
file_hash = hashlib.sha256(content).hexdigest()
|
|
dest_dir = UPLOAD_ROOT / str(document_id)
|
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
|
original_filename = _safe_original_filename(file.filename)
|
|
unique_name = f"{uuid.uuid4()}{Path(original_filename).suffix}"
|
|
dest_path = dest_dir / unique_name
|
|
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.EFFECTIVE,
|
|
effective_at=effective_at,
|
|
file_uri=str(dest_path),
|
|
original_filename=original_filename,
|
|
file_hash=file_hash,
|
|
file_size=len(content),
|
|
mime_type=file.content_type,
|
|
change_summary=change_summary,
|
|
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,
|
|
entity_type="DOCUMENT_VERSION",
|
|
entity_id=version.id,
|
|
action="VERSION_CREATED",
|
|
detail=_audit_detail(None, _version_snapshot(version)),
|
|
operator_id=current_user.id,
|
|
operator_role=await get_operator_role_label(db, doc.trial_id, 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 delete_version(
|
|
db: AsyncSession,
|
|
version_id: uuid.UUID,
|
|
current_user,
|
|
) -> 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="delete_document")
|
|
|
|
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)
|
|
|
|
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_DELETED",
|
|
detail=_audit_detail(before, None),
|
|
operator_id=current_user.id,
|
|
operator_role=await get_operator_role_label(db, doc.trial_id, current_user),
|
|
)
|
|
)
|
|
await db.commit()
|
|
|
|
file_path = Path(version.file_uri)
|
|
if file_path.exists():
|
|
try:
|
|
file_path.unlink()
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
async def delete_document(
|
|
db: AsyncSession,
|
|
document_id: uuid.UUID,
|
|
current_user,
|
|
) -> Document:
|
|
doc = await document_crud.get(db, 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="delete_document")
|
|
|
|
if doc.status == DocumentStatus.ARCHIVED:
|
|
return doc
|
|
|
|
before = _doc_snapshot(doc)
|
|
doc.status = DocumentStatus.ARCHIVED
|
|
db.add(doc)
|
|
db.add(
|
|
AuditLog(
|
|
study_id=doc.trial_id,
|
|
entity_type="DOCUMENT",
|
|
entity_id=doc.id,
|
|
action="DOCUMENT_ARCHIVED",
|
|
detail=_audit_detail(before, _doc_snapshot(doc)),
|
|
operator_id=current_user.id,
|
|
operator_role=await get_operator_role_label(db, doc.trial_id, current_user),
|
|
)
|
|
)
|
|
await db.commit()
|
|
await db.refresh(doc)
|
|
return doc
|
|
|
|
|
|
async def get_version_download_response(
|
|
db: AsyncSession,
|
|
version_id: uuid.UUID,
|
|
current_user,
|
|
) -> FileResponse:
|
|
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="view")
|
|
|
|
file_path = Path(version.file_uri)
|
|
if not file_path.exists():
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文件不存在")
|
|
filename = version.original_filename or _legacy_download_filename(version, doc)
|
|
return FileResponse(
|
|
path=str(file_path),
|
|
media_type=version.mime_type or "application/octet-stream",
|
|
headers={"Content-Disposition": _content_disposition(filename)},
|
|
)
|
|
|
|
|
|
async def create_distributions(
|
|
db: AsyncSession,
|
|
version_id: uuid.UUID,
|
|
payload: DistributionCreate,
|
|
current_user,
|
|
) -> list[Distribution]:
|
|
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="distribute")
|
|
|
|
if version.status != DocumentVersionStatus.EFFECTIVE:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="只有生效版本可分发")
|
|
|
|
created: list[Distribution] = []
|
|
for target in payload.targets:
|
|
distribution = Distribution(
|
|
document_id=version.document_id,
|
|
version_id=version.id,
|
|
target_type=target.target_type,
|
|
target_id=target.target_id,
|
|
due_at=payload.due_at,
|
|
created_by=current_user.id,
|
|
)
|
|
db.add(distribution)
|
|
created.append(distribution)
|
|
db.add(
|
|
AuditLog(
|
|
study_id=doc.trial_id,
|
|
entity_type="DISTRIBUTION",
|
|
entity_id=distribution.id,
|
|
action="DISTRIBUTION_CREATED",
|
|
detail=_audit_detail(None, {"version_id": str(version.id), "target_id": target.target_id}),
|
|
operator_id=current_user.id,
|
|
operator_role=await get_operator_role_label(db, doc.trial_id, current_user),
|
|
)
|
|
)
|
|
await db.commit()
|
|
for distribution in created:
|
|
await db.refresh(distribution)
|
|
return created
|
|
|
|
|
|
async def list_distributions(
|
|
db: AsyncSession,
|
|
version_id: uuid.UUID,
|
|
current_user,
|
|
) -> list[DistributionRead]:
|
|
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_member(db, doc.trial_id, current_user)
|
|
|
|
distributions = await distribution_crud.list_by_version(db, version_id)
|
|
result: list[DistributionRead] = []
|
|
for distribution in distributions:
|
|
stats = await _stats_for_distribution(db, distribution.id, distribution.due_at)
|
|
result.append(
|
|
DistributionRead(
|
|
id=distribution.id,
|
|
document_id=distribution.document_id,
|
|
version_id=distribution.version_id,
|
|
target_type=distribution.target_type,
|
|
target_id=distribution.target_id,
|
|
status=distribution.status,
|
|
due_at=distribution.due_at,
|
|
created_by=distribution.created_by,
|
|
created_at=distribution.created_at,
|
|
stats=stats,
|
|
)
|
|
)
|
|
return result
|
|
|
|
|
|
async def create_acknowledgement(
|
|
db: AsyncSession,
|
|
distribution_id: uuid.UUID,
|
|
payload: AcknowledgementCreate,
|
|
current_user,
|
|
) -> Acknowledgement:
|
|
distribution = await distribution_crud.get(db, distribution_id)
|
|
if not distribution:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="分发不存在")
|
|
version = await version_crud.get(db, distribution.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="文档不存在")
|
|
membership = await _ensure_study_access(db, doc.trial_id, current_user, action="ack")
|
|
|
|
if distribution.target_type == DistributionTargetType.USER and distribution.target_id != str(current_user.id):
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="不在分发范围内")
|
|
if distribution.target_type == DistributionTargetType.ROLE:
|
|
if distribution.target_id not in ("ADMIN" if is_system_admin(current_user) else "", getattr(membership, "role_in_study", "")):
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="不在分发范围内")
|
|
if distribution.target_type == DistributionTargetType.SITE and not is_system_admin(current_user):
|
|
site_ids = await site_crud.list_ids_by_contact_user(db, doc.trial_id, current_user.id)
|
|
if distribution.target_id not in {str(site_id) for site_id in site_ids}:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="不在分发中心范围内")
|
|
|
|
if payload.ack_type != AcknowledgementType.RECEIVED:
|
|
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="仅支持已接收回执")
|
|
|
|
existing = await acknowledgement_crud.get_existing(
|
|
db,
|
|
distribution_id=distribution_id,
|
|
user_id=current_user.id,
|
|
ack_type=payload.ack_type,
|
|
)
|
|
if existing:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="回执已存在")
|
|
|
|
ack = Acknowledgement(
|
|
distribution_id=distribution_id,
|
|
user_id=current_user.id,
|
|
site_id=None,
|
|
ack_type=payload.ack_type,
|
|
due_at=distribution.due_at,
|
|
acked_at=datetime.now(timezone.utc),
|
|
note=payload.evidence,
|
|
)
|
|
db.add(ack)
|
|
db.add(
|
|
AuditLog(
|
|
study_id=doc.trial_id,
|
|
entity_type="ACKNOWLEDGEMENT",
|
|
entity_id=ack.id,
|
|
action="ACK_CREATED",
|
|
detail=_audit_detail(None, {"distribution_id": str(distribution.id), "ack_type": payload.ack_type}),
|
|
operator_id=current_user.id,
|
|
operator_role=await get_operator_role_label(db, doc.trial_id, current_user),
|
|
)
|
|
)
|
|
await notification_service.resolve_source_notifications(
|
|
db,
|
|
source_type="DOCUMENT_DISTRIBUTION",
|
|
source_id=str(distribution.id),
|
|
recipient_id=current_user.id,
|
|
)
|
|
await db.commit()
|
|
await db.refresh(ack)
|
|
return ack
|
|
|
|
|
|
async def _stats_for_distribution(
|
|
db: AsyncSession,
|
|
distribution_id: uuid.UUID,
|
|
due_at: datetime | None,
|
|
) -> DistributionStats:
|
|
acks = await acknowledgement_crud.list_by_distribution(db, distribution_id)
|
|
return _aggregate_ack_stats(acks, due_at, target_count=1)
|
|
|
|
|
|
async def _aggregate_distribution_stats(
|
|
db: AsyncSession,
|
|
version_id: uuid.UUID | None,
|
|
) -> DistributionStats:
|
|
if not version_id:
|
|
return DistributionStats()
|
|
distributions = await distribution_crud.list_by_version(db, version_id)
|
|
stats = DistributionStats()
|
|
for distribution in distributions:
|
|
acks = await acknowledgement_crud.list_by_distribution(db, distribution.id)
|
|
dist_stats = _aggregate_ack_stats(acks, distribution.due_at, target_count=1)
|
|
stats.total += dist_stats.total
|
|
stats.received += dist_stats.received
|
|
# only RECEIVED is supported
|
|
stats.overdue += dist_stats.overdue
|
|
return stats
|
|
|
|
|
|
def _aggregate_ack_stats(
|
|
acks: Iterable[Acknowledgement],
|
|
due_at: datetime | None,
|
|
target_count: int,
|
|
) -> DistributionStats:
|
|
stats = DistributionStats()
|
|
stats.total = target_count
|
|
for ack in acks:
|
|
if ack.ack_type == AcknowledgementType.RECEIVED:
|
|
stats.received += 1
|
|
if due_at:
|
|
now = datetime.now(timezone.utc)
|
|
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]:
|
|
membership = None
|
|
if not is_system_admin(current_user):
|
|
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,
|
|
Study.name.label("study_name"),
|
|
DesktopNotificationDelivery.delivered_at,
|
|
DesktopNotificationDelivery.read_at,
|
|
)
|
|
.join(DocumentVersion, Distribution.version_id == DocumentVersion.id)
|
|
.join(Document, Distribution.document_id == Document.id)
|
|
.join(Study, Document.trial_id == Study.id)
|
|
.outerjoin(
|
|
Notification,
|
|
and_(
|
|
Notification.source_type == "DOCUMENT_DISTRIBUTION",
|
|
Notification.source_id == cast(Distribution.id, String),
|
|
Notification.recipient_id == current_user.id,
|
|
),
|
|
)
|
|
.outerjoin(
|
|
DesktopNotificationDelivery,
|
|
and_(
|
|
DesktopNotificationDelivery.notification_id == Notification.id,
|
|
DesktopNotificationDelivery.user_id == current_user.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,
|
|
study_id=study_id,
|
|
study_name=row.study_name,
|
|
delivered_at=row.delivered_at,
|
|
read_at=row.read_at,
|
|
)
|
|
)
|
|
return items
|