"""主案、子方案只读查询、DTO和查看审计。""" from __future__ import annotations import json import logging from datetime import datetime, timezone from typing import Any, Mapping from sqlalchemy import String, and_, cast, select from sqlalchemy.orm import defer from sqlalchemy.orm.attributes import set_committed_value from dms.common.enums import ( AuditAction, AuditTarget, DocumentStatus, DocumentType, RoleCode, SecurityLevel, ) from dms.common.errors import ( DocumentViewForbiddenError, InvalidArgumentError, MainPlanNotFoundError, ResourceNotFoundError, SecurityLevelForbiddenError, ) from dms.common.pagination import PageRequest, page_result from dms.common.response import serialize_id from dms.common.time_range import parse_datetime, parse_updated_range from dms.database.transaction import transaction from dms.extensions import db from dms.models import ( LOCK_DEFER_LOB, AttachmentBinding, Category, Document, SubPlanBinding, ) from dms.security.auth_context import AuthContext, get_auth_context from dms.services.audit_service import business_audit from dms.services.authorization_service import ( AccessMemo, PlanAccess, _active_permissions, evaluate_plan_access, plan_allowed_actions, ) from dms.services.document_password_service import ( is_view_locked, protected_actions_of, view_password_verified, ) logger = logging.getLogger(__name__) PLAN_TYPES = (DocumentType.MAIN.value, DocumentType.SUB_PLAN.value) # LOCK_DEFER_LOB 定义于 dms.models.document(供密码锁等下游服务复用,避免循环导入)。 DOCUMENT_SORTS = { "documentName": Document.document_name, "createdAt": Document.created_at, "updatedAt": Document.updated_at, "viewCount": Document.view_count, "downloadCount": Document.download_count, "sortNo": Document.sort_no, } def parse_string_id(value: str, *, field: str = "id") -> int: if not value.isdecimal() or int(value) <= 0: raise InvalidArgumentError(f"{field}必须是正整数形式的字符串ID") return int(value) def _integer( params: Mapping[str, str], name: str, default: int, ) -> int: raw = params.get(name) if raw is None: return default if not raw.isdecimal(): raise InvalidArgumentError(f"{name}必须是正整数") return int(raw) def _page(params: Mapping[str, str]) -> PageRequest: return PageRequest( page=_integer(params, "page", 1), page_size=_integer(params, "pageSize", 20), ) def _enum(value: str | None, enum_type, name: str) -> str | None: if value is None or value == "": return None try: return enum_type(value).value except ValueError as exc: raise InvalidArgumentError(f"{name}不是有效枚举值") from exc def _date(value: str | None, name: str) -> datetime | None: return parse_datetime(value, name=name) def _sort( params: Mapping[str, str], *, allowed: dict[str, Any] = DOCUMENT_SORTS, default: str = "updatedAt", ) -> tuple[Any, str]: sort_by = params.get("sortBy", default) direction = params.get("sortDirection", "desc").lower() if sort_by not in allowed: raise InvalidArgumentError("sortBy不是允许的排序字段") if direction not in {"asc", "desc"}: raise InvalidArgumentError("sortDirection必须是asc或desc") column = allowed[sort_by] return (column.asc() if direction == "asc" else column.desc()), direction def _iso(value: datetime) -> str: return value.replace(tzinfo=timezone.utc).isoformat(timespec="milliseconds").replace( "+00:00", "Z" ) def _permission_summary( access: PlanAccess, document: Document, memo: AccessMemo | None = None, ) -> tuple[dict[str, object], str]: """借阅授权摘要(契约1.19):当前生效的借阅人数与名单文本。 到期过滤与 evaluate_plan_access 惰性判定同口径(expires_at > now), 过期未清理的条目不计入摘要。 """ source = access.source if source is None: return { "borrowerCount": 0, "inheritedFromMainPlan": False, }, "未配置" now = datetime.now(timezone.utc).replace(tzinfo=None) borrowers = [ item for item in _active_permissions(source.id, memo) if item.can_view and item.expires_at > now ] summary = ( "、".join(item.subject_name for item in borrowers) if borrowers else "未配置" ) return { "borrowerCount": len(borrowers), "inheritedFromMainPlan": document.document_type == DocumentType.SUB_PLAN.value, }, summary _SNIPPET_BEFORE = 40 _SNIPPET_AFTER = 80 def _content_snippet(content_text: str | None, keyword: str | None) -> str | None: """当 keyword 命中正文时,截取包含命中区域的片段。 前后各保留若干字符作为上下文;若未触达正文首尾则以"……"标识截断。 未命中或任一参数为空时返回 None。 """ if not content_text or not keyword or not keyword.strip(): return None kw = keyword.strip() pos = content_text.lower().find(kw.lower()) if pos < 0: return None start = max(pos - _SNIPPET_BEFORE, 0) end = min(pos + len(kw) + _SNIPPET_AFTER, len(content_text)) snippet = content_text[start:end] if start > 0: snippet = "……" + snippet if end < len(content_text): snippet += "……" return snippet def _fill_content_text(documents: list[Document]) -> None: """批量补查正文:列表查询 defer 大字段后,仅当前页需要正文摘要时回填。 使用 set_committed_value 预填 deferred 属性,既不触发逐行 lazy load, 也不将实例标记为 dirty(避免误产生 UPDATE)。 """ if not documents: return by_id = {document.id: document for document in documents} rows = db.session.execute( select(Document.id, Document.content_text).where(Document.id.in_(by_id)) ).all() for document_id, content_text in rows: set_committed_value(by_id[document_id], "content_text", content_text) def document_summary( document: Document, context: AuthContext, access: PlanAccess, keyword: str | None = None, memo: AccessMemo | None = None, ) -> dict[str, object]: # 无权限方案也在列表展示(需求:全员可见列表),但借阅名单属敏感信息, # borrowSummary 以占位符返回;正文片段同样只在有权限时生成。 if access.allowed: _, borrow_summary = _permission_summary(access, document, memo) else: borrow_summary = "-" view_locked = is_view_locked(document, context) result: dict[str, object] = { "id": serialize_id(document.id), "documentName": document.document_name, "summary": document.summary, "description": document.description if access.allowed else None, "documentType": document.document_type, "status": document.document_status, "securityLevel": document.security_level, "borrowSummary": borrow_summary, "attachmentType": document.attachment_type, "categoryId": serialize_id(document.category_id), "categoryName": document.category_name, "categoryPath": document.category_path, "organizationId": serialize_id(document.organization_id), "organizationName": document.organization_name, "parentDocumentId": serialize_id(document.parent_document_id), "rootDocumentId": serialize_id( document.id if document.document_type == DocumentType.MAIN.value else document.root_document_id ), "tags": document.tags or [], "fileExtension": document.file_extension, "childCount": document.child_count, "attachmentCount": document.attachment_count, "viewCount": document.view_count, "downloadCount": document.download_count, "createdByName": document.created_by_name, "createdAt": _iso(document.created_at), "updatedAt": _iso(document.updated_at), "rowVersion": document.row_version, "allowedActions": plan_allowed_actions(document, context, access), "hasPassword": document.password_hash is not None, "passwordProtectedActions": protected_actions_of(document), "viewLocked": view_locked, } # 列表查询 defer 了 content_text:仅在有正文关键词时才访问(此时已批量回填), # 无关键词时 _content_snippet 本就返回 None,守卫不改变契约且避免逐行 lazy load。 if not view_locked and keyword: snippet = _content_snippet(document.content_text, keyword) if snippet is not None: result["contentSnippet"] = snippet return result def document_detail( document: Document, context: AuthContext, access: PlanAccess, ) -> dict[str, object]: result = document_summary(document, context, access) permission_summary, _ = _permission_summary(access, document) verified = view_password_verified(document, context) result.update( { "originalFileName": document.original_file_name, "mimeType": document.mime_type, "fileSize": document.file_size, "fileHash": document.file_hash, "permissionSummary": permission_summary, "viewPasswordVerified": verified, "contentText": document.content_text if verified else None, } ) # 提取状态/失败原因仅供管理员排障("编辑"界面提示与重新解析) if context.role_code == RoleCode.ADMIN: result.update( { "contentExtractStatus": document.content_extract_status, "contentExtractError": document.content_extract_error, "contentExtractedAt": ( _iso(document.content_extracted_at) if document.content_extracted_at else None ), } ) return result def _require_plan_access(document: Document, context: AuthContext) -> PlanAccess: access = evaluate_plan_access(document, context) if access.allowed: return access if access.reason == "SECURITY": raise SecurityLevelForbiddenError() raise DocumentViewForbiddenError() def _escape_like(keyword: str) -> str: return keyword.strip().replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") def _title_keyword(statement, keyword: str | None): """简单关键词:仅模糊匹配标题(document_name)。""" if not keyword or not keyword.strip(): return statement pattern = f"%{_escape_like(keyword)}%" return statement.where(Document.document_name.like(pattern, escape="\\")) def _security_levels(params: Mapping[str, str]) -> list[str] | None: """解析 ``securityLevels`` 逗号分隔多选(勾选项内 OR)。""" raw = params.get("securityLevels") if raw is None or not raw.strip(): return None values = [item.strip() for item in raw.split(",") if item.strip()] if not values: return None for item in values: try: SecurityLevel(item) except ValueError as exc: raise InvalidArgumentError("securityLevels包含无效密级值") from exc if len(set(values)) != len(values): raise InvalidArgumentError("securityLevels不允许重复值") return values def _advanced_filters(statement, params: Mapping[str, str]): """高级检索三条件(正文关键词 / 密级多选 / 部门模糊),全部 AND 叠加。""" content_keyword = params.get("contentKeyword") if content_keyword and content_keyword.strip(): pattern = f"%{_escape_like(content_keyword)}%" # 正文关键词严格匹配提取正文:search_text 含名称/概述/标签兜底词, # OR 匹配会把正文解析失败(content_text 为空)但名称撞词的方案 # 错误召回,也会放大"仅元数据命中"的误报。 statement = statement.where( Document.content_text.like(pattern, escape="\\") ) levels = _security_levels(params) if levels: statement = statement.where(Document.security_level.in_(levels)) organization_keyword = params.get("organizationKeyword") if organization_keyword and organization_keyword.strip(): pattern = f"%{_escape_like(organization_keyword)}%" statement = statement.where( Document.organization_name.like(pattern, escape="\\") ) return statement def _title_contains(document_name: str | None, keyword: str) -> bool: return keyword.lower() in (document_name or "").lower() def _child_title_hit_counts( context: AuthContext, main_ids: list[int], keyword: str, memo: AccessMemo | None = None, ) -> dict[int, list[int]]: """统计各主案下标题命中简单关键词的可见子案/附件数量。 - 子案经 ``evaluate_plan_access`` 逐条校验,不可见的命中不计入; - 附件随挂载主案权限可见(主案本身已通过校验),直接计数; - 返回 ``{main_id: [子案命中数, 附件命中数]}``。 """ if not main_ids or not keyword.strip(): return {} pattern = f"%{_escape_like(keyword)}%" counts: dict[int, list[int]] = {main_id: [0, 0] for main_id in main_ids} children = db.session.scalars( select(Document) .where( Document.parent_document_id.in_(main_ids), Document.document_type == DocumentType.SUB_PLAN.value, Document.is_deleted.is_(False), Document.document_name.like(pattern, escape="\\"), ) .options(defer(Document.content_text), defer(Document.search_text)) ).all() for child in children: if evaluate_plan_access(child, context, memo).allowed: counts[child.parent_document_id][0] += 1 bindings = db.session.execute( select(AttachmentBinding.main_document_id, Document.id) .join(Document, Document.id == AttachmentBinding.attachment_document_id) .where( AttachmentBinding.main_document_id.in_(main_ids), AttachmentBinding.is_deleted.is_(False), Document.document_type == DocumentType.ATTACHMENT.value, Document.is_deleted.is_(False), Document.document_name.like(pattern, escape="\\"), ) ).all() for main_id, _ in bindings: counts[main_id][1] += 1 return counts def _tags_filter( statement, tags_value: str | None, match_value: str | None, ): """按 ``tags`` JSON 数组精确匹配;``tagsMatch=any|all`` 决定 OR/AND 语义。""" if not tags_value or not tags_value.strip(): return statement tags = [item.strip() for item in tags_value.split(",") if item.strip()] if not tags: return statement match = (match_value or "any").strip().lower() if match not in {"any", "all"}: raise InvalidArgumentError("tagsMatch只允许any或all") # 达梦无 JSON_CONTAINS:对 JSON 编码后的标签(含引号边界)做 LIKE 包含。 # ensure_ascii=False 保持原文中文,与达梦存量数据及 Java 版写入格式一致; # _escape_like 处理 %、_ 与反斜杠。 conditions = [ Document.tags.like( f"%{_escape_like(json.dumps(tag, ensure_ascii=False))}%", escape="\\" ) for tag in tags ] combiner = and_ if match == "all" else or_ return statement.where(combiner(*conditions)) def _boolean(value: str | None, name: str, default: bool = False) -> bool: if value is None: return default if value not in {"true", "false"}: raise InvalidArgumentError(f"{name}必须是true或false") return value == "true" def _category_scope(category_id: int, include_descendants: bool) -> set[int]: categories = db.session.scalars( select(Category).where( Category.is_deleted.is_(False), Category.status == "ENABLED", ) ).all() if not any(category.id == category_id for category in categories): raise ResourceNotFoundError("方案分类不存在或不可用") if not include_descendants: return {category_id} children: dict[int | None, list[int]] = {} for category in categories: children.setdefault(category.parent_id, []).append(category.id) result = {category_id} pending = [category_id] while pending: current = pending.pop() for child_id in children.get(current, []): if child_id not in result: result.add(child_id) pending.append(child_id) return result def list_documents(params: Mapping[str, str]) -> dict[str, object]: context = get_auth_context() page = _page(params) raw_types = params.get("documentType", "MAIN") document_types = [item.strip() for item in raw_types.split(",") if item.strip()] if ( not document_types or any(item not in PLAN_TYPES for item in document_types) or len(set(document_types)) != len(document_types) ): raise InvalidArgumentError("documentType只允许MAIN、SUB_PLAN或二者组合") statement = select(Document).options( # 列表主体查询不携带正文大字段;仅 contentKeyword 检索生成当前页 # 摘要时按页 ID 批量补查(_fill_content_text)。 defer(Document.content_text), defer(Document.search_text), ).where( Document.is_deleted.is_(False), Document.document_type.in_(document_types), ) include_descendants = _boolean( params.get("includeDescendants"), "includeDescendants" ) category_id = params.get("categoryId") if category_id: parsed_category_id = parse_string_id(category_id, field="categoryId") statement = statement.where( Document.category_id.in_( _category_scope(parsed_category_id, include_descendants) ) ) aggregate = _boolean(params.get("aggregateHits"), "aggregateHits") if aggregate and document_types != ["MAIN"]: raise InvalidArgumentError("aggregateHits只允许在documentType=MAIN时使用") if not aggregate: statement = _title_keyword(statement, params.get("keyword")) statement = _advanced_filters(statement, params) statement = _tags_filter( statement, params.get("tags"), params.get("tagsMatch") ) for name, enum_type, column in ( ("securityLevel", SecurityLevel, Document.security_level), ("status", DocumentStatus, Document.document_status), ): value = _enum(params.get(name), enum_type, name) if value: statement = statement.where(column == value) updated_from, updated_to = parse_updated_range(params) if updated_from: statement = statement.where(Document.updated_at >= updated_from) if updated_to: statement = statement.where(Document.updated_at <= updated_to) order, _ = _sort(params) documents = db.session.scalars(statement.order_by(order, Document.id.asc())).all() # 列表全量可见(需求3):无权限方案同样返回,allowedActions 为空集, # 前端据此禁用查看类按钮;预览/详情/下载接口仍独立拦截。 memo = AccessMemo() listed: list[tuple[Document, PlanAccess]] = [] for document in documents: access = evaluate_plan_access(document, context, memo) listed.append((document, access)) total = len(listed) keyword = (params.get("keyword") or "").strip() hit_counts: dict[int, list[int]] = {} if aggregate: if keyword: # 子案/附件命中统计只覆盖有权限主案,避免借计数泄露无权限主案的挂载信息。 hit_counts = _child_title_hit_counts( context, [document.id for document, access in listed if access.allowed], keyword, memo ) listed = [ (document, access) for document, access in listed if _title_contains(document.document_name, keyword) or any(hit_counts.get(document.id)) ] total = len(listed) selected = listed[page.offset : page.offset + page.page_size] content_keyword = params.get("contentKeyword") if content_keyword and content_keyword.strip(): # 仅当前页有权行需要正文摘要,按页 ID 批量补查(避免逐行 lazy load)。 _fill_content_text( [document for document, access in selected if access.allowed] ) items = [ document_summary( document, context, access, keyword=content_keyword if access.allowed else None, memo=memo, ) for document, access in selected ] if aggregate: for item, (document, access) in zip(items, selected, strict=True): counts = hit_counts.get(document.id, [0, 0]) if access.allowed else [0, 0] item["titleHit"] = not keyword or _title_contains( document.document_name, keyword ) item["subHitCount"] = counts[0] item["attachmentHitCount"] = counts[1] return page_result( items, page=page.page, page_size=page.page_size, total=total, ) def _active_plan(document_id: int) -> Document: document = db.session.scalar( select(Document).where( Document.id == document_id, Document.is_deleted.is_(False), Document.document_type.in_(PLAN_TYPES), ) ) if document is None: raise ResourceNotFoundError("方案文档不存在") return document def _record_view(document: Document) -> int: old_count = document.view_count document_id = document.id document_name = document.document_name target = ( AuditTarget.ATTACHMENT if document.document_type == DocumentType.ATTACHMENT.value else AuditTarget.DOCUMENT ) db.session.rollback() try: with transaction() as session: current = session.scalar( select(Document) .where( Document.id == document_id, Document.is_deleted.is_(False), ) .options(*LOCK_DEFER_LOB) .with_for_update() ) if current is None: raise ResourceNotFoundError("文档不存在") current.view_count += 1 session.add( business_audit( action=AuditAction.VIEW_DOCUMENT, target=target, target_id=document_id, target_name=document_name, detail={"documentType": current.document_type}, ) ) new_count = current.view_count return new_count except Exception: db.session.rollback() logger.exception("文档查看计数或审计写入失败:document_id=%s", document_id) return old_count def get_document(document_id: int) -> dict[str, object]: context = get_auth_context() document = _active_plan(document_id) access = _require_plan_access(document, context) result = document_detail(document, context, access) result["viewCount"] = _record_view(document) return result def list_sub_plans( main_document_id: int, params: Mapping[str, str], ) -> dict[str, object]: context = get_auth_context() main = db.session.scalar( select(Document).where( Document.id == main_document_id, Document.document_type == DocumentType.MAIN.value, Document.is_deleted.is_(False), ) ) if main is None: raise MainPlanNotFoundError() # 主案无权限时也返回子案列表(列表全量可见,与主案列表语义一致); # 子案权限继承自原主案(root_document_id 不变),allowedActions 自然为 # 空集,内容由预览/详情接口拦截。 memo = AccessMemo() evaluate_plan_access(main, context, memo) page = _page(params) def _apply_filters(statement): statement = _title_keyword(statement, params.get("keyword")) statement = _advanced_filters(statement, params) status = _enum(params.get("status"), DocumentStatus, "status") if status: statement = statement.where(Document.document_status == status) updated_from, updated_to = parse_updated_range(params) if updated_from: statement = statement.where(Document.updated_at >= updated_from) if updated_to: statement = statement.where(Document.updated_at <= updated_to) return statement order, _ = _sort(params) # 自有子案:parent_document_id 指向该主案,按请求排序(默认 sortNo)。 own_children = db.session.scalars( _apply_filters( select(Document).where( Document.parent_document_id == main.id, Document.document_type == DocumentType.SUB_PLAN.value, Document.is_deleted.is_(False), ) ) .order_by(order, Document.id.asc()) .options(defer(Document.content_text), defer(Document.search_text)) ).all() # 挂载子案:其他主案的子案经 doc_sub_plan_binding 建立展示关系, # 按挂载顺序(binding.sort_no)固定追加在自有子案之后,不参与拖动排序。 mount_rows = db.session.execute( select(SubPlanBinding, Document) .join(Document, Document.id == SubPlanBinding.sub_document_id) .where( SubPlanBinding.main_document_id == main.id, SubPlanBinding.is_deleted.is_(False), Document.is_deleted.is_(False), Document.document_type == DocumentType.SUB_PLAN.value, ) .order_by(SubPlanBinding.sort_no.asc(), SubPlanBinding.id.asc()) .options(defer(Document.content_text), defer(Document.search_text)) ).all() mounted: list[tuple[SubPlanBinding, Document]] = [] if mount_rows: filtered_ids = set( db.session.scalars( _apply_filters( select(Document.id).where( Document.id.in_([doc.id for _, doc in mount_rows]) ) ) ).all() ) mounted = [row for row in mount_rows if row[0].sub_document_id in filtered_ids] merged: list[tuple[Document, int | None]] = [ *[(child, None) for child in own_children], *[(doc, binding.id) for binding, doc in mounted], ] total = len(merged) selected = merged[page.offset : page.offset + page.page_size] # 逐条评估子案权限(memo 复用同根主案与授权列表缓存)。 accesses = [evaluate_plan_access(child, context, memo) for child, _ in selected] content_keyword = params.get("contentKeyword") if content_keyword and content_keyword.strip(): # 仅当前页有权行需要正文摘要,按页 ID 批量补查(避免逐行 lazy load)。 _fill_content_text( [child for (child, _), access in zip(selected, accesses) if access.allowed] ) items: list[dict[str, object]] = [] for (child, mount_id), access in zip(selected, accesses): summary = document_summary( child, context, access, keyword=content_keyword if access.allowed else None, memo=memo, ) if mount_id is not None: summary["mountId"] = serialize_id(mount_id) items.append(summary) return page_result( items, page=page.page, page_size=page.page_size, total=total, ) def list_main_plan_tags() -> dict[str, object]: """聚合当前用户可见的所有主案标签。 - 全局视角:标签集合不随分类、关键词等其他筛选变化; - 权限:通过 ``evaluate_plan_access`` 在 Python 层逐条过滤, 与 ``list_documents`` 保持一致的可见性语义,避免泄露被 ACL/密级隔离的主案标签; - 顺序:按 Unicode 升序,便于前端稳定渲染。 """ context = get_auth_context() statement = select(Document).options( defer(Document.content_text), defer(Document.search_text), ).where( Document.is_deleted.is_(False), Document.document_type == DocumentType.MAIN.value, ) documents = db.session.scalars(statement).all() tags_set: set[str] = set() memo = AccessMemo() for document in documents: access = evaluate_plan_access(document, context, memo) if not access.allowed: continue if document.tags: tags_set.update(document.tags) tags = sorted(tags_set) return {"items": tags, "total": len(tags)} __all__ = [ "DOCUMENT_SORTS", "LOCK_DEFER_LOB", "_advanced_filters", "_content_snippet", "_date", "_escape_like", "_fill_content_text", "_enum", "_iso", "_page", "_record_view", "_sort", "_tags_filter", "_title_keyword", "document_detail", "document_summary", "get_document", "list_documents", "list_main_plan_tags", "list_sub_plans", "parse_string_id", ]