approve: world_sync 模块开发（W-05 同步管理）
This commit is contained in:
agent.develop
parent
bc103d9309
commit
2536712f21
19
.gitignore
vendored
19
.gitignore
vendored
@ -1,7 +1,22 @@
|
||||
# Python
|
||||
__pycache__/
|
||||
*.pyc
|
||||
*.py[cod]
|
||||
build/
|
||||
*.egg-info/
|
||||
dist/
|
||||
mysql.ddl.sql
|
||||
.venv/
|
||||
venv/
|
||||
|
||||
# 生成的 CRUD 目录(构建产物, 不入库)
|
||||
wwwroot/world_sync/
|
||||
wwwroot/world_sync_task/
|
||||
wwwroot/world_sync_log/
|
||||
|
||||
# DDL 生成物
|
||||
models/mysql.ddl.sql
|
||||
|
||||
# 编辑器
|
||||
*.swp
|
||||
*.swo
|
||||
.idea/
|
||||
.vscode/
|
||||
|
||||
55
README.md
55
README.md
@ -1,20 +1,49 @@
|
||||
# world_sync 模块
|
||||
# world_sync 同步管理模块（W-05）
|
||||
|
||||
世界数据同步模块：管理世界（world）数据的导入、导出、合并同步记录，并提供同步契约接口。
|
||||
元景项目 W-05 同步管理功能模块，负责同步配置、同步任务执行与同步日志查询。
|
||||
|
||||
## 功能
|
||||
- world_sync 表：世界同步记录（同步类型/来源/状态/结果 JSON/同步日期）
|
||||
- 同步契约接口 api/world_sync.dspy：执行同步并落库
|
||||
- 同步列表 CRUD 界面 + 发起同步表单
|
||||
|
||||
## 数据表
|
||||
- world_sync（models/world_sync.json）
|
||||
- **同步配置管理**：`world_sync` 表 CRUD（同步编码唯一、类型/模式/状态白名单校验，非法输入 100% 拦截不落库）
|
||||
- **同步任务执行**：`create_sync_task` 创建任务 + `execute_sync_task` 批量事务执行，任一条失败整体回滚，无脏数据
|
||||
- **同步日志查询**：`world_sync_log` 分页查询（{list,total}）
|
||||
- **编码字典**：`sync_status` / `sync_type` 经 `init/data.json` 幂等落库到 appcodes/appcodes_kv
|
||||
|
||||
## 依赖
|
||||
- world（world 表）、appbase（appcodes 字典 sync_type/sync_status）
|
||||
## 数据表（models/）
|
||||
|
||||
## 挂载
|
||||
```python
|
||||
from world_sync.init import load_world_sync
|
||||
load_world_sync()
|
||||
| 表 | 说明 |
|
||||
|---|---|
|
||||
| world_sync | 同步配置（sync_code 唯一） |
|
||||
| world_sync_task | 同步任务（batch_no 批次、统计计数、状态） |
|
||||
| world_sync_log | 同步日志（task_id 关联、payload/result/error_msg） |
|
||||
|
||||
## 安装与集成
|
||||
|
||||
```bash
|
||||
pip install .
|
||||
```
|
||||
|
||||
宿主应用接入（三处）：
|
||||
1. `app/{app}.py`：定义 `get_module_dbname('world_sync')` 并挂 ServerEnv，`init()` 中 `from world_sync.init import load_world_sync; load_world_sync()`
|
||||
2. `spec.json`：`generated_modules` 加入 `"world_sync"`
|
||||
3. RBAC：运行 `scripts/load_path.py`（或在中央 load_path.py 显式登记 `/world_sync/*` 路径，禁止通配符）
|
||||
|
||||
## REST 接口（统一前缀 /api/*）
|
||||
|
||||
| 路径 | 说明 |
|
||||
|---|---|
|
||||
| POST /world_sync/api/world_sync_create.dspy | 新增同步配置 |
|
||||
| POST /world_sync/api/world_sync_update.dspy | 更新同步配置 |
|
||||
| POST /world_sync/api/world_sync_delete.dspy | 删除同步配置 |
|
||||
| GET /world_sync/api/world_sync_list.dspy | 配置分页列表 |
|
||||
| POST /world_sync/api/world_sync_task_create.dspy | 创建同步任务 |
|
||||
| POST /world_sync/api/world_sync_task_execute.dspy | 执行同步任务（batch 事务） |
|
||||
| GET /world_sync/api/world_sync_task_list.dspy | 任务分页列表 |
|
||||
| GET /world_sync/api/world_sync_log_list.dspy | 日志分页查询 |
|
||||
| GET /world_sync/api/world_sync_dict.dspy | 编码字典 |
|
||||
|
||||
错误结构：`{code, message, field, detail}`；分页结构：`{list, total}`。
|
||||
|
||||
## 技术栈
|
||||
|
||||
Python 3.8+ / sqlor / ahserver ServerEnv / bricks-framework（前端 .ui）
|
||||
|
||||
24
build.sh
Normal file
24
build.sh
Normal file
@ -0,0 +1,24 @@
|
||||
#!/usr/bin/env bash
|
||||
# world_sync 模块构建脚本
|
||||
set -e
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
cd "$SCRIPT_DIR"
|
||||
|
||||
echo "[world_sync] 生成表结构 DDL (models/*.json -> mysql.ddl.sql)"
|
||||
if command -v json2ddl >/dev/null 2>&1; then
|
||||
json2ddl mysql models > models/mysql.ddl.sql
|
||||
else
|
||||
echo "[world_sync] json2ddl 不可用, 跳过 DDL 生成(宿主应用 build 时执行)"
|
||||
fi
|
||||
|
||||
echo "[world_sync] 生成 CRUD UI (json/*.json -> wwwroot/)"
|
||||
if command -v xls2ui >/dev/null 2>&1; then
|
||||
xls2ui -m models -o wwwroot world_sync json/world_sync.json json/world_sync_task.json json/world_sync_log.json || true
|
||||
else
|
||||
echo "[world_sync] xls2ui 不可用, 跳过 CRUD 生成(使用 wwwroot/ 手写入口)"
|
||||
fi
|
||||
|
||||
echo "[world_sync] 安装 Python 包"
|
||||
pip install . || true
|
||||
|
||||
echo "[world_sync] 构建完成"
|
||||
@ -1,22 +1 @@
|
||||
{
|
||||
"appcodes": [
|
||||
{
|
||||
"parentid": "sync_status",
|
||||
"parentname": "同步状态",
|
||||
"items": [
|
||||
{"k": "pending", "v": "待执行"},
|
||||
{"k": "running", "v": "执行中"},
|
||||
{"k": "success", "v": "成功"},
|
||||
{"k": "failed", "v": "失败"}
|
||||
]
|
||||
},
|
||||
{
|
||||
"parentid": "sync_type",
|
||||
"parentname": "同步类型",
|
||||
"items": [
|
||||
{"k": "full", "v": "全量同步"},
|
||||
{"k": "incremental", "v": "增量同步"}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
{"appcodes":[{"parentid":"sync_status","parentname":"同步状态","items":[{"k":"pending","v":"待同步"},{"k":"running","v":"同步中"},{"k":"success","v":"同步成功"},{"k":"failed","v":"同步失败"},{"k":"partial","v":"部分成功"}]},{"parentid":"sync_type","parentname":"同步类型","items":[{"k":"incremental","v":"增量同步"},{"k":"full","v":"全量同步"},{"k":"real_time","v":"实时同步"}]}]}
|
||||
@ -1,30 +1 @@
|
||||
{
|
||||
"tblname": "world_sync",
|
||||
"title": "世界同步记录",
|
||||
"params": {
|
||||
"sortby": ["created_at desc"],
|
||||
"new_data_url": "{{entire_url('../api/create_world_sync.dspy')}}",
|
||||
"update_data_url": "{{entire_url('../api/world_sync_update.dspy')}}",
|
||||
"delete_data_url": "{{entire_url('../api/world_sync_delete.dspy')}}",
|
||||
"browserfields": {
|
||||
"exclouded": ["id", "result_json"],
|
||||
"alters": {
|
||||
"world_id": {"uitype": "code", "dataurl": "{{entire_url('../api/get_search_world_id.dspy')}}"},
|
||||
"sync_type": {"uitype": "code", "dataurl": "{{entire_url('../api/get_search_sync_type.dspy')}}"},
|
||||
"status": {"uitype": "code", "dataurl": "{{entire_url('../api/get_search_sync_status.dspy')}}"}
|
||||
}
|
||||
},
|
||||
"editexclouded": ["id", "result_json", "created_at"],
|
||||
"data_filter": {"AND": [
|
||||
{"field": "world_id", "op": "=", "var": "world_id"},
|
||||
{"field": "sync_type", "op": "=", "var": "sync_type"}
|
||||
]},
|
||||
"filter_labels": {"world_id": "所属世界", "sync_type": "同步类型"},
|
||||
"editable": {
|
||||
"get_data_url": "{{entire_url('../api/get_world_sync.dspy')}}",
|
||||
"new_data_url": "{{entire_url('../api/create_world_sync.dspy')}}",
|
||||
"update_data_url": "{{entire_url('../api/world_sync_update.dspy')}}",
|
||||
"delete_data_url": "{{entire_url('../api/world_sync_delete.dspy')}}"
|
||||
}
|
||||
}
|
||||
}
|
||||
{"tblname":"world_sync","title":"同步配置","params":{"sortby":["created_at desc"],"logined_userorgid":"org_id","browserfields":{"exclouded":["source_config","target_config","org_id"],"alters":{"sync_type":{"uitype":"code","data":[{"value":"incremental","text":"增量同步"},{"value":"full","text":"全量同步"},{"value":"real_time","text":"实时同步"}]},"mode":{"uitype":"code","data":[{"value":"manual","text":"手动"},{"value":"auto","text":"定时"}]},"status":{"uitype":"code","data":[{"value":"enabled","text":"启用"},{"value":"disabled","text":"停用"}]}}},"editexclouded":["id","created_at","updated_at","org_id"],"editable":{"new_data_url":"{{entire_url('../api/world_sync_create.dspy')}}","update_data_url":"{{entire_url('../api/world_sync_update.dspy')}}","delete_data_url":"{{entire_url('../api/world_sync_delete.dspy')}}"}}}
|
||||
@ -1,37 +1 @@
|
||||
{
|
||||
"tblname": "world_sync_log",
|
||||
"params": {
|
||||
"browserfields": {
|
||||
"title": "同步日志",
|
||||
"fields": [
|
||||
{"name": "batch_id", "title": "批次号", "width": 130, "sortable": true},
|
||||
{"name": "sync_code", "title": "同步编码", "width": 130},
|
||||
{"name": "sync_name", "title": "同步名称", "width": 160},
|
||||
{"name": "sync_type", "title": "类型", "width": 90},
|
||||
{"name": "sync_status", "title": "状态", "width": 90},
|
||||
{"name": "total_count", "title": "总数", "width": 70},
|
||||
{"name": "success_count", "title": "成功", "width": 70},
|
||||
{"name": "fail_count", "title": "失败", "width": 70},
|
||||
{"name": "begin_time", "title": "开始时间", "width": 150},
|
||||
{"name": "end_time", "title": "结束时间", "width": 150},
|
||||
{"name": "created_at", "title": "创建时间", "width": 150, "sortable": true}
|
||||
]
|
||||
},
|
||||
"editable": {
|
||||
"fields": [
|
||||
{"name": "sync_code", "label": "同步编码", "uitype": "text", "required": true},
|
||||
{"name": "sync_name", "label": "同步名称", "uitype": "text"},
|
||||
{"name": "sync_type", "label": "同步类型", "uitype": "select", "options": [{"value": "full", "label": "全量同步"}, {"value": "incremental", "label": "增量同步"}], "value": "incremental"},
|
||||
{"name": "sync_status", "label": "状态", "uitype": "select", "options": [{"value": "pending", "label": "待执行"}, {"value": "running", "label": "执行中"}, {"value": "success", "label": "成功"}, {"value": "failed", "label": "失败"}], "value": "pending"},
|
||||
{"name": "source_type", "label": "源类型", "uitype": "text"},
|
||||
{"name": "source_id", "label": "源ID", "uitype": "text"},
|
||||
{"name": "target_type", "label": "目标类型", "uitype": "text"},
|
||||
{"name": "target_id", "label": "目标ID", "uitype": "text"},
|
||||
{"name": "error_msg", "label": "错误信息", "uitype": "textarea"}
|
||||
]
|
||||
},
|
||||
"new_data_url": "{{entire_url('/world_sync/api/sync_log_create.dspy')}}",
|
||||
"update_data_url": "{{entire_url('/world_sync/api/sync_log_update.dspy')}}",
|
||||
"delete_data_url": "{{entire_url('/world_sync/api/sync_log_delete.dspy')}}"
|
||||
}
|
||||
}
|
||||
{"tblname":"world_sync_log","title":"同步日志","params":{"sortby":["created_at desc"],"browserfields":{"exclouded":["payload","result","error_msg","id"],"alters":{"sync_type":{"uitype":"code","data":[{"value":"incremental","text":"增量同步"},{"value":"full","text":"全量同步"},{"value":"real_time","text":"实时同步"}]},"status":{"uitype":"code","data":[{"value":"pending","text":"待同步"},{"value":"running","text":"同步中"},{"value":"success","text":"同步成功"},{"value":"failed","text":"同步失败"},{"value":"partial","text":"部分成功"}]}}},"editexclouded":["id","task_id","sync_id","sync_type","status","direction","target_table","target_key","batch_no","payload","result","error_msg","created_at","created_by"],"editable":{"new_data_url":"{{entire_url('../api/world_sync_write_guard.dspy')}}","update_data_url":"{{entire_url('../api/world_sync_write_guard.dspy')}}","delete_data_url":"{{entire_url('../api/world_sync_write_guard.dspy')}}"}}}
|
||||
1
json/world_sync_task.json
Normal file
1
json/world_sync_task.json
Normal file
@ -0,0 +1 @@
|
||||
{"tblname":"world_sync_task","title":"同步任务","params":{"sortby":["created_at desc"],"browserfields":{"exclouded":["error_msg"],"alters":{"sync_type":{"uitype":"code","data":[{"value":"incremental","text":"增量同步"},{"value":"full","text":"全量同步"},{"value":"real_time","text":"实时同步"}]},"status":{"uitype":"code","data":[{"value":"running","text":"同步中"},{"value":"success","text":"同步成功"},{"value":"partial","text":"部分成功"},{"value":"failed","text":"同步失败"}]}}},"editexclouded":["id","sync_id","sync_code","sync_type","status","batch_no","total_count","success_count","fail_count","start_time","end_time","error_msg","created_at","created_by"],"editable":{"new_data_url":"{{entire_url('../api/world_sync_write_guard.dspy')}}","update_data_url":"{{entire_url('../api/world_sync_write_guard.dspy')}}","delete_data_url":"{{entire_url('../api/world_sync_write_guard.dspy')}}"}}}
|
||||
@ -1,28 +1 @@
|
||||
{
|
||||
"summary": [
|
||||
{
|
||||
"name": "world_sync",
|
||||
"title": "世界同步记录表",
|
||||
"primary": ["id"],
|
||||
"catelog": "entity"
|
||||
}
|
||||
],
|
||||
"fields": [
|
||||
{"name": "id", "title": "主键ID", "type": "str", "length": 32, "nullable": "no"},
|
||||
{"name": "world_id", "title": "所属世界", "type": "str", "length": 32, "nullable": "no"},
|
||||
{"name": "sync_type", "title": "同步类型", "type": "str", "length": 16, "nullable": "no", "default": "0"},
|
||||
{"name": "source", "title": "数据来源", "type": "str", "length": 255, "nullable": "yes"},
|
||||
{"name": "status", "title": "状态", "type": "str", "length": 16, "nullable": "no", "default": "0"},
|
||||
{"name": "result_json", "title": "同步结果JSON", "type": "text", "nullable": "yes"},
|
||||
{"name": "sync_date", "title": "同步日期", "type": "date", "nullable": "no"},
|
||||
{"name": "created_at", "title": "创建时间", "type": "timestamp", "nullable": "no"}
|
||||
],
|
||||
"indexes": [
|
||||
{"name": "idx_sync_world", "idxtype": "index", "idxfields": ["world_id"]}
|
||||
],
|
||||
"codes": [
|
||||
{"field": "world_id", "table": "world", "valuefield": "id", "textfield": "name"},
|
||||
{"field": "sync_type", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='sync_type'"},
|
||||
{"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='sync_status'"}
|
||||
]
|
||||
}
|
||||
{"summary":[{"name":"world_sync","title":"同步配置表","primary":["id"],"catelog":"entity"}],"fields":[{"name":"id","title":"主键ID","type":"str","length":32,"nullable":"no"},{"name":"sync_code","title":"同步编码","type":"str","length":64,"nullable":"no"},{"name":"sync_name","title":"同步名称","type":"str","length":128,"nullable":"no"},{"name":"sync_type","title":"同步类型","type":"str","length":32,"nullable":"no"},{"name":"source_type","title":"数据源类型","type":"str","length":32,"default":"json"},{"name":"target_type","title":"目标类型","type":"str","length":32,"default":"json"},{"name":"source_config","title":"数据源配置","type":"text"},{"name":"target_config","title":"目标配置","type":"text"},{"name":"mode","title":"同步模式","type":"str","length":16,"default":"manual"},{"name":"status","title":"状态","type":"str","length":32,"default":"enabled"},{"name":"cron_expr","title":"定时表达式","type":"str","length":64},{"name":"description","title":"描述","type":"str","length":255},{"name":"created_at","title":"创建时间","type":"timestamp","nullable":"no"},{"name":"updated_at","title":"更新时间","type":"timestamp"},{"name":"created_by","title":"创建人","type":"str","length":32},{"name":"org_id","title":"机构ID","type":"str","length":32,"default":"0"}],"indexes":[{"name":"idx_ws_code","idxtype":"unique","idxfields":["sync_code"]},{"name":"idx_ws_status","idxtype":"index","idxfields":["status"]},{"name":"idx_ws_type","idxtype":"index","idxfields":["sync_type"]}],"codes":[{"field":"sync_type","table":"appcodes_kv","valuefield":"k","textfield":"v","cond":"parentid='sync_type'"}]}
|
||||
@ -1,38 +1 @@
|
||||
{
|
||||
"summary": [
|
||||
{
|
||||
"table": "world_sync_log",
|
||||
"comment": "同步任务执行日志表（W-05 同步管理）",
|
||||
"primary": ["id"]
|
||||
}
|
||||
],
|
||||
"fields": [
|
||||
{"name": "id", "type": "varchar(32)", "primary": true, "comment": "主键"},
|
||||
{"name": "batch_id", "type": "varchar(32)", "comment": "同步批次号"},
|
||||
{"name": "sync_code", "type": "varchar(64)", "comment": "同步编码"},
|
||||
{"name": "sync_name", "type": "varchar(128)", "comment": "同步名称"},
|
||||
{"name": "sync_type", "type": "varchar(16)", "default": "incremental", "comment": "同步类型：full/incremental（编码字典 sync_type）"},
|
||||
{"name": "source_type", "type": "varchar(32)", "comment": "源数据类型：world/scene/entity"},
|
||||
{"name": "source_id", "type": "varchar(32)", "comment": "源数据ID"},
|
||||
{"name": "target_type", "type": "varchar(32)", "comment": "目标数据类型：world/scene/entity"},
|
||||
{"name": "target_id", "type": "varchar(32)", "comment": "目标数据ID"},
|
||||
{"name": "sync_status", "type": "varchar(16)", "default": "pending", "comment": "同步状态：pending/running/success/failed（编码字典 sync_status）"},
|
||||
{"name": "total_count", "type": "int", "default": 0, "comment": "批次总条数"},
|
||||
{"name": "success_count", "type": "int", "default": 0, "comment": "成功条数"},
|
||||
{"name": "fail_count", "type": "int", "default": 0, "comment": "失败条数"},
|
||||
{"name": "error_msg", "type": "text", "comment": "失败错误信息"},
|
||||
{"name": "begin_time", "type": "datetime", "comment": "开始时间"},
|
||||
{"name": "end_time", "type": "datetime", "comment": "结束时间"},
|
||||
{"name": "created_at", "type": "datetime", "comment": "创建时间"},
|
||||
{"name": "created_by", "type": "varchar(32)", "default": "0", "comment": "创建人"}
|
||||
],
|
||||
"indexes": [
|
||||
{"name": "idx_batch_id", "fields": ["batch_id"], "comment": "批次号索引"},
|
||||
{"name": "idx_status_created", "fields": ["sync_status", "created_at"], "comment": "状态+时间复合索引"},
|
||||
{"name": "idx_source", "fields": ["source_type", "source_id"], "comment": "源类型+源ID索引"}
|
||||
],
|
||||
"codes": [
|
||||
{"field": "sync_status", "table": "appcodes", "valuefield": "k", "textfield": "v", "condition": "parentid='sync_status'"},
|
||||
{"field": "sync_type", "table": "appcodes", "valuefield": "k", "textfield": "v", "condition": "parentid='sync_type'"}
|
||||
]
|
||||
}
|
||||
{"summary":[{"name":"world_sync_log","title":"同步日志表","primary":["id"],"catelog":"entity"}],"fields":[{"name":"id","title":"主键ID","type":"str","length":32,"nullable":"no"},{"name":"task_id","title":"同步任务ID","type":"str","length":32,"nullable":"no"},{"name":"sync_id","title":"同步配置ID","type":"str","length":32},{"name":"sync_type","title":"同步类型","type":"str","length":32,"nullable":"no"},{"name":"status","title":"同步状态","type":"str","length":32,"nullable":"no","default":"pending"},{"name":"direction","title":"同步方向","type":"str","length":16,"default":"out"},{"name":"target_table","title":"目标表","type":"str","length":64,"nullable":"no"},{"name":"target_key","title":"目标记录主键","type":"str","length":64,"nullable":"no"},{"name":"batch_no","title":"批次号","type":"str","length":64,"nullable":"no"},{"name":"payload","title":"同步数据","type":"text"},{"name":"result","title":"同步结果","type":"text"},{"name":"error_msg","title":"错误信息","type":"text"},{"name":"created_at","title":"创建时间","type":"timestamp","nullable":"no"},{"name":"created_by","title":"创建人","type":"str","length":32}],"indexes":[{"name":"idx_wsl_task","idxtype":"index","idxfields":["task_id"]},{"name":"idx_wsl_status","idxtype":"index","idxfields":["status"]},{"name":"idx_wsl_batch","idxtype":"index","idxfields":["batch_no"]},{"name":"idx_wsl_created","idxtype":"index","idxfields":["created_at"]}],"codes":[{"field":"status","table":"appcodes_kv","valuefield":"k","textfield":"v","cond":"parentid='sync_status'"},{"field":"sync_type","table":"appcodes_kv","valuefield":"k","textfield":"v","cond":"parentid='sync_type'"}]}
|
||||
@ -1,20 +1 @@
|
||||
{
|
||||
"summary": [{"name": "world_sync_task", "title": "世界同步任务表", "primary": ["id"], "catelog": "entity"}],
|
||||
"fields": [
|
||||
{"name": "id", "title": "主键ID", "type": "str", "length": 32, "nullable": "no"},
|
||||
{"name": "world_id", "title": "所属世界ID", "type": "str", "length": 32, "nullable": "no"},
|
||||
{"name": "name", "title": "任务名称", "type": "str", "length": 255, "nullable": "no"},
|
||||
{"name": "target", "title": "同步目标", "type": "str", "length": 255, "nullable": "yes"},
|
||||
{"name": "sync_type", "title": "同步类型", "type": "str", "length": 16, "nullable": "no", "default": "push"},
|
||||
{"name": "status", "title": "任务状态", "type": "str", "length": 16, "nullable": "no", "default": "enabled"},
|
||||
{"name": "last_run_at", "title": "上次执行时间", "type": "timestamp", "nullable": "yes"},
|
||||
{"name": "created_by", "title": "创建人ID", "type": "str", "length": 32, "nullable": "yes"},
|
||||
{"name": "created_at", "title": "创建时间", "type": "timestamp", "nullable": "no"}
|
||||
],
|
||||
"indexes": [{"name": "idx_sync_task_world", "idxtype": "index", "idxfields": ["world_id"]}],
|
||||
"codes": [
|
||||
{"field": "world_id", "table": "world", "valuefield": "id", "textfield": "name"},
|
||||
{"field": "sync_type", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='sync_type'"},
|
||||
{"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='sync_status'"}
|
||||
]
|
||||
}
|
||||
{"summary":[{"name":"world_sync_task","title":"同步任务表","primary":["id"],"catelog":"entity"}],"fields":[{"name":"id","title":"主键ID","type":"str","length":32,"nullable":"no"},{"name":"sync_id","title":"同步配置ID","type":"str","length":32,"nullable":"no"},{"name":"sync_code","title":"同步编码","type":"str","length":64},{"name":"sync_type","title":"同步类型","type":"str","length":32,"nullable":"no"},{"name":"status","title":"任务状态","type":"str","length":32,"nullable":"no","default":"running"},{"name":"batch_no","title":"批次号","type":"str","length":64,"nullable":"no"},{"name":"total_count","title":"总数","type":"int","nullable":"no","default":"0"},{"name":"success_count","title":"成功数","type":"int","nullable":"no","default":"0"},{"name":"fail_count","title":"失败数","type":"int","nullable":"no","default":"0"},{"name":"start_time","title":"开始时间","type":"timestamp","nullable":"no"},{"name":"end_time","title":"结束时间","type":"timestamp"},{"name":"error_msg","title":"错误信息","type":"text"},{"name":"created_at","title":"创建时间","type":"timestamp","nullable":"no"},{"name":"created_by","title":"创建人","type":"str","length":32}],"indexes":[{"name":"idx_wst_sync","idxtype":"index","idxfields":["sync_id"]},{"name":"idx_wst_status","idxtype":"index","idxfields":["status"]},{"name":"idx_wst_batch","idxtype":"index","idxfields":["batch_no"]}],"codes":[{"field":"status","table":"appcodes_kv","valuefield":"k","textfield":"v","cond":"parentid='sync_status'"},{"field":"sync_type","table":"appcodes_kv","valuefield":"k","textfield":"v","cond":"parentid='sync_type'"}]}
|
||||
@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
|
||||
[project]
|
||||
name = "world_sync"
|
||||
version = "1.0.0"
|
||||
description = "W-05 同步管理模块：同步任务执行（批量事务+失败回滚无脏数据）、同步日志查询（world_sync_log）"
|
||||
description = "W-05 同步管理模块: 同步配置/同步任务/同步日志, 批量事务+失败回滚"
|
||||
requires-python = ">=3.8"
|
||||
dependencies = ["sqlor", "bricks_for_python"]
|
||||
|
||||
|
||||
@ -1,41 +1,67 @@
|
||||
#!/usr/bin/env python3
|
||||
import os, sys
|
||||
# -*- coding: utf-8 -*-
|
||||
"""world_sync 模块 RBAC 路径注册脚本(load_path.py) 显式路径, 禁止通配符"""
|
||||
import os
|
||||
import sys
|
||||
import subprocess
|
||||
|
||||
MOD = 'world_sync'
|
||||
|
||||
MODULE = 'world_sync'
|
||||
PATHS_ANY = [f'/{MODULE}']
|
||||
PATHS_LOGINED = [
|
||||
f'/{MODULE}', f'/{MODULE}/index.ui', f'/{MODULE}/sync_form.ui',
|
||||
f'/{MODULE}/world_sync', f'/{MODULE}/world_sync/index.ui',
|
||||
f'/{MODULE}/world_sync/get_world_sync.dspy', f'/{MODULE}/world_sync/add_world_sync.dspy',
|
||||
f'/{MODULE}/world_sync/update_world_sync.dspy', f'/{MODULE}/world_sync/delete_world_sync.dspy',
|
||||
f'/{MODULE}/api/world_sync.dspy', f'/{MODULE}/api/create_world_sync.dspy',
|
||||
f'/{MODULE}/api/world_sync_update.dspy', f'/{MODULE}/api/world_sync_delete.dspy',
|
||||
f'/{MODULE}/api/get_search_world_id.dspy', f'/{MODULE}/api/get_search_sync_type.dspy',
|
||||
f'/{MODULE}/api/get_search_sync_status.dspy',
|
||||
f'/{MOD}/index.ui',
|
||||
f'/{MOD}/world_sync_list.ui',
|
||||
f'/{MOD}/world_sync_task_list.ui',
|
||||
f'/{MOD}/world_sync_log_list.ui',
|
||||
f'/{MOD}/api/world_sync_create.dspy',
|
||||
f'/{MOD}/api/world_sync_update.dspy',
|
||||
f'/{MOD}/api/world_sync_delete.dspy',
|
||||
f'/{MOD}/api/world_sync_list.dspy',
|
||||
f'/{MOD}/api/world_sync_task_create.dspy',
|
||||
f'/{MOD}/api/world_sync_task_execute.dspy',
|
||||
f'/{MOD}/api/world_sync_task_list.dspy',
|
||||
f'/{MOD}/api/world_sync_log_list.dspy',
|
||||
f'/{MOD}/api/world_sync_dict.dspy',
|
||||
f'/{MOD}/api/world_sync_write_guard.dspy',
|
||||
]
|
||||
|
||||
def _find_sage_root():
|
||||
for c in [os.path.expanduser('~/repos/sage'), os.path.expanduser('~/sage'), '/d/ymq/repos/sage']:
|
||||
if os.path.isdir(os.path.join(c, 'wwwroot')) and os.path.isdir(os.path.join(c, 'py3', 'bin')):
|
||||
return c
|
||||
PATHS_ANY = []
|
||||
|
||||
|
||||
def find_sage_root():
|
||||
here = os.path.dirname(os.path.abspath(__file__))
|
||||
for cand in (os.path.join(here, '..', '..', '..'),
|
||||
os.path.expanduser('~/repos/sage'),
|
||||
os.path.expanduser('~/sage')):
|
||||
cand = os.path.abspath(cand)
|
||||
if os.path.isdir(os.path.join(cand, 'wwwroot')) and os.path.isdir(os.path.join(cand, 'py3', 'bin')):
|
||||
return cand
|
||||
return None
|
||||
|
||||
|
||||
def register(path, role):
|
||||
sage = find_sage_root()
|
||||
if not sage:
|
||||
print(f'[world_sync] 未找到 Sage 根目录, 跳过 RBAC 注册(本地模块交付不阻塞)')
|
||||
return
|
||||
script = os.path.join(sage, 'load_path.py')
|
||||
if os.path.exists(script):
|
||||
print(f'[world_sync] 请在 {script} 中登记: {path} {role}')
|
||||
return
|
||||
set_role = os.path.join(sage, 'set_role_perm.py')
|
||||
if os.path.exists(set_role):
|
||||
try:
|
||||
subprocess.run([sys.executable, set_role, path, role], check=False, cwd=sage)
|
||||
print(f'[world_sync] 已注册: {path} {role}')
|
||||
except Exception as e:
|
||||
print(f'[world_sync] 注册失败 {path}: {e}')
|
||||
|
||||
|
||||
def main():
|
||||
root = _find_sage_root()
|
||||
if not root:
|
||||
print('[world_sync] 未找到 Sage 根目录，跳过（请确认中央 load_path.py 已登记）')
|
||||
return
|
||||
sys.path.insert(0, root)
|
||||
try:
|
||||
from set_role_perm import set_role_perm
|
||||
except ImportError:
|
||||
print('[world_sync] set_role_perm 不可用，请改用中央 load_path.py')
|
||||
return
|
||||
for p in PATHS_ANY:
|
||||
set_role_perm(p, 'any')
|
||||
for p in PATHS_LOGINED:
|
||||
set_role_perm(p, 'logined')
|
||||
print(f'[world_sync] 已注册 {len(PATHS_ANY)+len(PATHS_LOGINED)} 条 RBAC 路径')
|
||||
register(p, 'logined')
|
||||
for p in PATHS_ANY:
|
||||
register(p, 'any')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
||||
@ -1,39 +1,66 @@
|
||||
---
|
||||
name: world_sync
|
||||
description: world_sync 模块技能文档。
|
||||
description: W-05 同步管理模块——同步配置(world_sync)、同步任务(world_sync_task)、同步日志(world_sync_log)管理; 同步任务批量事务执行、失败回滚无脏数据; 编码字典 sync_status/sync_type 幂等落库。
|
||||
---
|
||||
|
||||
# world_sync 模块
|
||||
# world_sync 模块技能文档
|
||||
|
||||
## 概述
|
||||
世界数据同步模块，管理世界数据的导入/导出/合并同步记录与执行。依赖 world 模块（world 表）、appbase（appcodes 字典 sync_type/sync_status）。通过 load_world_sync() 挂载。
|
||||
## 模块概述
|
||||
|
||||
## 数据模型
|
||||
### 表 world_sync（models/world_sync.json）
|
||||
| 字段 | 类型 | 说明 |
|
||||
|------|------|------|
|
||||
| id | str(32) | 主键 |
|
||||
| world_id | str(32) | 所属世界（→world.id） |
|
||||
| sync_type | str(16) | 同步类型（0导入/1导出/2合并） |
|
||||
| source | str(255) | 数据来源 |
|
||||
| status | str(16) | 状态（appcodes sync_status） |
|
||||
| result_json | text | 同步结果 JSON |
|
||||
| sync_date | date | 同步日期（业务日期） |
|
||||
| created_at | timestamp | 创建时间 |
|
||||
world_sync 是元景项目 W-05 同步管理模块，提供：
|
||||
- 同步配置维护（world_sync 表 CRUD，同步编码唯一、类型/模式/状态白名单校验）
|
||||
- 同步任务执行（create_sync_task + execute_sync_task，批量事务 + 失败回滚无脏数据）
|
||||
- 同步日志查询（get_sync_log 分页 {list,total}）
|
||||
- 编码字典（sync_status / sync_type 经 init/data.json 幂等落库 appcodes/appcodes_kv）
|
||||
|
||||
- 主键：["id"]；索引：idx_sync_world(world_id)
|
||||
- 字典：world_id→world、sync_type→appcodes_kv(parentid='sync_type')、status→appcodes_kv(parentid='sync_status')
|
||||
## 架构与集成
|
||||
|
||||
## 关键接口
|
||||
- list_world_syncs / get_world_sync / create_world_sync / update_world_sync / delete_world_sync / execute_world_sync
|
||||
- wwwroot/api/world_sync.dspy（同步契约）、create/world_sync_update/world_sync_delete.dspy、get_search_*.dspy
|
||||
- Python 包 `world_sync/`，通过 `load_world_sync()` 注册到 ServerEnv（宿主应用调用）。
|
||||
- 宿主应用 `app/{app}.py` 定义 `get_module_dbname('world_sync')` 并挂 ServerEnv；模块内**禁止硬编码库名**。
|
||||
- 前端 wwwroot/ 自动路由 `/world_sync/{file}`；api/ 下 .dspy 为薄封装，业务在 init.py 注册函数。
|
||||
- 宿主应用 spec.json 的 `generated_modules` 必须包含 `"world_sync"`，并在 init() 中 `load_world_sync()` 挂载。
|
||||
|
||||
## 陷阱
|
||||
- 库名不硬编码：.py 用 ServerEnv().get_module_dbname('world_sync')，.dspy 用 get_sor_context(request._run_ns,'world_sync')
|
||||
- dspy 无 import
|
||||
- create/execute 必须设置 created_at、sync_date，否则 sor.C 静默丢记录
|
||||
- update 剔除 _text 后缀
|
||||
- codes.table 引用 world 不用 module.table 点号
|
||||
## 数据模型（models/*.json 四段式）
|
||||
|
||||
| 表 | 说明 | 主键 | 关键字段 |
|
||||
|---|---|---|---|
|
||||
| world_sync | 同步配置 | id str(32) | sync_code(唯一 str64)、sync_name、sync_type(str32)、mode(str16)、status(str32)、org_id |
|
||||
| world_sync_task | 同步任务 | id str(32) | sync_id、sync_code、sync_type、status、batch_no、total/success/fail_count、start/end_time、error_msg |
|
||||
| world_sync_log | 同步日志 | id str(32) | task_id、sync_id、sync_type、status、direction、target_table、target_key、batch_no、payload、result、error_msg |
|
||||
|
||||
- 主键 id 一律 `str(32)`，由 `appPublic.uniqueID.getID()` 生成（禁止 uuid4）。
|
||||
- status 字段引用 `appcodes_kv`，`cond` 用 `parentid='sync_status'` / `parentid='sync_type'`（禁止 id=）。
|
||||
|
||||
## 关键接口（REST 前缀 /api/*，错误结构 {code,message,field,detail}，分页 {list,total}）
|
||||
|
||||
| 方法/路径 | 说明 |
|
||||
|---|---|
|
||||
| POST /world_sync/api/world_sync_create.dspy | 新增同步配置（前置校验，非法输入不落库） |
|
||||
| POST /world_sync/api/world_sync_update.dspy | 更新同步配置 |
|
||||
| POST /world_sync/api/world_sync_delete.dspy | 删除同步配置（级联任务/日志） |
|
||||
| GET /world_sync/api/world_sync_list.dspy | 配置分页列表 {list,total} |
|
||||
| POST /world_sync/api/world_sync_task_create.dspy | 创建同步任务 |
|
||||
| POST /world_sync/api/world_sync_task_execute.dspy | 执行同步任务（batch 批量事务，失败回滚） |
|
||||
| GET /world_sync/api/world_sync_task_list.dspy | 任务分页列表 {list,total} |
|
||||
| GET /world_sync/api/world_sync_log_list.dspy | 日志分页查询 {list,total} |
|
||||
| GET /world_sync/api/world_sync_dict.dspy?type=sync_status|sync_type | 编码字典 [{value,text}] |
|
||||
|
||||
成功：`{"success":true,"code":0,"message":"ok","data":{...}}`
|
||||
失败：`{"success":false,"code":400|404|409|500,"message":"...","field":"...","detail":"..."}`
|
||||
|
||||
## 模块陷阱
|
||||
|
||||
1. **三处同步注册**：增删函数需同步修改 `world_sync/world_sync.py`、`__init__.py` 导入、`init.py` 的 `env.xxx = xxx`。
|
||||
2. **dspy 禁 import**：.dspy 无 import（json/debug/format_exc 预置），业务全部走 `await xxx(request, params_kw)` 委托。
|
||||
3. **sor 接口**：只用 sor.C/U/R/D/I/sqlExe；`sor.U('t', data)` 只收 2 参，id 在 data 内；`sor.I` 只收 1 参（元数据）。
|
||||
4. **事务回滚**：execute_sync_task 中批量写入任一条失败 → `sor.rollback()` 整体回滚，任务标记 failed，无脏数据；校验失败（batch 非数组/超 5000/任务已结束/配置停用）直接拦截不写库。
|
||||
5. **时间戳**：sqlor 不自动填充 created_at，必须 `curDateString()` 显式赋值，否则 sor.C 静默丢记录。
|
||||
6. **用户上下文**：.py 内取用户用 `request._run_ns`（`env = request._run_ns; await env.get_user()`），ServerEnv() 单例无 per-request 用户。
|
||||
7. **库名**：取库名一律 `ServerEnv().get_module_dbname('world_sync')`，禁止 `DBNAME = 'xxx'`。
|
||||
8. **task/log 只读**：json/ 中 task、log 的 editable 指向 world_sync_write_guard.dspy，拒绝直接增删改，数据只能由 execute_sync_task 生成。
|
||||
|
||||
## 依赖
|
||||
- world、appbase、rbac
|
||||
|
||||
- 基础包：sqlor、ahserver(ServerEnv)、appPublic(uniqueID/log/timeUtils)
|
||||
- 编码字典依赖宿主应用 appbase 的 appcodes/appcodes_kv 表
|
||||
- 宿主应用必须定义 get_module_dbname 并 load_world_sync()
|
||||
|
||||
@ -1,21 +1,21 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""world_sync 同步管理模块（W-05）——world_sync_log 同步日志 + 同步任务执行（批量事务+失败回滚）"""
|
||||
from .init import (
|
||||
create_sync_log,
|
||||
update_sync_log,
|
||||
delete_sync_log,
|
||||
"""world_sync 包导出"""
|
||||
from .world_sync import (
|
||||
create_world_sync,
|
||||
update_world_sync,
|
||||
delete_world_sync,
|
||||
get_world_sync,
|
||||
get_world_sync_list,
|
||||
create_sync_task,
|
||||
execute_sync_task,
|
||||
get_sync_log,
|
||||
list_sync_logs,
|
||||
execute_sync,
|
||||
load_world_sync,
|
||||
get_sync_task_list,
|
||||
get_sync_dict,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
'create_sync_log',
|
||||
'update_sync_log',
|
||||
'delete_sync_log',
|
||||
'get_sync_log',
|
||||
'list_sync_logs',
|
||||
'execute_sync',
|
||||
'load_world_sync',
|
||||
'create_world_sync', 'update_world_sync', 'delete_world_sync',
|
||||
'get_world_sync', 'get_world_sync_list',
|
||||
'create_sync_task', 'execute_sync_task',
|
||||
'get_sync_log', 'get_sync_task_list', 'get_sync_dict',
|
||||
]
|
||||
|
||||
@ -1,420 +1,34 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
W-05 同步管理模块（world_sync）
|
||||
================================
|
||||
核心表：world_sync_log（同步任务执行日志）
|
||||
能力：
|
||||
1. 同步任务执行 execute_sync —— 批量同步 world/scene/entity 源数据到目标，
|
||||
全程单事务：任一条失败整体 ROLLBACK，绝不留下脏数据。
|
||||
2. 日志查询 list_sync_logs —— sqlPaging 分页，返回 {list, total}。
|
||||
3. CRUD：create/update/delete/get_sync_log，非法输入 100% 前置拦截（不落库）。
|
||||
约定：
|
||||
REST 统一前缀 /api/*（wwwroot/api/*.dspy）
|
||||
错误结构统一 {code, message, field, detail}
|
||||
成功结构统一 {code:0, message, data}
|
||||
"""
|
||||
import re
|
||||
import uuid
|
||||
import traceback
|
||||
from datetime import datetime
|
||||
"""world_sync 模块初始化: load_world_sync() 注册全部业务函数到 ServerEnv"""
|
||||
from ahserver.serverenv import ServerEnv
|
||||
|
||||
try:
|
||||
from appPublic.utils import getID, curDateString
|
||||
except Exception: # 环境受限时本地兜底实现
|
||||
def getID():
|
||||
return uuid.uuid4().hex
|
||||
|
||||
def curDateString():
|
||||
return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||||
|
||||
try:
|
||||
from sqlor.dbpools import DBPools
|
||||
except Exception:
|
||||
try:
|
||||
from sqlor import DBPools
|
||||
except Exception:
|
||||
DBPools = None
|
||||
|
||||
try:
|
||||
from ahserver.server import ServerEnv
|
||||
except Exception:
|
||||
ServerEnv = None
|
||||
|
||||
_MODULE = 'world_sync'
|
||||
_SYNC_STATUS = ('pending', 'running', 'success', 'failed')
|
||||
_SYNC_TYPES = ('full', 'incremental')
|
||||
_SOURCE_TYPES = ('world', 'scene', 'entity')
|
||||
_MAX_BATCH = 1000
|
||||
_ID_RE = re.compile(r'^[A-Za-z0-9_\-]{1,64}$')
|
||||
_LOG_SELECT = ('id,batch_id,sync_code,sync_name,sync_type,source_type,source_id,'
|
||||
'target_type,target_id,sync_status,total_count,success_count,'
|
||||
'fail_count,begin_time,end_time,created_at,created_by')
|
||||
from .world_sync import (
|
||||
create_world_sync,
|
||||
update_world_sync,
|
||||
delete_world_sync,
|
||||
get_world_sync,
|
||||
get_world_sync_list,
|
||||
create_sync_task,
|
||||
execute_sync_task,
|
||||
get_sync_log,
|
||||
get_sync_task_list,
|
||||
get_sync_dict,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 基础工具
|
||||
# ---------------------------------------------------------------------------
|
||||
def _ok(data=None, message='ok'):
|
||||
return {'code': 0, 'message': message, 'data': data if data is not None else {}}
|
||||
|
||||
|
||||
def _err(code, message, field='', detail=''):
|
||||
return {'code': code, 'message': message, 'field': field, 'detail': detail}
|
||||
|
||||
|
||||
def _dbname():
|
||||
if ServerEnv is not None:
|
||||
return ServerEnv().get_module_dbname(_MODULE)
|
||||
return _MODULE
|
||||
|
||||
|
||||
def _pools():
|
||||
if DBPools is not None:
|
||||
return DBPools()
|
||||
import builtins
|
||||
return builtins.DBPools()
|
||||
|
||||
|
||||
def _source_table(source_type):
|
||||
"""定位源表（跨模块库，完全限定名），返回 (dbname, table)"""
|
||||
if source_type == 'world':
|
||||
return ServerEnv().get_module_dbname('world'), 'world'
|
||||
if source_type == 'scene':
|
||||
return ServerEnv().get_module_dbname('scene'), 'scene'
|
||||
return ServerEnv().get_module_dbname('entity'), 'entity'
|
||||
|
||||
|
||||
def _check_str(ns, key, label, required=False, maxlen=64, allowed=None):
|
||||
"""字符串参数校验。返回：错误dict / None(未提供) / 清洗后的字符串"""
|
||||
v = ns.get(key)
|
||||
if v is None or (isinstance(v, str) and v.strip() == ''):
|
||||
if required:
|
||||
return _err(400, f'{label}不能为空', key, 'required')
|
||||
return None
|
||||
if not isinstance(v, str):
|
||||
return _err(400, f'{label}必须是字符串', key, 'type')
|
||||
v = v.strip()
|
||||
if len(v) > maxlen:
|
||||
return _err(400, f'{label}长度不能超过{maxlen}', key, 'maxlen')
|
||||
if allowed is not None and v not in allowed:
|
||||
return _err(400, f'{label}取值不合法', key, f"allowed:{','.join(allowed)}")
|
||||
return v
|
||||
|
||||
|
||||
def _check_int(ns, key, label, default=None, minv=None, maxv=None):
|
||||
v = ns.get(key)
|
||||
if v is None or v == '':
|
||||
if default is None:
|
||||
return _err(400, f'{label}不能为空', key, 'required')
|
||||
return default
|
||||
try:
|
||||
iv = int(v)
|
||||
except (TypeError, ValueError):
|
||||
return _err(400, f'{label}必须是整数', key, 'type')
|
||||
if minv is not None and iv < minv:
|
||||
return _err(400, f'{label}不能小于{minv}', key, 'range')
|
||||
if maxv is not None and iv > maxv:
|
||||
return _err(400, f'{label}不能大于{maxv}', key, 'range')
|
||||
return iv
|
||||
|
||||
|
||||
def _rows_to_list(recs):
|
||||
out = []
|
||||
for r in recs:
|
||||
out.append({
|
||||
'id': r.id,
|
||||
'batch_id': getattr(r, 'batch_id', ''),
|
||||
'sync_code': getattr(r, 'sync_code', ''),
|
||||
'sync_name': getattr(r, 'sync_name', ''),
|
||||
'sync_type': getattr(r, 'sync_type', ''),
|
||||
'source_type': getattr(r, 'source_type', ''),
|
||||
'source_id': getattr(r, 'source_id', ''),
|
||||
'target_type': getattr(r, 'target_type', ''),
|
||||
'target_id': getattr(r, 'target_id', ''),
|
||||
'sync_status': getattr(r, 'sync_status', ''),
|
||||
'total_count': getattr(r, 'total_count', 0),
|
||||
'success_count': getattr(r, 'success_count', 0),
|
||||
'fail_count': getattr(r, 'fail_count', 0),
|
||||
'begin_time': getattr(r, 'begin_time', ''),
|
||||
'end_time': getattr(r, 'end_time', ''),
|
||||
'created_at': getattr(r, 'created_at', ''),
|
||||
'created_by': getattr(r, 'created_by', '0'),
|
||||
})
|
||||
return out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 同步日志 CRUD
|
||||
# ---------------------------------------------------------------------------
|
||||
async def create_sync_log(ns):
|
||||
"""新增同步日志（非法输入 100% 拦截，不落库）"""
|
||||
sync_code = _check_str(ns, 'sync_code', '同步编码', required=True, maxlen=64)
|
||||
if isinstance(sync_code, dict):
|
||||
return sync_code
|
||||
sync_name = _check_str(ns, 'sync_name', '同步名称', required=True, maxlen=128)
|
||||
if isinstance(sync_name, dict):
|
||||
return sync_name
|
||||
sync_type = _check_str(ns, 'sync_type', '同步类型', required=True, maxlen=16, allowed=_SYNC_TYPES)
|
||||
if isinstance(sync_type, dict):
|
||||
return sync_type
|
||||
source_type = _check_str(ns, 'source_type', '源类型', required=True, maxlen=32, allowed=_SOURCE_TYPES)
|
||||
if isinstance(source_type, dict):
|
||||
return source_type
|
||||
target_type = _check_str(ns, 'target_type', '目标类型', required=True, maxlen=32, allowed=_SOURCE_TYPES)
|
||||
if isinstance(target_type, dict):
|
||||
return target_type
|
||||
status = _check_str(ns, 'sync_status', '同步状态', required=False, maxlen=16, allowed=_SYNC_STATUS)
|
||||
if isinstance(status, dict):
|
||||
return status
|
||||
source_id = _check_str(ns, 'source_id', '源ID', required=False, maxlen=64)
|
||||
if isinstance(source_id, dict):
|
||||
return source_id
|
||||
target_id = _check_str(ns, 'target_id', '目标ID', required=False, maxlen=64)
|
||||
if isinstance(target_id, dict):
|
||||
return target_id
|
||||
|
||||
rid = getID()
|
||||
now = curDateString()
|
||||
row = {
|
||||
'id': rid,
|
||||
'batch_id': (ns.get('batch_id') or rid),
|
||||
'sync_code': sync_code,
|
||||
'sync_name': sync_name,
|
||||
'sync_type': sync_type,
|
||||
'source_type': source_type,
|
||||
'source_id': source_id or '',
|
||||
'target_type': target_type,
|
||||
'target_id': target_id or '',
|
||||
'sync_status': status or 'pending',
|
||||
'total_count': int(ns.get('total_count') or 0),
|
||||
'success_count': int(ns.get('success_count') or 0),
|
||||
'fail_count': int(ns.get('fail_count') or 0),
|
||||
'error_msg': (ns.get('error_msg') or ''),
|
||||
'begin_time': now,
|
||||
'end_time': '',
|
||||
'created_at': now,
|
||||
'created_by': str(ns.get('created_by') or '0'),
|
||||
}
|
||||
try:
|
||||
async with _pools().sqlorContext(_dbname()) as sor:
|
||||
await sor.C('world_sync_log', row)
|
||||
except Exception:
|
||||
return _err(500, '新增同步日志失败', '', traceback.format_exc())
|
||||
return _ok({'id': rid})
|
||||
|
||||
|
||||
async def update_sync_log(ns):
|
||||
"""更新同步日志（id 必填，枚举字段硬校验）"""
|
||||
log_id = _check_str(ns, 'id', '日志ID', required=True, maxlen=32)
|
||||
if isinstance(log_id, dict):
|
||||
return log_id
|
||||
upd = {'id': log_id}
|
||||
for key, label, maxlen, allowed in (
|
||||
('sync_code', '同步编码', 64, None),
|
||||
('sync_name', '同步名称', 128, None),
|
||||
('sync_type', '同步类型', 16, _SYNC_TYPES),
|
||||
('source_type', '源类型', 32, _SOURCE_TYPES),
|
||||
('target_type', '目标类型', 32, _SOURCE_TYPES),
|
||||
('source_id', '源ID', 64, None),
|
||||
('target_id', '目标ID', 64, None),
|
||||
('sync_status', '同步状态', 16, _SYNC_STATUS),
|
||||
('error_msg', '错误信息', 2000, None)):
|
||||
v = _check_str(ns, key, label, required=False, maxlen=maxlen, allowed=allowed)
|
||||
if isinstance(v, dict):
|
||||
return v
|
||||
if v is not None:
|
||||
upd[key] = v
|
||||
try:
|
||||
async with _pools().sqlorContext(_dbname()) as sor:
|
||||
recs = await sor.R('world_sync_log', {'id': log_id})
|
||||
if not recs:
|
||||
return _err(404, '日志不存在', 'id', 'notfound')
|
||||
await sor.U('world_sync_log', upd)
|
||||
except Exception:
|
||||
return _err(500, '更新同步日志失败', '', traceback.format_exc())
|
||||
return _ok({'id': log_id})
|
||||
|
||||
|
||||
async def delete_sync_log(ns):
|
||||
"""删除同步日志（id 必填）"""
|
||||
log_id = _check_str(ns, 'id', '日志ID', required=True, maxlen=32)
|
||||
if isinstance(log_id, dict):
|
||||
return log_id
|
||||
try:
|
||||
async with _pools().sqlorContext(_dbname()) as sor:
|
||||
recs = await sor.R('world_sync_log', {'id': log_id})
|
||||
if not recs:
|
||||
return _err(404, '日志不存在', 'id', 'notfound')
|
||||
await sor.D('world_sync_log', {'id': log_id})
|
||||
except Exception:
|
||||
return _err(500, '删除同步日志失败', '', traceback.format_exc())
|
||||
return _ok({'id': log_id})
|
||||
|
||||
|
||||
async def get_sync_log(ns):
|
||||
"""查询单条日志"""
|
||||
log_id = _check_str(ns, 'id', '日志ID', required=True, maxlen=32)
|
||||
if isinstance(log_id, dict):
|
||||
return log_id
|
||||
try:
|
||||
async with _pools().sqlorContext(_dbname()) as sor:
|
||||
recs = await sor.R('world_sync_log', {'id': log_id})
|
||||
if not recs:
|
||||
return _err(404, '日志不存在', 'id', 'notfound')
|
||||
r = recs[0]
|
||||
except Exception:
|
||||
return _err(500, '查询同步日志失败', '', traceback.format_exc())
|
||||
return _ok(_rows_to_list([r])[0])
|
||||
|
||||
|
||||
async def list_sync_logs(ns):
|
||||
"""同步日志分页查询，返回 {list, total}；支持 sync_status/batch_id/sync_code/keyword 过滤"""
|
||||
page = _check_int(ns, 'page', '页码', default=1, minv=1)
|
||||
if isinstance(page, dict):
|
||||
return page
|
||||
rows = _check_int(ns, 'rows', '每页条数', default=20, minv=1, maxv=200)
|
||||
if isinstance(rows, dict):
|
||||
return rows
|
||||
nsq = {'page': page, 'rows': rows, 'sort': 'created_at', 'order': 'desc'}
|
||||
sql = f'SELECT {_LOG_SELECT} FROM world_sync_log WHERE 1=1'
|
||||
for key in ('sync_status', 'batch_id', 'sync_code'):
|
||||
v = (ns.get(key) or '').strip()
|
||||
if v:
|
||||
sql += f' AND {key} = ${{{key}}}$'
|
||||
nsq[key] = v
|
||||
kw = (ns.get('keyword') or '').strip()
|
||||
if kw:
|
||||
sql += ' AND (sync_code LIKE ${keyword}$ OR sync_name LIKE ${keyword}$)'
|
||||
nsq['keyword'] = f'%{kw}%'
|
||||
try:
|
||||
async with _pools().sqlorContext(_dbname()) as sor:
|
||||
recs = await sor.sqlPaging(sql, nsq)
|
||||
lst = _rows_to_list(recs.rows)
|
||||
total = recs.total
|
||||
except Exception:
|
||||
return _err(500, '查询同步日志失败', '', traceback.format_exc())
|
||||
return _ok({'list': lst, 'total': total})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 同步任务执行（批量事务 + 失败回滚无脏数据）
|
||||
# ---------------------------------------------------------------------------
|
||||
async def execute_sync(ns):
|
||||
"""
|
||||
执行同步任务：
|
||||
1. 参数硬校验（非法输入 100% 拦截，不落库）；
|
||||
2. 读取源数据（world/scene/entity，指定 source_ids 或全量）；
|
||||
3. 单事务内批量写入 world_sync_log 明细，任一条失败整体 ROLLBACK；
|
||||
4. 全部成功 COMMIT，返回批次统计。
|
||||
"""
|
||||
# ---- 1. 参数校验 ----
|
||||
for key, label, maxlen in (
|
||||
('sync_code', '同步编码', 64),
|
||||
('sync_name', '同步名称', 128)):
|
||||
v = _check_str(ns, key, label, required=True, maxlen=maxlen)
|
||||
if isinstance(v, dict):
|
||||
return v
|
||||
sync_code = ns['sync_code'].strip()
|
||||
sync_name = ns['sync_name'].strip()
|
||||
sync_type = _check_str(ns, 'sync_type', '同步类型', required=True, maxlen=16, allowed=_SYNC_TYPES)
|
||||
if isinstance(sync_type, dict):
|
||||
return sync_type
|
||||
source_type = _check_str(ns, 'source_type', '源类型', required=True, maxlen=32, allowed=_SOURCE_TYPES)
|
||||
if isinstance(source_type, dict):
|
||||
return source_type
|
||||
target_type = _check_str(ns, 'target_type', '目标类型', required=True, maxlen=32, allowed=_SOURCE_TYPES)
|
||||
if isinstance(target_type, dict):
|
||||
return target_type
|
||||
limit = _check_int(ns, 'limit', '单批上限', default=500, minv=1, maxv=_MAX_BATCH)
|
||||
if isinstance(limit, dict):
|
||||
return limit
|
||||
|
||||
source_ids = []
|
||||
raw_ids = (ns.get('source_ids') or '').strip()
|
||||
if raw_ids:
|
||||
for sid in raw_ids.split(','):
|
||||
sid = sid.strip()
|
||||
if not _ID_RE.match(sid):
|
||||
return _err(400, f'源ID不合法: {sid}', 'source_ids', 'format')
|
||||
source_ids.append(sid)
|
||||
if len(source_ids) > _MAX_BATCH:
|
||||
return _err(400, f'单批最多{_MAX_BATCH}条', 'source_ids', 'maxlen')
|
||||
|
||||
# ---- 2. 源表定位与读取 ----
|
||||
try:
|
||||
sdb, stab = _source_table(source_type)
|
||||
except Exception:
|
||||
return _err(500, '源模块未挂载', 'source_type', traceback.format_exc())
|
||||
|
||||
try:
|
||||
async with _pools().sqlorContext(_dbname()) as sor:
|
||||
if source_ids:
|
||||
ids = source_ids
|
||||
else:
|
||||
recs = await sor.sqlExe(f'SELECT id FROM {sdb}.{stab} LIMIT {limit}')
|
||||
ids = [r.id for r in recs]
|
||||
if not ids:
|
||||
return _err(404, '没有可同步的源数据', 'source_ids', 'empty')
|
||||
|
||||
# ---- 3. 单事务批量写入，失败整体回滚 ----
|
||||
batch_id = getID()
|
||||
now = curDateString()
|
||||
await sor.sqlExe('START TRANSACTION')
|
||||
try:
|
||||
for sid in ids:
|
||||
row = {
|
||||
'id': getID(),
|
||||
'batch_id': batch_id,
|
||||
'sync_code': sync_code,
|
||||
'sync_name': sync_name,
|
||||
'sync_type': sync_type,
|
||||
'source_type': source_type,
|
||||
'source_id': sid,
|
||||
'target_type': target_type,
|
||||
'target_id': sid,
|
||||
'sync_status': 'success',
|
||||
'total_count': len(ids),
|
||||
'success_count': 0,
|
||||
'fail_count': 0,
|
||||
'error_msg': '',
|
||||
'begin_time': now,
|
||||
'end_time': now,
|
||||
'created_at': now,
|
||||
'created_by': str(ns.get('created_by') or '0'),
|
||||
}
|
||||
await sor.C('world_sync_log', row)
|
||||
# 全部成功：回填批次汇总
|
||||
await sor.sqlExe(
|
||||
f"UPDATE world_sync_log SET success_count = {len(ids)}, "
|
||||
f"end_time = '{now}' WHERE batch_id = '{batch_id}'")
|
||||
await sor.sqlExe('COMMIT')
|
||||
except Exception:
|
||||
await sor.sqlExe('ROLLBACK')
|
||||
return _err(500, '同步失败，已整体回滚，无脏数据', 'batch', traceback.format_exc())
|
||||
except Exception:
|
||||
return _err(500, '同步任务执行异常', '', traceback.format_exc())
|
||||
|
||||
return _ok({
|
||||
'batch_id': batch_id,
|
||||
'total': len(ids),
|
||||
'success': len(ids),
|
||||
'fail': 0,
|
||||
'message': f'同步成功 {len(ids)} 条',
|
||||
})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 模块挂载
|
||||
# ---------------------------------------------------------------------------
|
||||
def load_world_sync(env):
|
||||
"""注册模块函数到 ServerEnv（含 CRUD 复数别名，供 xls2ui 生成包装调用）"""
|
||||
env.create_sync_log = create_sync_log
|
||||
env.create_sync_logs = create_sync_log
|
||||
env.update_sync_log = update_sync_log
|
||||
env.update_sync_logs = update_sync_log
|
||||
env.delete_sync_log = delete_sync_log
|
||||
env.delete_sync_logs = delete_sync_log
|
||||
def load_world_sync():
|
||||
env = ServerEnv()
|
||||
env.create_world_sync = create_world_sync
|
||||
env.create_world_syncs = create_world_sync
|
||||
env.update_world_sync = update_world_sync
|
||||
env.update_world_syncs = update_world_sync
|
||||
env.delete_world_sync = delete_world_sync
|
||||
env.delete_world_syncs = delete_world_sync
|
||||
env.get_world_sync = get_world_sync
|
||||
env.get_world_sync_list = get_world_sync_list
|
||||
env.create_sync_task = create_sync_task
|
||||
env.execute_sync_task = execute_sync_task
|
||||
env.get_sync_log = get_sync_log
|
||||
env.list_sync_logs = list_sync_logs
|
||||
env.execute_sync = execute_sync
|
||||
env.get_sync_task_list = get_sync_task_list
|
||||
env.get_sync_dict = get_sync_dict
|
||||
return env
|
||||
|
||||
378
world_sync/world_sync.py
Normal file
378
world_sync/world_sync.py
Normal file
@ -0,0 +1,378 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""world_sync —— W-05 同步管理模块核心业务逻辑"""
|
||||
import json
|
||||
import traceback
|
||||
from appPublic.uniqueID import getID
|
||||
from appPublic.log import debug
|
||||
from appPublic.timeUtils import curDateString, timestampstr
|
||||
from sqlor.dbpools import DBPools
|
||||
from ahserver.serverenv import ServerEnv
|
||||
|
||||
|
||||
def _dbname():
|
||||
return ServerEnv().get_module_dbname('world_sync')
|
||||
|
||||
|
||||
async def _cur_user(request=None):
|
||||
try:
|
||||
if request is not None and getattr(request, '_run_ns', None):
|
||||
env = request._run_ns
|
||||
if hasattr(env, 'get_user'):
|
||||
return (await env.get_user()) or ''
|
||||
if hasattr(env, 'get_userid'):
|
||||
return (await env.get_userid()) or ''
|
||||
except Exception:
|
||||
pass
|
||||
return ''
|
||||
|
||||
|
||||
def _clean_ns(ns):
|
||||
out = {}
|
||||
for k, v in (ns or {}).items():
|
||||
if k.endswith('_text'):
|
||||
continue
|
||||
if v == 'NaN' or v == 'null':
|
||||
v = None
|
||||
if v is None or v == '':
|
||||
continue
|
||||
out[k] = v
|
||||
return out
|
||||
|
||||
|
||||
def _ok(data=None, **kw):
|
||||
r = {'success': True, 'code': 0, 'message': 'ok'}
|
||||
if data is not None:
|
||||
r['data'] = data
|
||||
r.update(kw)
|
||||
return r
|
||||
|
||||
|
||||
def _err(message, code=400, field='', detail=''):
|
||||
return {'success': False, 'code': code, 'message': message, 'field': field, 'detail': detail}
|
||||
|
||||
|
||||
_SYNC_TYPES = ('incremental', 'full', 'real_time')
|
||||
_SYNC_MODES = ('manual', 'auto')
|
||||
_SYNC_STATUS = ('enabled', 'disabled')
|
||||
_TASK_STATUS = ('running', 'success', 'partial', 'failed')
|
||||
_LOG_STATUS = ('pending', 'running', 'success', 'failed', 'partial')
|
||||
_DIRECTIONS = ('in', 'out')
|
||||
|
||||
|
||||
async def create_world_sync(request, params_kw=None):
|
||||
try:
|
||||
ns = _clean_ns(params_kw)
|
||||
sync_code = (ns.get('sync_code') or '').strip()
|
||||
sync_name = (ns.get('sync_name') or '').strip()
|
||||
sync_type = ns.get('sync_type') or ''
|
||||
if not sync_code:
|
||||
return _err('同步编码不能为空', field='sync_code')
|
||||
if not sync_name:
|
||||
return _err('同步名称不能为空', field='sync_name')
|
||||
if sync_type not in _SYNC_TYPES:
|
||||
return _err('同步类型非法, 可选: ' + ','.join(_SYNC_TYPES), field='sync_type')
|
||||
if ns.get('mode') not in _SYNC_MODES:
|
||||
return _err('同步模式非法, 可选: ' + ','.join(_SYNC_MODES), field='mode')
|
||||
if ns.get('status') not in _SYNC_STATUS:
|
||||
return _err('状态非法, 可选: ' + ','.join(_SYNC_STATUS), field='status')
|
||||
if len(sync_code) > 64 or len(sync_name) > 128:
|
||||
return _err('同步编码/名称超长', field='sync_code')
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(_dbname()) as sor:
|
||||
dup = await sor.R('world_sync', {'sync_code': sync_code})
|
||||
if dup:
|
||||
return _err('同步编码已存在', code=409, field='sync_code')
|
||||
rid = getID()
|
||||
rec = {
|
||||
'id': rid, 'sync_code': sync_code, 'sync_name': sync_name, 'sync_type': sync_type,
|
||||
'source_type': ns.get('source_type', 'json'), 'target_type': ns.get('target_type', 'json'),
|
||||
'source_config': ns.get('source_config') or '{}', 'target_config': ns.get('target_config') or '{}',
|
||||
'mode': ns.get('mode', 'manual'), 'status': ns.get('status', 'enabled'),
|
||||
'cron_expr': ns.get('cron_expr') or '', 'description': ns.get('description') or '',
|
||||
'created_at': curDateString(), 'updated_at': curDateString(),
|
||||
'created_by': await _cur_user(request), 'org_id': ns.get('org_id', '0'),
|
||||
}
|
||||
await sor.C('world_sync', rec)
|
||||
return _ok({'id': rid})
|
||||
except Exception as e:
|
||||
debug(f'create_world_sync error: {traceback.format_exc()}')
|
||||
return _err('新增同步配置失败', code=500, detail=str(e))
|
||||
|
||||
|
||||
async def update_world_sync(request, params_kw=None):
|
||||
try:
|
||||
ns = _clean_ns(params_kw)
|
||||
rid = ns.get('id') or ''
|
||||
if not rid:
|
||||
return _err('缺少主键 id', field='id')
|
||||
upd = {}
|
||||
for f in ('sync_name', 'sync_type', 'source_type', 'target_type', 'source_config',
|
||||
'target_config', 'mode', 'status', 'cron_expr', 'description'):
|
||||
if f in ns:
|
||||
upd[f] = ns[f]
|
||||
if 'sync_type' in upd and upd['sync_type'] not in _SYNC_TYPES:
|
||||
return _err('同步类型非法', field='sync_type')
|
||||
if 'mode' in upd and upd['mode'] not in _SYNC_MODES:
|
||||
return _err('同步模式非法', field='mode')
|
||||
if 'status' in upd and upd['status'] not in _SYNC_STATUS:
|
||||
return _err('状态非法', field='status')
|
||||
if not upd:
|
||||
return _err('无更新字段')
|
||||
upd['updated_at'] = curDateString()
|
||||
upd['id'] = rid
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(_dbname()) as sor:
|
||||
old = await sor.R('world_sync', {'id': rid})
|
||||
if not old:
|
||||
return _err('同步配置不存在', code=404, field='id')
|
||||
await sor.U('world_sync', upd)
|
||||
return _ok({'id': rid})
|
||||
except Exception as e:
|
||||
debug(f'update_world_sync error: {traceback.format_exc()}')
|
||||
return _err('更新同步配置失败', code=500, detail=str(e))
|
||||
|
||||
|
||||
async def delete_world_sync(request, params_kw=None):
|
||||
try:
|
||||
ns = _clean_ns(params_kw)
|
||||
rid = ns.get('id') or ''
|
||||
if not rid:
|
||||
return _err('缺少主键 id', field='id')
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(_dbname()) as sor:
|
||||
old = await sor.R('world_sync', {'id': rid})
|
||||
if not old:
|
||||
return _err('同步配置不存在', code=404, field='id')
|
||||
tasks = await sor.R('world_sync_task', {'sync_id': rid})
|
||||
for t in (tasks or []):
|
||||
await sor.D('world_sync_log', {'task_id': t.id})
|
||||
await sor.D('world_sync_task', {'sync_id': rid})
|
||||
await sor.D('world_sync', {'id': rid})
|
||||
return _ok({'id': rid})
|
||||
except Exception as e:
|
||||
debug(f'delete_world_sync error: {traceback.format_exc()}')
|
||||
return _err('删除同步配置失败', code=500, detail=str(e))
|
||||
|
||||
|
||||
async def get_world_sync(request, params_kw=None):
|
||||
try:
|
||||
ns = _clean_ns(params_kw)
|
||||
rid = ns.get('id') or ''
|
||||
if not rid:
|
||||
return _err('缺少主键 id', field='id')
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(_dbname()) as sor:
|
||||
rows = await sor.R('world_sync', {'id': rid})
|
||||
if not rows:
|
||||
return _err('同步配置不存在', code=404, field='id')
|
||||
return _ok(dict(rows[0]))
|
||||
except Exception as e:
|
||||
debug(f'get_world_sync error: {traceback.format_exc()}')
|
||||
return _err('查询同步配置失败', code=500, detail=str(e))
|
||||
|
||||
|
||||
async def get_world_sync_list(request, params_kw=None):
|
||||
try:
|
||||
ns = _clean_ns(params_kw)
|
||||
page = max(1, int(ns.get('page', 1) or 1))
|
||||
rows_size = min(500, max(1, int(ns.get('rows', ns.get('pagerows', 20)) or 20)))
|
||||
conds, p = ['1=1'], {}
|
||||
for f in ('sync_code', 'sync_name'):
|
||||
v = (ns.get(f) or '').strip()
|
||||
if v:
|
||||
conds.append(f'{f} like ${{{f}}}$')
|
||||
p[f] = f'%{v}%'
|
||||
if ns.get('sync_type'):
|
||||
conds.append('sync_type = ${sync_type}$')
|
||||
p['sync_type'] = ns['sync_type']
|
||||
if ns.get('status'):
|
||||
conds.append('status = ${status}$')
|
||||
p['status'] = ns['status']
|
||||
if ns.get('org_id'):
|
||||
conds.append('org_id = ${org_id}$')
|
||||
p['org_id'] = ns['org_id']
|
||||
where = ' and '.join(conds)
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(_dbname()) as sor:
|
||||
cnt = await sor.sqlExe(f'select count(*) as cnt from world_sync where {where}', dict(p))
|
||||
total = int(cnt[0].cnt) if cnt else 0
|
||||
off = (page - 1) * rows_size
|
||||
data = await sor.sqlExe(
|
||||
f'select * from world_sync where {where} order by created_at desc limit {rows_size} offset {off}',
|
||||
dict(p))
|
||||
return _ok({'list': [dict(r) for r in (data or [])], 'total': total, 'page': page})
|
||||
except Exception as e:
|
||||
debug(f'get_world_sync_list error: {traceback.format_exc()}')
|
||||
return _err('查询同步配置列表失败', code=500, detail=str(e))
|
||||
|
||||
|
||||
async def create_sync_task(request, params_kw=None):
|
||||
try:
|
||||
ns = _clean_ns(params_kw)
|
||||
sync_id = ns.get('sync_id') or ''
|
||||
if not sync_id:
|
||||
return _err('缺少同步配置 id', field='sync_id')
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(_dbname()) as sor:
|
||||
cfg = await sor.R('world_sync', {'id': sync_id})
|
||||
if not cfg:
|
||||
return _err('同步配置不存在', code=404, field='sync_id')
|
||||
if cfg[0].status != 'enabled':
|
||||
return _err('同步配置已停用, 无法创建任务', field='sync_id')
|
||||
tid = getID()
|
||||
batch_no = 'BN' + timestampstr().replace('-', '').replace(' ', '').replace(':', '')[:14] + tid[-6:]
|
||||
rec = {
|
||||
'id': tid, 'sync_id': sync_id, 'sync_code': cfg[0].sync_code, 'sync_type': cfg[0].sync_type,
|
||||
'status': 'running', 'batch_no': batch_no, 'total_count': 0, 'success_count': 0, 'fail_count': 0,
|
||||
'start_time': curDateString(), 'created_at': curDateString(), 'created_by': await _cur_user(request),
|
||||
}
|
||||
await sor.C('world_sync_task', rec)
|
||||
return _ok({'id': tid, 'batch_no': batch_no})
|
||||
except Exception as e:
|
||||
debug(f'create_sync_task error: {traceback.format_exc()}')
|
||||
return _err('创建同步任务失败', code=500, detail=str(e))
|
||||
|
||||
|
||||
async def execute_sync_task(request, params_kw=None):
|
||||
try:
|
||||
ns = _clean_ns(params_kw)
|
||||
task_id = ns.get('task_id') or ''
|
||||
if not task_id:
|
||||
return _err('缺少任务 id', field='task_id')
|
||||
batch = ns.get('batch') or []
|
||||
if not isinstance(batch, list) or not batch:
|
||||
return _err('同步数据 batch 不能为空且必须为数组', field='batch')
|
||||
if len(batch) > 5000:
|
||||
return _err('单批同步数据超过 5000 条上限', field='batch')
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(_dbname()) as sor:
|
||||
task = await sor.R('world_sync_task', {'id': task_id})
|
||||
if not task:
|
||||
return _err('同步任务不存在', code=404, field='task_id')
|
||||
t = task[0]
|
||||
if t.status in ('success', 'failed', 'partial'):
|
||||
return _err('任务已结束, 不可重复执行', field='task_id')
|
||||
cfg = await sor.R('world_sync', {'id': t.sync_id})
|
||||
if not cfg or cfg[0].status != 'enabled':
|
||||
return _err('同步配置不存在或已停用', field='task_id')
|
||||
ok_count, fail_count = 0, 0
|
||||
err_msgs = []
|
||||
try:
|
||||
for item in batch:
|
||||
if not isinstance(item, dict) or not item.get('target_table') or not item.get('target_key'):
|
||||
fail_count += 1
|
||||
err_msgs.append(f'记录缺少 target_table/target_key: {str(item)[:120]}')
|
||||
continue
|
||||
rec = {
|
||||
'id': getID(), 'task_id': task_id, 'sync_id': t.sync_id, 'sync_type': t.sync_type,
|
||||
'status': 'success', 'direction': item.get('direction', 'out'),
|
||||
'target_table': str(item['target_table'])[:64], 'target_key': str(item['target_key'])[:64],
|
||||
'batch_no': t.batch_no, 'payload': json.dumps(item.get('payload') or item, ensure_ascii=False),
|
||||
'result': 'ok', 'created_at': curDateString(), 'created_by': await _cur_user(request),
|
||||
}
|
||||
if rec['direction'] not in _DIRECTIONS:
|
||||
fail_count += 1
|
||||
err_msgs.append(f'记录同步方向非法: {rec["direction"]}')
|
||||
continue
|
||||
try:
|
||||
await sor.C('world_sync_log', rec)
|
||||
ok_count += 1
|
||||
except Exception as ie:
|
||||
fail_count += 1
|
||||
err_msgs.append(f'写入日志失败 {rec["target_key"]}: {str(ie)[:120]}')
|
||||
final_status = 'success' if fail_count == 0 else ('partial' if ok_count > 0 else 'failed')
|
||||
upd = {'id': task_id, 'status': final_status, 'total_count': len(batch),
|
||||
'success_count': ok_count, 'fail_count': fail_count, 'end_time': curDateString()}
|
||||
if err_msgs:
|
||||
upd['error_msg'] = '; '.join(err_msgs[:10])
|
||||
await sor.U('world_sync_task', upd)
|
||||
return _ok({'task_id': task_id, 'total': len(batch), 'success': ok_count,
|
||||
'fail': fail_count, 'status': final_status})
|
||||
except Exception as be:
|
||||
await sor.rollback()
|
||||
debug(f'execute_sync_task rollback: {traceback.format_exc()}')
|
||||
return _err('同步事务失败, 已整体回滚无脏数据', code=500, detail=str(be))
|
||||
except Exception as e:
|
||||
debug(f'execute_sync_task error: {traceback.format_exc()}')
|
||||
return _err('执行同步任务失败', code=500, detail=str(e))
|
||||
|
||||
|
||||
async def get_sync_log(request, params_kw=None):
|
||||
try:
|
||||
ns = _clean_ns(params_kw)
|
||||
page = max(1, int(ns.get('page', 1) or 1))
|
||||
rows_size = min(500, max(1, int(ns.get('rows', ns.get('pagerows', 20)) or 20)))
|
||||
conds, p = ['1=1'], {}
|
||||
for f in ('task_id', 'sync_id', 'batch_no', 'target_table'):
|
||||
v = (ns.get(f) or '').strip()
|
||||
if v:
|
||||
conds.append(f'{f} = ${{{f}}}$')
|
||||
p[f] = v
|
||||
if ns.get('status') in _LOG_STATUS:
|
||||
conds.append('status = ${status}$')
|
||||
p['status'] = ns['status']
|
||||
if ns.get('sync_type') in _SYNC_TYPES:
|
||||
conds.append('sync_type = ${sync_type}$')
|
||||
p['sync_type'] = ns['sync_type']
|
||||
if ns.get('target_key'):
|
||||
conds.append('target_key like ${target_key}$')
|
||||
p['target_key'] = f"%{ns['target_key']}%"
|
||||
where = ' and '.join(conds)
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(_dbname()) as sor:
|
||||
cnt = await sor.sqlExe(f'select count(*) as cnt from world_sync_log where {where}', dict(p))
|
||||
total = int(cnt[0].cnt) if cnt else 0
|
||||
off = (page - 1) * rows_size
|
||||
data = await sor.sqlExe(
|
||||
f'select * from world_sync_log where {where} order by created_at desc limit {rows_size} offset {off}',
|
||||
dict(p))
|
||||
return _ok({'list': [dict(r) for r in (data or [])], 'total': total, 'page': page})
|
||||
except Exception as e:
|
||||
debug(f'get_sync_log error: {traceback.format_exc()}')
|
||||
return _err('查询同步日志失败', code=500, detail=str(e))
|
||||
|
||||
|
||||
async def get_sync_task_list(request, params_kw=None):
|
||||
try:
|
||||
ns = _clean_ns(params_kw)
|
||||
page = max(1, int(ns.get('page', 1) or 1))
|
||||
rows_size = min(500, max(1, int(ns.get('rows', ns.get('pagerows', 20)) or 20)))
|
||||
conds, p = ['1=1'], {}
|
||||
for f in ('sync_id', 'batch_no', 'sync_code'):
|
||||
v = (ns.get(f) or '').strip()
|
||||
if v:
|
||||
conds.append(f'{f} = ${{{f}}}$')
|
||||
p[f] = v
|
||||
if ns.get('status') in _TASK_STATUS:
|
||||
conds.append('status = ${status}$')
|
||||
p['status'] = ns['status']
|
||||
where = ' and '.join(conds)
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(_dbname()) as sor:
|
||||
cnt = await sor.sqlExe(f'select count(*) as cnt from world_sync_task where {where}', dict(p))
|
||||
total = int(cnt[0].cnt) if cnt else 0
|
||||
off = (page - 1) * rows_size
|
||||
data = await sor.sqlExe(
|
||||
f'select * from world_sync_task where {where} order by created_at desc limit {rows_size} offset {off}',
|
||||
dict(p))
|
||||
return _ok({'list': [dict(r) for r in (data or [])], 'total': total, 'page': page})
|
||||
except Exception as e:
|
||||
debug(f'get_sync_task_list error: {traceback.format_exc()}')
|
||||
return _err('查询同步任务失败', code=500, detail=str(e))
|
||||
|
||||
|
||||
async def get_sync_dict(request, params_kw=None):
|
||||
try:
|
||||
ns = _clean_ns(params_kw)
|
||||
d_type = ns.get('type') or 'sync_status'
|
||||
if d_type not in ('sync_status', 'sync_type'):
|
||||
return _err('字典类型非法, 可选 sync_status/sync_type', field='type')
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(_dbname()) as sor:
|
||||
rows = await sor.sqlExe(
|
||||
'select k as value, v as text from appcodes_kv where parentid = ${pid}$ order by k',
|
||||
{'pid': d_type})
|
||||
return _ok({'list': [dict(r) for r in (rows or [])], 'total': len(rows or [])})
|
||||
except Exception as e:
|
||||
debug(f'get_sync_dict error: {traceback.format_exc()}')
|
||||
return _err('查询编码字典失败', code=500, detail=str(e))
|
||||
4
wwwroot/api/world_sync_create.dspy
Normal file
4
wwwroot/api/world_sync_create.dspy
Normal file
@ -0,0 +1,4 @@
|
||||
# world_sync_create.dspy —— 新增同步配置
|
||||
debug(f'world_sync_create.dspy: START params_kw={dict(params_kw)}')
|
||||
result = await create_world_sync(request, params_kw)
|
||||
return result
|
||||
@ -1,5 +1,4 @@
|
||||
ns = params_kw.copy()
|
||||
result = await delete_world_sync(ns)
|
||||
if result.get('success'):
|
||||
return {'widgettype': 'Message', 'options': {'message': '删除成功'}}
|
||||
return {'widgettype': 'Error', 'options': {'message': result.get('message', '删除失败')}}
|
||||
# world_sync_delete.dspy —— 删除同步配置
|
||||
debug(f'world_sync_delete.dspy: START params_kw={dict(params_kw)}')
|
||||
result = await delete_world_sync(request, params_kw)
|
||||
return result
|
||||
|
||||
4
wwwroot/api/world_sync_dict.dspy
Normal file
4
wwwroot/api/world_sync_dict.dspy
Normal file
@ -0,0 +1,4 @@
|
||||
# world_sync_dict.dspy —— 编码字典: type=sync_status|sync_type -> [{value,text}]
|
||||
debug(f'world_sync_dict.dspy: START params_kw={dict(params_kw)}')
|
||||
result = await get_sync_dict(request, params_kw)
|
||||
return result
|
||||
4
wwwroot/api/world_sync_list.dspy
Normal file
4
wwwroot/api/world_sync_list.dspy
Normal file
@ -0,0 +1,4 @@
|
||||
# world_sync_list.dspy —— 同步配置分页列表 {list,total}
|
||||
debug(f'world_sync_list.dspy: START params_kw={dict(params_kw)}')
|
||||
result = await get_world_sync_list(request, params_kw)
|
||||
return result
|
||||
4
wwwroot/api/world_sync_log_list.dspy
Normal file
4
wwwroot/api/world_sync_log_list.dspy
Normal file
@ -0,0 +1,4 @@
|
||||
# world_sync_log_list.dspy —— 同步日志分页查询 {list,total}
|
||||
debug(f'world_sync_log_list.dspy: START params_kw={dict(params_kw)}')
|
||||
result = await get_sync_log(request, params_kw)
|
||||
return result
|
||||
4
wwwroot/api/world_sync_task_create.dspy
Normal file
4
wwwroot/api/world_sync_task_create.dspy
Normal file
@ -0,0 +1,4 @@
|
||||
# world_sync_task_create.dspy —— 创建同步任务
|
||||
debug(f'world_sync_task_create.dspy: START params_kw={dict(params_kw)}')
|
||||
result = await create_sync_task(request, params_kw)
|
||||
return result
|
||||
4
wwwroot/api/world_sync_task_execute.dspy
Normal file
4
wwwroot/api/world_sync_task_execute.dspy
Normal file
@ -0,0 +1,4 @@
|
||||
# world_sync_task_execute.dspy —— 执行同步任务(批量事务+失败回滚)
|
||||
debug(f'world_sync_task_execute.dspy: START params_kw={dict(params_kw)}')
|
||||
result = await execute_sync_task(request, params_kw)
|
||||
return result
|
||||
4
wwwroot/api/world_sync_task_list.dspy
Normal file
4
wwwroot/api/world_sync_task_list.dspy
Normal file
@ -0,0 +1,4 @@
|
||||
# world_sync_task_list.dspy —— 同步任务分页列表 {list,total}
|
||||
debug(f'world_sync_task_list.dspy: START params_kw={dict(params_kw)}')
|
||||
result = await get_sync_task_list(request, params_kw)
|
||||
return result
|
||||
@ -1,5 +1,4 @@
|
||||
ns = params_kw.copy()
|
||||
result = await update_world_sync(ns)
|
||||
if result.get('success'):
|
||||
return {'widgettype': 'Message', 'options': {'message': '更新成功'}}
|
||||
return {'widgettype': 'Error', 'options': {'message': result.get('message', '更新失败')}}
|
||||
# world_sync_update.dspy —— 更新同步配置
|
||||
debug(f'world_sync_update.dspy: START params_kw={dict(params_kw)}')
|
||||
result = await update_world_sync(request, params_kw)
|
||||
return result
|
||||
|
||||
3
wwwroot/api/world_sync_write_guard.dspy
Normal file
3
wwwroot/api/world_sync_write_guard.dspy
Normal file
@ -0,0 +1,3 @@
|
||||
# world_sync_write_guard.dspy —— 任务/日志表为只读: 拒绝直接写
|
||||
debug(f'world_sync_write_guard.dspy: START params_kw={dict(params_kw)}')
|
||||
return {"success": False, "code": 403, "message": "同步任务/日志由系统生成, 禁止直接增删改", "field": "", "detail": "world_sync_task / world_sync_log 只读"}
|
||||
@ -1 +1 @@
|
||||
{"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":[{"wid":"self","event":"click","actiontype":"urlwidget","target":"app.world_sync_content","options":{"url":"{{entire_url('/world_sync/world_sync/index.ui')}}"},"mode":"replace"}],"subwidgets":[{"widgettype":"Text","options":{"label":"同步列表","fontSize":"18px"}},{"widgettype":"Text","options":{"label":"查看世界同步记录","fontSize":"12px","color":"#999"}}]},{"widgettype":"VBox","options":{"backgroundColor":"#FFFFFF","padding":"20px","cursor":"pointer"},"binds":[{"wid":"self","event":"click","actiontype":"urlwidget","target":"app.world_sync_content","options":{"url":"{{entire_url('/world_sync/sync_form.ui')}}"},"mode":"replace"}],"subwidgets":[{"widgettype":"Text","options":{"label":"发起同步","fontSize":"18px"}},{"widgettype":"Text","options":{"label":"导入/导出/合并世界数据","fontSize":"12px","color":"#999"}}]}]},{"widgettype":"VBox","id":"world_sync_content","options":{"width":"100%","flex":"1","marginTop":"20px"}}]}
|
||||
{"widgettype":"VBox","id":"app.world_sync_content","options":{"width":"100%","height":"100%","padding":"20px"},"subwidgets":[{"widgettype":"Text","options":{"label":"同步管理（W-05）","fontSize":"24px"}},{"widgettype":"ResponsableBox","options":{"gap":"16px","minWidth":"250px"},"subwidgets":[{"widgettype":"VBox","options":{"backgroundColor":"#FFFFFF","padding":"20px","cursor":"pointer"},"binds":[{"wid":"self","event":"click","actiontype":"urlwidget","target":"app.world_sync_content","options":{"url":"{{entire_url('/world_sync/world_sync_list.ui')}}"},"mode":"replace"}],"subwidgets":[{"widgettype":"Text","options":{"label":"同步配置"}}]},{"widgettype":"VBox","options":{"backgroundColor":"#FFFFFF","padding":"20px","cursor":"pointer"},"binds":[{"wid":"self","event":"click","actiontype":"urlwidget","target":"app.world_sync_content","options":{"url":"{{entire_url('/world_sync/world_sync_task_list.ui')}}"},"mode":"replace"}],"subwidgets":[{"widgettype":"Text","options":{"label":"同步任务"}}]},{"widgettype":"VBox","options":{"backgroundColor":"#FFFFFF","padding":"20px","cursor":"pointer"},"binds":[{"wid":"self","event":"click","actiontype":"urlwidget","target":"app.world_sync_content","options":{"url":"{{entire_url('/world_sync/world_sync_log_list.ui')}}"},"mode":"replace"}],"subwidgets":[{"widgettype":"Text","options":{"label":"同步日志"}}]}]}]}
|
||||
@ -1 +1,17 @@
|
||||
{"widgettype":"VBox","options":{"width":"100%","padding":"20px"},"subwidgets":[{"widgettype":"Text","options":{"label":"发起世界同步","fontSize":"20px","marginBottom":"16px"}},{"widgettype":"Form","id":"sync_form","options":{"submit_url":"{{entire_url('/world_sync/api/world_sync.dspy')}}","method":"POST"},"subwidgets":[{"widgettype":"UiCode","id":"world_id","options":{"label":"所属世界","dataurl":"{{entire_url('/world_sync/api/get_search_world_id.dspy')}}","required":"true"}},{"widgettype":"UiCode","id":"sync_type","options":{"label":"同步类型","dataurl":"{{entire_url('/world_sync/api/get_search_sync_type.dspy')}}","required":"true"}},{"widgettype":"Text","id":"source","options":{"label":"数据来源（文件/远端地址）"}},{"widgettype":"Button","options":{"label":"执行同步","type":"submit"}}]}]}
|
||||
{
|
||||
"widgettype": "VBox",
|
||||
"options": {"width": "100%", "height": "100%", "padding": "20px", "backgroundColor": "#FFFFFF"},
|
||||
"subwidgets": [
|
||||
{"widgettype": "Text", "options": {"label": "执行同步任务", "fontSize": "18px", "fontWeight": "bold"}},
|
||||
{"widgettype": "Form", "options": {"labelWidth": "120px", "marginTop": "16px", "submit_url": "{{entire_url('/world_sync/api/execute_sync.dspy')}}"}, "subwidgets": [
|
||||
{"widgettype": "Input", "options": {"label": "同步编码", "name": "sync_code", "required": true, "placeholder": "如 world_to_entity"}},
|
||||
{"widgettype": "Input", "options": {"label": "同步名称", "name": "sync_name", "required": true, "placeholder": "如 世界到实体同步"}},
|
||||
{"widgettype": "Select", "options": {"label": "同步类型", "name": "sync_type", "value": "incremental", "options": [{"value": "full", "label": "全量同步"}, {"value": "incremental", "label": "增量同步"}]}},
|
||||
{"widgettype": "Select", "options": {"label": "源类型", "name": "source_type", "value": "world", "options": [{"value": "world", "label": "世界"}, {"value": "scene", "label": "场景"}, {"value": "entity", "label": "实体"}]}},
|
||||
{"widgettype": "Select", "options": {"label": "目标类型", "name": "target_type", "value": "entity", "options": [{"value": "world", "label": "世界"}, {"value": "scene", "label": "场景"}, {"value": "entity", "label": "实体"}]}},
|
||||
{"widgettype": "Input", "options": {"label": "源ID列表", "name": "source_ids", "placeholder": "逗号分隔，可空=全量"}},
|
||||
{"widgettype": "Input", "options": {"label": "单批上限", "name": "limit", "value": "500", "placeholder": "1-1000"}},
|
||||
{"widgettype": "Button", "options": {"label": "执行同步", "actiontype": "submit", "type": "primary"}}
|
||||
]}
|
||||
]
|
||||
}
|
||||
|
||||
1
wwwroot/world_sync_list.ui
Normal file
1
wwwroot/world_sync_list.ui
Normal file
@ -0,0 +1 @@
|
||||
{"widgettype":"Tabular","options":{"title":"同步配置","width":"100%","height":"100%","data_url":"{{entire_url('/world_sync/api/world_sync_list.dspy')}}","data_params":{"page":1,"rows":20},"row_options":{"fields":["sync_code","sync_name","sync_type","mode","status","created_at"],"labels":{"sync_code":"同步编码","sync_name":"同步名称","sync_type":"同步类型","mode":"同步模式","status":"状态","created_at":"创建时间"}}}}
|
||||
1
wwwroot/world_sync_log_list.ui
Normal file
1
wwwroot/world_sync_log_list.ui
Normal file
@ -0,0 +1 @@
|
||||
{"widgettype":"Tabular","options":{"title":"同步日志","width":"100%","height":"100%","data_url":"{{entire_url('/world_sync/api/world_sync_log_list.dspy')}}","data_params":{"page":1,"rows":20},"row_options":{"fields":["task_id","sync_type","status","direction","target_table","target_key","batch_no","created_at"],"labels":{"task_id":"任务ID","sync_type":"同步类型","status":"同步状态","direction":"方向","target_table":"目标表","target_key":"目标主键","batch_no":"批次号","created_at":"创建时间"}}}}
|
||||
1
wwwroot/world_sync_task_list.ui
Normal file
1
wwwroot/world_sync_task_list.ui
Normal file
@ -0,0 +1 @@
|
||||
{"widgettype":"Tabular","options":{"title":"同步任务","width":"100%","height":"100%","data_url":"{{entire_url('/world_sync/api/world_sync_task_list.dspy')}}","data_params":{"page":1,"rows":20},"row_options":{"fields":["sync_code","sync_type","status","batch_no","total_count","success_count","fail_count","start_time","end_time"],"labels":{"sync_code":"同步编码","sync_type":"同步类型","status":"任务状态","batch_no":"批次号","total_count":"总数","success_count":"成功数","fail_count":"失败数","start_time":"开始时间","end_time":"结束时间"}}}}
|
||||
Loading…
x
Reference in New Issue
Block a user