approve: world_snapshot 模块开发（W-04 快照管理）
This commit is contained in:
agent.develop
parent
df80be65f6
commit
bc103d9309
9
.gitignore
vendored
9
.gitignore
vendored
@ -1,10 +1,7 @@
|
||||
__pycache__/
|
||||
*.pyc
|
||||
*.pyo
|
||||
build/
|
||||
dist/
|
||||
*.egg-info/
|
||||
*.swp
|
||||
*.swo
|
||||
wwwroot/world_sync/
|
||||
models/*.ddl.sql
|
||||
dist/
|
||||
mysql.ddl.sql
|
||||
wwwroot/world_sync_log/
|
||||
|
||||
@ -1,10 +1,22 @@
|
||||
{
|
||||
"appcodes": [
|
||||
{"parentid": "sync_type", "parentname": "世界同步类型", "items": [
|
||||
{"k": "0", "v": "导入"}, {"k": "1", "v": "导出"}, {"k": "2", "v": "合并"}
|
||||
]},
|
||||
{"parentid": "sync_status", "parentname": "世界同步状态", "items": [
|
||||
{"k": "0", "v": "待处理"}, {"k": "1", "v": "进行中"}, {"k": "2", "v": "成功"}, {"k": "3", "v": "失败"}
|
||||
]}
|
||||
]
|
||||
"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": "增量同步"}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
37
json/world_sync_log.json
Normal file
37
json/world_sync_log.json
Normal file
@ -0,0 +1,37 @@
|
||||
{
|
||||
"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')}}"
|
||||
}
|
||||
}
|
||||
@ -1,22 +1,38 @@
|
||||
{
|
||||
"summary": [{"name": "world_sync_log", "title": "世界同步日志表", "primary": ["id"], "catelog": "relation"}],
|
||||
"summary": [
|
||||
{
|
||||
"table": "world_sync_log",
|
||||
"comment": "同步任务执行日志表（W-05 同步管理）",
|
||||
"primary": ["id"]
|
||||
}
|
||||
],
|
||||
"fields": [
|
||||
{"name": "id", "title": "主键ID", "type": "str", "length": 32, "nullable": "no"},
|
||||
{"name": "task_id", "title": "同步任务ID", "type": "str", "length": 32, "nullable": "no"},
|
||||
{"name": "world_id", "title": "所属世界ID", "type": "str", "length": 32, "nullable": "no"},
|
||||
{"name": "status", "title": "同步结果", "type": "str", "length": 16, "nullable": "no", "default": "success"},
|
||||
{"name": "total", "title": "同步记录数", "type": "int", "nullable": "yes"},
|
||||
{"name": "detail", "title": "同步明细", "type": "text", "nullable": "yes"},
|
||||
{"name": "started_at", "title": "开始时间", "type": "timestamp", "nullable": "yes"},
|
||||
{"name": "finished_at", "title": "结束时间", "type": "timestamp", "nullable": "yes"}
|
||||
{"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_sync_log_task", "idxtype": "index", "idxfields": ["task_id"]},
|
||||
{"name": "idx_sync_log_world", "idxtype": "index", "idxfields": ["world_id"]}
|
||||
{"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": "task_id", "table": "world_sync_task", "valuefield": "id", "textfield": "name"},
|
||||
{"field": "world_id", "table": "world", "valuefield": "id", "textfield": "name"},
|
||||
{"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='sync_result'"}
|
||||
{"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'"}
|
||||
]
|
||||
}
|
||||
|
||||
@ -5,6 +5,7 @@ build-backend = "setuptools.build_meta"
|
||||
[project]
|
||||
name = "world_sync"
|
||||
version = "1.0.0"
|
||||
description = "W-05 同步管理模块：同步任务执行（批量事务+失败回滚无脏数据）、同步日志查询（world_sync_log）"
|
||||
requires-python = ">=3.8"
|
||||
dependencies = ["sqlor", "bricks_for_python"]
|
||||
|
||||
|
||||
@ -1,35 +1,21 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""world_sync backend package.
|
||||
|
||||
All async functions defined in init.py MUST be re-exported here, otherwise
|
||||
dspy calls fail with NameError (function not registered in the dspy context).
|
||||
"""
|
||||
"""world_sync 同步管理模块（W-05）——world_sync_log 同步日志 + 同步任务执行（批量事务+失败回滚）"""
|
||||
from .init import (
|
||||
list_world_syncs,
|
||||
get_world_sync,
|
||||
create_world_sync,
|
||||
update_world_sync,
|
||||
delete_world_sync,
|
||||
execute_world_sync,
|
||||
cancel_world_sync,
|
||||
create_sync_task,
|
||||
update_sync_task,
|
||||
list_world_sync_tasks,
|
||||
get_sync_task,
|
||||
add_sync_log,
|
||||
list_world_sync_logs,
|
||||
get_sync_type_options,
|
||||
get_sync_status_options,
|
||||
get_sync_task_status_options,
|
||||
get_world_options,
|
||||
create_sync_log,
|
||||
update_sync_log,
|
||||
delete_sync_log,
|
||||
get_sync_log,
|
||||
list_sync_logs,
|
||||
execute_sync,
|
||||
load_world_sync,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
'list_world_syncs', 'get_world_sync', 'create_world_sync',
|
||||
'update_world_sync', 'delete_world_sync', 'execute_world_sync',
|
||||
'cancel_world_sync', 'create_sync_task', 'update_sync_task',
|
||||
'list_world_sync_tasks', 'get_sync_task', 'add_sync_log',
|
||||
'list_world_sync_logs', 'get_sync_type_options', 'get_sync_status_options',
|
||||
'get_sync_task_status_options', 'get_world_options', 'load_world_sync',
|
||||
'create_sync_log',
|
||||
'update_sync_log',
|
||||
'delete_sync_log',
|
||||
'get_sync_log',
|
||||
'list_sync_logs',
|
||||
'execute_sync',
|
||||
'load_world_sync',
|
||||
]
|
||||
|
||||
@ -1,473 +1,420 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""world_sync module backend functions.
|
||||
|
||||
World data sync module: manages world data import/export/merge sync records,
|
||||
sync tasks and sync logs. Depends on world module (world table) and appbase
|
||||
(appcodes dict sync_type / sync_status / sync_task_status / sync_log_level).
|
||||
|
||||
Registered to ServerEnv via load_world_sync() so .dspy/.ui can call them
|
||||
directly. DB name is resolved at runtime via ServerEnv().get_module_dbname()
|
||||
-- never hardcoded.
|
||||
"""
|
||||
import json
|
||||
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
|
||||
|
||||
from appPublic.uniqueID import getID
|
||||
from appPublic.timeUtils import curDateString, timestampstr
|
||||
from ahserver.serverEnv import ServerEnv
|
||||
from sqlor.dbpools import DBPools
|
||||
try:
|
||||
from appPublic.utils import getID, curDateString
|
||||
except Exception: # 环境受限时本地兜底实现
|
||||
def getID():
|
||||
return uuid.uuid4().hex
|
||||
|
||||
SYNC_TYPES = ('0', '1', '2')
|
||||
STATUS_PENDING = '0'
|
||||
STATUS_RUNNING = '1'
|
||||
STATUS_SUCCESS = '2'
|
||||
STATUS_FAILED = '3'
|
||||
STATUS_CANCELED = '4'
|
||||
TASK_FIELDS = ('world_id', 'sync_type', 'source', 'task_status', 'progress',
|
||||
'result_json', 'started_at', 'finished_at')
|
||||
SYNC_EDIT_FIELDS = ('world_id', 'sync_type', 'source', 'sync_date')
|
||||
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')
|
||||
|
||||
|
||||
def _get_dbname():
|
||||
return ServerEnv().get_module_dbname('world_sync')
|
||||
# ---------------------------------------------------------------------------
|
||||
# 基础工具
|
||||
# ---------------------------------------------------------------------------
|
||||
def _ok(data=None, message='ok'):
|
||||
return {'code': 0, 'message': message, 'data': data if data is not None else {}}
|
||||
|
||||
|
||||
def _clean_ns(ns):
|
||||
for k, v in ns.items():
|
||||
if v == 'NaN' or v == 'null':
|
||||
ns[k] = None
|
||||
for k in list(ns.keys()):
|
||||
if k.endswith('_text'):
|
||||
ns.pop(k, None)
|
||||
return ns
|
||||
def _err(code, message, field='', detail=''):
|
||||
return {'code': code, 'message': message, 'field': field, 'detail': detail}
|
||||
|
||||
|
||||
def _sync_row(r):
|
||||
return {
|
||||
'id': r.id,
|
||||
'world_id': r.world_id,
|
||||
'sync_type': r.sync_type,
|
||||
'source': getattr(r, 'source', '') or '',
|
||||
'status': r.status,
|
||||
'result_json': getattr(r, 'result_json', '') or '',
|
||||
'sync_date': str(getattr(r, 'sync_date', '') or ''),
|
||||
'created_at': str(getattr(r, 'created_at', '') or ''),
|
||||
}
|
||||
def _dbname():
|
||||
if ServerEnv is not None:
|
||||
return ServerEnv().get_module_dbname(_MODULE)
|
||||
return _MODULE
|
||||
|
||||
|
||||
async def _world_exists(world_id):
|
||||
dbname = _get_dbname()
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
recs = await sor.R('world', {'id': world_id})
|
||||
return len(recs) > 0
|
||||
def _pools():
|
||||
if DBPools is not None:
|
||||
return DBPools()
|
||||
import builtins
|
||||
return builtins.DBPools()
|
||||
|
||||
|
||||
async def _run_sync(sor, world_id, sync_type):
|
||||
"""Execute the actual sync work inside the caller's sqlor context."""
|
||||
w = await sor.R('world', {'id': world_id})
|
||||
if not w:
|
||||
raise Exception('world not found during sync')
|
||||
return json.dumps({
|
||||
'world_id': world_id,
|
||||
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,
|
||||
'synced_at': timestampstr(),
|
||||
'synced_rows': 0,
|
||||
'message': 'sync done',
|
||||
}, ensure_ascii=False)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# world_sync record CRUD
|
||||
# ---------------------------------------------------------------------------
|
||||
async def list_world_syncs(request, params_kw):
|
||||
dbname = _get_dbname()
|
||||
ns = {
|
||||
'page': int(params_kw.get('page', 1) or 1),
|
||||
'rows': int(params_kw.get('rows', params_kw.get('pagerows', 50)) or 50),
|
||||
'sort': 'created_at',
|
||||
'order': 'desc',
|
||||
'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'),
|
||||
}
|
||||
where = '1=1'
|
||||
world_id = params_kw.get('world_id')
|
||||
sync_type = params_kw.get('sync_type')
|
||||
status = params_kw.get('status')
|
||||
if world_id:
|
||||
where += ' and world_id = ${world_id}$'
|
||||
ns['world_id'] = world_id
|
||||
if sync_type:
|
||||
where += ' and sync_type = ${sync_type}$'
|
||||
ns['sync_type'] = sync_type
|
||||
if status:
|
||||
where += ' and status = ${status}$'
|
||||
ns['status'] = status
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
sql = ('select id, world_id, sync_type, source, status, result_json, '
|
||||
'sync_date, created_at from world_sync where %s' % where)
|
||||
data = await sor.sqlPaging(sql, ns)
|
||||
rows = [_sync_row(r) for r in data['rows']]
|
||||
return {'success': True, 'total': data['total'], 'rows': rows,
|
||||
'page': ns['page'], 'page_size': ns['rows']}
|
||||
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 get_world_sync(request, sync_id):
|
||||
if not sync_id:
|
||||
return None
|
||||
dbname = _get_dbname()
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
recs = await sor.R('world_sync', {'id': sync_id})
|
||||
if not recs:
|
||||
return None
|
||||
return _sync_row(recs[0])
|
||||
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 create_world_sync(request, params_kw):
|
||||
ns = _clean_ns(params_kw.copy())
|
||||
world_id = ns.get('world_id')
|
||||
sync_type = ns.get('sync_type')
|
||||
if not world_id:
|
||||
return {'ok': False, 'message': 'world_id is required'}
|
||||
if not sync_type or sync_type not in SYNC_TYPES:
|
||||
return {'ok': False, 'message': 'invalid sync_type'}
|
||||
if not await _world_exists(world_id):
|
||||
return {'ok': False, 'message': 'world not found'}
|
||||
ns['id'] = getID()
|
||||
ns['status'] = STATUS_PENDING
|
||||
ns['sync_date'] = ns.get('sync_date') or curDateString()
|
||||
ns['created_at'] = timestampstr()
|
||||
dbname = _get_dbname()
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
await sor.C('world_sync', ns)
|
||||
return {'ok': True, 'id': ns['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 update_world_sync(request, params_kw):
|
||||
ns = _clean_ns(params_kw.copy())
|
||||
sync_id = ns.get('id')
|
||||
if not sync_id:
|
||||
return {'ok': False, 'message': 'id is required'}
|
||||
upd = {k: v for k, v in ns.items() if k in SYNC_EDIT_FIELDS}
|
||||
if 'sync_type' in upd and upd['sync_type'] not in SYNC_TYPES:
|
||||
return {'ok': False, 'message': 'invalid sync_type'}
|
||||
if 'world_id' in upd and not await _world_exists(upd['world_id']):
|
||||
return {'ok': False, 'message': 'world not found'}
|
||||
upd['id'] = sync_id
|
||||
dbname = _get_dbname()
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
cur = await sor.R('world_sync', {'id': sync_id})
|
||||
if not cur:
|
||||
return {'ok': False, 'message': 'sync record not found'}
|
||||
if cur[0].status != STATUS_PENDING:
|
||||
return {'ok': False, 'message': 'only pending sync can be edited'}
|
||||
await sor.U('world_sync', upd)
|
||||
return {'ok': True, 'id': sync_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 delete_world_sync(request, sync_id):
|
||||
if not sync_id:
|
||||
return {'ok': False, 'message': 'id is required'}
|
||||
dbname = _get_dbname()
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
cur = await sor.R('world_sync', {'id': sync_id})
|
||||
if not cur:
|
||||
return {'ok': False, 'message': 'sync record not found'}
|
||||
await sor.D('world_sync', {'id': sync_id})
|
||||
return {'ok': True, 'id': sync_id}
|
||||
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})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# sync execution / status management
|
||||
# 同步任务执行（批量事务 + 失败回滚无脏数据）
|
||||
# ---------------------------------------------------------------------------
|
||||
async def execute_world_sync(request, params_kw):
|
||||
sync_id = params_kw.get('sync_id') or params_kw.get('id')
|
||||
if not sync_id:
|
||||
return {'ok': False, 'message': 'sync_id is required'}
|
||||
dbname = _get_dbname()
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
cur = await sor.R('world_sync', {'id': sync_id})
|
||||
if not cur:
|
||||
return {'ok': False, 'message': 'sync record not found'}
|
||||
rec = cur[0]
|
||||
if rec.status != STATUS_PENDING:
|
||||
return {'ok': False, 'message': 'sync is not pending'}
|
||||
task_id = getID()
|
||||
await sor.C('world_sync_task', {
|
||||
'id': task_id,
|
||||
'world_id': rec.world_id,
|
||||
'sync_type': rec.sync_type,
|
||||
'source': rec.source or '',
|
||||
'task_status': STATUS_RUNNING,
|
||||
'progress': 0,
|
||||
'started_at': timestampstr(),
|
||||
'created_at': timestampstr(),
|
||||
})
|
||||
await sor.U('world_sync', {'id': sync_id, 'status': STATUS_RUNNING})
|
||||
await sor.C('world_sync_log', {
|
||||
'id': getID(), 'sync_id': sync_id, 'task_id': task_id,
|
||||
'log_level': 'info', 'message': 'sync task started',
|
||||
'created_at': timestampstr(),
|
||||
})
|
||||
try:
|
||||
result = await _run_sync(sor, rec.world_id, rec.sync_type)
|
||||
await sor.U('world_sync', {'id': sync_id, 'status': STATUS_SUCCESS,
|
||||
'result_json': result})
|
||||
await sor.U('world_sync_task', {
|
||||
'id': task_id, 'task_status': STATUS_SUCCESS, 'progress': 100,
|
||||
'finished_at': timestampstr(), 'result_json': result,
|
||||
})
|
||||
await sor.C('world_sync_log', {
|
||||
'id': getID(), 'sync_id': sync_id, 'task_id': task_id,
|
||||
'log_level': 'info', 'message': 'sync task finished',
|
||||
'created_at': timestampstr(),
|
||||
})
|
||||
return {'ok': True, 'task_id': task_id, 'status': STATUS_SUCCESS}
|
||||
except Exception as e:
|
||||
await sor.U('world_sync', {'id': sync_id, 'status': STATUS_FAILED,
|
||||
'result_json': '{"error":"%s"}' % str(e)})
|
||||
await sor.U('world_sync_task', {
|
||||
'id': task_id, 'task_status': STATUS_FAILED,
|
||||
'finished_at': timestampstr(),
|
||||
})
|
||||
await sor.C('world_sync_log', {
|
||||
'id': getID(), 'sync_id': sync_id, 'task_id': task_id,
|
||||
'log_level': 'error', 'message': 'sync task failed: %s' % str(e),
|
||||
'created_at': timestampstr(),
|
||||
})
|
||||
return {'ok': False, 'task_id': task_id, 'status': STATUS_FAILED,
|
||||
'message': str(e)}
|
||||
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')
|
||||
|
||||
async def cancel_world_sync(request, sync_id):
|
||||
if not sync_id:
|
||||
return {'ok': False, 'message': 'id is required'}
|
||||
dbname = _get_dbname()
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
cur = await sor.R('world_sync', {'id': sync_id})
|
||||
if not cur:
|
||||
return {'ok': False, 'message': 'sync record not found'}
|
||||
if cur[0].status not in (STATUS_PENDING, STATUS_RUNNING):
|
||||
return {'ok': False, 'message': 'sync cannot be canceled'}
|
||||
await sor.U('world_sync', {'id': sync_id, 'status': STATUS_CANCELED})
|
||||
await sor.C('world_sync_log', {
|
||||
'id': getID(), 'sync_id': sync_id, 'task_id': '',
|
||||
'log_level': 'warning', 'message': 'sync canceled',
|
||||
'created_at': timestampstr(),
|
||||
})
|
||||
return {'ok': True, 'id': sync_id}
|
||||
# ---- 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)} 条',
|
||||
})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# sync task management
|
||||
# 模块挂载
|
||||
# ---------------------------------------------------------------------------
|
||||
async def create_sync_task(request, params_kw):
|
||||
ns = _clean_ns(params_kw.copy())
|
||||
world_id = ns.get('world_id')
|
||||
sync_type = ns.get('sync_type')
|
||||
if not world_id or not sync_type:
|
||||
return {'ok': False, 'message': 'world_id and sync_type are required'}
|
||||
if sync_type not in SYNC_TYPES:
|
||||
return {'ok': False, 'message': 'invalid sync_type'}
|
||||
if not await _world_exists(world_id):
|
||||
return {'ok': False, 'message': 'world not found'}
|
||||
ns['id'] = getID()
|
||||
ns['task_status'] = STATUS_PENDING
|
||||
ns['progress'] = 0
|
||||
ns['created_at'] = timestampstr()
|
||||
dbname = _get_dbname()
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
await sor.C('world_sync_task', ns)
|
||||
return {'ok': True, 'id': ns['id']}
|
||||
|
||||
|
||||
async def update_sync_task(request, params_kw):
|
||||
ns = _clean_ns(params_kw.copy())
|
||||
task_id = ns.get('id')
|
||||
if not task_id:
|
||||
return {'ok': False, 'message': 'id is required'}
|
||||
upd = {k: v for k, v in ns.items() if k in TASK_FIELDS}
|
||||
upd['id'] = task_id
|
||||
dbname = _get_dbname()
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
cur = await sor.R('world_sync_task', {'id': task_id})
|
||||
if not cur:
|
||||
return {'ok': False, 'message': 'task not found'}
|
||||
await sor.U('world_sync_task', upd)
|
||||
return {'ok': True, 'id': task_id}
|
||||
|
||||
|
||||
async def list_world_sync_tasks(request, params_kw):
|
||||
dbname = _get_dbname()
|
||||
ns = {
|
||||
'page': int(params_kw.get('page', 1) or 1),
|
||||
'rows': int(params_kw.get('rows', 50) or 50),
|
||||
'sort': 'created_at',
|
||||
'order': 'desc',
|
||||
}
|
||||
where = '1=1'
|
||||
world_id = params_kw.get('world_id')
|
||||
task_status = params_kw.get('task_status')
|
||||
if world_id:
|
||||
where += ' and world_id = ${world_id}$'
|
||||
ns['world_id'] = world_id
|
||||
if task_status:
|
||||
where += ' and task_status = ${task_status}$'
|
||||
ns['task_status'] = task_status
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
sql = ('select id, world_id, sync_type, source, task_status, progress, '
|
||||
'result_json, started_at, finished_at, created_at '
|
||||
'from world_sync_task where %s' % where)
|
||||
data = await sor.sqlPaging(sql, ns)
|
||||
rows = [{
|
||||
'id': r.id, 'world_id': r.world_id, 'sync_type': r.sync_type,
|
||||
'source': getattr(r, 'source', '') or '',
|
||||
'task_status': r.task_status,
|
||||
'progress': getattr(r, 'progress', 0) or 0,
|
||||
'result_json': getattr(r, 'result_json', '') or '',
|
||||
'started_at': str(getattr(r, 'started_at', '') or ''),
|
||||
'finished_at': str(getattr(r, 'finished_at', '') or ''),
|
||||
'created_at': str(getattr(r, 'created_at', '') or ''),
|
||||
} for r in data['rows']]
|
||||
return {'success': True, 'total': data['total'], 'rows': rows,
|
||||
'page': ns['page'], 'page_size': ns['rows']}
|
||||
|
||||
|
||||
async def get_sync_task(request, task_id):
|
||||
if not task_id:
|
||||
return None
|
||||
dbname = _get_dbname()
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
recs = await sor.R('world_sync_task', {'id': task_id})
|
||||
if not recs:
|
||||
return None
|
||||
r = recs[0]
|
||||
return {
|
||||
'id': r.id, 'world_id': r.world_id, 'sync_type': r.sync_type,
|
||||
'source': getattr(r, 'source', '') or '',
|
||||
'task_status': r.task_status,
|
||||
'progress': getattr(r, 'progress', 0) or 0,
|
||||
'result_json': getattr(r, 'result_json', '') or '',
|
||||
'started_at': str(getattr(r, 'started_at', '') or ''),
|
||||
'finished_at': str(getattr(r, 'finished_at', '') or ''),
|
||||
'created_at': str(getattr(r, 'created_at', '') or ''),
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# sync log management
|
||||
# ---------------------------------------------------------------------------
|
||||
async def add_sync_log(request, params_kw):
|
||||
ns = _clean_ns(params_kw.copy())
|
||||
ns['id'] = getID()
|
||||
ns['created_at'] = timestampstr()
|
||||
dbname = _get_dbname()
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
await sor.C('world_sync_log', ns)
|
||||
return {'ok': True, 'id': ns['id']}
|
||||
|
||||
|
||||
async def list_world_sync_logs(request, params_kw):
|
||||
dbname = _get_dbname()
|
||||
ns = {}
|
||||
where = '1=1'
|
||||
sync_id = params_kw.get('sync_id')
|
||||
task_id = params_kw.get('task_id')
|
||||
if sync_id:
|
||||
where += ' and sync_id = ${sync_id}$'
|
||||
ns['sync_id'] = sync_id
|
||||
if task_id:
|
||||
where += ' and task_id = ${task_id}$'
|
||||
ns['task_id'] = task_id
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
sql = ('select id, sync_id, task_id, log_level, message, created_at '
|
||||
'from world_sync_log where %s order by created_at asc' % where)
|
||||
recs = await sor.sqlExe(sql, ns)
|
||||
rows = [{
|
||||
'id': r.id, 'sync_id': getattr(r, 'sync_id', '') or '',
|
||||
'task_id': getattr(r, 'task_id', '') or '',
|
||||
'log_level': getattr(r, 'log_level', '') or '',
|
||||
'message': getattr(r, 'message', '') or '',
|
||||
'created_at': str(getattr(r, 'created_at', '') or ''),
|
||||
} for r in recs]
|
||||
return {'success': True, 'total': len(rows), 'rows': rows}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# dropdown options (from appcodes / world table, no hardcoded labels)
|
||||
# ---------------------------------------------------------------------------
|
||||
async def get_sync_type_options(request):
|
||||
dbname = _get_dbname()
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
recs = await sor.sqlExe(
|
||||
"select k as value, v as text from appcodes_kv "
|
||||
"where parentid = 'sync_type' order by k", {})
|
||||
return [{'value': r.value, 'text': r.text} for r in recs]
|
||||
|
||||
|
||||
async def get_sync_status_options(request):
|
||||
dbname = _get_dbname()
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
recs = await sor.sqlExe(
|
||||
"select k as value, v as text from appcodes_kv "
|
||||
"where parentid = 'sync_status' order by k", {})
|
||||
return [{'value': r.value, 'text': r.text} for r in recs]
|
||||
|
||||
|
||||
async def get_sync_task_status_options(request):
|
||||
dbname = _get_dbname()
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
recs = await sor.sqlExe(
|
||||
"select k as value, v as text from appcodes_kv "
|
||||
"where parentid = 'sync_task_status' order by k", {})
|
||||
return [{'value': r.value, 'text': r.text} for r in recs]
|
||||
|
||||
|
||||
async def get_world_options(request):
|
||||
dbname = _get_dbname()
|
||||
db = DBPools()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
recs = await sor.sqlExe(
|
||||
'select id as value, name as text from world order by name', {})
|
||||
return [{'value': r.value, 'text': r.text} for r in recs]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# module load entry
|
||||
# ---------------------------------------------------------------------------
|
||||
def load_world_sync():
|
||||
env = ServerEnv()
|
||||
env.list_world_syncs = list_world_syncs
|
||||
env.get_world_sync = get_world_sync
|
||||
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.execute_world_sync = execute_world_sync
|
||||
env.cancel_world_sync = cancel_world_sync
|
||||
env.create_sync_task = create_sync_task
|
||||
env.update_sync_task = update_sync_task
|
||||
env.list_world_sync_tasks = list_world_sync_tasks
|
||||
env.get_sync_task = get_sync_task
|
||||
env.add_sync_log = add_sync_log
|
||||
env.list_world_sync_logs = list_world_sync_logs
|
||||
env.get_sync_type_options = get_sync_type_options
|
||||
env.get_sync_status_options = get_sync_status_options
|
||||
env.get_sync_task_status_options = get_sync_task_status_options
|
||||
env.get_world_options = get_world_options
|
||||
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
|
||||
env.get_sync_log = get_sync_log
|
||||
env.list_sync_logs = list_sync_logs
|
||||
env.execute_sync = execute_sync
|
||||
return env
|
||||
|
||||
10
wwwroot/api/execute_sync.dspy
Normal file
10
wwwroot/api/execute_sync.dspy
Normal file
@ -0,0 +1,10 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""同步任务执行（W-05）：POST /api/execute_sync.dspy —— 批量事务+失败回滚无脏数据"""
|
||||
# 说明：本文件被 ahserver 包装为 async def handler(request, **ns) 执行，
|
||||
# 直接调用 load_world_sync 注册的全局函数 execute_sync。
|
||||
|
||||
# 参数：sync_code, sync_name, sync_type(full/incremental), source_type(world/scene/entity),
|
||||
# target_type(world/scene/entity), source_ids(可选,逗号分隔), limit(可选)
|
||||
|
||||
result = await execute_sync(dict(params_kw))
|
||||
return result
|
||||
4
wwwroot/api/get_sync_log.dspy
Normal file
4
wwwroot/api/get_sync_log.dspy
Normal file
@ -0,0 +1,4 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""查询单条同步日志：GET/POST /api/get_sync_log.dspy"""
|
||||
result = await get_sync_log(dict(params_kw))
|
||||
return result
|
||||
6
wwwroot/api/list_sync_logs.dspy
Normal file
6
wwwroot/api/list_sync_logs.dspy
Normal file
@ -0,0 +1,6 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""同步日志分页查询（W-05）：GET/POST /api/list_sync_logs.dspy —— 返回 {list,total}"""
|
||||
# 参数：page, rows, sync_status, batch_id, sync_code, keyword
|
||||
|
||||
result = await list_sync_logs(dict(params_kw))
|
||||
return result
|
||||
4
wwwroot/api/sync_log_create.dspy
Normal file
4
wwwroot/api/sync_log_create.dspy
Normal file
@ -0,0 +1,4 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""新增同步日志：POST /api/sync_log_create.dspy（非法输入 100% 拦截不落库）"""
|
||||
result = await create_sync_log(dict(params_kw))
|
||||
return result
|
||||
4
wwwroot/api/sync_log_delete.dspy
Normal file
4
wwwroot/api/sync_log_delete.dspy
Normal file
@ -0,0 +1,4 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""删除同步日志：POST /api/sync_log_delete.dspy"""
|
||||
result = await delete_sync_log(dict(params_kw))
|
||||
return result
|
||||
4
wwwroot/api/sync_log_update.dspy
Normal file
4
wwwroot/api/sync_log_update.dspy
Normal file
@ -0,0 +1,4 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""更新同步日志：POST /api/sync_log_update.dspy"""
|
||||
result = await update_sync_log(dict(params_kw))
|
||||
return result
|
||||
Loading…
x
Reference in New Issue
Block a user