246 lines
9.8 KiB
Python
246 lines
9.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
pbl_blueprint · M1b 模板「平台公共部分」（tenant_id NULL）读写与解析
|
||
|
||
规则（本文件是 M1b 平台公共模板的唯一实现口径）
|
||
1. `pbl_template.tenant_id IS NULL` => 平台公共模板（platform scope）：
|
||
- 对**全部租户只读可见**（list/get 自动并入）；
|
||
- 仅平台管理员可写/归档（`is_platform_admin(ctx)` 门禁，非管理员写 -> PBL_TPL_PLATFORM_FORBIDDEN）；
|
||
- 租户**不得**修改/删除平台公共行，只能 fork 成租户私有模板（新 template_code 或同 code 的租户行）。
|
||
2. 解析优先级：租户私有 > 平台公共（同 template_code + template_version 时租户行覆盖平台行）。
|
||
3. 唯一性：MySQL 8 函数索引 uk((IFNULL(tenant_id,'__PLATFORM__')), template_code, template_version)，
|
||
保证平台公共行同样受唯一约束（NULL 不参与 MySQL 唯一索引比较的坑由函数索引消除）。
|
||
4. 实例化平台公共模板时：产出的蓝图/子对象/实例化日志 **tenant_id 一律为发起租户**（平台公共只存在于模板层，
|
||
不产生跨租户数据）；`pbl_template_instance_log.template_scope='platform'` 留痕来源。
|
||
5. 离线兜底模板（offline_flag='Y'）由 build.sh 种子落库为**平台公共行**，确保 test 环境冷启动即可用。
|
||
|
||
所有查询 tenant 维度打头；无 FOREIGN KEY；不直写 M1a 表。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from .models.pbl_template import (
|
||
PLATFORM_TENANT_KEY,
|
||
is_platform_scope,
|
||
normalize_tenant_key,
|
||
)
|
||
|
||
#: 平台公共模板可见性范围标识
|
||
SCOPE_PLATFORM = "platform"
|
||
SCOPE_TENANT = "tenant"
|
||
|
||
#: 解析顺序：租户私有优先，平台公共兜底
|
||
RESOLVE_ORDER = (SCOPE_TENANT, SCOPE_PLATFORM)
|
||
|
||
|
||
class TemplateError(Exception):
|
||
"""模板域业务异常基类（错误码与 M1a errors.py 同口径）。"""
|
||
|
||
code = "PBL_TPL_ERROR"
|
||
|
||
def __init__(self, message, **detail):
|
||
super(TemplateError, self).__init__(message)
|
||
self.message = message
|
||
self.detail = detail
|
||
|
||
|
||
class TemplateNotFound(TemplateError):
|
||
code = "PBL_TEMPLATE_NOT_FOUND"
|
||
|
||
|
||
class TemplatePlatformForbidden(TemplateError):
|
||
code = "PBL_TPL_PLATFORM_FORBIDDEN"
|
||
|
||
|
||
class TemplateSchemaInvalid(TemplateError):
|
||
code = "PBL_TPL_SCHEMA_INVALID"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 租户上下文（与 M1a tenant.py 同口径：tenant_id 强制打头，缺失即 fail-closed）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def require_tenant(tenant_ctx):
|
||
"""
|
||
取 tenant_id；缺失抛 PBL_TENANT_MISSING（fail-closed，不静默、不默认全局）。
|
||
平台管理员操作平台公共模板时同样必须携带 tenant_ctx（可为平台租户）。
|
||
"""
|
||
tenant_id = None
|
||
if isinstance(tenant_ctx, dict):
|
||
tenant_id = tenant_ctx.get("tenant_id")
|
||
else:
|
||
tenant_id = getattr(tenant_ctx, "tenant_id", None)
|
||
if tenant_id in (None, ""):
|
||
err = TemplateError("tenant_id 缺失")
|
||
err.code = "PBL_TENANT_MISSING"
|
||
raise err
|
||
return str(tenant_id)
|
||
|
||
|
||
def is_platform_admin(tenant_ctx):
|
||
"""
|
||
是否平台管理员（可写平台公共模板）。
|
||
兼容三种上下文表达：is_platform_admin / platform_admin / role 列表含 'platform_admin'。
|
||
"""
|
||
if isinstance(tenant_ctx, dict):
|
||
ctx = tenant_ctx
|
||
else:
|
||
ctx = getattr(tenant_ctx, "__dict__", {}) or {}
|
||
if ctx.get("is_platform_admin") or ctx.get("platform_admin"):
|
||
return True
|
||
roles = ctx.get("roles") or []
|
||
if isinstance(roles, str):
|
||
roles = [roles]
|
||
return any(str(r) in ("platform_admin", "owner.platform", "admin.platform") for r in roles)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# SQL 片段（tenant 维度打头；平台公共 = tenant_id IS NULL）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def scope_where(tenant_id, include_platform=True, alias=""):
|
||
"""
|
||
可见范围 WHERE 片段：租户私有 + 平台公共。
|
||
|
||
:return: (sql_fragment, params)
|
||
"""
|
||
col = "%stenant_id" % alias
|
||
if include_platform:
|
||
return "(%s = %%s OR %s IS NULL)" % (col, col), [tenant_id]
|
||
return "%s = %%s" % col, [tenant_id]
|
||
|
||
|
||
def build_list_sql(tenant_id, filters=None, include_platform=True):
|
||
"""
|
||
模板库浏览与筛选（F-CP-0x）：租户私有优先排序，平台公共兜底。
|
||
filters: {subject, grade, tpl_status, offline_flag, template_name(like), scope}
|
||
"""
|
||
filters = filters or {}
|
||
include_platform = include_platform and filters.get("scope") != SCOPE_TENANT
|
||
where, params = scope_where(tenant_id, include_platform=include_platform)
|
||
conds = [where]
|
||
if filters.get("scope") == SCOPE_PLATFORM:
|
||
conds = ["tenant_id IS NULL"]
|
||
params = []
|
||
for key, col in (("subject", "subject"), ("grade", "grade"),
|
||
("tpl_status", "tpl_status"), ("offline_flag", "offline_flag")):
|
||
if filters.get(key) not in (None, ""):
|
||
conds.append("%s = %%s" % col)
|
||
params.append(filters[key])
|
||
if filters.get("template_name"):
|
||
conds.append("template_name LIKE %s")
|
||
params.append("%%%s%%" % filters["template_name"])
|
||
if not filters.get("tpl_status"):
|
||
conds.append("tpl_status <> 'archived'")
|
||
sql = (
|
||
"SELECT id, tenant_id, template_code, template_version, template_name, subject, grade, "
|
||
"tpl_hash, offline_flag, tpl_status, create_time, update_time, "
|
||
"CASE WHEN tenant_id IS NULL THEN '%s' ELSE '%s' END AS template_scope "
|
||
"FROM pbl_template WHERE %s "
|
||
"ORDER BY (tenant_id IS NULL) ASC, template_code ASC, template_version DESC"
|
||
% (SCOPE_PLATFORM, SCOPE_TENANT, " AND ".join(conds))
|
||
)
|
||
return sql, params
|
||
|
||
|
||
def build_resolve_sql(tenant_id, template_code, template_version=None):
|
||
"""
|
||
模板解析（租户私有 > 平台公共）：一次查询取回候选行，由 pick_template 选定。
|
||
template_version 为 None 时取各 scope 的最新 active 版本。
|
||
"""
|
||
sql = (
|
||
"SELECT id, tenant_id, template_code, template_version, template_name, subject, grade, "
|
||
"tpl_json, tpl_hash, offline_flag, tpl_status, create_time, update_time "
|
||
"FROM pbl_template "
|
||
"WHERE (tenant_id = %s OR tenant_id IS NULL) AND template_code = %s "
|
||
"AND tpl_status = 'active' "
|
||
)
|
||
params = [tenant_id, template_code]
|
||
if template_version is not None:
|
||
sql += "AND template_version = %s "
|
||
params.append(int(template_version))
|
||
sql += ("ORDER BY (tenant_id IS NULL) ASC, template_version DESC LIMIT 2")
|
||
return sql, params
|
||
|
||
|
||
def pick_template(rows):
|
||
"""
|
||
从候选行按 RESOLVE_ORDER 选定模板行；无候选抛 PBL_TEMPLATE_NOT_FOUND。
|
||
|
||
:param rows: build_resolve_sql 的结果（已按 租户优先 + 版本倒序 排序）
|
||
:return: (row, scope) scope ∈ {'tenant','platform'}
|
||
"""
|
||
if not rows:
|
||
raise TemplateNotFound("模板不存在或已归档")
|
||
tenant_rows = [r for r in rows if not is_platform_scope(_get(r, "tenant_id"))]
|
||
platform_rows = [r for r in rows if is_platform_scope(_get(r, "tenant_id"))]
|
||
buckets = {SCOPE_TENANT: tenant_rows, SCOPE_PLATFORM: platform_rows}
|
||
for scope in RESOLVE_ORDER:
|
||
if buckets.get(scope):
|
||
return buckets[scope][0], scope
|
||
raise TemplateNotFound("模板不存在或已归档")
|
||
|
||
|
||
def build_offline_sql(tenant_id, subject=None, grade=None):
|
||
"""
|
||
离线兜底模板选取（M1b-annex §4）：offline_flag='Y' 且 active，租户私有优先、平台公共兜底。
|
||
"""
|
||
sql = (
|
||
"SELECT id, tenant_id, template_code, template_version, template_name, subject, grade, "
|
||
"tpl_json, tpl_hash, offline_flag, tpl_status "
|
||
"FROM pbl_template "
|
||
"WHERE (tenant_id = %s OR tenant_id IS NULL) AND offline_flag = 'Y' "
|
||
"AND tpl_status = 'active' "
|
||
)
|
||
params = [tenant_id]
|
||
if subject:
|
||
sql += "AND (subject = %s OR subject IS NULL) "
|
||
params.append(subject)
|
||
if grade:
|
||
sql += "AND (grade = %s OR grade IS NULL) "
|
||
params.append(grade)
|
||
sql += "ORDER BY (tenant_id IS NULL) ASC, template_version DESC LIMIT 1"
|
||
return sql, params
|
||
|
||
|
||
def build_platform_write_guard(tenant_ctx, target_tenant_id):
|
||
"""
|
||
写入门禁：写平台公共行（target_tenant_id 为 None/'')必须平台管理员。
|
||
|
||
:return: scope（'platform'/'tenant'）
|
||
:raises TemplatePlatformForbidden: 非平台管理员写平台公共模板
|
||
"""
|
||
if is_platform_scope(target_tenant_id):
|
||
if not is_platform_admin(tenant_ctx):
|
||
raise TemplatePlatformForbidden(
|
||
"平台公共模板（tenant_id NULL）仅平台管理员可写/归档")
|
||
return SCOPE_PLATFORM
|
||
# 租户行：tenant_ctx.tenant_id 必须与目标一致（禁止跨租户写）
|
||
ctx_tenant = require_tenant(tenant_ctx)
|
||
if target_tenant_id and str(target_tenant_id) != ctx_tenant:
|
||
raise TemplatePlatformForbidden(
|
||
"跨租户写模板被拒绝: ctx=%s target=%s" % (ctx_tenant, target_tenant_id))
|
||
return SCOPE_TENANT
|
||
|
||
|
||
def tenant_key_for_insert(target_tenant_id):
|
||
"""
|
||
插入时的归一键（MySQL 5.7 回退列 tenant_key 用；MySQL 8 函数索引下无需写该列）。
|
||
"""
|
||
return normalize_tenant_key(target_tenant_id)
|
||
|
||
|
||
def _get(row, key, default=None):
|
||
if isinstance(row, dict):
|
||
return row.get(key, default)
|
||
return getattr(row, key, default)
|
||
|
||
|
||
__all__ = [
|
||
"PLATFORM_TENANT_KEY", "SCOPE_PLATFORM", "SCOPE_TENANT", "RESOLVE_ORDER",
|
||
"TemplateError", "TemplateNotFound", "TemplatePlatformForbidden", "TemplateSchemaInvalid",
|
||
"require_tenant", "is_platform_admin", "scope_where", "build_list_sql",
|
||
"build_resolve_sql", "pick_template", "build_offline_sql",
|
||
"build_platform_write_guard", "tenant_key_for_insert",
|
||
]
|