Cheng Zhou
c1dd8def2c
新增与项目平级的协作台账，提供空白模板、独立账号授权和在线编辑。 复用共享库的访问权限与历史版本界面，拆分信息维护入口并统一工作台卡片样式。 补充在线文档字体构建支持、数据库表结构升级和回归测试。 不包含本地台账内容、账号授权记录或历史文件迁移数据。
260 lines
12 KiB
Python
260 lines
12 KiB
Python
"""Global ledger metadata and grants; reuse collaboration revisions and sessions."""
|
|
import io
|
|
import uuid
|
|
import zipfile
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import aiofiles
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import or_, select, text
|
|
|
|
from app.core.config import settings
|
|
from app.core.deps import is_system_admin
|
|
from app.models.collaboration import CollaborationFile, CollaborationMember, CollaborationRevision
|
|
from app.models.user import User, UserStatus
|
|
from app.schemas.ledger import LedgerRead, LedgerSettings
|
|
from app.services import collaboration_service as collaboration
|
|
from app.services import ledger_access
|
|
|
|
LEDGER_TEMPLATES = (
|
|
("ledger-1", "医学事务部合同台账明细表.xlsx", "contract-ledger.xlsx", "记录合同编号、项目代码与合同内容"),
|
|
("ledger-2", "医学事务部临床运营项目编号.xlsx", "clinical-project-register.xlsx", "记录项目编号、产品信息与生效日期"),
|
|
)
|
|
TEMPLATE_ROOT = Path(__file__).resolve().parents[1] / "templates" / "ledgers"
|
|
|
|
|
|
async def get_ledger(db, ledger_id, user, *, manage=False, lock=False):
|
|
stmt = select(CollaborationFile).where(
|
|
CollaborationFile.id == ledger_id, CollaborationFile.scope == "LEDGER",
|
|
).execution_options(populate_existing=True)
|
|
if lock:
|
|
stmt = stmt.with_for_update()
|
|
item = await db.scalar(stmt)
|
|
if not item:
|
|
raise HTTPException(404, "台账不存在或未获授权")
|
|
await ledger_access.require_access(db, item, user, manage=manage)
|
|
return item
|
|
|
|
|
|
async def read_ledger(db, item, user):
|
|
role = await ledger_access.role_for(db, item, user)
|
|
await ledger_access.require_access(db, item, user)
|
|
owner = await db.get(User, item.owner_id)
|
|
revision = await db.get(CollaborationRevision, item.current_revision_id) if item.current_revision_id else None
|
|
return LedgerRead(
|
|
id=item.id, title=item.title, description=item.description, status=item.status,
|
|
owner_id=item.owner_id, owner_name=owner.full_name if owner else "",
|
|
generation=item.generation, updated_at=item.updated_at,
|
|
current_revision_id=item.current_revision_id,
|
|
current_revision_no=revision.revision_no if revision else None,
|
|
current_revision_file_size=revision.file_size if revision else None,
|
|
current_revision_mime_type=revision.mime_type if revision else None,
|
|
current_revision_created_at=revision.created_at if revision else None,
|
|
role=role, can_edit=await ledger_access.can_edit(db, item, user),
|
|
can_manage=role == "MANAGER", can_export=await ledger_access.can_export(db, item, user),
|
|
allow_export=item.allow_export, allow_sheet_structure_edit=item.allow_sheet_structure_edit,
|
|
allow_edit_request=item.allow_edit_request,
|
|
can_request_edit=await collaboration.can_request_edit_file(db, item, user),
|
|
edit_request_status=await collaboration.edit_request_status(db, item, user.id),
|
|
can_transfer_ownership=await collaboration.can_transfer_ownership(db, item, user),
|
|
)
|
|
|
|
|
|
async def list_ledgers(db, user):
|
|
stmt = select(CollaborationFile).where(CollaborationFile.scope == "LEDGER")
|
|
if not is_system_admin(user):
|
|
granted = select(CollaborationMember.file_id).where(
|
|
CollaborationMember.user_id == user.id,
|
|
CollaborationMember.role.in_(["VIEWER", "EDITOR", "MANAGER"]),
|
|
)
|
|
stmt = stmt.where(or_(CollaborationFile.owner_id == user.id, CollaborationFile.id.in_(granted)))
|
|
rows = (await db.scalars(stmt.order_by(CollaborationFile.ledger_key))).all()
|
|
return [await read_ledger(db, item, user) for item in rows]
|
|
|
|
|
|
async def initialize_ledgers(db, user):
|
|
if not is_system_admin(user):
|
|
raise HTTPException(403, "仅系统管理员可以初始化台账")
|
|
# Serialize initialization across administrators; unique keys also prevent duplicates.
|
|
if db.get_bind().dialect.name == "postgresql":
|
|
await db.execute(text("SELECT pg_advisory_xact_lock(2026090301)"))
|
|
for key, title, template_name, description in LEDGER_TEMPLATES:
|
|
if await db.scalar(select(CollaborationFile.id).where(CollaborationFile.ledger_key == key)):
|
|
continue
|
|
item = CollaborationFile(
|
|
id=uuid.uuid4(), scope="LEDGER", ledger_key=key, study_id=None, folder_id=None,
|
|
title=title, description=description, file_type="cell", extension="xlsx", owner_id=user.id,
|
|
status="ACTIVE", generation=1, allow_sheet_structure_edit=False, allow_export=False,
|
|
)
|
|
db.add(item)
|
|
await db.flush()
|
|
content, backup = collaboration.apply_workbook_structure_policy(
|
|
(TEMPLATE_ROOT / template_name).read_bytes(), file_id=item.id,
|
|
allow_sheet_structure_edit=False, protection_backup=None,
|
|
)
|
|
item.sheet_structure_protection_backup = backup
|
|
await collaboration.append_revision(db, item, content, source="CREATE", created_by=user.id)
|
|
await db.commit()
|
|
return await list_ledgers(db, user)
|
|
|
|
|
|
def check_generation(item, generation):
|
|
if item.generation != generation:
|
|
raise HTTPException(409, "台账已发生变化，请刷新后重试")
|
|
|
|
|
|
def advance_generation(item):
|
|
item.generation += 1
|
|
item.updated_at = datetime.now(timezone.utc)
|
|
|
|
|
|
async def active_user(db, user_id):
|
|
user = await db.get(User, user_id)
|
|
if not user or not user.is_active:
|
|
raise HTTPException(422, "请选择有效的系统账号")
|
|
return user
|
|
|
|
|
|
async def update_settings(db, item, payload, user):
|
|
await ledger_access.require_access(db, item, user, manage=True)
|
|
check_generation(item, payload.generation)
|
|
if payload.owner_id != item.owner_id:
|
|
if not is_system_admin(user):
|
|
raise HTTPException(403, "仅系统管理员可以变更台账负责人")
|
|
await active_user(db, payload.owner_id)
|
|
if payload.allow_sheet_structure_edit != item.allow_sheet_structure_edit:
|
|
revision = await db.get(CollaborationRevision, item.current_revision_id)
|
|
if not revision or not Path(revision.file_uri).is_file():
|
|
raise HTTPException(404, "台账内容不存在")
|
|
async with aiofiles.open(revision.file_uri, "rb") as stream:
|
|
content = await stream.read(settings.COLLABORATION_MAX_FILE_BYTES + 1)
|
|
content, backup = collaboration.apply_workbook_structure_policy(
|
|
content, file_id=item.id, allow_sheet_structure_edit=payload.allow_sheet_structure_edit,
|
|
protection_backup=item.sheet_structure_protection_backup,
|
|
)
|
|
item.sheet_structure_protection_backup = backup
|
|
await collaboration.append_revision(db, item, content, source="PERMISSION_CHANGE", created_by=user.id)
|
|
item.title = collaboration._safe_title(payload.title, "xlsx")
|
|
item.description = payload.description.strip()
|
|
item.status = payload.status
|
|
item.owner_id = payload.owner_id
|
|
item.allow_export = payload.allow_export
|
|
if payload.allow_edit_request is not None:
|
|
item.allow_edit_request = payload.allow_edit_request
|
|
item.allow_sheet_structure_edit = payload.allow_sheet_structure_edit
|
|
advance_generation(item)
|
|
await db.commit()
|
|
return await read_ledger(db, item, user)
|
|
|
|
|
|
async def update_access_settings(db, item, payload, user):
|
|
values = dict(title=item.title, description=item.description, status=item.status,
|
|
owner_id=item.owner_id, allow_export=item.allow_export,
|
|
allow_edit_request=item.allow_edit_request,
|
|
allow_sheet_structure_edit=item.allow_sheet_structure_edit)
|
|
values.update(payload.model_dump(exclude_none=True))
|
|
return await update_settings(db, item, LedgerSettings(**values), user)
|
|
|
|
|
|
async def list_members(db, item):
|
|
rows = (await db.execute(select(CollaborationMember, User).join(
|
|
User, User.id == CollaborationMember.user_id,
|
|
).where(CollaborationMember.file_id == item.id).order_by(User.full_name))).all()
|
|
grants = {user.id: dict(user_id=user.id, full_name=user.full_name, email=user.email,
|
|
role=member.role, is_active=user.is_active) for member, user in rows}
|
|
implicit = (await db.scalars(select(User).where(or_(
|
|
User.id == item.owner_id,
|
|
(User.is_admin.is_(True)) & (User.status == UserStatus.ACTIVE),
|
|
)).order_by(User.full_name))).all()
|
|
for user in implicit:
|
|
grants[user.id] = dict(user_id=user.id, full_name=user.full_name, email=user.email,
|
|
role="MANAGER", is_active=user.is_active,
|
|
protected_label="所有者" if user.id == item.owner_id else "系统管理员")
|
|
return sorted(grants.values(), key=lambda member: (
|
|
member["user_id"] != item.owner_id, member["role"] != "MANAGER", member["full_name"],
|
|
))
|
|
|
|
|
|
async def set_member(db, item, payload, user):
|
|
await ledger_access.require_access(db, item, user, manage=True)
|
|
target = await active_user(db, payload.user_id)
|
|
if target.id == item.owner_id or is_system_admin(target):
|
|
raise HTTPException(422, "负责人和系统管理员已拥有管理权限")
|
|
member = await db.scalar(select(CollaborationMember).where(
|
|
CollaborationMember.file_id == item.id, CollaborationMember.user_id == target.id,
|
|
))
|
|
if member and member.role == payload.role:
|
|
return
|
|
if member:
|
|
member.role = payload.role
|
|
else:
|
|
db.add(CollaborationMember(file_id=item.id, user_id=target.id, role=payload.role, invited_by=user.id))
|
|
advance_generation(item)
|
|
await db.commit()
|
|
|
|
|
|
async def remove_member(db, item, user_id, user):
|
|
await ledger_access.require_access(db, item, user, manage=True)
|
|
if user_id == item.owner_id:
|
|
raise HTTPException(422, "不能移除台账负责人")
|
|
target = await db.get(User, user_id)
|
|
if target and is_system_admin(target):
|
|
raise HTTPException(422, "系统管理员始终拥有管理权限")
|
|
member = await db.scalar(select(CollaborationMember).where(
|
|
CollaborationMember.file_id == item.id, CollaborationMember.user_id == user_id,
|
|
))
|
|
if member:
|
|
await db.delete(member)
|
|
advance_generation(item)
|
|
await db.commit()
|
|
|
|
|
|
async def candidates(db, keyword):
|
|
stmt = select(User).where(User.status == UserStatus.ACTIVE)
|
|
if keyword:
|
|
pattern = f"%{keyword}%"
|
|
stmt = stmt.where(or_(User.full_name.ilike(pattern), User.email.ilike(pattern)))
|
|
rows = (await db.scalars(stmt.order_by(User.full_name, User.id))).all()
|
|
return [dict(user_id=user.id, full_name=user.full_name, email=user.email,
|
|
role_in_study="系统管理员" if user.is_admin else "系统账号") for user in rows]
|
|
|
|
|
|
async def import_content(db, item, upload, generation, user):
|
|
await ledger_access.require_access(db, item, user, manage=True)
|
|
check_generation(item, generation)
|
|
if item.status != "ACTIVE":
|
|
raise HTTPException(409, "请先启用台账再导入")
|
|
if Path(upload.filename or "").suffix.lower() != ".xlsx":
|
|
raise HTTPException(415, "台账仅支持 XLSX 文件")
|
|
content = await upload.read(settings.COLLABORATION_MAX_FILE_BYTES + 1)
|
|
if not content or len(content) > settings.COLLABORATION_MAX_FILE_BYTES:
|
|
raise HTTPException(413, "台账文件为空或超出大小限制")
|
|
try:
|
|
with zipfile.ZipFile(io.BytesIO(content)) as package:
|
|
if "xl/workbook.xml" not in package.namelist():
|
|
raise ValueError()
|
|
if sum(info.file_size for info in package.infolist()) > settings.COLLABORATION_MAX_FILE_BYTES * 20:
|
|
raise ValueError()
|
|
if any("vbaproject" in name.lower() for name in package.namelist()):
|
|
raise ValueError()
|
|
except (zipfile.BadZipFile, ValueError):
|
|
raise HTTPException(422, "XLSX 文件无效、包含宏或解压后过大")
|
|
content, backup = collaboration.apply_workbook_structure_policy(
|
|
content, file_id=item.id, allow_sheet_structure_edit=item.allow_sheet_structure_edit,
|
|
protection_backup=None,
|
|
)
|
|
item.sheet_structure_protection_backup = backup
|
|
await collaboration.append_revision(db, item, content, source="IMPORT", created_by=user.id)
|
|
advance_generation(item)
|
|
await db.commit()
|
|
return await read_ledger(db, item, user)
|
|
|
|
|
|
async def restore_revision(db, item, revision_id, generation, user):
|
|
await ledger_access.require_access(db, item, user, manage=True)
|
|
check_generation(item, generation)
|
|
if item.status != "ACTIVE":
|
|
raise HTTPException(409, "请先启用台账再恢复版本")
|
|
return await collaboration.restore_revision(db, item, revision_id, user, None)
|