mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 04:01:58 +00:00
ThearchyHelios
19329c873e
外部系统（门户、OA、BI）和个人脚本要用一个身份调接口，以前只能拿个人会话 cookie 去模拟。这次加访问密钥：
- 签发（A1）：users.kind 区分人和服务账号（svc- 开头、没有口令、不能登录、角色固定使用者，管理员建）；access_keys 只存
sha256 和开头一截，明文 agl_ 加 32 字节随机串只显示一次；到期必填（最长 365 天、缺省 90 天）、可撤销、可轮换（新钥 + 旧钥
24 小时后到期）；个人访问密钥由管理员开关 auth.personal_keys 管，缺省关，关掉时已有的个人密钥一并不认
- 认人与门禁（A2）：Principal 加 via / key_id / key_name / scopes / app_ids（全部带默认值）；中间件认 Bearer agl_…，不看
cookie、不查 CSRF、不看 X-Actor，认不出的各回 401 带各自的 code、绝不退回匿名；WebSocket 带密钥一律拒；关登录时 409
keys_need_login。KEY_ROUTES（方法 + 路由模板 → scope）一张表，FastAPI(dependencies=[Depends(key_gate)]) 挂一次；不在表里的
403 key_route_forbidden，缺 scope、限定应用、个人应用和试运行各有各的 403；每把密钥每分钟 120 次（429 + Retry-After）
- 封顶（A3）：按密钥认出来的一律是使用者（拥有者现在的组），现有的使用者规则和遮罩全部自动生效，管理员的密钥也一样
- 写操作（A4）：发起入口、刷新视图必须带 Idempotency-Key，api_receipts 按（拥有者, 键）去重，轮换后照样认得出；状态机从
process/hooks.py 抽到 process/receipts.py，钩子和密钥共用（钩子现有测试原样通过），另按请求指纹拒掉同一个键办另一件事；
有文件控件的入口 422；回复待办只给个人密钥、只限审批 / 补充输入 / 编辑稿件；run.origin、task.responded 记 via=key、哪把密钥
- 视图数据（A5）：GET /api/apps/{id}/views/{view}/data（agentlab.viewdata/1），印章、清单哈希、口径卡指标原值和渲染文字，
遮罩的 value 为 null（和页面同一个判断），按需视图只读缓存、不发起运行，只给正式运行
- 留痕（A6）：建 / 撤销 / 轮换密钥、建 / 停用服务账号、开关个人密钥写审计；请求记录进程内缓冲、调度循环一拍里批量写、
保留 90 天；last_used_at 至少隔 1 分钟写一次；钩子收件记录接口（manage），验签失败只在进程内按钩子计数
- 名录与派单：服务账号在名录里标出，岗位绑定、选人候选、提交审核、改派里排除，应用授权照收；服务账号发起的实例里派给发起人
的待办交给任何人
- 接口说明（A7）：GET /api/access-keys/reference 从 KEY_ROUTES 生成（要登录）
测试：签发与各种 401、门禁合约（遍历全站路由）、CSRF 与身份来源、封顶（管理员密钥取工件也是遮过的）、幂等（含轮换后、
半路杀进程）、留痕、待办种类、视图数据（真跑的遮罩会员库）、限流、钩子收件记录。
788 lines
44 KiB
Python
788 lines
44 KiB
Python
"""应用的发布检查（§3.13 的 app.* 编号）。问题格式同 engine/schema.py 的 ValidationIssue：
|
||
{message, level: error|warning, code, path}。path 指向 spec 里的位置（如 pages[0].body[2].bind）。
|
||
|
||
编号是给程序认的（编辑器按它把问题落到属性面板、助手按它修），文案可以改，编号不改。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from typing import Any
|
||
|
||
from pydantic import ValidationError
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.apps.spec import APP_ROOTS, DISPOSITIONS, parse_spec
|
||
from app.apps.tree import bind_kind, components, input_fields, node_map, pins_by_alias, resolve_form, slot_refs
|
||
|
||
#: 按需视图缓存时效：默认 600 秒，最多一天（§2.3）
|
||
TTL_DEFAULT = 600
|
||
TTL_MAX = 86_400
|
||
#: 运行信息绑定能取的几项
|
||
RUN_FIELDS = ("status", "tier", "finished_at")
|
||
#: 入口映射里能取的当前用户的几项
|
||
USER_FIELDS = ("username", "display_name")
|
||
|
||
_DIGIT = re.compile(r"\d")
|
||
#: 文字里的一个数：千分位的逗号算在数里（「5,000」就是 5000）
|
||
_NUMBER = re.compile(r"\d[\d,，]*(?:\.\d+)?")
|
||
#: 岗位在图里的用处（tree.SLOT_ROLES），检查的说明里写清是哪一步要它
|
||
_SLOT_USE = {"assignee": "人工节点「{title}」派给", "escalation": "人工节点「{title}」超时升级给",
|
||
"confirm": "调用工具节点「{title}」的发送确认派给"}
|
||
_REPORT_TEXT = re.compile(r"^\s*\{\{\s*nodes\.([A-Za-z0-9_-]+)\.text\s*\}\}\s*$")
|
||
_PLACEHOLDER = re.compile(r"\{([A-Za-z0-9_㐀-鿿]+)\}")
|
||
|
||
|
||
class _Issues:
|
||
def __init__(self) -> None:
|
||
self.items: list[dict[str, Any]] = []
|
||
|
||
def error(self, code: str, message: str, path: str | None = None, **extra: Any) -> None:
|
||
self.items.append({"message": message, "level": "error", "code": code, "path": path, **extra})
|
||
|
||
def warning(self, code: str, message: str, path: str | None = None, **extra: Any) -> None:
|
||
self.items.append({"message": message, "level": "warning", "code": code, "path": path, **extra})
|
||
|
||
|
||
def parse_map_source(source: Any) -> tuple[str, Any] | None:
|
||
"""入口映射的一项：("form", 字段) / ("user", 属性) / ("const", 值)；写法不对是 None。
|
||
|
||
常量写成 JSON 字面量（数、true / false、"文字"）或单引号括起来的文字（'报销'）。"""
|
||
if not isinstance(source, str) or not source.strip():
|
||
return None
|
||
text = source.strip()
|
||
if text.startswith("form."):
|
||
return ("form", text[5:]) if text[5:] else None
|
||
if text.startswith("user."):
|
||
return ("user", text[5:]) if text[5:] in USER_FIELDS else None
|
||
if len(text) >= 2 and text[0] == text[-1] == "'":
|
||
return ("const", text[1:-1])
|
||
try:
|
||
return ("const", json.loads(text))
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def needs_disposition(previous: Any | None, pins: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""钉住项相对上一个应用版本换了版本的：[{alias, from, to}]。按别名对上；别名换了的，按同一个工作流对上（唯一时）。"""
|
||
if previous is None:
|
||
return []
|
||
before = [p for p in (getattr(previous, "pins", None) or []) if isinstance(p, dict)]
|
||
out: list[dict[str, Any]] = []
|
||
for pin in pins:
|
||
if not isinstance(pin, dict):
|
||
continue
|
||
old = next((p for p in before if p.get("alias") == pin.get("alias")), None)
|
||
if old is None:
|
||
same = [p for p in before if p.get("workflow_id") == pin.get("workflow_id")]
|
||
old = same[0] if len(same) == 1 else None
|
||
if old is not None and old.get("workflow_id") == pin.get("workflow_id") and old.get("version") != pin.get("version"):
|
||
out.append({"alias": pin.get("alias"), "from": old.get("version"), "to": pin.get("version")})
|
||
return out
|
||
|
||
|
||
async def lint_app(session: AsyncSession, spec: dict[str, Any], bindings: dict[str, Any], *,
|
||
previous: Any | None = None, dispositions: dict[str, str] | None = None,
|
||
pins: dict[str, Any] | None = None, scope: str = "shared") -> list[dict[str, Any]]:
|
||
"""检查一份待发布的应用。previous 是上一个应用版本（判断钉住项升版要不要处置）。
|
||
|
||
pins 给了就按它认钉住项（别名 → PinState），不读库：发布单的预检按「将要发布的那一版」检查（版本已经存下、还没发布，
|
||
blueprints/publish_plan.py 把它当作已发布的样子交进来），其余调用方不给。
|
||
|
||
scope 是应用的范围（App.scope）。个人应用（personal，docs/design/normal-user-apps.md N7）固定的是它自己的蓝图、视图只取
|
||
探索运行、图要满足使用者的约束（_lint_personal）；共享应用里出现蓝图、探索运行的照实报错。"""
|
||
from app.apps.pins import load_pins
|
||
|
||
out = _Issues()
|
||
try:
|
||
parse_spec(spec)
|
||
except ValidationError as e:
|
||
for err in e.errors()[:20]:
|
||
loc = ".".join(str(x) for x in err.get("loc") or ())
|
||
out.error("app.format", f"应用描述的格式不对：{loc}（{err.get('msg')}）", loc or None)
|
||
return out.items
|
||
pins = pins if pins is not None else await load_pins(session, spec)
|
||
level = _level(pins)
|
||
_lint_unique(spec, out)
|
||
_lint_pins(spec, pins, out, scope)
|
||
await _lint_subgraph_pins(session, spec, pins, out)
|
||
_lint_upgrade(spec, previous, dispositions or {}, out)
|
||
views = spec.get("views") or {}
|
||
await _lint_views(session, spec, pins, out, scope)
|
||
_lint_components(spec, pins, views, level, out)
|
||
await _lint_records_rows(session, spec, pins, out)
|
||
_lint_hidden(spec, out)
|
||
if scope == "personal":
|
||
_lint_personal(spec, pins, out)
|
||
_lint_entries(spec, pins, out)
|
||
await _lint_process(session, spec, pins, bindings or {}, out)
|
||
_lint_hooks(spec, out)
|
||
await _lint_notify(session, spec, out)
|
||
return out.items
|
||
|
||
|
||
def _level(pins: dict[str, Any]) -> str:
|
||
from app.apps.tree import app_level
|
||
|
||
return app_level([{"level": p.level} for p in pins.values()])
|
||
|
||
|
||
def _lint_unique(spec: dict[str, Any], out: _Issues) -> None:
|
||
seen: set[str] = set()
|
||
for i, page in enumerate(spec.get("pages") or []):
|
||
if page.get("id") in seen:
|
||
out.error("app.format", f"页面「{page.get('id')}」重名", f"pages[{i}].id")
|
||
seen.add(page.get("id"))
|
||
for comp, _, at in components(spec):
|
||
if comp.get("id") in seen:
|
||
out.error("app.format", f"组件「{comp.get('id')}」和别的组件或页面重名", f"{at}.id")
|
||
seen.add(comp.get("id"))
|
||
aliases: set[str] = set()
|
||
for i, pin in enumerate(spec.get("pins") or []):
|
||
if pin.get("alias") in aliases:
|
||
out.error("app.format", f"引用的工作流版本别名「{pin.get('alias')}」重复", f"pins[{i}].alias")
|
||
aliases.add(pin.get("alias"))
|
||
entries: set[str] = set()
|
||
for i, entry in enumerate(spec.get("entries") or []):
|
||
if entry.get("id") in entries:
|
||
out.error("app.format", f"入口「{entry.get('id')}」重名", f"entries[{i}].id")
|
||
entries.add(entry.get("id"))
|
||
|
||
|
||
def _lint_pins(spec: dict[str, Any], pins: dict[str, Any], out: _Issues, scope: str = "shared") -> None:
|
||
for i, pin in enumerate(spec.get("pins") or []):
|
||
state = pins.get(pin.get("alias"))
|
||
at = f"pins[{i}]"
|
||
if pin.get("blueprint") or scope == "personal":
|
||
# 个人应用的蓝图在 _lint_personal 里查；两种范围和两种固定版本对不上的在这里报
|
||
if pin.get("blueprint") and scope != "personal":
|
||
out.error("app.pin_unpublished", f"引用的「{pin.get('alias')}」是蓝图里还没发布的图：共享应用只能引用已发布的"
|
||
"工作流版本", at)
|
||
elif not pin.get("blueprint"):
|
||
out.error("personal.policy", "个人应用只能引用它自己的蓝图，不能引用工作流版本", at)
|
||
continue
|
||
if state is None or state.row is None:
|
||
out.error("app.pin_unpublished", f"引用的工作流版本「{pin.get('alias')}」不存在（工作流或 v{pin.get('version')} 已删除）", at)
|
||
elif state.retired:
|
||
out.error("app.pin_retired", f"引用的「{state.label}」已退役，请升级到一个在用的发布版本", at)
|
||
elif not state.published:
|
||
out.error("app.pin_unpublished", f"引用的「{state.label}」没有发布过，应用只能引用已发布的版本", at)
|
||
elif state.hash != pin.get("graph_hash"):
|
||
out.error("app.pin_hash", f"引用的「{state.label}」的内容指纹和应用里记的不一致，请重新选择这一版", at)
|
||
|
||
|
||
async def _lint_subgraph_pins(session: AsyncSession, spec: dict[str, Any], pins: dict[str, Any], out: _Issues) -> None:
|
||
"""钉住的版本里，子工作流固定的版本不是「已发布且未退役」的（D-q，apps/versions.subgraph_pin_problems）：各种用法都会
|
||
发起这一版的正式运行（入口、流程、按需视图、给最新封存视图供数的定时任务、补算），每一次都会在发起时被拒，所以报错。
|
||
钉住项本身不能用的（上面 _lint_pins 报过）不再往里看。"""
|
||
from app.apps.versions import subgraph_pin_problems
|
||
from app.engine.schema import GraphSpec
|
||
|
||
for i, pin in enumerate(spec.get("pins") or []):
|
||
state = pins.get(pin.get("alias"))
|
||
if state is None or state.row is None or not state.published:
|
||
continue
|
||
try:
|
||
graph = GraphSpec.model_validate(state.graph)
|
||
except ValueError:
|
||
continue
|
||
for problem in await subgraph_pin_problems(session, graph):
|
||
out.error(problem.code, f"「{state.label}」里的{problem.who}固定的{problem.reason}：这一版的正式运行"
|
||
"发起时会被拒绝。请在那个工作流里改好并发布新版本，再把这里的引用换成新版本", f"pins[{i}]",
|
||
node_id=problem.node_id)
|
||
|
||
|
||
def _lint_upgrade(spec: dict[str, Any], previous: Any | None, dispositions: dict[str, str], out: _Issues) -> None:
|
||
from app.engine.governance import UPGRADE_POLICY_LABELS
|
||
|
||
choices = "、".join(f"「{v}」" for v in UPGRADE_POLICY_LABELS.values())
|
||
for item in needs_disposition(previous, spec.get("pins") or []):
|
||
if dispositions.get(item["alias"]) not in DISPOSITIONS:
|
||
index = next(i for i, p in enumerate(spec.get("pins") or []) if p.get("alias") == item["alias"])
|
||
out.error("app.upgrade_policy",
|
||
f"引用的工作流版本「{item['alias']}」从 v{item['from']} 升到了 v{item['to']}，口径升版必须处置："
|
||
f"请选择 {choices} 中的一种", f"pins[{index}]", alias=item["alias"])
|
||
|
||
|
||
async def _lint_views(session: AsyncSession, spec: dict[str, Any], pins: dict[str, Any], out: _Issues,
|
||
scope: str = "shared") -> None:
|
||
from app.apps.pins import reads_live
|
||
from app.apps.resolve import latest_sealed_run
|
||
from app.db.models import Schedule
|
||
|
||
for vid, view in (spec.get("views") or {}).items():
|
||
source = (view or {}).get("source") or {}
|
||
at = f"views.{vid}.source"
|
||
kind = source.get("kind")
|
||
if (kind == "exploratory") != (scope == "personal"):
|
||
# 探索运行只在个人应用里合法（N7）：共享应用的视图取了它，等于把没发布的图的结果当正式结果给人看；个人应用
|
||
# 只有蓝图上的图、没有正式运行，取别的来源永远是空的
|
||
out.error("app.view_source_personal",
|
||
f"视图「{vid}」取的是探索运行：只有个人应用能这样取数，共享应用的视图要取正式运行（最近一次出数、"
|
||
"按条件查看或本实例）" if kind == "exploratory" else
|
||
f"视图「{vid}」：个人应用的视图只能取它自己的探索运行（每次刷新的结果）", f"{at}.kind")
|
||
continue
|
||
if kind == "instance":
|
||
continue
|
||
state = pins.get(source.get("pin"))
|
||
if state is None:
|
||
out.error("app.binding_target_missing", f"视图「{vid}」引用的工作流版本「{source.get('pin')}」不在应用里", at)
|
||
continue
|
||
if kind == "exploratory":
|
||
continue
|
||
if kind == "latest_sealed":
|
||
origin = source.get("from")
|
||
if origin not in (None, "formal") and not (isinstance(origin, str) and origin.startswith("schedule:")):
|
||
out.error("app.format", f"视图「{vid}」的来源只能是正式运行或某个定时任务", f"{at}.from")
|
||
continue
|
||
if isinstance(origin, str) and origin.startswith("schedule:") \
|
||
and await session.get(Schedule, origin[len("schedule:"):]) is None:
|
||
out.error("app.binding_target_missing", f"视图「{vid}」取数的定时任务已不存在", f"{at}.from")
|
||
continue
|
||
if state.row is not None and await latest_sealed_run(session, state, origin) is None:
|
||
out.warning("app.view_source_empty", f"视图「{vid}」暂时没有可读的封存运行：页面上会写明还没有数据", at)
|
||
elif kind == "on_demand":
|
||
ttl = source.get("ttl_s")
|
||
if ttl is not None and not (isinstance(ttl, int) and 0 < ttl <= TTL_MAX):
|
||
out.error("app.on_demand_ttl", f"视图「{vid}」的缓存时效需要在 1 到 {TTL_MAX:,} 秒之间", f"{at}.ttl_s")
|
||
elif ttl is None and state.row is not None and await reads_live(session, state.graph):
|
||
out.error("app.on_demand_ttl",
|
||
f"视图「{vid}」按需运行时会查活库，需要设置缓存时效（建议 {TTL_DEFAULT} 秒）：不然同样的参数会一直用旧结果",
|
||
f"{at}.ttl_s")
|
||
if source.get("params") is not None:
|
||
form = resolve_form(spec, source.get("params"))
|
||
problems = ["引用的表单不存在"] if form is None else _form_problems(form)
|
||
for p in problems[:3]:
|
||
out.error("app.format", f"视图「{vid}」的参数表单：{p}", f"{at}.params")
|
||
|
||
|
||
def _form_problems(form: dict[str, Any]) -> list[str]:
|
||
from app.apps.forms import validate_form_schema
|
||
|
||
return validate_form_schema(form)
|
||
|
||
|
||
def _view_graphs(spec: dict[str, Any], pins: dict[str, Any], view_id: str | None) -> list[dict[str, Any]]:
|
||
"""视图背后的图：钉住项那一版的图；实例视图是各入口引用的那几版。"""
|
||
view = (spec.get("views") or {}).get(view_id or "") or {}
|
||
source = view.get("source") or {}
|
||
if source.get("kind") == "instance":
|
||
aliases = {e.get("pin") for e in spec.get("entries") or []} or set(pins)
|
||
return [pins[a].graph for a in aliases if a in pins and pins[a].row is not None]
|
||
state = pins.get(source.get("pin"))
|
||
# 个人应用的固定版本是蓝图：图在蓝图上（row 为空）
|
||
return [state.graph] if state is not None and (state.row is not None or state.blueprint is not None) else []
|
||
|
||
|
||
def _lint_components(spec: dict[str, Any], pins: dict[str, Any], views: dict[str, Any], level: str,
|
||
out: _Issues) -> None:
|
||
from jsonschema import Draft202012Validator
|
||
|
||
from app.apps.catalog import BY_NAME
|
||
|
||
constants = _graph_numbers(pins)
|
||
for comp, _, at in components(spec):
|
||
ctype = comp.get("type")
|
||
entry = BY_NAME.get(str(ctype))
|
||
if entry is None:
|
||
out.error("app.component_unknown", f"组件「{comp.get('id')}」的类型「{ctype}」不在组件目录里", f"{at}.type")
|
||
continue
|
||
for err in list(Draft202012Validator(entry["props"]).iter_errors(comp.get("props") or {}))[:3]:
|
||
where = ".".join(str(x) for x in err.absolute_path)
|
||
out.error("app.props_invalid", f"组件「{comp.get('id')}」（{entry['label']}）的属性{('「' + where + '」') if where else ''}不对",
|
||
f"{at}.props{('.' + where) if where else ''}")
|
||
if comp.get("children") and not entry["children"]:
|
||
out.error("app.props_invalid", f"组件「{comp.get('id')}」（{entry['label']}）不能放子组件", f"{at}.children")
|
||
_lint_text(comp, entry, at, level, out, constants)
|
||
_lint_bind(spec, comp, entry, pins, views, at, out)
|
||
if level == "governed":
|
||
_lint_metric_origin(spec, comp, pins, at, out)
|
||
_lint_visible(comp, views, at, out)
|
||
_lint_refs(spec, comp, at, out)
|
||
|
||
|
||
def _numbers(text: str) -> list[str]:
|
||
return [m.replace(",", "").replace("，", "").rstrip(".") for m in _NUMBER.findall(text)]
|
||
|
||
|
||
def _graph_numbers(pins: dict[str, Any]) -> set[str]:
|
||
"""固定的版本里节点配置写着的数：条件里的金额线、时限、会签人数、表单的上下限这些流程规则。只看 config，不看标题。"""
|
||
found: set[str] = set()
|
||
|
||
def walk(value: Any) -> None:
|
||
if isinstance(value, bool):
|
||
return
|
||
if isinstance(value, (int, float)):
|
||
found.update(_numbers(f"{value:g}" if isinstance(value, float) else str(value)))
|
||
elif isinstance(value, str):
|
||
found.update(_numbers(value))
|
||
elif isinstance(value, dict):
|
||
for v in value.values():
|
||
walk(v)
|
||
elif isinstance(value, list):
|
||
for v in value:
|
||
walk(v)
|
||
|
||
for state in pins.values():
|
||
if getattr(state, "row", None) is None:
|
||
continue
|
||
for node in node_map(state.graph).values():
|
||
walk(node.get("config"))
|
||
return found
|
||
|
||
|
||
def _lint_text(comp: dict[str, Any], entry: dict[str, Any], at: str, level: str, out: _Issues,
|
||
constants: set[str]) -> None:
|
||
"""静态文字里有数字：叙述层无算术、数字只来自证据。受管级报错，已发布级警告。
|
||
|
||
数据要绑定指标或单元格；流程规则（审批的金额线、时限、会签人数）的出处是固定版本里的节点配置——文字里的每个数都在
|
||
那里写着才算有出处。规则只写在说明里、流程并不照着执行的，照样报：那句话是假的。"""
|
||
props = comp.get("props") or {}
|
||
text = props.get("markdown") if comp.get("type") == "text" else props.get("text") if comp.get("type") == "notice" else None
|
||
if not isinstance(text, str) or not _DIGIT.search(text):
|
||
return
|
||
loose = [n for n in dict.fromkeys(_numbers(text)) if n not in constants]
|
||
if not loose:
|
||
return
|
||
add = out.error if level == "governed" else out.warning
|
||
add("app.text_number", f"组件「{comp.get('id')}」（{entry['label']}）的静态文字里写了数字「{'、'.join(loose[:5])}」，"
|
||
"没有出处：数据请改为绑定指标或单元格；流程规则（如审批的金额线、时限）要先写进工作流，文字里的数和工作流里的一致"
|
||
"才算有出处", f"{at}.props")
|
||
|
||
|
||
def _lint_bind(spec: dict[str, Any], comp: dict[str, Any], entry: dict[str, Any], pins: dict[str, Any],
|
||
views: dict[str, Any], at: str, out: _Issues) -> None:
|
||
bind = comp.get("bind")
|
||
if bind is None:
|
||
return
|
||
where = f"{at}.bind"
|
||
cid = comp.get("id")
|
||
kind = bind_kind(bind)
|
||
if kind is not None and kind not in entry["bindings"]:
|
||
allowed = "、".join(entry["bindings"]) or "任何数据"
|
||
out.error("app.props_invalid", f"组件「{cid}」（{entry['label']}）不能绑定这种数据，只能绑定：{allowed}", where)
|
||
return
|
||
if kind in ("task", "action"):
|
||
graphs = _view_graphs(spec, pins, next((v for v, d in views.items()
|
||
if ((d or {}).get("source") or {}).get("kind") == "instance"), None))
|
||
_lint_process_bind(cid, kind, bind[kind], graphs, where, out)
|
||
return
|
||
view_id = bind.get("view")
|
||
if view_id not in views:
|
||
out.error("app.binding_target_missing", f"组件「{cid}」绑定的视图「{view_id}」不存在", where)
|
||
return
|
||
if kind is None:
|
||
return
|
||
graphs = _view_graphs(spec, pins, view_id)
|
||
if not graphs:
|
||
return
|
||
target = bind[kind]
|
||
if not any(_target_ok(kind, target, g, out, cid, where) for g in graphs):
|
||
out.error("app.binding_target_missing", f"组件「{cid}」绑定的{_KIND_LABEL[kind]}在引用的工作流版本里不存在", where)
|
||
|
||
|
||
def _lint_metric_origin(spec: dict[str, Any], comp: dict[str, Any], pins: dict[str, Any], at: str, out: _Issues) -> None:
|
||
"""受管级应用里绑定的指标（指标块、趋势）：表达式的输入追不到原始数据（查询快照里的格、运行输入）的给提示。
|
||
|
||
指标照样是口径卡确定性算出来的、面板上照样「有出处」，但点开追到某个节点就断了：受管级的应用要的是一路追到数据。
|
||
和运行时、证据面板同一套认法（engine/lineage.static_stops），只看图不看值。转抄别处口径卡的指标在上游那一版里，
|
||
不在这里追；目标不存在的由 _lint_bind 报。"""
|
||
from app.engine.labels import type_label
|
||
from app.engine.lineage import static_stops
|
||
|
||
bind = comp.get("bind")
|
||
kind = bind_kind(bind) if isinstance(bind, dict) else None
|
||
if kind not in ("metric", "series"):
|
||
return
|
||
node_id, _, metric_id = str(bind.get(kind) or "").partition(".")
|
||
for graph in _view_graphs(spec, pins, bind.get("view")):
|
||
nodes = node_map(graph)
|
||
card = nodes.get(node_id)
|
||
if card is None or card["type"] != "metrics" or isinstance(card["config"].get("caliber_from"), dict):
|
||
continue
|
||
metric = next((m for m in card["config"].get("metrics") or [] if isinstance(m, dict) and m.get("id") == metric_id),
|
||
None)
|
||
if metric is None:
|
||
continue
|
||
stops = static_stops(nodes, str(metric.get("expression") or ""))
|
||
if not stops:
|
||
continue
|
||
where = "；".join(
|
||
f"{s.path} 停在「{nodes[s.node_id]['title']}」（{type_label(nodes[s.node_id]['type'])}）" if s.node_id in nodes
|
||
else f"{s.path} 无法确定来自哪个节点" for s in stops[:3])
|
||
out.warning("app.metric_untraced",
|
||
f"组件「{comp.get('id')}」绑定的指标「{metric.get('name') or metric_id}」的输入没有追到原始数据（{where}）："
|
||
"数是口径卡算出来的，但点开出处只能追到那里，看不到查询快照里的格。请在口径卡里用 cell() 按格取数，"
|
||
"或让数据整形只原样解析调用工具节点的结果", f"{at}.bind", node_id=node_id)
|
||
return
|
||
|
||
|
||
_KIND_LABEL = {"metric": "指标", "series": "指标", "cell": "单元格", "query": "查询结果", "report": "报告",
|
||
"input": "运行输入", "run": "运行信息", "output": "成果字段", "task": "待办", "action": "动作"}
|
||
|
||
|
||
def _target_ok(kind: str, target: Any, graph: dict[str, Any], out: _Issues, cid: Any, where: str) -> bool:
|
||
nodes = node_map(graph)
|
||
if kind in ("metric", "series"):
|
||
node_id, _, metric = str(target or "").partition(".")
|
||
node = nodes.get(node_id)
|
||
if node is None or node["type"] != "metrics" or not metric:
|
||
return False
|
||
listed = [m.get("id") for m in node["config"].get("metrics") or [] if isinstance(m, dict)]
|
||
# 转抄别处口径卡的（caliber_from）指标在上游那一版里，发布时引用的那一版不变，这里不去追
|
||
return metric in listed or (not listed and isinstance(node["config"].get("caliber_from"), dict))
|
||
if kind in ("cell", "query"):
|
||
node = nodes.get(str((target or {}).get("node") if isinstance(target, dict) else ""))
|
||
if kind == "cell" and isinstance(target, dict) and not (
|
||
isinstance(target.get("row"), int) and isinstance(target.get("col"), str)):
|
||
return False
|
||
return node is not None and node["type"] in ("tool", "agent", "merge")
|
||
if kind == "report":
|
||
node = nodes.get(str(target))
|
||
return node is not None and node["type"] == "report"
|
||
if kind == "input":
|
||
fields = input_fields(graph)
|
||
return not fields or any(f["name"].strip() == target for f in fields)
|
||
if kind == "run":
|
||
return target in RUN_FIELDS
|
||
if kind == "output":
|
||
for node in nodes.values():
|
||
if node["type"] != "output":
|
||
continue
|
||
for f in node["config"].get("fields") or []:
|
||
if isinstance(f, dict) and f.get("name") == target:
|
||
value = str(f.get("value") or "")
|
||
report = _REPORT_TEXT.match(value)
|
||
if report and (nodes.get(report.group(1)) or {}).get("type") == "report":
|
||
return True
|
||
if "{{" not in value and not _DIGIT.search(value):
|
||
return True
|
||
out.error("app.binding_not_evidenced",
|
||
f"组件「{cid}」绑定的成果字段「{target}」不是逐字取自报告的文字，里面的数字没有证据。"
|
||
"请改为绑定指标、单元格或报告", where)
|
||
return True
|
||
return False
|
||
return True
|
||
|
||
|
||
def _lint_process_bind(cid: Any, kind: str, target: Any, graphs: list[dict[str, Any]], where: str, out: _Issues) -> None:
|
||
want = "human" if kind == "task" else "tool"
|
||
if not any((node_map(g).get(str(target)) or {}).get("type") == want for g in graphs):
|
||
out.error("app.binding_target_missing", f"组件「{cid}」绑定的{_KIND_LABEL[kind]}节点「{target}」在入口引用的工作流版本里不存在",
|
||
where)
|
||
|
||
|
||
def _lint_visible(comp: dict[str, Any], views: dict[str, Any], at: str, out: _Issues) -> None:
|
||
from app.engine.expressions import ExpressionArithmetic, ExpressionError, parse_expression
|
||
|
||
if comp.get("scope") is not None and comp.get("scope") not in views:
|
||
out.error("app.binding_target_missing", f"组件「{comp.get('id')}」的可见条件引用的视图「{comp.get('scope')}」不存在",
|
||
f"{at}.scope")
|
||
expr = comp.get("visible")
|
||
if not expr:
|
||
return
|
||
where = f"{at}.visible"
|
||
try:
|
||
_, _, unknown = parse_expression(str(expr), roots=APP_ROOTS, arithmetic=False)
|
||
except ExpressionArithmetic as e:
|
||
out.error("app.expr_arith", f"组件「{comp.get('id')}」的可见条件：{e}", where)
|
||
return
|
||
except ExpressionError as e:
|
||
out.error("app.expr_invalid", f"组件「{comp.get('id')}」的可见条件写法不对：{e}", where)
|
||
return
|
||
if unknown:
|
||
roots = "、".join(sorted(APP_ROOTS))
|
||
out.error("app.expr_root", f"组件「{comp.get('id')}」的可见条件用了不认识的名字：{'、'.join(unknown)}。能用的有 {roots}",
|
||
where)
|
||
|
||
|
||
def _lint_refs(spec: dict[str, Any], comp: dict[str, Any], at: str, out: _Issues) -> None:
|
||
"""组件属性里按名字引用的入口、表单得在。"""
|
||
props = comp.get("props") or {}
|
||
if comp.get("type") in ("entry_button", "instance_list") and props.get("entry") is not None \
|
||
and not any(e.get("id") == props["entry"] for e in spec.get("entries") or []):
|
||
out.error("app.binding_target_missing", f"组件「{comp.get('id')}」引用的入口「{props['entry']}」不存在", f"{at}.props.entry")
|
||
if comp.get("type") == "form" and props.get("form") not in (spec.get("forms") or {}):
|
||
out.error("app.binding_target_missing", f"组件「{comp.get('id')}」引用的表单「{props.get('form')}」不存在", f"{at}.props.form")
|
||
|
||
|
||
#: SQL 里出现这些就算汇总过了：分组，或者聚合函数
|
||
_AGGREGATED = re.compile(r"\bgroup\s+by\b|\b(?:count|sum|avg|min|max|total|group_concat)\s*\(", re.I)
|
||
|
||
|
||
async def _lint_records_rows(session: AsyncSession, spec: dict[str, Any], pins: dict[str, Any], out: _Issues) -> None:
|
||
"""表格组件绑定的查询直接列出记录表的逐行、没有汇总（app.records_rows_exposed，docs/design/normal-user-apps.md N8）：
|
||
看板读的是正式运行的结果，正式运行看得到全部行，这样等于所有使用者都能看到别人的记录，绕过了行级可见。只提醒：
|
||
作者可能就是要公开（排行榜）；逐行查看请用记录列表组件，它按看的人过滤。只看调用工具节点写在图里的 SQL。"""
|
||
from app.data.guard import _blank_quoted
|
||
from app.db.models import DataSource
|
||
from app.records import RECORDS_ORIGIN
|
||
from app.tools.datasource import QUERY_PREFIX
|
||
|
||
records: set[str] | None = None
|
||
for comp, _, at in components(spec):
|
||
bind = comp.get("bind") if isinstance(comp.get("bind"), dict) else None
|
||
if comp.get("type") != "table" or bind is None or bind_kind(bind) != "query":
|
||
continue
|
||
target = bind.get("query") if isinstance(bind.get("query"), dict) else {}
|
||
for graph in _view_graphs(spec, pins, bind.get("view")):
|
||
node = node_map(graph).get(str(target.get("node") or ""))
|
||
tool = str(((node or {}).get("config") or {}).get("tool") or "")
|
||
sql = (((node or {}).get("config") or {}).get("args") or {}).get("sql")
|
||
if node is None or node["type"] != "tool" or not tool.startswith(QUERY_PREFIX) or not isinstance(sql, str):
|
||
continue
|
||
if records is None:
|
||
records = set((await session.execute(select(DataSource.name).where(
|
||
DataSource.origin == RECORDS_ORIGIN))).scalars())
|
||
if tool[len(QUERY_PREFIX):] not in records or _AGGREGATED.search(_blank_quoted(sql, dialect="sqlite")):
|
||
continue
|
||
out.warning("app.records_rows_exposed",
|
||
f"组件「{comp.get('id')}」（表格）直接列出了记录表的逐行：看板的数据是正式运行查的，看得到全部行，"
|
||
"所有能用这个应用的人都会看到别人提交的记录。要逐行查看请改用「记录列表」组件（按看的人过滤）；"
|
||
"看板上的表格请改成汇总", f"{at}.bind", node_id=str(target.get("node")))
|
||
break
|
||
|
||
|
||
def _lint_hidden(spec: dict[str, Any], out: _Issues) -> None:
|
||
"""隐藏了的组件是唯一绑定某个视图的组件（N6，app.hidden_bound）：隐藏之后这个视图的数在页面上一处都看不到了——可能
|
||
隐藏错了，或者这个视图连同组件可以删掉。只提醒：作者也许就是暂时不想让人看。放在隐藏了的布局组件里的算隐藏。"""
|
||
binders: dict[str, list[tuple[bool, str, str]]] = {}
|
||
|
||
def walk(items: list[Any], hidden: bool, at: str) -> None:
|
||
for i, comp in enumerate(items):
|
||
if not isinstance(comp, dict):
|
||
continue
|
||
here = f"{at}[{i}]"
|
||
off = hidden or bool(comp.get("hidden"))
|
||
bind = comp.get("bind") if isinstance(comp.get("bind"), dict) else {}
|
||
if isinstance(bind.get("view"), str):
|
||
binders.setdefault(bind["view"], []).append((off, str(comp.get("id")), here))
|
||
walk(comp.get("children") or [], off, f"{here}.children")
|
||
|
||
for pi, page in enumerate(spec.get("pages") or []):
|
||
if isinstance(page, dict):
|
||
walk(page.get("body") or [], False, f"pages[{pi}].body")
|
||
for vid, items in binders.items():
|
||
if not all(off for off, _, _ in items):
|
||
continue
|
||
_, cid, at = items[0]
|
||
what = f"组件「{cid}」是唯一绑定视图「{vid}」的组件" if len(items) == 1 else f"绑定视图「{vid}」的组件都"
|
||
out.warning("app.hidden_bound", f"{what}，隐藏之后页面上一处都看不到这个视图的数了。确实不要了可以删掉它",
|
||
f"{at}.hidden")
|
||
|
||
|
||
def _lint_personal(spec: dict[str, Any], pins: dict[str, Any], out: _Issues) -> None:
|
||
"""个人应用（N7）：固定版本是它自己的蓝图（还在、图的指纹和应用记的一样），图满足使用者的约束（同问数据的白名单，
|
||
blueprints/personal.policy_problem，personal.policy），不能有入口、钩子、通知订阅、岗位。"""
|
||
from app.blueprints.personal import NOT_IN_PERSONAL, policy_problem
|
||
|
||
for i, pin in enumerate(spec.get("pins") or []):
|
||
if not pin.get("blueprint"):
|
||
continue
|
||
state = pins.get(pin.get("alias"))
|
||
at = f"pins[{i}]"
|
||
if state is None or state.blueprint is None:
|
||
out.error("personal.policy", "个人应用出自的蓝图已经不在了：没法再刷新，可以归档这个应用", at)
|
||
continue
|
||
if state.hash != pin.get("graph_hash"):
|
||
out.error("app.pin_hash", "蓝图在存成个人应用之后又改过：请回到蓝图重新「存成我的应用」", at)
|
||
if problem := policy_problem(state.graph):
|
||
out.error("personal.policy", problem, at)
|
||
declared = [(key, label) for key, label in NOT_IN_PERSONAL if spec.get(key)]
|
||
if declared:
|
||
out.error("personal.policy", f"个人应用不发起实例、不派待办，不能有{'、'.join(label for _, label in declared)}：请删掉",
|
||
declared[0][0])
|
||
|
||
|
||
def _lint_entries(spec: dict[str, Any], pins: dict[str, Any], out: _Issues) -> None:
|
||
for i, entry in enumerate(spec.get("entries") or []):
|
||
at = f"entries[{i}]"
|
||
label = f"入口「{entry.get('label') or entry.get('id')}」"
|
||
form = resolve_form(spec, entry.get("form"))
|
||
if form is None:
|
||
out.error("app.entry_form_mismatch", f"{label}引用的表单不存在", f"{at}.form")
|
||
continue
|
||
for p in _form_problems(form)[:3]:
|
||
out.error("app.entry_form_mismatch", f"{label}的表单：{p}", f"{at}.form")
|
||
state = pins.get(entry.get("pin"))
|
||
if state is None:
|
||
out.error("app.binding_target_missing", f"{label}引用的工作流版本「{entry.get('pin')}」不在应用里", f"{at}.pin")
|
||
continue
|
||
fields = {f["name"].strip(): f for f in input_fields(state.graph)} if state.row is not None else {}
|
||
mapping = entry.get("map") or {}
|
||
props = form.get("properties") or {}
|
||
for target, source in mapping.items():
|
||
parsed = parse_map_source(source)
|
||
if fields and target not in fields:
|
||
out.error("app.entry_form_mismatch", f"{label}映射的「{target}」不是「{state.label}」的入口字段", f"{at}.map.{target}")
|
||
if parsed is None:
|
||
out.error("app.entry_form_mismatch",
|
||
f"{label}映射「{target}」的取值写法不对：只能取表单字段、当前用户的用户名或显示名，或者写一个常量",
|
||
f"{at}.map.{target}")
|
||
elif parsed[0] == "form" and parsed[1] not in props:
|
||
out.error("app.entry_form_mismatch", f"{label}映射「{target}」取的表单字段「{parsed[1]}」不在表单里",
|
||
f"{at}.map.{target}")
|
||
for name, f in fields.items():
|
||
if f.get("required") and f.get("default") in (None, "") and name not in mapping:
|
||
out.error("app.entry_unmapped_required", f"{label}没有给「{state.label}」的必填字段「{name}」取值", f"{at}.map")
|
||
title = entry.get("instance_title")
|
||
if isinstance(title, dict):
|
||
listed = set(title.get("fields") or [])
|
||
used = set(_PLACEHOLDER.findall(str(title.get("pattern") or "")))
|
||
if bad := sorted((listed | used) - set(mapping)):
|
||
out.error("app.entry_form_mismatch", f"{label}的实例标题引用了没有映射的字段：{'、'.join(bad)}",
|
||
f"{at}.instance_title")
|
||
elif used - listed:
|
||
out.error("app.entry_form_mismatch", f"{label}的实例标题用到的字段要写在字段列表里：{'、'.join(sorted(used - listed))}",
|
||
f"{at}.instance_title")
|
||
|
||
|
||
async def _lint_process(session: AsyncSession, spec: dict[str, Any], pins: dict[str, Any], bindings: dict[str, Any],
|
||
out: _Issues) -> None:
|
||
"""流程钉住项：图里引用到的岗位（tree.slot_refs：派给谁、超时升级给谁、动作确认派给谁）都声明了、绑了人，写入动作
|
||
有人核实，图本身经得起升级（process.* 升成 app.process_unsafe）。岗位没人，流程走到那一步就派给了「任何人」：升级、
|
||
确认悄悄失效，所以报错。哪些调用工具节点是写入（开发送确认卡、要人核实）按 pins.write_nodes 认，和运行时、派生同一个
|
||
认法。"""
|
||
from app.apps.pins import write_nodes
|
||
from app.engine.governance import publish_issues
|
||
from app.engine.lint_process import load_subgraphs, with_subgraphs
|
||
from app.engine.schema import GraphSpec
|
||
|
||
process = [s for s in pins.values() if s.pin.get("use") == "process" and s.row is not None]
|
||
if not process:
|
||
return
|
||
bound = (bindings or {}).get("slots") or {}
|
||
slots = dict(spec.get("slots") or {})
|
||
used: dict[str, list[str]] = {}
|
||
writes = {state.alias: await write_nodes(session, state.graph) for state in process}
|
||
for state in process:
|
||
for ref in slot_refs(state.graph, writes[state.alias]):
|
||
slot, title = ref["slot"], ref["title"]
|
||
what = _SLOT_USE[ref["role"]].format(title=title)
|
||
used.setdefault(slot, [])
|
||
if f"{what}它" not in used[slot]:
|
||
used[slot].append(f"{what}它")
|
||
if slot not in slots:
|
||
out.error("app.slot_unbound", f"「{state.label}」里{what}岗位「{slot}」，应用里没有声明这个岗位",
|
||
"slots", node_id=ref["node_id"])
|
||
slots[slot] = {"label": slot}
|
||
for slot, info in slots.items():
|
||
if not bound.get(slot):
|
||
label = (info or {}).get("label") or slot
|
||
why = f"（{'；'.join(used[slot][:3])}）：没有人，流程走到这一步就派给了任何人" if used.get(slot) else ""
|
||
out.error("app.slot_unbound", f"岗位「{label}」还没有绑定到人或组{why}", f"slots.{slot}")
|
||
await _lint_slot_principals(session, slots, bound, out)
|
||
for state in process:
|
||
nodes = node_map(state.graph)
|
||
for node_id, cap in writes[state.alias].items():
|
||
if not getattr(cap, "lookup", None) and not bound.get("reconciler"):
|
||
out.error("app.reconciler_unbound", f"「{state.label}」的「{nodes[node_id]['title']}」会写外部系统，写入的能力"
|
||
"没有登记查询，结果未知时要人核实：请绑定核实岗位", "slots", node_id=node_id)
|
||
try:
|
||
graph = GraphSpec.model_validate(state.graph)
|
||
except ValueError:
|
||
continue
|
||
# 子工作流里有没有人工节点要读库才知道（门禁本身不查库）：先读好放进来，process.human_in_subgraph 才真正生效
|
||
with with_subgraphs(await load_subgraphs(session, graph)):
|
||
issues = publish_issues(graph, level=state.level or "published")
|
||
for issue in issues:
|
||
if (issue.code or "").startswith("process."):
|
||
out.error("app.process_unsafe", f"「{state.label}」作为流程不安全：{issue.message}",
|
||
f"pins[{_pin_index(spec, state.alias)}]", node_id=issue.node_id)
|
||
|
||
|
||
async def _lint_slot_principals(session: AsyncSession, slots: dict[str, Any], bound: dict[str, Any],
|
||
out: _Issues) -> None:
|
||
"""岗位绑定的人和组要在名录里（auth/directory.py，同选人控件的名录）：绑了名录里没有的、已停用的，流程走到这一步就
|
||
派给了没人能处理的名字——界面上选人时拦着，旧客户端、直接调接口、绑好之后人被停用或删除的照样会漏过来，所以发布时
|
||
再核一次。账号密码登录报错；单点登录的人第一次登录后才进名录，名录里还没有的只提醒；关登录不核对。"""
|
||
from app.api.deps import auth_mode
|
||
from app.auth.directory import load_directory, mismatch
|
||
|
||
mode = auth_mode()
|
||
if mode == "off" or not any(bound.get(s) for s in slots):
|
||
return
|
||
known = await load_directory(session)
|
||
for slot, info in slots.items():
|
||
label = (info or {}).get("label") or slot
|
||
for item in bound.get(slot) or []:
|
||
m = mismatch(known, mode, item.get("kind"), str(item.get("name"))) if isinstance(item, dict) else None
|
||
if m is None:
|
||
continue
|
||
if m.blocking:
|
||
out.error("app.slot_principal_unknown", f"岗位「{label}」绑定的{m.text}：派给这个名字的待办没有人能处理"
|
||
"（全体会签时整条待办办不完）。请从候选中重新选择，或从岗位里移除", f"slots.{slot}",
|
||
problem=m.problem)
|
||
else:
|
||
out.warning("app.slot_principal_unknown", f"岗位「{label}」绑定的{m.text}", f"slots.{slot}",
|
||
problem=m.problem)
|
||
|
||
|
||
def _pin_index(spec: dict[str, Any], alias: str) -> int:
|
||
return next((i for i, p in enumerate(spec.get("pins") or []) if p.get("alias") == alias), 0)
|
||
|
||
|
||
def _lint_hooks(spec: dict[str, Any], out: _Issues) -> None:
|
||
entries = {e.get("id") for e in spec.get("entries") or []}
|
||
for i, hook in enumerate(spec.get("hooks") or []):
|
||
if hook.get("kind") == "start" and hook.get("entry") not in entries:
|
||
out.error("app.hook_entry_missing", f"钩子「{hook.get('name')}」要按入口发起实例，但入口「{hook.get('entry')}」不存在",
|
||
f"hooks[{i}].entry")
|
||
|
||
|
||
async def _lint_notify(session: AsyncSession, spec: dict[str, Any], out: _Issues) -> None:
|
||
from app.db.models_process import NotifyChannel
|
||
|
||
wanted = [n.get("channel") for n in spec.get("notify") or []]
|
||
if not wanted:
|
||
return
|
||
known = {r for (r,) in await session.execute(select(NotifyChannel.id).where(NotifyChannel.id.in_(wanted)))}
|
||
for i, channel in enumerate(wanted):
|
||
if channel not in known:
|
||
out.warning("app.notify_channel_missing", f"订阅的通知渠道「{channel}」不存在", f"notify[{i}].channel")
|
||
|
||
|
||
async def acl_issues(session: AsyncSession, app_id: str) -> list[dict[str, Any]]:
|
||
"""开了登录却没有任何 use 以上的授权：除了管理员和搭建者，谁都用不了这个应用。关登录时不报。
|
||
|
||
授权给的人和组要在名录里（同岗位绑定，见 _lint_slot_principals）：授给名录里没有的名字，以后建了同名的人就直接拿到
|
||
这一档权限；授给停用的人，重新启用就又能用——都说清是哪一档、哪个名字。账号密码登录报错，单点登录名录里还没有的只提醒。"""
|
||
from app.api.deps import auth_mode
|
||
from app.apps.access import ROLE_TEXT, grants_of
|
||
from app.auth.directory import load_directory, mismatch
|
||
|
||
mode = auth_mode()
|
||
if mode == "off":
|
||
return []
|
||
grants = await grants_of(session, app_id)
|
||
if not grants:
|
||
return [{"message": "还没有授权给任何人或组使用这个应用：只有管理员和搭建者能打开它", "level": "warning",
|
||
"code": "app.acl_empty", "path": None}]
|
||
out = _Issues()
|
||
known = await load_directory(session)
|
||
for g in grants:
|
||
# 服务账号能拿授权（只是不能被派待办，岗位绑定那边才算错）
|
||
if (m := mismatch(known, mode, g.principal_kind, g.principal, allow_service=True)) is None:
|
||
continue
|
||
what = f"「{ROLE_TEXT.get(g.role, g.role)}」授权给的{m.text}"
|
||
if m.blocking:
|
||
out.error("app.acl_principal_unknown", f"{what}：{_GRANT_RISK[m.problem]}", "grants", problem=m.problem)
|
||
else:
|
||
out.warning("app.acl_principal_unknown", what, "grants", problem=m.problem)
|
||
return out.items
|
||
|
||
|
||
#: 授权给了名录里对不上的名字，会出什么事
|
||
_GRANT_RISK = {"missing": "以后建了同名的人会直接拿到这一档权限。请从授权里移除，或从候选中重新选择",
|
||
"disabled": "重新启用后他又能用这个应用。请从授权里移除"}
|
||
|
||
|
||
def pins_with_state(spec: dict[str, Any], pins: dict[str, Any]) -> list[dict[str, Any]]:
|
||
"""发布时冻结的钉住项：按库里的那一版补上级别（不信 spec 里自报的）。"""
|
||
out = []
|
||
for alias, pin in pins_by_alias(spec).items():
|
||
state = pins.get(alias)
|
||
out.append({**pin, "level": state.level if state is not None else None})
|
||
return out
|