document_query_service.py 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546
  1. """主案、子方案只读查询、DTO和查看审计。"""
  2. from __future__ import annotations
  3. import json
  4. import logging
  5. from datetime import datetime, timezone
  6. from typing import Any, Mapping
  7. from sqlalchemy import String, and_, cast, func, or_, select
  8. from dms.common.enums import (
  9. AuditAction,
  10. AuditTarget,
  11. DocumentStatus,
  12. DocumentType,
  13. RoleCode,
  14. SecurityLevel,
  15. VisibilityType,
  16. )
  17. from dms.common.errors import (
  18. DocumentViewForbiddenError,
  19. InvalidArgumentError,
  20. MainPlanNotFoundError,
  21. ResourceNotFoundError,
  22. SecurityLevelForbiddenError,
  23. )
  24. from dms.common.pagination import PageRequest, page_result
  25. from dms.common.response import serialize_id
  26. from dms.common.time_range import parse_datetime, parse_updated_range
  27. from dms.database.transaction import transaction
  28. from dms.extensions import db
  29. from dms.models import Category, Document, Permission
  30. from dms.security.auth_context import AuthContext, get_auth_context
  31. from dms.services.audit_service import business_audit
  32. from dms.services.authorization_service import (
  33. PlanAccess,
  34. evaluate_plan_access,
  35. plan_allowed_actions,
  36. )
  37. logger = logging.getLogger(__name__)
  38. PLAN_TYPES = (DocumentType.MAIN.value, DocumentType.SUB_PLAN.value)
  39. DOCUMENT_SORTS = {
  40. "documentName": Document.document_name,
  41. "createdAt": Document.created_at,
  42. "updatedAt": Document.updated_at,
  43. "viewCount": Document.view_count,
  44. "downloadCount": Document.download_count,
  45. }
  46. def parse_string_id(value: str, *, field: str = "id") -> int:
  47. if not value.isdecimal() or int(value) <= 0:
  48. raise InvalidArgumentError(f"{field}必须是正整数形式的字符串ID")
  49. return int(value)
  50. def _integer(
  51. params: Mapping[str, str],
  52. name: str,
  53. default: int,
  54. ) -> int:
  55. raw = params.get(name)
  56. if raw is None:
  57. return default
  58. if not raw.isdecimal():
  59. raise InvalidArgumentError(f"{name}必须是正整数")
  60. return int(raw)
  61. def _page(params: Mapping[str, str]) -> PageRequest:
  62. return PageRequest(
  63. page=_integer(params, "page", 1),
  64. page_size=_integer(params, "pageSize", 20),
  65. )
  66. def _enum(value: str | None, enum_type, name: str) -> str | None:
  67. if value is None or value == "":
  68. return None
  69. try:
  70. return enum_type(value).value
  71. except ValueError as exc:
  72. raise InvalidArgumentError(f"{name}不是有效枚举值") from exc
  73. def _date(value: str | None, name: str) -> datetime | None:
  74. return parse_datetime(value, name=name)
  75. def _sort(
  76. params: Mapping[str, str],
  77. *,
  78. allowed: dict[str, Any] = DOCUMENT_SORTS,
  79. default: str = "updatedAt",
  80. ) -> tuple[Any, str]:
  81. sort_by = params.get("sortBy", default)
  82. direction = params.get("sortDirection", "desc").lower()
  83. if sort_by not in allowed:
  84. raise InvalidArgumentError("sortBy不是允许的排序字段")
  85. if direction not in {"asc", "desc"}:
  86. raise InvalidArgumentError("sortDirection必须是asc或desc")
  87. column = allowed[sort_by]
  88. return (column.asc() if direction == "asc" else column.desc()), direction
  89. def _iso(value: datetime) -> str:
  90. return value.replace(tzinfo=timezone.utc).isoformat(timespec="milliseconds").replace(
  91. "+00:00", "Z"
  92. )
  93. def _permission_summary(
  94. access: PlanAccess,
  95. document: Document,
  96. ) -> tuple[dict[str, object], str]:
  97. source = access.source
  98. if source is None:
  99. return {
  100. "organizationCount": 0,
  101. "userCount": 0,
  102. "inheritedFromMainPlan": False,
  103. }, "未配置"
  104. permissions = db.session.scalars(
  105. select(Permission).where(
  106. Permission.document_id == source.id,
  107. Permission.is_deleted.is_(False),
  108. )
  109. ).all()
  110. organization_count = sum(item.subject_type == "ORG" for item in permissions)
  111. user_count = sum(item.subject_type == "USER" for item in permissions)
  112. visibility = VisibilityType(source.visibility_type)
  113. if visibility == VisibilityType.ALL_AUTHENTICATED:
  114. visibility_summary = "全部已登录用户"
  115. else:
  116. names = [item.subject_name for item in permissions if item.can_view]
  117. visibility_summary = "、".join(names) if names else "未配置"
  118. return {
  119. "organizationCount": organization_count,
  120. "userCount": user_count,
  121. "inheritedFromMainPlan": document.document_type
  122. == DocumentType.SUB_PLAN.value,
  123. }, visibility_summary
  124. _SNIPPET_BEFORE = 40
  125. _SNIPPET_AFTER = 80
  126. def _content_snippet(content_text: str | None, keyword: str | None) -> str | None:
  127. """当 keyword 命中正文时,截取包含命中区域的片段。
  128. 前后各保留若干字符作为上下文;若未触达正文首尾则以"……"标识截断。
  129. 未命中或任一参数为空时返回 None。
  130. """
  131. if not content_text or not keyword or not keyword.strip():
  132. return None
  133. kw = keyword.strip()
  134. pos = content_text.lower().find(kw.lower())
  135. if pos < 0:
  136. return None
  137. start = max(pos - _SNIPPET_BEFORE, 0)
  138. end = min(pos + len(kw) + _SNIPPET_AFTER, len(content_text))
  139. snippet = content_text[start:end]
  140. if start > 0:
  141. snippet = "……" + snippet
  142. if end < len(content_text):
  143. snippet += "……"
  144. return snippet
  145. def document_summary(
  146. document: Document,
  147. context: AuthContext,
  148. access: PlanAccess,
  149. keyword: str | None = None,
  150. ) -> dict[str, object]:
  151. _, visibility_summary = _permission_summary(access, document)
  152. source = access.source or document
  153. result: dict[str, object] = {
  154. "id": serialize_id(document.id),
  155. "documentName": document.document_name,
  156. "summary": document.summary,
  157. "documentType": document.document_type,
  158. "status": document.document_status,
  159. "securityLevel": document.security_level,
  160. "visibilityType": source.visibility_type,
  161. "visibilitySummary": visibility_summary,
  162. "attachmentType": document.attachment_type,
  163. "categoryId": serialize_id(document.category_id),
  164. "categoryName": document.category_name,
  165. "categoryPath": document.category_path,
  166. "parentDocumentId": serialize_id(document.parent_document_id),
  167. "rootDocumentId": serialize_id(
  168. document.id
  169. if document.document_type == DocumentType.MAIN.value
  170. else document.root_document_id
  171. ),
  172. "tags": document.tags or [],
  173. "fileExtension": document.file_extension,
  174. "childCount": document.child_count,
  175. "attachmentCount": document.attachment_count,
  176. "viewCount": document.view_count,
  177. "downloadCount": document.download_count,
  178. "createdByName": document.created_by_name,
  179. "createdAt": _iso(document.created_at),
  180. "updatedAt": _iso(document.updated_at),
  181. "rowVersion": document.row_version,
  182. "allowedActions": plan_allowed_actions(document, context, access),
  183. }
  184. snippet = _content_snippet(document.content_text, keyword)
  185. if snippet is not None:
  186. result["contentSnippet"] = snippet
  187. return result
  188. def document_detail(
  189. document: Document,
  190. context: AuthContext,
  191. access: PlanAccess,
  192. ) -> dict[str, object]:
  193. result = document_summary(document, context, access)
  194. permission_summary, _ = _permission_summary(access, document)
  195. result.update(
  196. {
  197. "originalFileName": document.original_file_name,
  198. "mimeType": document.mime_type,
  199. "fileSize": document.file_size,
  200. "fileHash": document.file_hash,
  201. "permissionSummary": permission_summary,
  202. "contentText": document.content_text,
  203. }
  204. )
  205. return result
  206. def _require_plan_access(document: Document, context: AuthContext) -> PlanAccess:
  207. access = evaluate_plan_access(document, context)
  208. if access.allowed:
  209. return access
  210. if access.reason == "SECURITY":
  211. raise SecurityLevelForbiddenError()
  212. raise DocumentViewForbiddenError()
  213. def _keyword(statement, keyword: str | None):
  214. if not keyword or not keyword.strip():
  215. return statement
  216. escaped = (
  217. keyword.strip().replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
  218. )
  219. pattern = f"%{escaped}%"
  220. return statement.where(
  221. or_(
  222. Document.document_name.like(pattern, escape="\\"),
  223. Document.summary.like(pattern, escape="\\"),
  224. Document.search_text.like(pattern, escape="\\"),
  225. cast(Document.tags, String).like(pattern, escape="\\"),
  226. )
  227. )
  228. def _tags_filter(
  229. statement,
  230. tags_value: str | None,
  231. match_value: str | None,
  232. ):
  233. """按 ``tags`` JSON 数组精确匹配;``tagsMatch=any|all`` 决定 OR/AND 语义。"""
  234. if not tags_value or not tags_value.strip():
  235. return statement
  236. tags = [item.strip() for item in tags_value.split(",") if item.strip()]
  237. if not tags:
  238. return statement
  239. match = (match_value or "any").strip().lower()
  240. if match not in {"any", "all"}:
  241. raise InvalidArgumentError("tagsMatch只允许any或all")
  242. # 使用 JSON_CONTAINS 做数组元素的精确匹配;json.dumps 保证特殊字符安全转义。
  243. conditions = [func.json_contains(Document.tags, json.dumps(tag)) for tag in tags]
  244. combiner = and_ if match == "all" else or_
  245. return statement.where(combiner(*conditions))
  246. def _boolean(value: str | None, name: str, default: bool = False) -> bool:
  247. if value is None:
  248. return default
  249. if value not in {"true", "false"}:
  250. raise InvalidArgumentError(f"{name}必须是true或false")
  251. return value == "true"
  252. def _category_scope(category_id: int, include_descendants: bool) -> set[int]:
  253. categories = db.session.scalars(
  254. select(Category).where(
  255. Category.is_deleted.is_(False),
  256. Category.status == "ENABLED",
  257. )
  258. ).all()
  259. if not any(category.id == category_id for category in categories):
  260. raise ResourceNotFoundError("方案分类不存在或不可用")
  261. if not include_descendants:
  262. return {category_id}
  263. children: dict[int | None, list[int]] = {}
  264. for category in categories:
  265. children.setdefault(category.parent_id, []).append(category.id)
  266. result = {category_id}
  267. pending = [category_id]
  268. while pending:
  269. current = pending.pop()
  270. for child_id in children.get(current, []):
  271. if child_id not in result:
  272. result.add(child_id)
  273. pending.append(child_id)
  274. return result
  275. def list_documents(params: Mapping[str, str]) -> dict[str, object]:
  276. context = get_auth_context()
  277. page = _page(params)
  278. raw_types = params.get("documentType", "MAIN")
  279. document_types = [item.strip() for item in raw_types.split(",") if item.strip()]
  280. if (
  281. not document_types
  282. or any(item not in PLAN_TYPES for item in document_types)
  283. or len(set(document_types)) != len(document_types)
  284. ):
  285. raise InvalidArgumentError("documentType只允许MAIN、SUB_PLAN或二者组合")
  286. statement = select(Document).where(
  287. Document.is_deleted.is_(False),
  288. Document.document_type.in_(document_types),
  289. )
  290. include_descendants = _boolean(
  291. params.get("includeDescendants"), "includeDescendants"
  292. )
  293. category_id = params.get("categoryId")
  294. if category_id:
  295. parsed_category_id = parse_string_id(category_id, field="categoryId")
  296. statement = statement.where(
  297. Document.category_id.in_(
  298. _category_scope(parsed_category_id, include_descendants)
  299. )
  300. )
  301. statement = _keyword(statement, params.get("keyword"))
  302. statement = _tags_filter(
  303. statement, params.get("tags"), params.get("tagsMatch")
  304. )
  305. visibility_filter = _enum(
  306. params.get("visibilityType"), VisibilityType, "visibilityType"
  307. )
  308. for name, enum_type, column in (
  309. ("securityLevel", SecurityLevel, Document.security_level),
  310. ("status", DocumentStatus, Document.document_status),
  311. ):
  312. value = _enum(params.get(name), enum_type, name)
  313. if value:
  314. statement = statement.where(column == value)
  315. updated_from, updated_to = parse_updated_range(params)
  316. if updated_from:
  317. statement = statement.where(Document.updated_at >= updated_from)
  318. if updated_to:
  319. statement = statement.where(Document.updated_at <= updated_to)
  320. order, _ = _sort(params)
  321. documents = db.session.scalars(statement.order_by(order, Document.id.asc())).all()
  322. visible: list[tuple[Document, PlanAccess]] = []
  323. for document in documents:
  324. access = evaluate_plan_access(document, context)
  325. if (
  326. access.allowed
  327. and (
  328. visibility_filter is None
  329. or (
  330. access.source is not None
  331. and access.source.visibility_type == visibility_filter
  332. )
  333. )
  334. ):
  335. visible.append((document, access))
  336. total = len(visible)
  337. selected = visible[page.offset : page.offset + page.page_size]
  338. return page_result(
  339. [
  340. document_summary(document, context, access, keyword=params.get("keyword"))
  341. for document, access in selected
  342. ],
  343. page=page.page,
  344. page_size=page.page_size,
  345. total=total,
  346. )
  347. def _active_plan(document_id: int) -> Document:
  348. document = db.session.scalar(
  349. select(Document).where(
  350. Document.id == document_id,
  351. Document.is_deleted.is_(False),
  352. Document.document_type.in_(PLAN_TYPES),
  353. )
  354. )
  355. if document is None:
  356. raise ResourceNotFoundError("方案文档不存在")
  357. return document
  358. def _record_view(document: Document) -> int:
  359. old_count = document.view_count
  360. document_id = document.id
  361. document_name = document.document_name
  362. target = (
  363. AuditTarget.ATTACHMENT
  364. if document.document_type == DocumentType.ATTACHMENT.value
  365. else AuditTarget.DOCUMENT
  366. )
  367. db.session.rollback()
  368. try:
  369. with transaction() as session:
  370. current = session.scalar(
  371. select(Document)
  372. .where(
  373. Document.id == document_id,
  374. Document.is_deleted.is_(False),
  375. )
  376. .with_for_update()
  377. )
  378. if current is None:
  379. raise ResourceNotFoundError("文档不存在")
  380. current.view_count += 1
  381. session.add(
  382. business_audit(
  383. action=AuditAction.VIEW_DOCUMENT,
  384. target=target,
  385. target_id=document_id,
  386. target_name=document_name,
  387. detail={"documentType": current.document_type},
  388. )
  389. )
  390. new_count = current.view_count
  391. return new_count
  392. except Exception:
  393. db.session.rollback()
  394. logger.exception("文档查看计数或审计写入失败:document_id=%s", document_id)
  395. return old_count
  396. def get_document(document_id: int) -> dict[str, object]:
  397. context = get_auth_context()
  398. document = _active_plan(document_id)
  399. access = _require_plan_access(document, context)
  400. result = document_detail(document, context, access)
  401. result["viewCount"] = _record_view(document)
  402. return result
  403. def list_sub_plans(
  404. main_document_id: int,
  405. params: Mapping[str, str],
  406. ) -> dict[str, object]:
  407. context = get_auth_context()
  408. main = db.session.scalar(
  409. select(Document).where(
  410. Document.id == main_document_id,
  411. Document.document_type == DocumentType.MAIN.value,
  412. Document.is_deleted.is_(False),
  413. )
  414. )
  415. if main is None:
  416. raise MainPlanNotFoundError()
  417. _require_plan_access(main, context)
  418. page = _page(params)
  419. statement = select(Document).where(
  420. Document.parent_document_id == main.id,
  421. Document.document_type == DocumentType.SUB_PLAN.value,
  422. Document.is_deleted.is_(False),
  423. )
  424. statement = _keyword(statement, params.get("keyword"))
  425. status = _enum(params.get("status"), DocumentStatus, "status")
  426. if status:
  427. statement = statement.where(Document.document_status == status)
  428. updated_from, updated_to = parse_updated_range(params)
  429. if updated_from:
  430. statement = statement.where(Document.updated_at >= updated_from)
  431. if updated_to:
  432. statement = statement.where(Document.updated_at <= updated_to)
  433. order, _ = _sort(params)
  434. children = db.session.scalars(
  435. statement.order_by(order, Document.id.asc())
  436. ).all()
  437. visible: list[tuple[Document, PlanAccess]] = []
  438. for child in children:
  439. access = evaluate_plan_access(child, context)
  440. if access.allowed:
  441. visible.append((child, access))
  442. total = len(visible)
  443. selected = visible[page.offset : page.offset + page.page_size]
  444. return page_result(
  445. [document_summary(child, context, access, keyword=params.get("keyword")) for child, access in selected],
  446. page=page.page,
  447. page_size=page.page_size,
  448. total=total,
  449. )
  450. def list_main_plan_tags() -> dict[str, object]:
  451. """聚合当前用户可见的所有主案标签。
  452. - 全局视角:标签集合不随分类、关键词等其他筛选变化;
  453. - 权限:通过 ``evaluate_plan_access`` 在 Python 层逐条过滤,
  454. 与 ``list_documents`` 保持一致的可见性语义,避免泄露被
  455. ACL/密级隔离的主案标签;
  456. - 顺序:按 Unicode 升序,便于前端稳定渲染。
  457. """
  458. context = get_auth_context()
  459. statement = select(Document).where(
  460. Document.is_deleted.is_(False),
  461. Document.document_type == DocumentType.MAIN.value,
  462. )
  463. documents = db.session.scalars(statement).all()
  464. tags_set: set[str] = set()
  465. for document in documents:
  466. access = evaluate_plan_access(document, context)
  467. if not access.allowed:
  468. continue
  469. if document.tags:
  470. tags_set.update(document.tags)
  471. tags = sorted(tags_set)
  472. return {"items": tags, "total": len(tags)}
  473. __all__ = [
  474. "DOCUMENT_SORTS",
  475. "_date",
  476. "_enum",
  477. "_iso",
  478. "_keyword",
  479. "_page",
  480. "_record_view",
  481. "_sort",
  482. "_tags_filter",
  483. "document_detail",
  484. "document_summary",
  485. "get_document",
  486. "list_documents",
  487. "list_main_plan_tags",
  488. "list_sub_plans",
  489. "parse_string_id",
  490. ]