feat(world): W-01 世界管理模块 — world 表 CRUD/列表查询/自动编码/mode 白名单与流转校验/编码字典幂等落库

This commit is contained in:
develop 2026-08-29 12:35:16 +08:00
parent ada95ae1d8
commit 22b18f556c
23 changed files with 970 additions and 731 deletions

8
.gitignore vendored
View File

@ -1,14 +1,8 @@
# Python
__pycache__/
*.py[cod]
*.egg-info/
build/
dist/
*.egg-info/
*.egg
# 生成物
models/mysql.ddl.sql
wwwroot/world/ # xls2ui 生成的 CRUD 目录build 产物)
*.swp
*.swo
.DS_Store

View File

@ -1,45 +1,46 @@
# world 世界管理模块
世界定义与管理模块W-01 世界管理),平台入口域根模块。
# world 世界定义与管理模块W-01 世界管理)
## 功能
- world 表 CRUD创建自动编码 W+时间戳+随机4位
- 列表查询sqlPaging 分页 + 复合索引org_id+mode+created_at过滤返回 {list, total}
- world.mode 白名单硬校验draft / active / paused / archived
- 模式流转规则校验draft→active⇄paused、active/paused→archived、archived→active
- 编码全局唯一:唯一索引 + 应用层前置校验 + 冲突重试3 次)
- world_mode / world_type 编码字典经 init/data.json 幂等落库appcodes + appcodes_kv
- world 表 CRUD创建/更新/删除/查询)
- 分页列表查询sqlPaging + 复合索引 idx_world_list
- 创建自动编码W + 时间戳 + 随机4位全局唯一冲突重试 3 次)
- mode 白名单硬校验draft/active/paused/archived与流转规则校验
- world_mode / world_type 编码字典幂等落库appcodes + appcodes_kv
## 数据表
- `world`世界定义表models/world.json四段式 summary/fields/indexes/codes
- 编码字典:`world_mode`draft/active/paused/archived`world_type`game/simulation/education/business/social
`world`models/world.json 四段式 summary/fields/indexes/codes
- id VARCHAR(32) PKstr 32getID() 生成,禁 uuid
- code VARCHAR(64) 唯一(唯一索引 idx_world_code
- name VARCHAR(255) NOT NULL
- description TEXT
- mode VARCHAR(32) NOT NULL DEFAULT 'draft'
- world_type VARCHAR(32)
- owner_id / org_id VARCHAR(32)
- created_at / updated_at TIMESTAMP
- 复合索引 idx_world_list(org_id, mode, created_at)
## 接口
REST 统一前缀 `/api/*`,错误结构 `{code, message, field, detail}`,分页 `{list, total}`
- `POST/GET /world/api/create_world.dspy` 创建自动编码、mode 校验)
- `POST /world/api/world_update.dspy` 更新mode 流转校验、code 唯一校验)
- `POST /world/api/world_delete.dspy` 删除
- `GET /world/api/get_world.dspy` 单条查询
- `GET /world/api/list_worlds.dspy` 列表(分页 + 过滤)
- `POST /world/api/set_world_mode.dspy` 模式切换
- `GET /world/api/get_search_status.dspy` / `get_search_world_type.dspy` 下拉数据源
## 安装集成
1. 模块仓库克隆到宿主应用 pkgs/`pip install .`
2. 宿主应用 `app/{app}.py` 引入 `from world.init import load_world` 并在 init() 调用 `load_world()`
3. 宿主应用定义 `get_module_dbname('world')` 映射到实际库名(禁止模块内硬编码 DBNAME
4. `scripts/load_path.py` 注册 RBAC 路径(禁通配符)
5. 构建:`bash build.sh`(生成 DDL、CRUD UI、链接 wwwroot
## 安装与集成
1. `pip install .`pyproject.toml依赖 sqlor / bricks_for_python
2. 宿主应用 `from world.init import load_world`init() 中 `load_world()` 挂载
3. 建表build.sh 中 json2ddl 依据 models/world.json 生成 DDL编码字典经 init/data.json 幂等落库
4. RBAC运行 `scripts/load_path.py`(显式路径注册,禁通配符)
5. 前端wwwroot/index.ui入口、menu.ui菜单、CRUD 由 json/world.json 经 xls2ui 生成
## 关键接口(经 load_world() 注册)
- `create_world(params)` → 自动编码mode 默认 draft非法 mode/流转 100% 拦截;编码冲突重试 3 次
- `update_world(params)` → 部分更新mode 变更走流转校验code 变更走全局唯一校验
- `delete_world(params)` / `get_world(params)`
- `list_worlds(params)` → 分页 {list, total, page, page_size}
- `set_world_mode(params)` → 模式切换严格校验draft→active⇄paused、active/paused→archived、archived→active
- `get_world_mode_options()` / `get_world_type_options()` → 下拉数据源 [{value, text}]
## 目录结构
```
world/
├── world/ # Python 包world.py 业务逻辑 / init.py 注册 / __init__.py 导出)
├── wwwroot/ # index.ui、menu.ui、api/*.dspy
├── models/world.json # 表定义(四段式)
├── json/world.json # CRUD 定义tblname+params
├── init/data.json # 编码字典种子数据
├── scripts/load_path.py # RBAC 注册
├── skill/SKILL.md # 模块技能文档
└── pyproject.toml # 打包
```
REST 前缀 `/api/*`create_world.dspy / world_update.dspy / world_delete.dspy /
get_world.dspy / list_worlds.dspy / set_world_mode.dspy /
get_search_status.dspy / get_search_world_type.dspy
## 错误与分页结构
- 错误:`{code, message, field, detail}`
- 分页:`{list, total, page, page_size}`
## 依赖
sqlorDBPools/sor.C/U/D/R/sqlPaging/sqlExe、ahserverServerEnv、appbaseappcodes/appcodes_kv

View File

@ -1,18 +1,37 @@
#!/bin/bash
# world 模块构建脚本四步安装①DDL 生成CRUD UI 生成编码字典落库RBAC 注册)
set -e
cd "$(dirname "$0")"
#!/usr/bin/env bash
# world 模块构建脚本:生成 DDL、生成 CRUD UI、链接 wwwroot 到宿主应用
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
MOD=world
echo "[world] ① 生成 DDL (json2ddl)"
json2ddl mysql ./models > models/mysql.ddl.sql 2>/dev/null || echo " json2ddl 不可用,跳过(部署时由应用 build.sh 统一执行)"
# 定位宿主 Sage 根目录
SAGE_ROOT=""
for candidate in "$SCRIPT_DIR/../.." "$HOME/repos/sage" "$HOME/sage"; do
if [ -d "$candidate/wwwroot" ] && [ -d "$candidate/py3/bin" ]; then
SAGE_ROOT="$(cd "$candidate" && pwd)"
break
fi
done
if [ -z "$SAGE_ROOT" ]; then
echo "ERROR: sage root not found" >&2
exit 1
fi
PY="$SAGE_ROOT/py3/bin"
echo "[world] ② 生成 CRUD UI (xls2ui)"
xls2ui -m ../models -o ../wwwroot world ./json/*.json 2>/dev/null || echo " xls2ui 不可用,跳过(部署时由应用 build.sh 统一执行)"
echo "==> 1/4 生成 DDL"
if [ -d "$SCRIPT_DIR/models" ]; then
"$PY/python3" "$PY/json2ddl" mysql "$SCRIPT_DIR/models" > "$SCRIPT_DIR/models/mysql.ddl.sql" 2>/dev/null \
|| "$PY/json2ddl" mysql "$SCRIPT_DIR/models" > "$SCRIPT_DIR/models/mysql.ddl.sql"
fi
echo "[world] ③ 编码字典落库init/data.json应用初始化时幂等执行"
echo " 由宿主应用初始化流程执行 appcodes/appcodes_kv 幂等插入"
echo "==> 2/4 生成 CRUD UIxls2ui"
if [ -d "$SCRIPT_DIR/json" ]; then
(cd "$SCRIPT_DIR" && "$PY/python3" "$PY/xls2ui" -m "$SCRIPT_DIR/models" -o "$SCRIPT_DIR/wwwroot" "$MOD" json/*.json) \
|| (cd "$SCRIPT_DIR" && "$PY/xls2ui" -m "$SCRIPT_DIR/models" -o "$SCRIPT_DIR/wwwroot" "$MOD" json/*.json)
fi
echo "[world] ④ RBAC 注册"
python3 scripts/load_path.py || echo " load_path.py 需在 Sage 环境执行"
echo "==> 3/4 链接 wwwroot 到宿主应用"
mkdir -p "$SAGE_ROOT/wwwroot/$MOD"
ln -sfn "$SCRIPT_DIR/wwwroot" "$SAGE_ROOT/wwwroot/$MOD"
echo "[world] build done"
echo "==> 4/4 完成world 模块构建成功"

46
docs/work-log.md Normal file
View File

@ -0,0 +1,46 @@
# Work Log — 2025-01-xx
## Scope / Background
- 任务W-01 世界管理world 模块new_devtask_id gvpV_sE3EQghCY2yNovFT元景项目初始迭代
- 仓库:机构 modules/world/
- 目标world 表 CRUD + 列表查询(sqlPaging+复合索引) + 创建自动编码 + mode 白名单/流转校验 +
world_mode 编码字典幂等落库 + /api/* REST + 统一错误结构 + 分页结构
## Timeline / Commits
git log --oneline 输出见下方验证区)
## Key Technical Decisions
1. mode 校验集中在 world.pyMODE_WHITELIST + MODE_TRANSITIONS所有入口create/update/set_world_mode统一拦截。
2. 自动编码 W+时间戳+随机4位唯一索引兜底 + Duplicate key 重试 3 次。
3. 列表查询count 与数据分离、显式列清单(排除 TEXT、idx_world_list(org_id,mode,created_at) 复合索引。
4. 下拉数据源从 appcodes_kv 读world_mode/world_type字典不可用回退英文标签避免硬编码中文。
5. init.py 单数/复数双注册create_world/create_worlds满足 CRUD 框架 dspy 包装器调用约定。
## VerificationQC 硬证据)
```
$ python3 -m py_compile world/world.py world/init.py world/__init__.py scripts/load_path.py
(无输出 = 通过)
$ grep -rn "^import\|^from" wwwroot/ --include='*.dspy'
(无输出 = 通过dspy 零 import
$ grep -rn "print\|uuid" wwwroot/ --include='*.dspy'
(无输出 = 通过)
$ python3 -c "import json; \
json.load(open('init/data.json')); json.load(open('models/world.json')); json.load(open('json/world.json'))"
(无输出 = 通过)
$ python3 -c "from setuptools import find_packages; print(find_packages(where='.'))"
['world'] # 仅包目录,根目录无 __init__.py不会误打包
$ git log --oneline
<提交记录见交付件>
```
## Environment-Limited Checks
- 本环境无 MySQL/Sage 运行时,无法现场起服务 curl 验证;已用 py_compile + JSON load +
dspy grep 审计 + 打包 find_packages 做静态闭环验证。部署时按 build.sh 生成 DDL/CRUD UI 并落库。
## Current State
- 分支/提交:见 git log模块代码已本地提交QC 核验硬证据)。

View File

@ -1,25 +1,52 @@
{
"appcodes": [
"appcodes": [
{
"parentid": "world_mode",
"parentname": "世界模式",
"items": [
{
"parentid": "world_mode",
"parentname": "世界模式",
"items": [
{"k": "draft", "v": "草稿"},
{"k": "active", "v": "运行中"},
{"k": "paused", "v": "已暂停"},
{"k": "archived", "v": "已归档"}
]
"k": "draft",
"v": "草稿"
},
{
"parentid": "world_type",
"parentname": "世界类型",
"items": [
{"k": "game", "v": "游戏世界"},
{"k": "simulation", "v": "仿真世界"},
{"k": "education", "v": "教育世界"},
{"k": "business", "v": "商业世界"},
{"k": "social", "v": "社交世界"}
]
"k": "active",
"v": "运行中"
},
{
"k": "paused",
"v": "已暂停"
},
{
"k": "archived",
"v": "已归档"
}
]
]
},
{
"parentid": "world_type",
"parentname": "世界类型",
"items": [
{
"k": "game",
"v": "游戏"
},
{
"k": "simulation",
"v": "仿真"
},
{
"k": "education",
"v": "教育"
},
{
"k": "business",
"v": "商业"
},
{
"k": "social",
"v": "社交"
}
]
}
]
}

View File

@ -1,41 +1,112 @@
{
"tblname": "world",
"title": "世界管理",
"params": {
"sortby": ["created_at desc"],
"new_data_url": "{{entire_url('/world/api/create_world.dspy')}}",
"update_data_url": "{{entire_url('/world/api/world_update.dspy')}}",
"delete_data_url": "{{entire_url('/world/api/world_delete.dspy')}}",
"editable": {
"new_data_url": "{{entire_url('../api/create_world.dspy')}}",
"update_data_url": "{{entire_url('../api/world_update.dspy')}}",
"delete_data_url": "{{entire_url('../api/world_delete.dspy')}}"
"tblname": "world",
"params": {
"browserfields": {
"title": "世界列表",
"fields": [
{
"name": "id",
"title": "ID",
"hidden": true
},
"data_filter": {
"AND": [
{"field": "name", "op": "LIKE", "var": "name_input"},
{"field": "mode", "op": "=", "var": "mode_input"},
{"field": "world_type", "op": "=", "var": "world_type_input"}
]
{
"name": "code",
"title": "世界编码",
"width": 130
},
"filter_labels": {
"name_input": "世界名称",
"mode_input": "世界模式",
"world_type_input": "世界类型"
{
"name": "name",
"title": "世界名称",
"width": 200
},
"browserfields": {
"exclouded": ["id"],
"alters": {
"mode": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_search_status.dspy')}}"
},
"world_type": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_search_world_type.dspy')}}"
}
}
{
"name": "mode",
"title": "世界模式",
"width": 100
},
"editexclouded": ["id", "code", "owner_id", "org_id", "created_at", "updated_at"]
}
{
"name": "world_type",
"title": "世界类型",
"width": 100
},
{
"name": "owner_id",
"title": "所属用户",
"width": 100,
"hidden": true
},
{
"name": "org_id",
"title": "所属组织",
"width": 100,
"hidden": true
},
{
"name": "created_at",
"title": "创建时间",
"width": 160
},
{
"name": "updated_at",
"title": "更新时间",
"width": 160
}
],
"sorts": [
{
"field": "created_at",
"type": "desc"
}
],
"dataurl": "{{entire_url('/world/api/list_worlds.dspy')}}",
"alters": [
{
"field": "mode",
"uitype": "code",
"dataurl": "{{entire_url('/world/api/get_search_status.dspy')}}"
},
{
"field": "world_type",
"uitype": "code",
"dataurl": "{{entire_url('/world/api/get_search_world_type.dspy')}}"
}
]
},
"editable": {
"fields": [
{
"name": "code",
"title": "世界编码",
"uitype": "text",
"required": false
},
{
"name": "name",
"title": "世界名称",
"uitype": "text",
"required": true
},
{
"name": "description",
"title": "世界描述",
"uitype": "textarea"
},
{
"name": "mode",
"title": "世界模式",
"uitype": "code",
"dataurl": "{{entire_url('/world/api/get_search_status.dspy')}}"
},
{
"name": "world_type",
"title": "世界类型",
"uitype": "code",
"dataurl": "{{entire_url('/world/api/get_search_world_type.dspy')}}"
}
]
},
"new_data_url": "{{entire_url('/world/api/create_world.dspy')}}",
"update_data_url": "{{entire_url('/world/api/world_update.dspy')}}",
"delete_data_url": "{{entire_url('/world/api/world_delete.dspy')}}"
}
}

View File

@ -1,104 +1,105 @@
{
"summary": [
{
"name": "world",
"title": "世界定义表",
"primary": ["id"],
"catelog": "entity"
}
],
"fields": [
{
"name": "id",
"title": "主键ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "code",
"title": "世界编码",
"type": "str",
"length": 64,
"nullable": "no"
},
{
"name": "name",
"title": "世界名称",
"type": "str",
"length": 255,
"nullable": "no"
},
{
"name": "description",
"title": "世界描述",
"type": "text"
},
{
"name": "mode",
"title": "世界模式",
"type": "str",
"length": 32,
"nullable": "no",
"default": "draft"
},
{
"name": "world_type",
"title": "世界类型",
"type": "str",
"length": 32
},
{
"name": "owner_id",
"title": "创建人",
"type": "str",
"length": 32
},
{
"name": "org_id",
"title": "所属机构",
"type": "str",
"length": 32,
"default": "0"
},
{
"name": "created_at",
"title": "创建时间",
"type": "timestamp",
"nullable": "no"
},
{
"name": "updated_at",
"title": "更新时间",
"type": "timestamp"
}
],
"indexes": [
{
"name": "idx_world_code",
"idxtype": "unique",
"idxfields": ["code"]
},
{
"name": "idx_world_list",
"idxtype": "index",
"idxfields": ["org_id", "mode", "created_at"]
}
],
"codes": [
{
"field": "mode",
"table": "appcodes_kv",
"valuefield": "k",
"textfield": "v",
"cond": "parentid='world_mode'"
},
{
"field": "world_type",
"table": "appcodes_kv",
"valuefield": "k",
"textfield": "v",
"cond": "parentid='world_type'"
}
]
"summary": [
{
"name": "world",
"title": "世界定义表",
"primary": ["id"],
"catelog": "entity"
}
],
"fields": [
{
"name": "id",
"title": "主键ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "code",
"title": "世界编码",
"type": "str",
"length": 64,
"nullable": "no"
},
{
"name": "name",
"title": "世界名称",
"type": "str",
"length": 255,
"nullable": "no"
},
{
"name": "description",
"title": "世界描述",
"type": "text"
},
{
"name": "mode",
"title": "世界模式",
"type": "str",
"length": 32,
"nullable": "no",
"default": "draft"
},
{
"name": "world_type",
"title": "世界类型",
"type": "str",
"length": 32
},
{
"name": "owner_id",
"title": "所属用户ID",
"type": "str",
"length": 32
},
{
"name": "org_id",
"title": "所属组织ID",
"type": "str",
"length": 32,
"default": "0"
},
{
"name": "created_at",
"title": "创建时间",
"type": "timestamp",
"nullable": "no"
},
{
"name": "updated_at",
"title": "更新时间",
"type": "timestamp",
"nullable": "no"
}
],
"indexes": [
{
"name": "idx_world_code",
"idxtype": "unique",
"idxfields": ["code"]
},
{
"name": "idx_world_list",
"idxtype": "index",
"idxfields": ["org_id", "mode", "created_at"]
}
],
"codes": [
{
"field": "mode",
"table": "appcodes_kv",
"valuefield": "k",
"textfield": "v",
"cond": "parentid='world_mode'"
},
{
"field": "world_type",
"table": "appcodes_kv",
"valuefield": "k",
"textfield": "v",
"cond": "parentid='world_type'"
}
]
}

View File

@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "world"
version = "1.0.0"
description = "世界定义与管理模块W-01 世界管理world 表 CRUD、列表查询、创建自动编码、mode 白名单/流转校验、world_mode 编码字典"
description = "世界定义与管理模块W-01 世界管理world 表 CRUD、列表查询、自动编码、mode 白名单与流转校验"
requires-python = ">=3.8"
dependencies = ["sqlor", "bricks_for_python"]

View File

@ -1,37 +1,27 @@
#!/usr/bin/env python3
"""world 模块 RBAC 权限注册脚本(显式路径,禁通配符)。
# -*- coding: utf-8 -*-
"""world 模块 RBAC 路径注册脚本(显式注册,禁通配符)。
Usage: cd ~/repos/sage && ./py3/bin/python ~/repos/world/scripts/load_path.py
包含全部新 API/页面路径wwwroot/index.uimenu.ui wwwroot/api/*.dspy
角色logined需登录menu.ui any预登录资源
执行python scripts/load_path.py
"""
import subprocess
import os
import sys
def find_sage_root():
for c in [os.path.expanduser('~/repos/sage'), os.path.expanduser('~/sage')]:
if os.path.isdir(os.path.join(c, 'py3')) and os.path.isdir(os.path.join(c, 'wwwroot')):
return c
return None
SAGE_ROOT = find_sage_root()
if not SAGE_ROOT:
print('ERROR: Cannot find Sage root')
sys.exit(1)
PYTHON = os.path.join(SAGE_ROOT, 'py3', 'bin', 'python')
SET_PERM_SCRIPT = os.path.join(SAGE_ROOT, 'set_role_perm.py')
MOD = 'world'
# 公开路径(无需登录)
PATHS_ANY = [
'/world/menu.ui',
]
# 登录路径(页面 + API + CRUD 自动生成目录,全部显式注册)
# 显式路径注册(禁止 % / * 通配符)
PATHS_LOGINED = [
'/world',
'/world/index.ui',
'/world/menu.ui',
'/world/world_list',
'/world/world_list/index.ui',
'/world/world_list/get_world_list.dspy',
'/world/world_list/add_world.dspy',
'/world/world_list/update_world.dspy',
'/world/world_list/delete_world.dspy',
'/world/api/create_world.dspy',
'/world/api/world_update.dspy',
'/world/api/world_delete.dspy',
@ -40,34 +30,50 @@ PATHS_LOGINED = [
'/world/api/set_world_mode.dspy',
'/world/api/get_search_status.dspy',
'/world/api/get_search_world_type.dspy',
# CRUD 自动生成目录 world/xls2ui 依据 json/world.json 生成)
'/world/world',
'/world/world/index.ui',
'/world/world/get_world.dspy',
'/world/world/add_world.dspy',
'/world/world/update_world.dspy',
'/world/world/delete_world.dspy',
]
PATHS_ANY = [
'/world/menu.ui',
]
def run_set_perm(role, path):
try:
r = subprocess.run([PYTHON, SET_PERM_SCRIPT, role, path], capture_output=True, text=True)
return r.returncode == 0
except Exception as e:
print(' set_role_perm error for %s: %s' % (path, e))
return False
def register_role_paths(role, paths):
count = sum(1 for p in paths if run_set_perm(role, p))
print(' %s: %d/%d paths registered' % (role, count, len(paths)))
return count
def find_sage_root():
candidates = [
os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..')),
os.path.expanduser('~/repos/sage'),
os.path.expanduser('~/sage'),
]
for cand in candidates:
if os.path.isdir(os.path.join(cand, 'wwwroot')) and os.path.isdir(os.path.join(cand, 'py3', 'bin')):
return cand
return None
def main():
total = register_role_paths('any', PATHS_ANY) + register_role_paths('logined', PATHS_LOGINED)
print('Done. Total %d permission entries registered.' % total)
sage_root = find_sage_root()
if not sage_root:
print('ERROR: sage root not found; register paths manually in central load_path.py')
sys.exit(1)
sys.path.insert(0, os.path.join(sage_root, 'py3', 'bin'))
try:
from load_path import register_paths # noqa
except ImportError:
# 中央 load_path.py 未提供可导入函数时,输出待注册清单供人工接入
print('INFO: central load_path.py has no importable register_paths; '
'falling back to central registration file (see sage/load_path.py)')
register_paths = None
if register_paths is not None:
for p in PATHS_ANY:
register_paths(p, 'any')
for p in PATHS_LOGINED:
register_paths(p, 'logined')
print('world RBAC paths registered: %d logined, %d any'
% (len(PATHS_LOGINED), len(PATHS_ANY)))
else:
print('world RBAC paths (register in sage/load_path.py):')
for p in PATHS_LOGINED:
print(' %s logined' % p)
for p in PATHS_ANY:
print(' %s any' % p)
if __name__ == '__main__':

View File

@ -40,12 +40,12 @@ REST 前缀 `/api/*`wwwroot/api/*.dspy
## 陷阱
- mode 硬校验:非法枚举一律返回 INVALID_MODE绝不落库流转规则见 MODE_TRANSITIONS
- 编码唯一:唯一索引兜底 + 前置校验 + Duplicate key 冲突重试(自动编码场景)
- 列表查询 LIMIT/OFFSET 用整数内联,禁止 `${param}$`(会生成引号字符串)
- 列表查询 LIMIT/OFFSET 用整数内联sqlPaging ns.page/ns.rows禁止 `${param}$` 生成引号字符串
- 所有 .dspy 无 importjson/exception/format_exc/get_module_dbname/DBPools 为预载全局)
- 取库名用 get_module_dbname('world'),禁止硬编码 DBNAME
- 根目录不放 __init__.py包目录为 world/world/三处同步注册world.py / __init__.py / init.py
## 依赖
- sqlorDBPools/sqlExe/sor.C/U/D/R
- sqlorDBPools/sqlExe/sor.C/U/D/R/sqlPaging
- ahserverServerEnv、get_module_dbname
- appbaseappcodes/appcodes_kv 编码字典)

View File

@ -1,24 +1,29 @@
"""world 模块包W-01 世界管理)。
# -*- coding: utf-8 -*-
"""world 世界定义与管理模块W-01 世界管理)。
包目录 world/world/模块根目录不放置 __init__.py规范要求
所有业务函数在此导出保证 .dspy 调用时 NameError 不出现
包目录 world/world/模块根目录不放 __init__.py规范要求
通过 load_world() 挂载到宿主应用ServerEnv 注册业务函数
"""
from world.world import (
from .init import load_world
from .world import (
create_world,
update_world,
delete_world,
get_world,
list_worlds,
set_world_mode,
get_world_mode_options,
get_world_type_options,
)
from world.init import load_world
__all__ = [
'load_world',
'create_world',
'update_world',
'delete_world',
'get_world',
'list_worlds',
'set_world_mode',
'load_world',
'get_world_mode_options',
'get_world_type_options',
]

View File

@ -1,28 +1,27 @@
"""world 模块初始化:将业务函数注册到 ServerEnv供 .dspy/.ui 直接调用。
# -*- coding: utf-8 -*-
"""world 模块初始化:注册全部业务函数到 ServerEnvload_world 为唯一集成点)。
三处同步注册
实现world/world.py本模块函数定义
导出world/__init__.pyimport
注册本文件 load_world()env.xxx = xxx
三处同步注册约定实现world/world.py 导出world/__init__.py 本文件 env.xxx = xxx
新增/删除函数必须三处同步更新
"""
from appPublic.log import info
from ahserver.serverenv import ServerEnv
from ahserver import ServerEnv
from world.world import (
from .world import (
create_world,
update_world,
delete_world,
get_world,
list_worlds,
set_world_mode,
get_world_mode_options,
get_world_type_options,
)
def load_world():
"""挂载 world 模块到 ServerEnv宿主应用 init() 中调用)"""
"""注册 world 模块全部业务函数到 ServerEnv供 .dspy/.ui 直接调用"""
env = ServerEnv()
# 单数 + CRUD 复数双注册xls2ui 生成的 add_world.dspy 包装用复数名)
# 单数/复数双注册CRUD 框架约定dspy 包装器用复数表名)
env.create_world = create_world
env.create_worlds = create_world
env.update_world = update_world
@ -30,8 +29,9 @@ def load_world():
env.delete_world = delete_world
env.delete_worlds = delete_world
env.get_world = get_world
env.get_worlds = get_world
env.list_worlds = list_worlds
env.set_world_mode = set_world_mode
info('world module loaded: create/update/delete/get/list_worlds/set_world_mode registered')
env.get_world_mode_options = get_world_mode_options
env.get_world_type_options = get_world_type_options
return env

View File

@ -1,398 +1,341 @@
"""world 模块核心业务逻辑W-01 世界管理)。
# -*- coding: utf-8 -*-
"""W-01 世界管理核心业务逻辑world 模块)。
职责
- world CRUD创建自动编码 / 更新 / 删除 / 单条查询
- 列表查询sqlPaging 分页 + 复合索引过滤P95 达标
- world.mode 白名单硬校验draft/active/paused/archived
- mode 流转规则校验draftactivepausedactive/pausedarchivedarchivedactive
- 编码全局唯一唯一索引兜底 + 应用层前置校验 + 冲突重试
- 统一错误结构 {code, message, field, detail}分页结构 {list, total}
提供 world CRUD分页列表查询sqlPaging + 复合索引创建自动编码
mode 白名单硬校验与流转规则校验编码唯一冲突重试编码字典下拉数据源
本文件为 .py 模块文件函数经 world/init.py load_world() 注册到 ServerEnv
wwwroot/api/*.dspy 薄包装直接调用
统一错误结构{code, message, field, detail}WorldError
分页结构{list, total, page, page_size}
依赖sqlorDBPoolsahserverServerEnvappbaseappcodes/appcodes_kv 编码字典
"""
import time
import random
from datetime import datetime
from ahserver.serverenv import ServerEnv
from appPublic.jsonConfig import getConfig
from appPublic.uniqueID import getID
from appPublic.timeUtils import curDateString
from appPublic.log import debug, error
from sqlor.dbpools import DBPools
from ahserver import ServerEnv
try:
from sqlor import DBPools
except ImportError:
from sqlor.dbpools import DBPools
try:
from appPublic.utils import getID, curDateString
except ImportError:
from appPublic import getID, curDateString
# ---------------------------------------------------------------------------
# 常量mode 白名单 + 流转规则
# ---------------------------------------------------------------------------
MODE_WHITELIST = ('draft', 'active', 'paused', 'archived')
WORLD_TYPE_WHITELIST = ('game', 'simulation', 'education', 'business', 'social')
# mode 流转规则from -> 允许切换到的目标集合
# 模式流转规则draft→active⇄paused、active/paused→archived、archived→active
MODE_TRANSITIONS = {
'draft': {'active'},
'active': {'paused', 'archived'},
'paused': {'active', 'archived'},
'archived': {'active'},
'draft': ('active',),
'active': ('paused', 'archived'),
'paused': ('active', 'archived'),
'archived': ('active',),
}
# 列表查询列(显式列清单,排除 description TEXT 大字段,避免全表扫描与 off-page I/O
LIST_FIELDS = 'id, code, name, mode, world_type, owner_id, org_id, created_at, updated_at'
# 错误码
ERR_VALIDATION = 'VALIDATION_ERROR'
ERR_INVALID_MODE = 'INVALID_MODE'
ERR_INVALID_TRANSITION = 'INVALID_TRANSITION'
ERR_CODE_EXISTS = 'CODE_EXISTS'
ERR_NOT_FOUND = 'NOT_FOUND'
ERR_INTERNAL = 'INTERNAL_ERROR'
CODE_RETRY = 3 # 自动编码唯一冲突重试次数
LIST_SORTS = ('created_at', 'code', 'name', 'mode') # 列表排序白名单,防注入
LIST_PAGE_SIZE_MAX = 200
# ---------------------------------------------------------------------------
# 工具函数
# ---------------------------------------------------------------------------
def _err(code, message, field='', detail=''):
"""统一错误结构:{code, message, field, detail}。"""
return {'code': code, 'message': message, 'field': field, 'detail': detail}
class WorldError(Exception):
"""业务异常,携带统一错误结构 {code, message, field, detail}。"""
def _ok(message='ok', data=None):
"""统一成功结构。"""
return {'code': 0, 'message': message, 'data': data or {}}
def __init__(self, code='INTERNAL_ERROR', message='', field='', detail=''):
super().__init__(message)
self.code = code
self.message = message
self.field = field
self.detail = detail
def _get_dbname():
env = ServerEnv()
return env.get_module_dbname('world')
def _get_pool():
db = DBPools()
config = getConfig()
db.databases = config.databases
return db
def _to_int(val, default):
try:
return int(val)
except (TypeError, ValueError):
return default
"""取 world 模块库名,禁止硬编码 DBNAME。"""
return ServerEnv().get_module_dbname('world')
def _gen_world_code():
"""生成世界编码W + 时间戳 + 4 位随机数(可读、全局唯一概率极高)。"""
return 'W' + time.strftime('%Y%m%d%H%M%S') + format(random.randint(0, 9999), '04d')
"""自动编码W + 时间戳(YYYYMMDDHHMMSS) + 随机4位数字。全局唯一由唯一索引+重试兜底。"""
ts = datetime.now().strftime('%Y%m%d%H%M%S')
suffix = ''.join(random.choice('0123456789') for _ in range(4))
return 'W' + ts + suffix
def _pick(row, fields):
"""从查询结果行DictObject/dict中按列名取值返回普通 dict。"""
out = {}
for f in fields:
out[f] = row.get(f)
return out
def _clean_ns(ns):
"""清洗提交参数:剔除 _text 后缀字段与空值Tabular 会发送 _text 展示列)。"""
clean = {}
for k, v in (ns or {}).items():
if k.endswith('_text'):
continue
if v is None or v == '':
continue
clean[k] = v
return clean
def _validate_mode(mode):
"""mode 白名单硬校验,返回 (ok, err)。非法枚举 100% 拦截。"""
if not mode:
return False, _err(ERR_VALIDATION, '世界模式不能为空', 'mode')
def _check_mode(mode):
if mode not in MODE_WHITELIST:
return False, _err(
ERR_INVALID_MODE,
'非法世界模式: %s,合法枚举为 %s' % (mode, '/'.join(MODE_WHITELIST)),
'mode',
)
return True, None
raise WorldError('INVALID_MODE', 'invalid world mode: ' + str(mode), 'mode',
'mode must be one of: ' + ', '.join(MODE_WHITELIST))
def _validate_transition(old_mode, new_mode):
"""mode 流转规则校验。"""
ok, err = _validate_mode(new_mode)
if not ok:
return False, err
allowed = MODE_TRANSITIONS.get(old_mode)
if allowed is None or new_mode not in allowed:
return False, _err(
ERR_INVALID_TRANSITION,
'不允许从 %s 切换到 %s(合法流转: draft→active⇄paused、active/paused→archived、archived→active'
% (old_mode, new_mode),
'mode',
)
return True, None
def _check_world_type(world_type):
if world_type and world_type not in WORLD_TYPE_WHITELIST:
raise WorldError('INVALID_WORLD_TYPE', 'invalid world_type: ' + str(world_type),
'world_type',
'world_type must be one of: ' + ', '.join(WORLD_TYPE_WHITELIST))
async def _check_code_unique(sor, code, exclude_id=None):
"""编码唯一性前置校验(唯一索引为最终兜底)。"""
if exclude_id:
rows = await sor.sqlExe(
'SELECT id FROM world WHERE code = ${code}$ AND id != ${exclude_id}$ LIMIT 1',
{'code': code, 'exclude_id': exclude_id},
)
else:
rows = await sor.sqlExe(
'SELECT id FROM world WHERE code = ${code}$ LIMIT 1',
{'code': code},
)
return not rows
def _check_transition(old_mode, new_mode):
_check_mode(old_mode)
_check_mode(new_mode)
if new_mode not in MODE_TRANSITIONS.get(old_mode, ()):
raise WorldError('INVALID_MODE_TRANSITION',
'mode transition not allowed: ' + old_mode + ' -> ' + new_mode,
'mode',
'allowed transitions: draft->active, active<->paused, '
'active/paused->archived, archived->active')
def _is_dup_key_error(msg):
"""判断是否为唯一索引冲突Duplicate entry ... for key ...)。"""
return 'Duplicate' in msg and 'key' in msg
def _to_int(val, default, minimum=1):
try:
n = int(val)
except (TypeError, ValueError):
n = default
return n if n >= minimum else minimum
# ---------------------------------------------------------------------------
# 业务函数CRUD + 模式切换
# ---------------------------------------------------------------------------
async def create_world(params=None):
"""创建世界:自动编码 + mode 白名单/流转校验 + 唯一索引冲突重试。
非法输入缺名称非法 mode非法流转100% 拦截不落库
"""
params = params or {}
ns = _clean_ns(params)
now = curDateString()
name = ns.get('name')
if not name:
return _err(ERR_VALIDATION, '世界名称不能为空', 'name')
# mode 校验:默认 draft显式传入则按 draft→mode 流转规则校验
mode = ns.get('mode', 'draft')
if mode != 'draft':
ok, err = _validate_transition('draft', mode)
if not ok:
return err
else:
ok, err = _validate_mode(mode)
if not ok:
return err
world_type = ns.get('world_type', '')
if world_type and len(world_type) > 32:
return _err(ERR_VALIDATION, '世界类型长度不能超过 32', 'world_type')
db = _get_pool()
dbname = _get_dbname()
user_id = ns.get('owner_id', '')
org_id = ns.get('org_id', '0')
last_err = ''
for _attempt in range(3): # 唯一索引冲突重试(最多 3 次)
code = ns.get('code') or _gen_world_code()
record = {
'id': getID(),
'code': code,
'name': name,
'mode': mode,
'world_type': world_type,
'description': ns.get('description', ''),
'owner_id': user_id,
'org_id': org_id,
'created_at': now,
'updated_at': now,
}
try:
async with db.sqlorContext(dbname) as sor:
if not await _check_code_unique(sor, code):
if ns.get('code'):
return _err(ERR_CODE_EXISTS, '世界编码已存在: %s' % code, 'code', '编码全局唯一')
last_err = 'auto code conflict'
continue # 自动编码冲突 → 重新生成重试
await sor.C('world', record)
return _ok('世界创建成功', {'id': record['id'], 'code': code, 'mode': mode})
except Exception as e:
msg = str(e)
if _is_dup_key_error(msg):
if ns.get('code'):
return _err(ERR_CODE_EXISTS, '世界编码已存在: %s' % code, 'code', msg)
last_err = msg
continue # 唯一索引兜底冲突 → 重新生成重试
error('create_world failed: %s' % msg)
return _err(ERR_INTERNAL, '创建世界失败: %s' % msg, 'detail', msg)
return _err(ERR_CODE_EXISTS, '自动编码冲突重试耗尽', 'code', last_err)
async def update_world(params=None):
"""更新世界(含 mode 流转校验、code 唯一校验)。部分更新语义。"""
params = params or {}
ns = _clean_ns(params)
world_id = ns.get('id') or ns.get('world_id')
if not world_id:
return _err(ERR_VALIDATION, '缺少世界 ID', 'id')
db = _get_pool()
dbname = _get_dbname()
async with db.sqlorContext(dbname) as sor:
rows = await sor.R('world', {'id': world_id})
if not rows:
return _err(ERR_NOT_FOUND, '世界不存在: %s' % world_id, 'id')
cur = rows[0]
upd = {}
if 'name' in ns:
if not ns['name']:
return _err(ERR_VALIDATION, '世界名称不能为空', 'name')
upd['name'] = ns['name']
if 'world_type' in ns:
upd['world_type'] = ns['world_type']
if 'description' in ns:
upd['description'] = ns['description']
# mode 切换校验(仅在变更时)
new_mode = ns.get('mode')
if new_mode and new_mode != cur['mode']:
ok, err = _validate_transition(cur['mode'], new_mode)
if not ok:
return err
upd['mode'] = new_mode
# code 变更校验(一般不允许改编码,若传入则校验全局唯一)
new_code = ns.get('code')
if new_code and new_code != cur['code']:
if not await _check_code_unique(sor, new_code, exclude_id=world_id):
return _err(ERR_CODE_EXISTS, '世界编码已存在: %s' % new_code, 'code', '编码全局唯一')
upd['code'] = new_code
if not upd:
return _err(ERR_VALIDATION, '没有可更新的字段', 'params')
upd['id'] = world_id
upd['updated_at'] = curDateString()
await sor.U('world', upd)
return _ok('世界更新成功', {'id': world_id})
async def delete_world(params=None):
"""删除世界。"""
params = params or {}
world_id = params.get('id') or params.get('world_id')
if not world_id:
return _err(ERR_VALIDATION, '缺少世界 ID', 'id')
db = _get_pool()
dbname = _get_dbname()
async with db.sqlorContext(dbname) as sor:
rows = await sor.R('world', {'id': world_id})
if not rows:
return _err(ERR_NOT_FOUND, '世界不存在: %s' % world_id, 'id')
await sor.D('world', {'id': world_id})
return _ok('世界删除成功', {'id': world_id})
async def get_world(params=None):
"""按 ID 查询单条世界。"""
params = params or {}
world_id = params.get('id') or params.get('world_id')
if not world_id:
return _err(ERR_VALIDATION, '缺少世界 ID', 'id')
db = _get_pool()
dbname = _get_dbname()
async with db.sqlorContext(dbname) as sor:
rows = await sor.R('world', {'id': world_id})
if not rows:
return _err(ERR_NOT_FOUND, '世界不存在: %s' % world_id, 'id')
return _ok('ok', _pick(rows[0], LIST_FIELDS.split(', ')))
async def list_worlds(params=None):
"""世界列表sqlPaging 分页 + 复合索引org_id+mode+created_at过滤。
返回 {code:0, message, list: [...], total, page, page_size}
采用count 与数据查询分离+ 显式列清单P95 达标
"""
params = params or {}
page = _to_int(params.get('page'), 1)
rows_per_page = _to_int(params.get('rows') or params.get('page_size'), 20)
if page < 1:
page = 1
if rows_per_page < 1:
rows_per_page = 20
if rows_per_page > 200:
rows_per_page = 200
# 过滤条件(优先走复合索引前导列 org_id/mode
conds = []
ns = {}
org_id = params.get('org_id')
mode = params.get('mode')
world_type = params.get('world_type')
keyword = params.get('keyword')
if org_id:
conds.append('org_id = ${org_id}$')
ns['org_id'] = org_id
if mode:
conds.append('mode = ${mode}$')
ns['mode'] = mode
if world_type:
conds.append('world_type = ${world_type}$')
ns['world_type'] = world_type
if keyword:
conds.append('(name LIKE ${keyword}$ OR code LIKE ${keyword}$)')
ns['keyword'] = '%%%s%%' % keyword
where = ''
if conds:
where = ' WHERE ' + ' AND '.join(conds)
db = _get_pool()
dbname = _get_dbname()
offset = (page - 1) * rows_per_page
async with db.sqlorContext(dbname) as sor:
# ① count 查询
cnt_rows = await sor.sqlExe('SELECT COUNT(*) AS cnt FROM world' + where, ns)
total = int(cnt_rows[0]['cnt'] or 0) if cnt_rows else 0
# ② 数据查询LIMIT/OFFSET 内联整数,避免 ${param}$ 引号问题)
sort = ' ORDER BY created_at DESC'
data_rows = await sor.sqlExe(
'SELECT %s FROM world%s%s LIMIT %d, %d' % (LIST_FIELDS, where, sort, offset, rows_per_page),
ns,
)
list_data = [_pick(r, LIST_FIELDS.split(', ')) for r in (data_rows or [])]
def _row_to_dict(r):
"""查询行DictObject转 dict显式列清单description 为 TEXT列表查询不取"""
return {
'code': 0,
'message': 'ok',
'list': list_data,
'total': total,
'page': page,
'page_size': rows_per_page,
'id': r.id,
'code': r.code,
'name': r.name,
'description': getattr(r, 'description', '') or '',
'mode': r.mode,
'world_type': getattr(r, 'world_type', '') or '',
'owner_id': getattr(r, 'owner_id', '') or '',
'org_id': getattr(r, 'org_id', '') or '',
'created_at': str(r.created_at),
'updated_at': str(r.updated_at),
}
async def set_world_mode(params=None):
"""模式切换严格按流转规则校验draft→active⇄paused、active/paused→archived、archived→active"""
params = params or {}
world_id = params.get('id') or params.get('world_id')
new_mode = params.get('mode')
if not world_id:
return _err(ERR_VALIDATION, '缺少世界 ID', 'id')
if not new_mode:
return _err(ERR_VALIDATION, '缺少目标模式', 'mode')
async def create_world(params):
"""创建世界。
db = _get_pool()
- 自动编码W+时间戳+随机4位可显式传入 code
- mode 默认 draft非法 mode 100% 拦截不落库
- 编码全局唯一唯一索引兜底 + 冲突重试 3
"""
dbname = _get_dbname()
async with db.sqlorContext(dbname) as sor:
rows = await sor.R('world', {'id': world_id})
name = (params.get('name') or '').strip()
if not name:
raise WorldError('INVALID_NAME', 'world name is required', 'name', '')
mode = params.get('mode') or 'draft'
_check_mode(mode)
if mode != 'draft':
raise WorldError('INVALID_MODE_TRANSITION', 'new world must start at draft', 'mode',
'new world mode must be draft, then draft->active to activate')
world_type = (params.get('world_type') or '').strip()
_check_world_type(world_type)
code = (params.get('code') or '').strip()
org_id = params.get('org_id') or '0'
owner_id = params.get('owner_id') or '0'
description = params.get('description') or ''
now = curDateString()
ns = {
'id': getID(),
'code': code or _gen_world_code(),
'name': name,
'description': description,
'mode': mode,
'world_type': world_type,
'owner_id': owner_id,
'org_id': org_id,
'created_at': now,
'updated_at': now,
}
last_err = None
for attempt in range(CODE_RETRY):
if attempt > 0:
ns['code'] = _gen_world_code()
try:
async with DBPools().sqlorContext(dbname) as sor:
await sor.C('world', dict(ns))
return dict(ns)
except Exception as e:
msg = str(e)
if 'Duplicate' in msg or '1062' in msg:
last_err = e
continue # 唯一冲突 → 换码重试
raise WorldError('DB_ERROR', 'create world failed', '', msg) from e
raise WorldError('CODE_CONFLICT', 'world code conflict after retries', 'code',
str(last_err) if last_err else '')
async def update_world(params):
"""部分更新。mode 变更走流转校验code 变更走全局唯一校验。"""
dbname = _get_dbname()
wid = params.get('id') or params.get('world_id')
if not wid:
raise WorldError('INVALID_ID', 'world id is required', 'id', '')
async with DBPools().sqlorContext(dbname) as sor:
rows = await sor.R('world', {'id': wid})
if not rows:
return _err(ERR_NOT_FOUND, '世界不存在: %s' % world_id, 'id')
cur_mode = rows[0]['mode']
if cur_mode == new_mode:
return _err(ERR_INVALID_TRANSITION, '世界已处于 %s 模式,无需切换' % new_mode, 'mode')
ok, err = _validate_transition(cur_mode, new_mode)
if not ok:
return err
await sor.U('world', {'id': world_id, 'mode': new_mode, 'updated_at': curDateString()})
return _ok('模式切换成功', {'id': world_id, 'mode': new_mode})
raise WorldError('WORLD_NOT_FOUND', 'world not found: ' + str(wid), 'id', '')
old = rows[0]
ns = {}
if 'name' in params:
name = (params.get('name') or '').strip()
if not name:
raise WorldError('INVALID_NAME', 'world name is required', 'name', '')
ns['name'] = name
if 'description' in params:
ns['description'] = params.get('description') or ''
if 'world_type' in params:
wt = (params.get('world_type') or '').strip()
_check_world_type(wt)
ns['world_type'] = wt
if 'code' in params:
new_code = (params.get('code') or '').strip()
if not new_code:
raise WorldError('INVALID_CODE', 'world code is required', 'code', '')
if new_code != old.code:
dup = await sor.sqlExe('SELECT id FROM world WHERE code = ${code}$ AND id <> ${id}$',
{'code': new_code, 'id': wid})
if dup:
raise WorldError('CODE_EXISTS', 'world code already exists: ' + new_code, 'code', '')
ns['code'] = new_code
if 'mode' in params:
new_mode = params.get('mode')
_check_transition(old.mode, new_mode)
ns['mode'] = new_mode
if not ns:
return {'id': wid, 'updated': False}
ns['id'] = wid
ns['updated_at'] = curDateString()
await sor.U('world', dict(ns))
return {'id': wid, 'updated': True}
async def delete_world(params):
"""删除世界。"""
dbname = _get_dbname()
wid = params.get('id') or params.get('world_id')
if not wid:
raise WorldError('INVALID_ID', 'world id is required', 'id', '')
async with DBPools().sqlorContext(dbname) as sor:
await sor.D('world', {'id': wid})
return {'id': wid, 'deleted': True}
async def get_world(params):
"""按 id或 code查询单个世界。"""
dbname = _get_dbname()
wid = params.get('id') or params.get('world_id') or params.get('code')
if not wid:
raise WorldError('INVALID_ID', 'world id or code is required', 'id', '')
async with DBPools().sqlorContext(dbname) as sor:
rows = await sor.R('world', {'id': wid})
if not rows:
rows = await sor.R('world', {'code': wid})
if not rows:
raise WorldError('WORLD_NOT_FOUND', 'world not found', 'id', '')
return _row_to_dict(rows[0])
async def list_worlds(params):
"""分页列表查询。
性能count 与数据查询分离显式列清单排除 TEXT description
复合索引 idx_world_list(org_id, mode, created_at) 支撑过滤+排序
"""
dbname = _get_dbname()
page = _to_int(params.get('page'), 1, 1)
page_size = _to_int(params.get('page_size') or params.get('rows'), 20, 1)
if page_size > LIST_PAGE_SIZE_MAX:
page_size = LIST_PAGE_SIZE_MAX
org_id = params.get('org_id') or '0'
mode = params.get('mode') or ''
keyword = params.get('keyword') or ''
where = ['org_id = ${org_id}$']
qargs = {'org_id': org_id}
if mode:
_check_mode(mode) # 非法 mode 100% 拦截
where.append('mode = ${mode}$')
qargs['mode'] = mode
if keyword:
where.append('(name LIKE ${keyword}$ OR code LIKE ${keyword}$)')
qargs['keyword'] = '%' + keyword + '%'
cond = ' AND '.join(where)
sort = params.get('sort') or 'created_at'
if sort not in LIST_SORTS:
sort = 'created_at'
order = (params.get('order') or 'desc').lower()
if order not in ('asc', 'desc'):
order = 'desc'
# count 与数据查询分离(显式列清单,排除 TEXT
count_sql = 'SELECT COUNT(*) AS cnt FROM world WHERE ' + cond
data_sql = ('SELECT id, code, name, mode, world_type, owner_id, org_id, created_at, updated_at '
'FROM world WHERE ' + cond)
async with DBPools().sqlorContext(dbname) as sor:
cnt_rows = await sor.sqlExe(count_sql, qargs)
total = int(cnt_rows[0].cnt) if cnt_rows else 0
ns = dict(qargs)
ns.update({'page': page, 'rows': page_size, 'sort': sort, 'order': order})
result = await sor.sqlPaging(data_sql, ns)
items = [{
'id': r.id,
'code': r.code,
'name': r.name,
'mode': r.mode,
'world_type': getattr(r, 'world_type', '') or '',
'owner_id': getattr(r, 'owner_id', '') or '',
'org_id': getattr(r, 'org_id', '') or '',
'created_at': str(r.created_at),
'updated_at': str(r.updated_at),
} for r in result.rows]
return {'list': items, 'total': total, 'page': page, 'page_size': page_size}
async def set_world_mode(params):
"""模式切换:严格校验白名单 + 流转规则draft→active⇄paused、active/paused→archived、archived→active"""
dbname = _get_dbname()
wid = params.get('id') or params.get('world_id')
new_mode = params.get('mode')
if not wid:
raise WorldError('INVALID_ID', 'world id is required', 'id', '')
if not new_mode:
raise WorldError('INVALID_MODE', 'mode is required', 'mode', '')
_check_mode(new_mode)
async with DBPools().sqlorContext(dbname) as sor:
rows = await sor.R('world', {'id': wid})
if not rows:
raise WorldError('WORLD_NOT_FOUND', 'world not found: ' + str(wid), 'id', '')
old_mode = rows[0].mode
_check_transition(old_mode, new_mode)
await sor.U('world', {'id': wid, 'mode': new_mode, 'updated_at': curDateString()})
return {'id': wid, 'mode': new_mode}
async def get_world_mode_options(prefix_all=True):
"""世界模式下拉数据源 [{value, text}],数据来自编码字典 world_mode。"""
return await _dict_options('world_mode', MODE_WHITELIST, prefix_all)
async def get_world_type_options(prefix_all=True):
"""世界类型下拉数据源 [{value, text}],数据来自编码字典 world_type。"""
return await _dict_options('world_type', WORLD_TYPE_WHITELIST, prefix_all)
async def _dict_options(parentid, fallback_keys, prefix_all):
"""从 appbase 编码字典 appcodes_kv 读下拉选项;字典不可用时回退英文标签(不硬编码中文)。"""
items = [{'value': '', 'text': 'All'}] if prefix_all else []
try:
dbname = ServerEnv().get_module_dbname('appbase')
async with DBPools().sqlorContext(dbname) as sor:
rows = await sor.sqlExe('SELECT k, v FROM appcodes_kv WHERE parentid = ${pid}$ ORDER BY id',
{'pid': parentid})
for r in rows:
items.append({'value': r.k, 'text': r.v})
except Exception:
# 字典表未就绪时回退英文标签,保证下拉可用
items = [{'value': '', 'text': 'All'}] if prefix_all else []
for k in fallback_keys:
items.append({'value': k, 'text': k})
return items

View File

@ -1,10 +1,15 @@
"""创建世界:自动编码 + mode 白名单/流转校验 + 编码唯一冲突重试。
REST 契约:错误结构 {code, message, field, detail};成功 {code:0, message, data}。
"""
# -*- coding: utf-8 -*-
# create_world.dspy —— 创建世界自动编码、mode 白名单硬校验、编码唯一冲突重试)
# 预载全局params_kw / json / format_exc / create_world / get_module_dbname
debug('create_world.dspy: START params_kw=%s' % json.dumps(dict(params_kw), ensure_ascii=False))
try:
result = await create_worlds(dict(params_kw))
return json.dumps(result, ensure_ascii=False)
result = await create_world(dict(params_kw))
return {"status": "ok", "data": result}
except Exception as e:
exception('create_world.dspy: %s' % e)
return json.dumps({'code': 'INTERNAL_ERROR', 'message': '创建世界失败', 'field': '', 'detail': format_exc()}, ensure_ascii=False)
code = getattr(e, 'code', 'INTERNAL_ERROR')
message = getattr(e, 'message', str(e)) or str(e)
field = getattr(e, 'field', '')
detail = getattr(e, 'detail', '') or ''
debug('create_world.dspy: ERROR %s %s' % (code, message))
exception(format_exc())
return {"status": "error", "code": code, "message": message, "field": field, "detail": detail}

View File

@ -1,13 +1,11 @@
"""世界模式下拉数据源world_mode 编码字典):返回 [{value, text}],首项为"全部"。"""
result = [{'value': '', 'text': '全部'}]
# -*- coding: utf-8 -*-
# get_search_status.dspy —— 世界模式下拉数据源 [{value, text}](首项"全部"
try:
dbname = get_module_dbname('world')
async with DBPools().sqlorContext(dbname) as sor:
rows = await sor.sqlExe(
'SELECT k AS value, v AS text FROM appcodes_kv WHERE parentid = ${pid}$ ORDER BY id',
{'pid': 'world_mode'},
)
result = result + [{'value': r['value'], 'text': r['text']} for r in (rows or [])]
items = await get_world_mode_options(True)
return {"status": "ok", "data": items}
except Exception as e:
exception('get_search_status.dspy: %s' % e)
return json.dumps(result, ensure_ascii=False)
debug('get_search_status.dspy: ERROR %s' % str(e))
exception(format_exc())
return {"status": "error", "code": getattr(e, 'code', 'INTERNAL_ERROR'),
"message": getattr(e, 'message', str(e)) or str(e),
"field": getattr(e, 'field', ''), "detail": getattr(e, 'detail', '') or ''}

View File

@ -1,13 +1,11 @@
"""世界类型下拉数据源world_type 编码字典):返回 [{value, text}],首项为"全部"。"""
result = [{'value': '', 'text': '全部'}]
# -*- coding: utf-8 -*-
# get_search_world_type.dspy —— 世界类型下拉数据源 [{value, text}](首项"全部"
try:
dbname = get_module_dbname('world')
async with DBPools().sqlorContext(dbname) as sor:
rows = await sor.sqlExe(
'SELECT k AS value, v AS text FROM appcodes_kv WHERE parentid = ${pid}$ ORDER BY id',
{'pid': 'world_type'},
)
result = result + [{'value': r['value'], 'text': r['text']} for r in (rows or [])]
items = await get_world_type_options(True)
return {"status": "ok", "data": items}
except Exception as e:
exception('get_search_world_type.dspy: %s' % e)
return json.dumps(result, ensure_ascii=False)
debug('get_search_world_type.dspy: ERROR %s' % str(e))
exception(format_exc())
return {"status": "error", "code": getattr(e, 'code', 'INTERNAL_ERROR'),
"message": getattr(e, 'message', str(e)) or str(e),
"field": getattr(e, 'field', ''), "detail": getattr(e, 'detail', '') or ''}

View File

@ -1,7 +1,14 @@
"""按 ID 查询单个世界。"""
# -*- coding: utf-8 -*-
# get_world.dspy —— 查询单个世界id 或 code
debug('get_world.dspy: START params_kw=%s' % json.dumps(dict(params_kw), ensure_ascii=False))
try:
result = await get_world(dict(params_kw))
return json.dumps(result, ensure_ascii=False)
return {"status": "ok", "data": result}
except Exception as e:
exception('get_world.dspy: %s' % e)
return json.dumps({'code': 'INTERNAL_ERROR', 'message': '查询世界失败', 'field': '', 'detail': format_exc()}, ensure_ascii=False)
code = getattr(e, 'code', 'INTERNAL_ERROR')
message = getattr(e, 'message', str(e)) or str(e)
field = getattr(e, 'field', '')
detail = getattr(e, 'detail', '') or ''
debug('get_world.dspy: ERROR %s %s' % (code, message))
exception(format_exc())
return {"status": "error", "code": code, "message": message, "field": field, "detail": detail}

View File

@ -1,10 +1,15 @@
"""世界列表查询sqlPaging + 复合索引(org_id+mode+created_at)过滤。
REST 契约:分页结构 {list, total, page, page_size};错误 {code, message, field, detail}。
"""
# -*- coding: utf-8 -*-
# list_worlds.dspy —— 分页列表查询sqlPaging + 复合索引 idx_world_list
# 分页结构:{list, total, page, page_size}
debug('list_worlds.dspy: START params_kw=%s' % json.dumps(dict(params_kw), ensure_ascii=False))
try:
data = await list_worlds(dict(params_kw))
return json.dumps(data, ensure_ascii=False)
result = await list_worlds(dict(params_kw))
return {"status": "ok", "data": result}
except Exception as e:
exception('list_worlds.dspy: %s' % e)
return json.dumps({'code': 'INTERNAL_ERROR', 'message': '查询世界列表失败', 'field': '', 'detail': format_exc()}, ensure_ascii=False)
code = getattr(e, 'code', 'INTERNAL_ERROR')
message = getattr(e, 'message', str(e)) or str(e)
field = getattr(e, 'field', '')
detail = getattr(e, 'detail', '') or ''
debug('list_worlds.dspy: ERROR %s %s' % (code, message))
exception(format_exc())
return {"status": "error", "code": code, "message": message, "field": field, "detail": detail}

View File

@ -1,10 +1,15 @@
"""世界模式切换:严格流转规则校验。
draft→active⇄paused、active/paused→archived、archived→active
"""
# -*- coding: utf-8 -*-
# set_world_mode.dspy —— 模式切换(严格校验白名单 + 流转规则)
# 流转规则:draft→active⇄paused、active/paused→archived、archived→active
debug('set_world_mode.dspy: START params_kw=%s' % json.dumps(dict(params_kw), ensure_ascii=False))
try:
result = await set_world_mode(dict(params_kw))
return json.dumps(result, ensure_ascii=False)
return {"status": "ok", "data": result}
except Exception as e:
exception('set_world_mode.dspy: %s' % e)
return json.dumps({'code': 'INTERNAL_ERROR', 'message': '模式切换失败', 'field': '', 'detail': format_exc()}, ensure_ascii=False)
code = getattr(e, 'code', 'INTERNAL_ERROR')
message = getattr(e, 'message', str(e)) or str(e)
field = getattr(e, 'field', '')
detail = getattr(e, 'detail', '') or ''
debug('set_world_mode.dspy: ERROR %s %s' % (code, message))
exception(format_exc())
return {"status": "error", "code": code, "message": message, "field": field, "detail": detail}

View File

@ -1,7 +1,14 @@
"""删除世界。"""
# -*- coding: utf-8 -*-
# world_delete.dspy —— 删除世界
debug('world_delete.dspy: START params_kw=%s' % json.dumps(dict(params_kw), ensure_ascii=False))
try:
result = await delete_worlds(dict(params_kw))
return json.dumps(result, ensure_ascii=False)
result = await delete_world(dict(params_kw))
return {"status": "ok", "data": result}
except Exception as e:
exception('world_delete.dspy: %s' % e)
return json.dumps({'code': 'INTERNAL_ERROR', 'message': '删除世界失败', 'field': '', 'detail': format_exc()}, ensure_ascii=False)
code = getattr(e, 'code', 'INTERNAL_ERROR')
message = getattr(e, 'message', str(e)) or str(e)
field = getattr(e, 'field', '')
detail = getattr(e, 'detail', '') or ''
debug('world_delete.dspy: ERROR %s %s' % (code, message))
exception(format_exc())
return {"status": "error", "code": code, "message": message, "field": field, "detail": detail}

View File

@ -1,7 +1,14 @@
"""更新世界:部分更新 + mode 流转校验 + code 唯一校验。"""
# -*- coding: utf-8 -*-
# world_update.dspy —— 更新世界部分更新mode 变更走流转校验code 变更走全局唯一校验)
debug('world_update.dspy: START params_kw=%s' % json.dumps(dict(params_kw), ensure_ascii=False))
try:
result = await update_worlds(dict(params_kw))
return json.dumps(result, ensure_ascii=False)
result = await update_world(dict(params_kw))
return {"status": "ok", "data": result}
except Exception as e:
exception('world_update.dspy: %s' % e)
return json.dumps({'code': 'INTERNAL_ERROR', 'message': '更新世界失败', 'field': '', 'detail': format_exc()}, ensure_ascii=False)
code = getattr(e, 'code', 'INTERNAL_ERROR')
message = getattr(e, 'message', str(e)) or str(e)
field = getattr(e, 'field', '')
detail = getattr(e, 'detail', '') or ''
debug('world_update.dspy: ERROR %s %s' % (code, message))
exception(format_exc())
return {"status": "error", "code": code, "message": message, "field": field, "detail": detail}

View File

@ -1,30 +1,91 @@
{
"widgettype": "VBox",
"options": {"width": "100%", "height": "100%", "padding": "20px"},
"subwidgets": [
{"widgettype": "Text", "options": {"label": "世界管理", "fontSize": "24px", "marginBottom": "12px"}},
{"widgettype": "ResponsableBox", "options": {"gap": "16px", "minWidth": "250px"}, "subwidgets": [
"widgettype": "VBox",
"options": {
"width": "100%",
"height": "100%",
"padding": "20px"
},
"subwidgets": [
{
"widgettype": "Text",
"options": {
"label": "世界管理",
"fontSize": "24px"
}
},
{
"widgettype": "ResponsableBox",
"options": {
"gap": "16px",
"minWidth": "250px"
},
"subwidgets": [
{
"widgettype": "VBox",
"options": {
"backgroundColor": "#FFFFFF",
"padding": "20px",
"cursor": "pointer"
},
"binds": [
{
"widgettype": "VBox",
"options": {"backgroundColor": "#FFFFFF", "padding": "20px", "cursor": "pointer", "borderRadius": "6px", "boxShadow": "0 1px 4px rgba(0,0,0,0.1)"},
"binds": [{"wid": "self", "event": "click", "actiontype": "urlwidget", "target": "app.world_content",
"options": {"url": "{{entire_url('/world/world')}}"}, "mode": "replace"}],
"subwidgets": [
{"widgettype": "Text", "options": {"label": "世界列表", "fontSize": "16px"}},
{"widgettype": "Text", "options": {"label": "世界定义 CRUD创建自动编码、模式白名单校验、编码全局唯一", "fontSize": "12px", "color": "#888"}}
]
},
{
"widgettype": "VBox",
"options": {"backgroundColor": "#FFFFFF", "padding": "20px", "cursor": "pointer", "borderRadius": "6px", "boxShadow": "0 1px 4px rgba(0,0,0,0.1)"},
"binds": [{"wid": "self", "event": "click", "actiontype": "urlwidget", "target": "app.world_content",
"options": {"url": "{{entire_url('/world/api/list_worlds.dspy')}}"}, "mode": "replace"}],
"subwidgets": [
{"widgettype": "Text", "options": {"label": "列表查询接口", "fontSize": "16px"}},
{"widgettype": "Text", "options": {"label": "sqlPaging 分页 + 复合索引过滤({list, total}", "fontSize": "12px", "color": "#888"}}
]
"wid": "self",
"event": "click",
"actiontype": "urlwidget",
"target": "app.world_content",
"options": {
"url": "{{entire_url('/world/world_list')}}"
},
"mode": "replace"
}
]},
{"widgettype": "VBox", "id": "world_content", "options": {"width": "100%", "flex": "1", "marginTop": "20px"}}
]
],
"subwidgets": [
{
"widgettype": "Text",
"options": {
"label": "世界列表"
}
}
]
},
{
"widgettype": "VBox",
"options": {
"backgroundColor": "#FFFFFF",
"padding": "20px",
"cursor": "pointer"
},
"binds": [
{
"wid": "self",
"event": "click",
"actiontype": "urlwidget",
"target": "app.world_content",
"options": {
"url": "{{entire_url('/world/api/set_world_mode.dspy')}}"
},
"mode": "replace"
}
],
"subwidgets": [
{
"widgettype": "Text",
"options": {
"label": "模式切换"
}
}
]
}
]
},
{
"widgettype": "VBox",
"id": "app.world_content",
"options": {
"width": "100%",
"flex": "1",
"marginTop": "20px"
}
}
]
}

View File

@ -1,7 +1,40 @@
{
"widgettype": "Menu",
"options": {"target": "PopupWindow", "popup_options": {"width": "70%", "height": "80%"}, "items": [
{"name": "world_list", "label": "世界列表", "url": "{{entire_url('/world/world')}}"},
{"name": "world_index", "label": "世界管理首页", "url": "{{entire_url('/world/index.ui')}}"}
]}
"widgettype": "VBox",
"options": {
"width": "100%",
"height": "100%",
"padding": "10px"
},
"subwidgets": [
{
"widgettype": "Text",
"options": {
"label": "世界管理",
"fontSize": "18px"
}
},
{
"widgettype": "Menu",
"options": {
"width": "100%",
"items": [
{
"name": "world_list",
"label": "世界列表",
"url": "{{entire_url('/world/world_list')}}"
},
{
"name": "world_create",
"label": "创建世界",
"url": "{{entire_url('/world/api/create_world.dspy')}}"
},
{
"name": "world_mode",
"label": "模式切换",
"url": "{{entire_url('/world/api/set_world_mode.dspy')}}"
}
]
}
}
]
}