213 lines
7.4 KiB
Python
213 lines
7.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""pbl_domain_ext 公共内核：常量、参数校验、租户上下文、审计字段。
|
||
|
||
设计依据：projects/pbls/docs/01-design/data-model.md §J1（基础域薄扩展）。
|
||
铁律：
|
||
1. 只维护 1 张关联表 ``pbl_domain_ref``，**不改** world/scene/entity 三张基表结构；
|
||
2. 所有读写 ``tenant_id`` 强制打头，缺失即抛 ``PBL_DE_TENANT_MISSING``；
|
||
3. 扩展字段权威名 ``ext_json``（LONGTEXT / 抽象类型 text），禁止写成旧名 ``ext``；
|
||
4. 主键统一 ``appPublic.uniqueID.getID()`` 生成（crud-spec Pitfall 11），禁止 uuid4；
|
||
5. 审计列名统一 ``creator_id / updater_id``（与 models/pbl_domain_ref.json 同名同型）。
|
||
"""
|
||
|
||
import json
|
||
import time
|
||
|
||
from .errors import PblDomainExtError
|
||
|
||
# 主键生成：平台标准 appPublic.uniqueID.getID()（QC#6）。
|
||
# 离线单测/无 appPublic 环境降级为 32 位 hex，保证契约函数可测（列宽 str(32) 一致）。
|
||
try: # pragma: no cover - 取决于宿主环境
|
||
from appPublic.uniqueID import getID as _platform_get_id
|
||
except Exception: # noqa: BLE001
|
||
_platform_get_id = None
|
||
|
||
__all__ = [
|
||
"TABLE", "BASE_TABLES", "REF_TYPES", "BIND_STATES", "EXT_FIELD",
|
||
"LIST_FIELDS", "AUDIT_FIELDS", "ALL_COLUMNS", "now_str", "gen_id",
|
||
"require_tenant", "check_ref_type", "check_bind_state", "normalize_ext_json",
|
||
"audit_on_create", "audit_on_update", "page_args", "as_text", "as_list",
|
||
]
|
||
|
||
#: 本模块唯一自有表
|
||
TABLE = "pbl_domain_ref"
|
||
|
||
#: 基础域基表（只读引用，禁止写入/改结构）
|
||
BASE_TABLES = {
|
||
"world": "world",
|
||
"scene": "scene",
|
||
"entity": "entity",
|
||
}
|
||
|
||
#: ref_type 合法枚举
|
||
REF_TYPES = ("world", "scene", "entity")
|
||
|
||
#: bind_state 合法枚举
|
||
BIND_STATES = ("bound", "unbound")
|
||
|
||
#: 扩展字段权威名（设计 §J1）——三处（models/sql/api）必须同名同型
|
||
EXT_FIELD = "ext_json"
|
||
|
||
#: 审计列名（QC#1：与 models/pbl_domain_ref.json fields 严格同名）
|
||
AUDIT_FIELDS = ["creator_id", "created_at", "updater_id", "updated_at"]
|
||
|
||
#: 列表返回字段白名单（避免 SELECT * 带出无关列）
|
||
#: 与 models/pbl_domain_ref.json 的 fields 一一对应（tests/test_models_contract.py 机械校验）
|
||
LIST_FIELDS = [
|
||
"id", "tenant_id", "ref_type", "ref_id", "ref_code", "ref_name",
|
||
"blueprint_id", "team_id", "class_id", "bind_state", "bind_at",
|
||
EXT_FIELD, "is_deleted",
|
||
"creator_id", "created_at", "updater_id", "updated_at",
|
||
]
|
||
|
||
#: 表全部列（= LIST_FIELDS，本表无隐藏列；供 DDL/契约一致性校验）
|
||
ALL_COLUMNS = list(LIST_FIELDS)
|
||
|
||
_MAX_PAGE_SIZE = 500
|
||
|
||
|
||
def now_str():
|
||
"""统一时间串格式 ``YYYY-MM-DD HH:MM:SS``。"""
|
||
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
|
||
|
||
|
||
def gen_id():
|
||
"""生成 32 位主键：优先平台标准 ``appPublic.uniqueID.getID()``（QC#6）。"""
|
||
if _platform_get_id is not None:
|
||
try:
|
||
val = _platform_get_id()
|
||
val = as_text(val).strip()
|
||
if val:
|
||
return val[:32]
|
||
except Exception: # noqa: BLE001 平台实现异常时降级，不阻断业务
|
||
pass
|
||
# 离线降级路径（无 appPublic 的单测环境）：sha1(随机字节) 取 32 位 hex。
|
||
# 禁止 uuid4 生成主键（crud-spec Pitfall 11）——降级实现同样不得引入 uuid。
|
||
import hashlib
|
||
import os as _os
|
||
return hashlib.sha1(_os.urandom(24)).hexdigest()[:32]
|
||
|
||
|
||
def as_text(value, default=""):
|
||
"""安全转字符串（None -> default）。"""
|
||
if value is None:
|
||
return default
|
||
if isinstance(value, bytes):
|
||
try:
|
||
return value.decode("utf-8")
|
||
except UnicodeDecodeError:
|
||
return value.decode("utf-8", "replace")
|
||
if isinstance(value, str):
|
||
return value
|
||
return str(value)
|
||
|
||
|
||
def as_list(value):
|
||
"""把 None/单值/列表统一成 list。"""
|
||
if value is None or value == "":
|
||
return []
|
||
if isinstance(value, (list, tuple, set)):
|
||
return list(value)
|
||
return [value]
|
||
|
||
|
||
def require_tenant(params):
|
||
"""取并校验 tenant_id（所有契约第一道门禁）。
|
||
|
||
:return: 去空白后的 tenant_id
|
||
:raises PblDomainExtError: PBL_DE_TENANT_MISSING
|
||
"""
|
||
tenant_id = ""
|
||
if isinstance(params, dict):
|
||
tenant_id = as_text(params.get("tenant_id")).strip()
|
||
if not tenant_id or len(tenant_id) > 32:
|
||
raise PblDomainExtError("PBL_DE_TENANT_MISSING",
|
||
detail="tenant_id=%r" % (tenant_id,))
|
||
return tenant_id
|
||
|
||
|
||
def check_ref_type(ref_type, allow_empty=False):
|
||
"""校验 ref_type ∈ REF_TYPES。"""
|
||
ref_type = as_text(ref_type).strip().lower()
|
||
if not ref_type:
|
||
if allow_empty:
|
||
return ""
|
||
raise PblDomainExtError("PBL_DE_REF_TYPE_INVALID", detail="ref_type 为空")
|
||
if ref_type not in REF_TYPES:
|
||
raise PblDomainExtError("PBL_DE_REF_TYPE_INVALID",
|
||
detail="ref_type=%r 允许=%s" % (ref_type, list(REF_TYPES)))
|
||
return ref_type
|
||
|
||
|
||
def check_bind_state(bind_state, allow_empty=True):
|
||
"""校验 bind_state ∈ BIND_STATES。"""
|
||
bind_state = as_text(bind_state).strip().lower()
|
||
if not bind_state:
|
||
if allow_empty:
|
||
return ""
|
||
return BIND_STATES[0]
|
||
if bind_state not in BIND_STATES:
|
||
raise PblDomainExtError("PBL_DE_STATE_INVALID",
|
||
detail="bind_state=%r" % (bind_state,))
|
||
return bind_state
|
||
|
||
|
||
def normalize_ext_json(raw):
|
||
"""把入参 ext_json 归一为 **合法 JSON 文本**（列类型 LONGTEXT，存字符串）。
|
||
|
||
接受：None/""（-> ""）、dict/list（-> json.dumps）、已是 JSON 文本的 str。
|
||
非法 JSON 文本抛 ``PBL_DE_EXT_JSON_INVALID``。
|
||
"""
|
||
if raw is None:
|
||
return ""
|
||
if isinstance(raw, (dict, list)):
|
||
return json.dumps(raw, ensure_ascii=False, sort_keys=True)
|
||
text = as_text(raw).strip()
|
||
if not text:
|
||
return ""
|
||
try:
|
||
parsed = json.loads(text)
|
||
except (ValueError, TypeError):
|
||
raise PblDomainExtError("PBL_DE_EXT_JSON_INVALID", detail=text[:120])
|
||
if not isinstance(parsed, (dict, list)):
|
||
raise PblDomainExtError("PBL_DE_EXT_JSON_INVALID",
|
||
detail="ext_json 顶层必须是 object/array")
|
||
return json.dumps(parsed, ensure_ascii=False, sort_keys=True)
|
||
|
||
|
||
def audit_on_create(row, operator_id):
|
||
"""补创建审计四件套（就地修改并返回 row）。列名 = creator_id/updater_id（QC#1）。"""
|
||
ts = now_str()
|
||
row["creator_id"] = as_text(operator_id)
|
||
row["created_at"] = ts
|
||
row["updater_id"] = as_text(operator_id)
|
||
row["updated_at"] = ts
|
||
row.setdefault("is_deleted", 0)
|
||
return row
|
||
|
||
|
||
def audit_on_update(row, operator_id):
|
||
"""补更新审计（就地修改并返回 row）。"""
|
||
row["updater_id"] = as_text(operator_id)
|
||
row["updated_at"] = now_str()
|
||
return row
|
||
|
||
|
||
def page_args(params):
|
||
"""解析分页参数，返回 (page, page_size, offset)。越界自动收敛。"""
|
||
try:
|
||
page = int(params.get("page") or 1)
|
||
except (TypeError, ValueError):
|
||
page = 1
|
||
try:
|
||
page_size = int(params.get("page_size") or params.get("limit") or 20)
|
||
except (TypeError, ValueError):
|
||
page_size = 20
|
||
if page < 1:
|
||
page = 1
|
||
if page_size < 1:
|
||
page_size = 1
|
||
if page_size > _MAX_PAGE_SIZE:
|
||
page_size = _MAX_PAGE_SIZE
|
||
return page, page_size, (page - 1) * page_size
|