106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
from sqlalchemy import select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import Document, DocumentTag, User
|
||
|
||
|
||
def _extract_doc_labels(doc: dict) -> set[str]:
|
||
labels: set[str] = set()
|
||
for attr in (doc.get("attr_labels") or []):
|
||
for lbl in (attr.get("labels") or []):
|
||
name = (lbl.get("label_name") or "").strip()
|
||
if name:
|
||
labels.add(name.replace("可见", ""))
|
||
return labels
|
||
|
||
|
||
def build_unit_chain(user: User) -> list[str]:
|
||
"""从用户自身单位一路向上到根节点，返回单位名列表。"""
|
||
chain: list[str] = []
|
||
seen: set[int] = set()
|
||
cur = user.unit
|
||
while cur is not None and cur.id not in seen:
|
||
seen.add(cur.id)
|
||
name = (cur.name or "").strip()
|
||
if name:
|
||
chain.append(name)
|
||
cur = cur.parent
|
||
return chain
|
||
|
||
|
||
def resolve_chat_doc_scope(user: User, adp_docs: list[dict]) -> dict:
|
||
"""按"就近单位优先、逐级向上回退"的规则计算问答阶段允许引用的文档集合。
|
||
|
||
- 超管/管理员：不做过滤。
|
||
- 普通用户：从自身单位开始，找到最近一层"有直接匹配文档"的单位；
|
||
allowed = 该层匹配文档 ∪ 所有"所有人可见"文档；
|
||
若整条链都没有匹配单位的文档，则 allowed = 仅公开文档。
|
||
"""
|
||
is_admin = user.is_superuser or any(
|
||
r.code in {"admin", "system_admin"} for r in user.roles
|
||
)
|
||
unit_chain = build_unit_chain(user)
|
||
result = {
|
||
"allowed_titles": None, # None 表示不过滤
|
||
"unit_chain": unit_chain,
|
||
"own_unit": unit_chain[0] if unit_chain else "",
|
||
"chosen_unit": "", # 实际命中的单位名，可能是祖先单位
|
||
"fell_back": False, # True = 走到了祖先单位
|
||
"public_only": False, # True = 整条链无匹配，仅公开
|
||
}
|
||
if is_admin or not unit_chain:
|
||
return result
|
||
|
||
public: set[str] = set()
|
||
per_unit: dict[str, set[str]] = {u: set() for u in unit_chain}
|
||
for doc in adp_docs:
|
||
title = (doc.get("file_name") or "").strip()
|
||
if not title:
|
||
continue
|
||
bases = _extract_doc_labels(doc)
|
||
if "所有人" in bases:
|
||
public.add(title)
|
||
for u in unit_chain:
|
||
if any(u in b for b in bases):
|
||
per_unit[u].add(title)
|
||
|
||
for idx, u in enumerate(unit_chain):
|
||
if per_unit[u]:
|
||
result["allowed_titles"] = public | per_unit[u]
|
||
result["chosen_unit"] = u
|
||
result["fell_back"] = idx > 0
|
||
return result
|
||
|
||
result["allowed_titles"] = public
|
||
result["public_only"] = True
|
||
return result
|
||
|
||
|
||
def get_or_create_tags(db: Session, names: list[str]) -> list[DocumentTag]:
|
||
tags: list[DocumentTag] = []
|
||
for raw_name in names:
|
||
name = raw_name.strip()
|
||
if not name:
|
||
continue
|
||
tag = db.scalar(select(DocumentTag).where(DocumentTag.name == name))
|
||
if not tag:
|
||
tag = DocumentTag(name=name)
|
||
db.add(tag)
|
||
db.flush()
|
||
tags.append(tag)
|
||
return tags
|
||
|
||
|
||
def serialize_document(document: Document) -> dict:
|
||
return {
|
||
"id": document.id,
|
||
"title": document.title,
|
||
"version": document.version,
|
||
"adp_doc_id": document.adp_doc_id,
|
||
"secrecy_level": document.secrecy_level,
|
||
"status": document.status,
|
||
"owner_dept": document.owner_dept,
|
||
"tags": [tag.name for tag in document.tags],
|
||
"created_at": document.created_at,
|
||
}
|