category_service.py 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533
  1. """方案分类树、写操作、路径冗余和占用规则。"""
  2. from __future__ import annotations
  3. from collections import deque
  4. from datetime import datetime, timezone
  5. from typing import Any
  6. from sqlalchemy import func, select, update
  7. from sqlalchemy.exc import IntegrityError
  8. from dms.common.enums import (
  9. AuditAction,
  10. AuditTarget,
  11. CategoryType,
  12. DocumentType,
  13. EnabledStatus,
  14. )
  15. from dms.common.errors import (
  16. CategoryInUseError,
  17. ConflictError,
  18. InvalidArgumentError,
  19. ResourceNotFoundError,
  20. )
  21. from dms.common.response import serialize_id
  22. from dms.database.transaction import transaction
  23. from dms.extensions import db
  24. from dms.models import Category, Document
  25. from dms.security.auth_context import get_auth_context
  26. from dms.services.audit_service import business_audit
  27. CREATE_FIELDS = {
  28. "categoryCode",
  29. "categoryName",
  30. "categoryType",
  31. "parentId",
  32. "sortNo",
  33. }
  34. UPDATE_FIELDS = {
  35. "categoryName",
  36. "categoryType",
  37. "parentId",
  38. "sortNo",
  39. "rowVersion",
  40. }
  41. def _utc_now() -> datetime:
  42. return datetime.now(timezone.utc).replace(tzinfo=None)
  43. def _string_id(value: Any, *, field: str, nullable: bool = False) -> int | None:
  44. if value is None and nullable:
  45. return None
  46. if not isinstance(value, str) or not value.isdecimal() or int(value) <= 0:
  47. raise InvalidArgumentError(f"{field}必须是正整数形式的字符串ID")
  48. return int(value)
  49. def _required_text(
  50. value: Any,
  51. *,
  52. field: str,
  53. max_length: int,
  54. ) -> str:
  55. if not isinstance(value, str) or not value.strip():
  56. raise InvalidArgumentError(f"{field}不能为空")
  57. normalized = value.strip()
  58. if len(normalized) > max_length:
  59. raise InvalidArgumentError(f"{field}长度不能超过{max_length}")
  60. return normalized
  61. def _category_type(value: Any) -> str:
  62. if not isinstance(value, str):
  63. raise InvalidArgumentError("categoryType必须是字符串")
  64. try:
  65. return CategoryType(value).value
  66. except ValueError as exc:
  67. raise InvalidArgumentError("categoryType不是有效枚举值") from exc
  68. def _sort_no(value: Any) -> int:
  69. if type(value) is not int:
  70. raise InvalidArgumentError("sortNo必须是整数")
  71. return value
  72. def _row_version(value: Any) -> int:
  73. if type(value) is not int or value < 0:
  74. raise InvalidArgumentError("rowVersion必须是非负整数")
  75. return value
  76. def _exact_payload(payload: Any, expected: set[str]) -> dict[str, Any]:
  77. if not isinstance(payload, dict) or set(payload) != expected:
  78. raise InvalidArgumentError(
  79. "请求体字段必须且只能包含:" + "、".join(sorted(expected))
  80. )
  81. return payload
  82. def _path(parent: Category | None, category_name: str) -> str:
  83. value = (
  84. f"{parent.category_path.rstrip('/')}/{category_name}"
  85. if parent is not None
  86. else f"/{category_name}"
  87. )
  88. if len(value) > 1000:
  89. raise InvalidArgumentError("categoryPath长度超过数据库限制")
  90. return value
  91. def _dto(
  92. category: Category,
  93. *,
  94. parent_id: int | None = None,
  95. document_count: int | None = None,
  96. children: list[dict[str, object]] | None = None,
  97. ) -> dict[str, object]:
  98. return {
  99. "id": serialize_id(category.id),
  100. "categoryCode": category.category_code,
  101. "categoryName": category.category_name,
  102. "categoryType": category.category_type,
  103. "parentId": serialize_id(
  104. category.parent_id if parent_id is None else parent_id
  105. ),
  106. "categoryPath": category.category_path,
  107. "sortNo": category.sort_no,
  108. "documentCount": (
  109. category.document_count
  110. if document_count is None
  111. else document_count
  112. ),
  113. "rowVersion": category.row_version,
  114. "children": children or [],
  115. }
  116. def _children_map(categories: list[Category]) -> dict[int | None, list[Category]]:
  117. children: dict[int | None, list[Category]] = {}
  118. for category in categories:
  119. children.setdefault(category.parent_id, []).append(category)
  120. return children
  121. def _descendants(
  122. category_id: int,
  123. children: dict[int | None, list[Category]],
  124. ) -> list[Category]:
  125. result: list[Category] = []
  126. pending = deque(children.get(category_id, []))
  127. seen = {category_id}
  128. while pending:
  129. category = pending.popleft()
  130. if category.id in seen:
  131. raise ConflictError("分类树存在循环关系")
  132. seen.add(category.id)
  133. result.append(category)
  134. pending.extend(children.get(category.id, []))
  135. return result
  136. def category_tree(
  137. *,
  138. keyword: str | None,
  139. status: str | None,
  140. ) -> list[dict[str, object]]:
  141. if status:
  142. try:
  143. normalized_status = EnabledStatus(status).value
  144. except ValueError as exc:
  145. raise InvalidArgumentError("status必须是ENABLED或DISABLED") from exc
  146. else:
  147. normalized_status = EnabledStatus.ENABLED.value
  148. categories = db.session.scalars(
  149. select(Category)
  150. .where(Category.is_deleted.is_(False))
  151. .order_by(Category.sort_no.asc(), Category.id.asc())
  152. ).all()
  153. main_document_counts = dict(
  154. db.session.execute(
  155. select(Document.category_id, func.count(Document.id))
  156. .where(
  157. Document.category_id.is_not(None),
  158. Document.document_type == DocumentType.MAIN.value,
  159. Document.is_deleted.is_(False),
  160. )
  161. .group_by(Document.category_id)
  162. ).all()
  163. )
  164. by_id = {item.id: item for item in categories}
  165. children = _children_map(categories)
  166. eligible_ids = {
  167. item.id for item in categories if item.status == normalized_status
  168. }
  169. if keyword and keyword.strip():
  170. needle = keyword.strip().casefold()
  171. matched_ids = {
  172. item.id
  173. for item in categories
  174. if item.id in eligible_ids
  175. and (
  176. needle in item.category_name.casefold()
  177. or needle in item.category_code.casefold()
  178. )
  179. }
  180. included_ids = set(matched_ids)
  181. for matched_id in tuple(matched_ids):
  182. included_ids.update(
  183. descendant.id
  184. for descendant in _descendants(matched_id, children)
  185. if descendant.id in eligible_ids
  186. )
  187. parent_id = by_id[matched_id].parent_id
  188. while parent_id is not None and parent_id in by_id:
  189. included_ids.add(parent_id)
  190. parent_id = by_id[parent_id].parent_id
  191. else:
  192. included_ids = eligible_ids
  193. nodes: dict[int, dict[str, object]] = {}
  194. for category in categories:
  195. if category.id in included_ids:
  196. response_parent_id = (
  197. category.parent_id if category.parent_id in included_ids else None
  198. )
  199. nodes[category.id] = _dto(
  200. category,
  201. document_count=main_document_counts.get(category.id, 0),
  202. )
  203. nodes[category.id]["parentId"] = serialize_id(response_parent_id)
  204. roots: list[dict[str, object]] = []
  205. for category in categories:
  206. node = nodes.get(category.id)
  207. if node is None:
  208. continue
  209. if category.parent_id in nodes:
  210. nodes[category.parent_id]["children"].append(node)
  211. else:
  212. roots.append(node)
  213. return roots
  214. def _valid_parent(session, parent_id: int | None) -> Category | None:
  215. if parent_id is None:
  216. return None
  217. parent = session.scalar(
  218. select(Category)
  219. .where(
  220. Category.id == parent_id,
  221. Category.is_deleted.is_(False),
  222. Category.status == EnabledStatus.ENABLED.value,
  223. )
  224. .with_for_update()
  225. )
  226. if parent is None:
  227. raise InvalidArgumentError("parentId必须指向有效且已启用的分类")
  228. return parent
  229. def create_category(payload: Any) -> dict[str, object]:
  230. body = _exact_payload(payload, CREATE_FIELDS)
  231. category_code = _required_text(
  232. body["categoryCode"],
  233. field="categoryCode",
  234. max_length=64,
  235. )
  236. category_name = _required_text(
  237. body["categoryName"],
  238. field="categoryName",
  239. max_length=128,
  240. )
  241. category_type = _category_type(body["categoryType"])
  242. parent_id = _string_id(body["parentId"], field="parentId", nullable=True)
  243. sort_no = _sort_no(body["sortNo"])
  244. actor_id = get_auth_context().user_id
  245. try:
  246. with transaction() as session:
  247. if session.scalar(
  248. select(Category.id).where(Category.category_code == category_code)
  249. ) is not None:
  250. raise ConflictError("分类编码已存在")
  251. parent = _valid_parent(session, parent_id)
  252. category = Category(
  253. category_code=category_code,
  254. category_name=category_name,
  255. category_type=category_type,
  256. parent_id=parent_id,
  257. category_path=_path(parent, category_name),
  258. sort_no=sort_no,
  259. document_count=0,
  260. status=EnabledStatus.ENABLED.value,
  261. created_by=actor_id,
  262. updated_by=actor_id,
  263. )
  264. session.add(category)
  265. session.flush()
  266. session.add(
  267. business_audit(
  268. action=AuditAction.CREATE_CATEGORY,
  269. target=AuditTarget.CATEGORY,
  270. target_id=category.id,
  271. target_name=category.category_name,
  272. detail={
  273. "categoryCode": category.category_code,
  274. "categoryType": category.category_type,
  275. "parentId": serialize_id(category.parent_id),
  276. "categoryPath": category.category_path,
  277. "sortNo": category.sort_no,
  278. },
  279. )
  280. )
  281. return _dto(category)
  282. except IntegrityError as exc:
  283. raise ConflictError("分类编码已存在") from exc
  284. def _locked_categories(session) -> list[Category]:
  285. return session.scalars(
  286. select(Category)
  287. .where(Category.is_deleted.is_(False))
  288. .order_by(Category.id.asc())
  289. .with_for_update()
  290. ).all()
  291. def update_category(category_id: int, payload: Any) -> dict[str, object]:
  292. body = _exact_payload(payload, UPDATE_FIELDS)
  293. category_name = _required_text(
  294. body["categoryName"],
  295. field="categoryName",
  296. max_length=128,
  297. )
  298. category_type = _category_type(body["categoryType"])
  299. parent_id = _string_id(body["parentId"], field="parentId", nullable=True)
  300. sort_no = _sort_no(body["sortNo"])
  301. expected_version = _row_version(body["rowVersion"])
  302. actor_id = get_auth_context().user_id
  303. now = _utc_now()
  304. with transaction() as session:
  305. categories = _locked_categories(session)
  306. by_id = {item.id: item for item in categories}
  307. category = by_id.get(category_id)
  308. if category is None:
  309. raise ResourceNotFoundError("分类不存在")
  310. if category.row_version != expected_version:
  311. raise ConflictError(
  312. "数据已被其他用户修改,请刷新后重试",
  313. details={"currentRowVersion": category.row_version},
  314. )
  315. children = _children_map(categories)
  316. descendants = _descendants(category.id, children)
  317. descendant_ids = {item.id for item in descendants}
  318. if parent_id == category.id:
  319. raise InvalidArgumentError("parentId不能等于当前分类ID")
  320. if parent_id in descendant_ids:
  321. raise InvalidArgumentError("分类不能移动到自己的后代节点下")
  322. parent = None
  323. if parent_id is not None:
  324. parent = by_id.get(parent_id)
  325. if (
  326. parent is None
  327. or parent.is_deleted
  328. or parent.status != EnabledStatus.ENABLED.value
  329. ):
  330. raise InvalidArgumentError("parentId必须指向有效且已启用的分类")
  331. before = {
  332. "categoryName": category.category_name,
  333. "categoryType": category.category_type,
  334. "parentId": serialize_id(category.parent_id),
  335. "categoryPath": category.category_path,
  336. "sortNo": category.sort_no,
  337. "rowVersion": category.row_version,
  338. }
  339. path_changed = (
  340. category_name != category.category_name
  341. or parent_id != category.parent_id
  342. )
  343. category.category_name = category_name
  344. category.category_type = category_type
  345. category.parent_id = parent_id
  346. category.category_path = _path(parent, category_name)
  347. category.sort_no = sort_no
  348. category.updated_by = actor_id
  349. category.updated_at = now
  350. category.row_version += 1
  351. affected_categories = [category]
  352. if path_changed:
  353. pending = deque([category])
  354. while pending:
  355. current = pending.popleft()
  356. for child in children.get(current.id, []):
  357. child.category_path = _path(current, child.category_name)
  358. child.updated_by = actor_id
  359. child.updated_at = now
  360. child.row_version += 1
  361. affected_categories.append(child)
  362. pending.append(child)
  363. affected_document_count = 0
  364. if path_changed:
  365. for affected in affected_categories:
  366. result = session.execute(
  367. update(Document)
  368. .where(
  369. Document.category_id == affected.id,
  370. Document.is_deleted.is_(False),
  371. Document.document_type.in_(
  372. [
  373. DocumentType.MAIN.value,
  374. DocumentType.SUB_PLAN.value,
  375. ]
  376. ),
  377. )
  378. .values(
  379. category_name=affected.category_name,
  380. category_path=affected.category_path,
  381. updated_by=actor_id,
  382. updated_at=now,
  383. row_version=Document.row_version + 1,
  384. )
  385. )
  386. affected_document_count += result.rowcount
  387. after = {
  388. "categoryName": category.category_name,
  389. "categoryType": category.category_type,
  390. "parentId": serialize_id(category.parent_id),
  391. "categoryPath": category.category_path,
  392. "sortNo": category.sort_no,
  393. "rowVersion": category.row_version,
  394. }
  395. session.add(
  396. business_audit(
  397. action=AuditAction.EDIT_CATEGORY,
  398. target=AuditTarget.CATEGORY,
  399. target_id=category.id,
  400. target_name=category.category_name,
  401. detail={
  402. "before": before,
  403. "after": after,
  404. "updatedDescendantCount": len(affected_categories) - 1,
  405. "updatedDocumentCount": affected_document_count,
  406. },
  407. )
  408. )
  409. return _dto(category)
  410. def delete_category(category_id: int, row_version: int) -> dict[str, object]:
  411. actor_id = get_auth_context().user_id
  412. now = _utc_now()
  413. with transaction() as session:
  414. categories = _locked_categories(session)
  415. by_id = {item.id: item for item in categories}
  416. category = by_id.get(category_id)
  417. if category is None:
  418. raise ResourceNotFoundError("分类不存在")
  419. if category.row_version != row_version:
  420. raise ConflictError(
  421. "数据已被其他用户修改,请刷新后重试",
  422. details={"currentRowVersion": category.row_version},
  423. )
  424. descendants = _descendants(category.id, _children_map(categories))
  425. child_count = len(descendants)
  426. document_count = (
  427. session.scalar(
  428. select(func.count(Document.id)).where(
  429. Document.category_id == category.id,
  430. Document.is_deleted.is_(False),
  431. Document.document_type.in_(
  432. [
  433. DocumentType.MAIN.value,
  434. DocumentType.SUB_PLAN.value,
  435. ]
  436. ),
  437. )
  438. )
  439. or 0
  440. )
  441. if child_count or document_count:
  442. raise CategoryInUseError(
  443. details={
  444. "categoryId": serialize_id(category.id),
  445. "childCategoryCount": child_count,
  446. "documentCount": document_count,
  447. }
  448. )
  449. target_name = category.category_name
  450. category.is_deleted = True
  451. category.deleted_at = now
  452. category.updated_at = now
  453. category.updated_by = actor_id
  454. category.row_version += 1
  455. session.add(
  456. business_audit(
  457. action=AuditAction.DELETE_CATEGORY,
  458. target=AuditTarget.CATEGORY,
  459. target_id=category.id,
  460. target_name=target_name,
  461. detail={
  462. "categoryCode": category.category_code,
  463. "categoryPath": category.category_path,
  464. "rowVersion": category.row_version,
  465. },
  466. )
  467. )
  468. return {"id": serialize_id(category.id), "deleted": True}
  469. def parse_category_id(value: str) -> int:
  470. result = _string_id(value, field="id")
  471. assert result is not None
  472. return result
  473. def parse_delete_row_version(value: str | None) -> int:
  474. if value is None or not value.isdecimal():
  475. raise InvalidArgumentError("rowVersion必须是非负整数")
  476. return int(value)