From fda236eb0cdb91aad96355b0df4e2bdccf182b86 Mon Sep 17 00:00:00 2001 From: TermiNexus Date: Thu, 13 Aug 2026 23:59:34 +0800 Subject: [PATCH] =?UTF-8?q?feat(msync):=20ATestCServer=20=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E5=85=A5=E5=BA=93=EF=BC=88=E5=BF=BD=E7=95=A5=20certs/?= =?UTF-8?q?db.sqlite3=EF=BC=89+=20=E8=BF=9C=E7=AB=AF=20App=20=E6=B8=85?= =?UTF-8?q?=E5=8D=95=E4=B8=8A=E6=8A=A5=E8=83=BD=E5=8A=9B=EF=BC=88ReportApp?= =?UTF-8?q?Manifests=20+=20report=5Fmanifest=20=E5=91=BD=E4=BB=A4=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 5 +- ATestCServer/ATestCServer/__init__.py | 0 ATestCServer/ATestCServer/asgi.py | 5 + ATestCServer/ATestCServer/settings.py | 153 +++++ ATestCServer/ATestCServer/urls.py | 21 + ATestCServer/ATestCServer/wsgi.py | 5 + ATestCServer/apps/__init__.py | 0 ATestCServer/apps/sub_server/__init__.py | 0 ATestCServer/apps/sub_server/apps.py | 72 +++ ATestCServer/apps/sub_server/form_service.py | 22 + .../apps/sub_server/local_handlers.py | 53 ++ .../apps/sub_server/management/__init__.py | 0 .../management/commands/__init__.py | 0 .../management/commands/msync_experiment.py | 202 +++++++ .../management/commands/report_manifest.py | 129 +++++ .../sub_server/migrations/0001_initial.py | 65 +++ .../apps/sub_server/migrations/__init__.py | 0 ATestCServer/apps/sub_server/models.py | 66 +++ ATestCServer/apps/sub_server/serializers.py | 34 ++ ATestCServer/apps/sub_server/service_auth.py | 81 +++ .../apps/sub_server/templates/status.html | 217 +++++++ ATestCServer/apps/sub_server/urls.py | 21 + ATestCServer/apps/sub_server/views.py | 420 ++++++++++++++ ATestCServer/manage.py | 30 + ATestCServer/msync_client/__init__.py | 9 + ATestCServer/msync_client/sio_client.py | 540 ++++++++++++++++++ ATestCServer/requirements.txt | 23 + ATestCServer/start.ps1 | 35 ++ 28 files changed, 2206 insertions(+), 2 deletions(-) create mode 100644 ATestCServer/ATestCServer/__init__.py create mode 100644 ATestCServer/ATestCServer/asgi.py create mode 100644 ATestCServer/ATestCServer/settings.py create mode 100644 ATestCServer/ATestCServer/urls.py create mode 100644 ATestCServer/ATestCServer/wsgi.py create mode 100644 ATestCServer/apps/__init__.py create mode 100644 ATestCServer/apps/sub_server/__init__.py create mode 100644 ATestCServer/apps/sub_server/apps.py create mode 100644 ATestCServer/apps/sub_server/form_service.py create mode 100644 ATestCServer/apps/sub_server/local_handlers.py create mode 100644 ATestCServer/apps/sub_server/management/__init__.py create mode 100644 ATestCServer/apps/sub_server/management/commands/__init__.py create mode 100644 ATestCServer/apps/sub_server/management/commands/msync_experiment.py create mode 100644 ATestCServer/apps/sub_server/management/commands/report_manifest.py create mode 100644 ATestCServer/apps/sub_server/migrations/0001_initial.py create mode 100644 ATestCServer/apps/sub_server/migrations/__init__.py create mode 100644 ATestCServer/apps/sub_server/models.py create mode 100644 ATestCServer/apps/sub_server/serializers.py create mode 100644 ATestCServer/apps/sub_server/service_auth.py create mode 100644 ATestCServer/apps/sub_server/templates/status.html create mode 100644 ATestCServer/apps/sub_server/urls.py create mode 100644 ATestCServer/apps/sub_server/views.py create mode 100644 ATestCServer/manage.py create mode 100644 ATestCServer/msync_client/__init__.py create mode 100644 ATestCServer/msync_client/sio_client.py create mode 100644 ATestCServer/requirements.txt create mode 100644 ATestCServer/start.ps1 diff --git a/.gitignore b/.gitignore index 48c3cf9b..76b5f587 100644 --- a/.gitignore +++ b/.gitignore @@ -12,8 +12,9 @@ /static/ /staticfiles/ -# 子服务器示例（含 HTTPS 私钥与 SQLite 数据，防入库） -/ATestCServer/ +# ATestCServer：代码入库，仅忽略敏感文件（HTTPS 私钥/本地 SQLite） +/ATestCServer/certs/ +/ATestCServer/db.sqlite3 # 用户上传目录 /media/ diff --git a/ATestCServer/ATestCServer/__init__.py b/ATestCServer/ATestCServer/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/ATestCServer/ATestCServer/asgi.py b/ATestCServer/ATestCServer/asgi.py new file mode 100644 index 00000000..bf1f42f4 --- /dev/null +++ b/ATestCServer/ATestCServer/asgi.py @@ -0,0 +1,5 @@ +import os +from django.core.asgi import get_asgi_application + +os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'ATestCServer.settings') +application = get_asgi_application() \ No newline at end of file diff --git a/ATestCServer/ATestCServer/settings.py b/ATestCServer/ATestCServer/settings.py new file mode 100644 index 00000000..eafcfc87 --- /dev/null +++ b/ATestCServer/ATestCServer/settings.py @@ -0,0 +1,153 @@ +"""ATestCServer — SMANAGE 子服务器 +通过 MSYNC 协议与 SMANAGE 主服务器（www.gvsds.com）桥接。 + +通信拓扑： + 子 → 父: https://www.gvsds.com/api/msync/* （MSYNC 协议） + 父 → 子: http://127.0.0.1:25080/api/sub/* （本地回环）""" + +import os +from pathlib import Path +from django.core.exceptions import ImproperlyConfigured + +BASE_DIR = Path(__file__).resolve().parent.parent + +SECRET_KEY = os.environ.get('DJANGO_SECRET_KEY') +if not SECRET_KEY: + raise ImproperlyConfigured('DJANGO_SECRET_KEY 环境变量未设置，禁止使用默认值启动') + +DEBUG = os.environ.get('DJANGO_DEBUG', 'False').lower() in ('true', '1', 'yes') + +ALLOWED_HOSTS = os.environ.get('ALLOWED_HOSTS', 'localhost,127.0.0.1').split(',') + +INSTALLED_APPS = [ + 'django.contrib.admin', + 'django.contrib.auth', + 'django.contrib.contenttypes', + 'django.contrib.sessions', + 'django.contrib.messages', + 'django.contrib.staticfiles', + 'rest_framework', + 'gvsdsdk', + 'apps.sub_server', +] + +MIDDLEWARE = [ + 'django.middleware.security.SecurityMiddleware', + 'django.contrib.sessions.middleware.SessionMiddleware', + 'django.middleware.common.CommonMiddleware', + 'django.middleware.csrf.CsrfViewMiddleware', + 'django.contrib.auth.middleware.AuthenticationMiddleware', + 'django.contrib.messages.middleware.MessageMiddleware', + 'django.middleware.clickjacking.XFrameOptionsMiddleware', +] + +ROOT_URLCONF = 'ATestCServer.urls' + +TEMPLATES = [ + { + 'BACKEND': 'django.template.backends.django.DjangoTemplates', + 'DIRS': [], + 'APP_DIRS': True, + 'OPTIONS': { + 'context_processors': [ + 'django.template.context_processors.debug', + 'django.template.context_processors.request', + 'django.contrib.auth.context_processors.auth', + 'django.contrib.messages.context_processors.messages', + ], + }, + }, +] + +WSGI_APPLICATION = 'ATestCServer.wsgi.application' + +DATABASES = { + 'default': { + 'ENGINE': 'django.db.backends.mysql', + 'NAME': os.environ.get('DB_NAME', 'atestc'), + 'USER': os.environ.get('DB_USER', 'root'), + 'PASSWORD': os.environ.get('DB_PASSWORD', ''), + 'HOST': os.environ.get('DB_HOST', '127.0.0.1'), + 'PORT': os.environ.get('DB_PORT', '3306'), + 'OPTIONS': { + 'charset': 'utf8mb4', + 'init_command': "SET sql_mode='STRICT_TRANS_TABLES'", + }, + }, +} + +DATABASE_ROUTERS = [] + +# gvsdsdk 表由 ATestCServer 自己管理（同源数据库） +GVSDSDK_MANAGE_TABLES = True + +AUTH_PASSWORD_VALIDATORS = [ + {'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator'}, + {'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator'}, + {'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator'}, + {'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator'}, +] + +LANGUAGE_CODE = 'zh-hans' +TIME_ZONE = 'Asia/Shanghai' +USE_I18N = True +USE_TZ = False + +STATIC_URL = 'static/' + +DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField' + +REST_FRAMEWORK = { + 'DEFAULT_RENDERER_CLASSES': [ + 'rest_framework.renderers.JSONRenderer', + ], + 'DEFAULT_AUTHENTICATION_CLASSES': [ + 'rest_framework.authentication.SessionAuthentication', + ], + 'DEFAULT_PERMISSION_CLASSES': [ + 'rest_framework.permissions.IsAuthenticated', + ], + 'UNAUTHENTICATED_USER': None, +} + +MSYNC_MASTER_URL = os.environ.get('MSYNC_MASTER_URL', 'https://www.gvsds.com') + +MSYNC_SERVICE_NAME = os.environ.get('MSYNC_SERVICE_NAME', 'atest-cserver') +MSYNC_SERVICE_DOMAIN = os.environ.get('MSYNC_SERVICE_DOMAIN', 'http://127.0.0.1:25080') +MSYNC_SERVICE_SECRET = os.environ.get('MSYNC_SERVICE_SECRET') +if not MSYNC_SERVICE_SECRET: + raise ImproperlyConfigured('MSYNC_SERVICE_SECRET 环境变量未设置，禁止使用默认值启动') + +MSYNC_SERVICE_SCOPE = [ + 'ogm:read', 'ogm:write', + 'registry:read', + 'tx:participate', +] + +MSYNC_HEARTBEAT_INTERVAL = int(os.environ.get('MSYNC_HEARTBEAT_INTERVAL', '30')) + +HTTP_PORT = int(os.environ.get('HTTP_PORT', '25080')) +HTTPS_PORT = int(os.environ.get('HTTPS_PORT', '25443')) + +LOGGING = { + 'version': 1, + 'disable_existing_loggers': False, + 'formatters': { + 'verbose': { + 'format': '[{levelname}] {asctime} {module} {message}', + 'style': '{', + }, + }, + 'handlers': { + 'console': { + 'class': 'logging.StreamHandler', + 'formatter': 'verbose', + }, + }, + 'loggers': { + '': { + 'handlers': ['console'], + 'level': 'INFO', + }, + }, +} \ No newline at end of file diff --git a/ATestCServer/ATestCServer/urls.py b/ATestCServer/ATestCServer/urls.py new file mode 100644 index 00000000..79a124b6 --- /dev/null +++ b/ATestCServer/ATestCServer/urls.py @@ -0,0 +1,21 @@ +"""根 URL 配置""" + +from django.urls import path, include +from django.shortcuts import redirect +from apps.sub_server.views import HealthView +from apps.sub_server import views + + +def root_redirect(request): + return redirect('/api/sub/status/') + + +urlpatterns = [ + path('', root_redirect), + path('health/', HealthView.as_view(), name='health'), + path('api/sub/', include('apps.sub_server.urls')), + # SDK 契约路径：主服务器 Saga 编排引擎按 /api/msync/local/* 回调 + path('api/msync/local/execute/', views.LocalExecuteView.as_view(), name='msync_local_execute'), + path('api/msync/local/compensate/', views.LocalCompensateView.as_view(), name='msync_local_compensate'), + path('api/msync/local/reset/', views.LocalResetView.as_view(), name='msync_local_reset'), +] \ No newline at end of file diff --git a/ATestCServer/ATestCServer/wsgi.py b/ATestCServer/ATestCServer/wsgi.py new file mode 100644 index 00000000..e5aa4e00 --- /dev/null +++ b/ATestCServer/ATestCServer/wsgi.py @@ -0,0 +1,5 @@ +import os +from django.core.wsgi import get_wsgi_application + +os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'ATestCServer.settings') +application = get_wsgi_application() \ No newline at end of file diff --git a/ATestCServer/apps/__init__.py b/ATestCServer/apps/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/ATestCServer/apps/sub_server/__init__.py b/ATestCServer/apps/sub_server/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/ATestCServer/apps/sub_server/apps.py b/ATestCServer/apps/sub_server/apps.py new file mode 100644 index 00000000..6e3fdab0 --- /dev/null +++ b/ATestCServer/apps/sub_server/apps.py @@ -0,0 +1,72 @@ +"""模块：子服务器 AppConfig +Django 启动时自动完成 MSYNC Socket.IO 桥接——连接、注册、认证、心跳。""" + +import logging +import threading +from django.apps import AppConfig +from django.conf import settings + +logger = logging.getLogger(__name__) + +# 与主服务器约定的 SDK 最低版本——版本不一致时启动警告 +REQUIRED_GVSDSDK_VERSION = '1.1.0' + + +class SubServerConfig(AppConfig): + default_auto_field = 'django.db.models.BigAutoField' + name = 'apps.sub_server' + verbose_name = 'SMANAGE 子服务器' + verbose_name_plural = 'SMANAGE 子服务器' + + def ready(self): + # 1) 校验 gvsdsdk 版本与主服务器一致 + self._check_sdk_version() + # 2) 注册实验本地事务处理器（供主服务器 Saga 编排回调） + self._register_local_handlers() + # 3) 启动 MSYNC 桥接 + self._connect_msync() + + def _register_local_handlers(self): + try: + from .local_handlers import RegisterExperimentHandlers + RegisterExperimentHandlers() + logger.info('实验本地事务处理器已注册: demo_a/demo_b/demo_fail/demo_flaky') + except Exception as e: + logger.error(f'实验本地事务处理器注册失败: {e}') + + def _check_sdk_version(self): + try: + import gvsdsdk + info = gvsdsdk.get_version_info() + logger.info(f'[SDK] 加载: v{info["Version"]} @ {info["Location"]}') + ok, msg = gvsdsdk.check_min_version(REQUIRED_GVSDSDK_VERSION) + if ok: + logger.info(f'[SDK] {msg}') + else: + logger.warning(f'[SDK] 版本不符！{msg}。双端 gvsdsdk 版本必须一致！') + except ImportError as e: + logger.error(f'[SDK] gvsdsdk 未安装: {e}') + except Exception as e: + logger.error(f'[SDK] 版本检查失败: {e}') + + def _connect_msync(self): + try: + from msync_client.sio_client import MSyncSocketClient + + msync_client = MSyncSocketClient( + master_url=settings.MSYNC_MASTER_URL, + service_name=settings.MSYNC_SERVICE_NAME, + service_domain=settings.MSYNC_SERVICE_DOMAIN, + service_secret=settings.MSYNC_SERVICE_SECRET, + scope=settings.MSYNC_SERVICE_SCOPE, + heartbeat_interval=settings.MSYNC_HEARTBEAT_INTERVAL, + metadata={'DBName': settings.DATABASES['default']['NAME']}, + ) + + setattr(settings, 'MSYNC_CLIENT', msync_client) + + t = threading.Thread(target=msync_client.Connect, daemon=True) + t.start() + logger.info('MSYNC Socket.IO 桥接线程已启动') + except Exception as e: + logger.error(f'MSYNC 桥接启动失败: {e}') diff --git a/ATestCServer/apps/sub_server/form_service.py b/ATestCServer/apps/sub_server/form_service.py new file mode 100644 index 00000000..8d960bc9 --- /dev/null +++ b/ATestCServer/apps/sub_server/form_service.py @@ -0,0 +1,22 @@ +"""模块：ATestCServer FormService +使用 gvsdsdk 通用 FormService 引擎暴露子服务器 ORM 表。""" + +from gvsdsdk.form_service import FormServiceBase +from .models import BusinessUser, BusinessOrder, SyncLog + + +class ATestCFormService(FormServiceBase): + TABLE_MAP = { + 'BusinessUser': BusinessUser, + 'BusinessOrder': BusinessOrder, + 'SyncLog': SyncLog, + } + + ENTITY_PICKER_MAP = { + 'BusinessUser': { + 'api': '/api/sub/form/action', + 'labelField': 'Username', + 'valueField': 'UserID', + 'placeholder': '输入用户名搜索...', + }, + } diff --git a/ATestCServer/apps/sub_server/local_handlers.py b/ATestCServer/apps/sub_server/local_handlers.py new file mode 100644 index 00000000..a97818fd --- /dev/null +++ b/ATestCServer/apps/sub_server/local_handlers.py @@ -0,0 +1,53 @@ +"""实验用本地事务处理器 + +随子服务器进程（runserver）启动即注册，供主服务器 Saga 编排回调。 +msync_experiment 管理命令与 HTTP 服务共用同一进程内处理器。 +""" + +import uuid + +from gvsdsdk.msync.local_tx import local_tx + +# demo_flaky 重试实验状态（进程内计数器，可通过 local/reset 端点清零） +_flaky_state = {'attempts': 0} + + +def _ActionA(payload): + return {'demo_a_uuid': str(uuid.uuid4()), 'received': payload} + + +def _ActionB(payload): + # 返回收到的完整 payload，供主服务器验证步骤间数据流 + return {'received_payload': payload} + + +def _CompensateA(payload, action_result): + return {'compensated': True, 'action_result': action_result} + + +def _ActionFail(payload): + raise RuntimeError('确定性失败（实验）') + + +def _ActionFlaky(payload): + _flaky_state['attempts'] += 1 + if _flaky_state['attempts'] == 1: + raise RuntimeError('首次失败（实验：验证自动重试）') + return {'retried': True, 'attempts': _flaky_state['attempts']} + + +def RegisterExperimentHandlers(): + """注册全部实验处理器（幂等：已存在则跳过）""" + for name, action, compensate in ( + ('demo_a', _ActionA, _CompensateA), + ('demo_b', _ActionB, None), + ('demo_fail', _ActionFail, None), + ('demo_flaky', _ActionFlaky, None), + ): + if not local_tx.HasHandler(name): + local_tx.register(name, action=action, compensate=compensate) + + +def ResetExperimentState(): + """清零重试实验状态（供 local/reset 端点调用）""" + _flaky_state['attempts'] = 0 diff --git a/ATestCServer/apps/sub_server/management/__init__.py b/ATestCServer/apps/sub_server/management/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/ATestCServer/apps/sub_server/management/commands/__init__.py b/ATestCServer/apps/sub_server/management/commands/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/ATestCServer/apps/sub_server/management/commands/msync_experiment.py b/ATestCServer/apps/sub_server/management/commands/msync_experiment.py new file mode 100644 index 00000000..cc0d80bc --- /dev/null +++ b/ATestCServer/apps/sub_server/management/commands/msync_experiment.py @@ -0,0 +1,202 @@ +"""msync_experiment — MSYNC 主从链路自动化实验与验证 + +用法: + python manage.py msync_experiment + +前置条件: + - 主服务器在线（SIO + HTTP），且本服务已通过 MSYNC 桥接注册成功 + - 本服务 local 端点可被主服务器访问（MSYNC_SERVICE_DOMAIN） + +自动执行并输出 PASS/FAIL 汇报表: + 1. 环境状态: SDK 版本 / SIO 连接 / 认证 / 注册 + 2. 正向 Saga: 多步骤成功 → Committed，且验证步骤间数据流 + 3. 失败补偿: 步骤失败 → 逆序补偿 → Compensated + 4. 自动重试: 步骤首次失败 → 主服务器按 CanRetry 自动重试成功 + 5. 幂等去重: local_tx 相同 Idempotency-Key 命中缓存 +""" + +import logging +import time + +from django.conf import settings +from django.core.management.base import BaseCommand + +logger = logging.getLogger(__name__) + + +def _get_msync(): + return getattr(settings, 'MSYNC_CLIENT', None) + + +class Command(BaseCommand): + help = 'MSYNC 主从链路自动化实验与验证' + + def add_arguments(self, parser): + parser.add_argument( + '--wait-seconds', type=int, default=10, + help='等待桥接就绪的秒数（默认 10）', + ) + + def handle(self, *args, **options): + self.results = [] + wait_seconds = options['wait_seconds'] + + self.out('=' * 60) + self.out('MSYNC 主从链路自动化实验') + self.out('=' * 60) + + msync = _get_msync() + if msync is None: + self.fail('MSYNC 未初始化', 'settings.MSYNC_CLIENT 不存在，桥接线程未启动') + return + + # 1. 等待桥接就绪（连接/注册/认证） + self.record('等待桥接就绪', self._wait_ready(msync, wait_seconds)) + self._report_status(msync) + + # 2. 注册本地实验步骤处理器 + self.record('注册本地步骤处理器', self._register_handlers(), 'demo_a/demo_b/demo_fail') + + # 3. 正向 Saga + 步骤数据流 + self.record('正向 Saga（数据流）', self._experiment_forward(msync)) + + # 4. 失败补偿 + self.record('失败补偿', self._experiment_compensate(msync)) + + # 5. 自动重试 + self.record('自动重试', self._experiment_retry(msync)) + + # 6. 幂等去重 + self.record('幂等去重', self._experiment_idempotency()) + + self._summary() + + # ─── 工具 ────────────────────────────────────────────── + + def out(self, msg): + self.stdout.write(msg) + + def record(self, name, ok, detail=''): + self.results.append((name, bool(ok), detail or '')) + mark = 'PASS' if ok else 'FAIL' + self.stdout.write(f' [{mark}] {name}' + (f' ({detail})' if detail else '')) + + def fail(self, name, detail=''): + self.record(name, False, detail) + + def _wait_ready(self, msync, seconds): + for _ in range(seconds * 2): + if msync.IsConnected and msync.IsAuthenticated and msync.IsRegistered: + return True + time.sleep(0.5) + return msync.IsConnected and msync.IsAuthenticated and msync.IsRegistered + + def _report_status(self, msync): + import gvsdsdk + info = gvsdsdk.get_version_info() + self.out(f'\n SDK 版本 : {info["Version"]} @ {info["Location"]}') + self.out(f' SIO 连接 : {msync.IsConnected}') + self.out(f' 密钥认证 : {msync.IsAuthenticated}') + self.out(f' 服务注册 : {msync.IsRegistered}') + self.out(f' 服务名称 : {msync.ServiceName}') + self.out(f' 主服务器 : {msync.MasterURL}') + + # ─── 实验步骤 ────────────────────────────────────────── + + def _register_handlers(self): + from apps.sub_server.local_handlers import RegisterExperimentHandlers + from gvsdsdk.msync.local_tx import local_tx + + RegisterExperimentHandlers() + return all(local_tx.HasHandler(n) for n in ('demo_a', 'demo_b')) + + def _create_and_execute(self, msync, name, steps, timeout_seconds=120): + created = msync.TxCreate(name, steps, timeout_seconds=timeout_seconds) + if not created: + return None, 'TxCreate 失败' + tx_code = created.get('TransactionCode') + detail = msync.TxExecute(tx_code, access_token=msync.AccessToken) + if detail is None: + return tx_code, 'TxExecute 失败' + return tx_code, detail + + def _experiment_forward(self, msync): + tx_code, detail = self._create_and_execute(msync, '实验-正向', [ + {'name': 'demo_a', 'ParticipantService': msync.ServiceName, 'ActionPayload': {'in': 'a'}}, + {'name': 'demo_b', 'ParticipantService': msync.ServiceName, 'ActionPayload': {'in': 'b'}}, + ]) + if not detail: + return False + + self.out(f'\n 正向事务 {tx_code}: Status={detail.get("Status")} / Code={detail.get("StatusCode")}') + committed = detail.get('StatusCode') == 2 + if not committed: + self.out(f' 事务未提交: {detail}') + return False + + steps = {s['Index']: s for s in detail.get('Steps', [])} + step_a, step_b = steps.get(0), steps.get(1) + if not step_a or not step_b: + return False + + result_b = (step_b.get('Result') or {}).get('received_payload') or {} + flow_ok = bool(result_b.get('demo_a_uuid')) # demo_b 收到 demo_a 的输出 → 数据流生效 + self.out(f' 数据流验证: demo_b 收到 demo_a_uuid = {result_b.get("demo_a_uuid")}') + return committed and flow_ok + + def _experiment_compensate(self, msync): + tx_code, detail = self._create_and_execute(msync, '实验-补偿', [ + {'name': 'demo_a', 'ParticipantService': msync.ServiceName}, + {'name': 'demo_fail', 'ParticipantService': msync.ServiceName, 'MaxRetries': 0}, + ]) + if not detail: + return False + self.out(f'\n 补偿事务 {tx_code}: Status={detail.get("Status")} / Code={detail.get("StatusCode")}') + steps = {s['Index']: s for s in detail.get('Steps', [])} + return detail.get('StatusCode') == 4 and steps.get(0, {}).get('Status') in ('已补偿', 'Compensated') + + def _experiment_retry(self, msync): + self._reset_flaky_state() + tx_code, detail = self._create_and_execute(msync, '实验-重试', [ + {'name': 'demo_flaky', 'ParticipantService': msync.ServiceName, 'MaxRetries': 2}, + ]) + if not detail: + return False + self.out(f'\n 重试事务 {tx_code}: Status={detail.get("Status")} / Code={detail.get("StatusCode")}') + steps = {s['Index']: s for s in detail.get('Steps', [])} + step = steps.get(0, {}) + ok = detail.get('StatusCode') == 2 and step.get('RetryCount', 0) >= 1 + self.out(f' 自动重试次数: {step.get("RetryCount")}（要求 >=1）') + return ok + + def _reset_flaky_state(self): + """重置 runserver 进程内的 demo_flaky 计数器（通过 local/reset 端点）""" + import requests + try: + reset_url = f'{settings.MSYNC_SERVICE_DOMAIN}/api/msync/local/reset/' + requests.post(reset_url, timeout=5) + except Exception: + pass + + def _experiment_idempotency(self): + from gvsdsdk.msync.local_tx import local_tx + + r1 = local_tx.ExecuteStep('no_such_demo', {}, 'EXP:1:0') + r2 = local_tx.ExecuteStep('no_such_demo', {}, 'EXP:1:0') + r3 = local_tx.ExecuteStep('no_such_demo', {}, 'EXP:1:1') + return r1 is r2 and r1 is not r3 + + # ─── 汇报 ────────────────────────────────────────────── + + def _summary(self): + self.out('\n' + '=' * 60) + self.out('实验汇报') + self.out('=' * 60) + passed = sum(1 for _, ok, _ in self.results if ok) + for name, ok, detail in self.results: + mark = '✓' if ok else '✗' + self.out(f' {mark} {name}' + (f' ({detail})' if detail else '')) + self.out(f'\n结果: {passed}/{len(self.results)} 通过') + if passed != len(self.results): + self.stdout.write(self.style.ERROR('存在未通过的实验环节，请查看上方明细')) + raise SystemExit(1) diff --git a/ATestCServer/apps/sub_server/management/commands/report_manifest.py b/ATestCServer/apps/sub_server/management/commands/report_manifest.py new file mode 100644 index 00000000..7caa658f --- /dev/null +++ b/ATestCServer/apps/sub_server/management/commands/report_manifest.py @@ -0,0 +1,129 @@ +"""report_manifest — MSYNC 远端 App 清单上报演示 + +用法: + python manage.py report_manifest [--app-code custom-demo] [--app-name 定制项目演示] + +前置条件: + - 主服务器在线（SIO + HTTP），本服务已通过 MSYNC 注册认证 + - 上报后主服务器应用中心可见该 App（Source=msync）， + 租户管理员可安装到指定公司，配合 RAM 授权角色粒度可见 + +动作: + 1. 构建定制项目 App 清单（页面由本服务器托管，iframe 加载） + 2. 通过 MSYNC ReportAppManifests 上报主服务器 + 3. 输出 Registered/Failed 汇报 +""" + +import logging + +from django.conf import settings +from django.core.management.base import BaseCommand + +logger = logging.getLogger(__name__) + + +def _get_msync(): + return getattr(settings, 'MSYNC_CLIENT', None) + + +def _DefaultManifest(app_code, app_name, service_domain): + """构建演示清单：App + 两个菜单（首页 + 数据页），挂载在 myself 下""" + base = service_domain.rstrip('/') + return [ + { + 'AppCode': app_code, + 'AppName': app_name, + 'AppIcon': 'Connection', + 'AppDesc': f'{app_name}——由子服务器 {settings.MSYNC_SERVICE_NAME} 通过 MSYNC 上报的定制页面', + 'AppType': 'enterprise', + 'AuthRequired': 0, + 'AppSort': 50, + 'FrontendRoute': f'/remote/{app_code}', + 'FrontendType': 'legacy', + 'LegacySrc': f'{base}/status/', + 'Menus': [ + { + 'MenuName': f'{app_code}', + 'MenuTitle': app_name, + 'MenuIcon': 'Connection', + 'MenuSort': 1, + }, + { + 'MenuName': f'{app_code}-home', + 'MenuTitle': f'{app_name}·首页', + 'MenuIcon': 'Home', + 'MenuLink': f'{base}/status/', + 'ParentMenuName': f'{app_code}', + 'MenuSort': 1, + }, + ], + 'ConfigKeys': [], + } + ] + + +class Command(BaseCommand): + help = 'MSYNC 远端 App 清单上报演示' + + def add_arguments(self, parser): + parser.add_argument( + '--app-code', default='custom-demo', + help='演示 AppCode（默认 custom-demo）', + ) + parser.add_argument( + '--app-name', default='定制项目演示', + help='演示 AppName（默认 定制项目演示）', + ) + parser.add_argument( + '--wait-seconds', type=int, default=10, + help='等待桥接就绪的秒数（默认 10）', + ) + + def handle(self, *args, **options): + app_code = options['app_code'] + app_name = options['app_name'] + wait_seconds = options['wait_seconds'] + + self.stdout.write('=' * 60) + self.stdout.write('MSYNC 远端 App 清单上报演示') + self.stdout.write('=' * 60) + + msync = _get_msync() + if msync is None: + self.stdout.write(self.style.ERROR('MSYNC 未初始化：settings.MSYNC_CLIENT 不存在')) + return + + # 1. 等待桥接就绪 + ready = self._wait_ready(msync, wait_seconds) + if not ready: + self.stdout.write(self.style.ERROR('桥接未就绪（连接/注册/认证失败），中止上报')) + return + self.stdout.write(self.style.SUCCESS( + f'桥接就绪: {msync.ServiceName} @ {msync.ServiceDomain}' + )) + + # 2. 构建清单并上报 + manifests = _DefaultManifest(app_code, app_name, msync.ServiceDomain) + self.stdout.write(f'上报清单: {app_code}（{app_name}），菜单 {len(manifests[0]["Menus"])} 个') + ok, result = msync.ReportAppManifests(manifests) + if not ok: + self.stdout.write(self.style.ERROR(f'上报失败: {result}')) + return + self.stdout.write(self.style.SUCCESS(f'上报结果: Registered={result.get("Registered")} Failed={result.get("Failed")}')) + + # 3. 提示后续步骤 + self.stdout.write('=' * 60) + self.stdout.write('后续步骤（主服务器侧）：') + self.stdout.write(' 1. 应用中心可见该 App（来源标记 msync）') + self.stdout.write(' 2. 租户管理员安装到指定公司') + self.stdout.write(' 3. RAMManager.GrantAccess 授权角色可见（可选，不授权则全员可见）') + self.stdout.write(' 4. 员工登录后菜单注入，点击 iframe 加载本服务器页面') + self.stdout.write('=' * 60) + + def _wait_ready(self, msync, wait_seconds): + for _ in range(wait_seconds): + if msync.IsConnected and msync.IsAuthenticated and msync.IsRegistered: + return True + import time + time.sleep(1) + return msync.IsConnected and msync.IsAuthenticated and msync.IsRegistered diff --git a/ATestCServer/apps/sub_server/migrations/0001_initial.py b/ATestCServer/apps/sub_server/migrations/0001_initial.py new file mode 100644 index 00000000..39ed3c06 --- /dev/null +++ b/ATestCServer/apps/sub_server/migrations/0001_initial.py @@ -0,0 +1,65 @@ +# Generated by Django 6.0.5 on 2026-06-11 01:44 + +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + + initial = True + + dependencies = [ + ] + + operations = [ + migrations.CreateModel( + name='BusinessUser', + fields=[ + ('UserID', models.AutoField(primary_key=True, serialize=False)), + ('Username', models.CharField(max_length=64, unique=True)), + ('Nickname', models.CharField(default='', max_length=128)), + ('OGMMappingID', models.CharField(db_index=True, default='', max_length=128)), + ('JoinTime', models.DateTimeField(auto_now_add=True)), + ('UserStatus', models.SmallIntegerField(default=1)), + ], + options={ + 'verbose_name': '业务用户', + 'verbose_name_plural': '业务用户', + 'db_table': 'sub_business_user', + }, + ), + migrations.CreateModel( + name='SyncLog', + fields=[ + ('LogID', models.AutoField(primary_key=True, serialize=False)), + ('EventType', models.CharField(max_length=64)), + ('EventDetail', models.JSONField(default=dict)), + ('EventStatus', models.SmallIntegerField(default=1)), + ('CreateTime', models.DateTimeField(auto_now_add=True)), + ], + options={ + 'verbose_name': '同步日志', + 'verbose_name_plural': '同步日志', + 'db_table': 'sub_sync_log', + 'ordering': ['-CreateTime'], + }, + ), + migrations.CreateModel( + name='BusinessOrder', + fields=[ + ('OrderID', models.AutoField(primary_key=True, serialize=False)), + ('OrderCode', models.CharField(db_index=True, max_length=64, unique=True)), + ('Amount', models.DecimalField(decimal_places=4, max_digits=18)), + ('OrderStatus', models.SmallIntegerField(default=0)), + ('OGMMappingID', models.CharField(db_index=True, default='', max_length=128)), + ('CreateTime', models.DateTimeField(auto_now_add=True)), + ('UpdateTime', models.DateTimeField(auto_now=True)), + ('User', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='orders', to='sub_server.businessuser')), + ], + options={ + 'verbose_name': '业务订单', + 'verbose_name_plural': '业务订单', + 'db_table': 'sub_business_order', + }, + ), + ] diff --git a/ATestCServer/apps/sub_server/migrations/__init__.py b/ATestCServer/apps/sub_server/migrations/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/ATestCServer/apps/sub_server/models.py b/ATestCServer/apps/sub_server/models.py new file mode 100644 index 00000000..58bd95bd --- /dev/null +++ b/ATestCServer/apps/sub_server/models.py @@ -0,0 +1,66 @@ +"""模块：子服务器数据模型 +模拟一个真实子服务器的业务数据结构。""" + +from django.db import models + + +class BusinessUser(models.Model): + """业务用户——模拟子服务器的本地用户数据""" + + UserID = models.AutoField(primary_key=True, verbose_name='用户ID') + Username = models.CharField(max_length=64, unique=True, verbose_name='用户名') + Nickname = models.CharField(max_length=128, default='', verbose_name='昵称') + OGMMappingID = models.CharField(max_length=128, default='', db_index=True, verbose_name='OGM映射ID') + JoinTime = models.DateTimeField(auto_now_add=True, verbose_name='加入时间') + UserStatus = models.SmallIntegerField(default=1, verbose_name='用户状态') + + class Meta: + db_table = 'sub_business_user' + verbose_name = '业务用户' + verbose_name_plural = '业务用户' + + def __str__(self): + return self.Username + + +class BusinessOrder(models.Model): + """业务订单——模拟子服务器的本地订单数据""" + + OrderID = models.AutoField(primary_key=True, verbose_name='订单ID') + OrderCode = models.CharField(max_length=64, unique=True, db_index=True, verbose_name='订单编号') + User = models.ForeignKey( + BusinessUser, on_delete=models.CASCADE, related_name='orders', + verbose_name='用户', + ) + Amount = models.DecimalField(max_digits=18, decimal_places=4, verbose_name='金额') + OrderStatus = models.SmallIntegerField(default=0, verbose_name='订单状态') + OGMMappingID = models.CharField(max_length=128, default='', db_index=True, verbose_name='OGM映射ID') + CreateTime = models.DateTimeField(auto_now_add=True, verbose_name='创建时间') + UpdateTime = models.DateTimeField(auto_now=True, verbose_name='更新时间') + + class Meta: + db_table = 'sub_business_order' + verbose_name = '业务订单' + verbose_name_plural = '业务订单' + + def __str__(self): + return self.OrderCode + + +class SyncLog(models.Model): + """同步日志——记录与主服务器 MSYNC 通信的事件""" + + LogID = models.AutoField(primary_key=True, verbose_name='日志ID') + EventType = models.CharField(max_length=64, verbose_name='事件类型') + EventDetail = models.JSONField(default=dict, verbose_name='事件详情') + EventStatus = models.SmallIntegerField(default=1, verbose_name='事件状态') + CreateTime = models.DateTimeField(auto_now_add=True, verbose_name='创建时间') + + class Meta: + db_table = 'sub_sync_log' + verbose_name = '同步日志' + verbose_name_plural = '同步日志' + ordering = ['-CreateTime'] + + def __str__(self): + return f'{self.EventType}@{self.CreateTime}' \ No newline at end of file diff --git a/ATestCServer/apps/sub_server/serializers.py b/ATestCServer/apps/sub_server/serializers.py new file mode 100644 index 00000000..c924cf2a --- /dev/null +++ b/ATestCServer/apps/sub_server/serializers.py @@ -0,0 +1,34 @@ +from rest_framework import serializers +from .models import BusinessUser, BusinessOrder, SyncLog + + +class BusinessUserSerializer(serializers.ModelSerializer): + class Meta: + model = BusinessUser + fields = ['UserID', 'Username', 'Nickname', 'OGMMappingID', 'JoinTime', 'UserStatus'] + read_only_fields = ['UserID', 'JoinTime', 'OGMMappingID'] + + +class BusinessOrderSerializer(serializers.ModelSerializer): + class Meta: + model = BusinessOrder + fields = ['OrderID', 'OrderCode', 'Amount', 'OrderStatus', 'OGMMappingID', 'CreateTime', 'UpdateTime', 'User'] + read_only_fields = ['OrderID', 'CreateTime', 'UpdateTime', 'OGMMappingID'] + + +class OrderCreateSerializer(serializers.Serializer): + username = serializers.CharField() + amount = serializers.DecimalField(max_digits=18, decimal_places=4) + + +class OGMRegisterSerializer(serializers.Serializer): + object_type = serializers.CharField() + local_id = serializers.CharField() + metadata = serializers.DictField(required=False, default=dict) + + +class SyncLogSerializer(serializers.ModelSerializer): + class Meta: + model = SyncLog + fields = ['LogID', 'EventType', 'EventDetail', 'EventStatus', 'CreateTime'] + read_only_fields = ['LogID', 'CreateTime'] \ No newline at end of file diff --git a/ATestCServer/apps/sub_server/service_auth.py b/ATestCServer/apps/sub_server/service_auth.py new file mode 100644 index 00000000..c9c7c307 --- /dev/null +++ b/ATestCServer/apps/sub_server/service_auth.py @@ -0,0 +1,81 @@ +"""子服务器服务认证 + +主服务器调用子服务器 local 端点时携带 `Authorization: Service `。 +子服务器无法本地验证令牌（无主库 ServiceToken 表 / MASTER_SECRET）， +因此通过主服务器 `/api/msync/auth/introspect/` 远程自省校验。 +""" + +import json +import logging +import urllib.request +import urllib.error + +from django.conf import settings +from rest_framework.authentication import BaseAuthentication +from rest_framework.exceptions import AuthenticationFailed + +logger = logging.getLogger(__name__) + +SERVICE_AUTH_HEADER = 'Authorization' +SERVICE_AUTH_PREFIX = 'Service' +INTROSPECT_PATH = '/api/msync/auth/introspect/' + + +class ServicePrincipal: + """服务主体——标识调用方服务""" + + def __init__(self, service_name): + self.service_name = service_name + self.service_scope = [] + + @property + def is_authenticated(self): + return True + + def __str__(self): + return f'Service:{self.service_name}' + + +class SubServerServiceAuthentication(BaseAuthentication): + """DRF 认证类——校验主服务器 Service 令牌（远程自省）""" + + def authenticate(self, request): + auth_header = request.headers.get(SERVICE_AUTH_HEADER, '') + if not auth_header.startswith(f'{SERVICE_AUTH_PREFIX} '): + return None + token = auth_header[len(SERVICE_AUTH_PREFIX) + 1:].strip() + if not token: + return None + + payload = self._Introspect(token) + if not payload: + raise AuthenticationFailed('服务令牌验证失败') + + return (ServicePrincipal(payload.get('sub', 'unknown')), token) + + def authenticate_header(self, request): + return 'Service' + + @staticmethod + def _Introspect(access_token): + """调用主服务器令牌自省端点""" + master_url = str(getattr(settings, 'MSYNC_MASTER_URL', '')).rstrip('/') + if not master_url: + return None + body = json.dumps({'AccessToken': access_token}).encode('utf-8') + req = urllib.request.Request( + f'{master_url}{INTROSPECT_PATH}', + data=body, + headers={'Content-Type': 'application/json'}, + method='POST', + ) + try: + with urllib.request.urlopen(req, timeout=10) as resp: + data = json.loads(resp.read().decode('utf-8')) + except (urllib.error.URLError, urllib.error.HTTPError, ValueError, OSError) as e: + logger.error(f'令牌自省失败: {e}') + return None + + if not data.get('Valid'): + return None + return data.get('Payload') or {} diff --git a/ATestCServer/apps/sub_server/templates/status.html b/ATestCServer/apps/sub_server/templates/status.html new file mode 100644 index 00000000..187cdd7b --- /dev/null +++ b/ATestCServer/apps/sub_server/templates/status.html @@ -0,0 +1,217 @@ + + + + + +ATestCServer — MSYNC 连接状态 + + + +
+
+

ATestCServer

+ SMANAGE 子服务器 · HTTP :25080 · HTTPS :25443 +
+
+ + 断开 +
+
+ +
+ +
+

连接状态

+
加载中...
+
加载中...
+
加载中...
+
加载中...
+
自动刷新中...
+
+ + +
+

服务标识

+
+
服务名—
+
域名—
+
主服务器—
+
Token—
+
+
+ + +
+

对等服务

+
  • 加载中...
+
+ + +
+

最近同步事件

+
  • 暂无事件
+
+ + +
+

原始响应

+ 
{}
+
+
+ +
+ MSYNC Bridge · Heartbeat/30s · ATestCServer © SMANAGE +
+ + + + \ No newline at end of file diff --git a/ATestCServer/apps/sub_server/urls.py b/ATestCServer/apps/sub_server/urls.py new file mode 100644 index 00000000..d3a8bb62 --- /dev/null +++ b/ATestCServer/apps/sub_server/urls.py @@ -0,0 +1,21 @@ +from django.urls import path +from . import views + +app_name = 'sub_server' + +urlpatterns = [ + path('health/', views.HealthView.as_view(), name='health'), + path('status/', views.StatusPageView, name='status_page'), + path('msync/status/', views.MSyncStatusView.as_view(), name='msync_status'), + path('users/', views.UserListView.as_view(), name='user_list'), + path('users//', views.UserDetailView.as_view(), name='user_detail'), + path('orders/', views.OrderListView.as_view(), name='order_list'), + path('orders//', views.OrderDetailView.as_view(), name='order_detail'), + path('ogm/register/', views.OGMRegisterView.as_view(), name='ogm_register'), + path('ogm/resolve//', views.OGMResolveView.as_view(), name='ogm_resolve'), + path('remote/call/', views.RemoteCallDemoView.as_view(), name='remote_call'), + path('sync/logs/', views.SyncLogView.as_view(), name='sync_logs'), + path('local/execute/', views.LocalExecuteView.as_view(), name='local_execute'), + path('local/compensate/', views.LocalCompensateView.as_view(), name='local_compensate'), + path('local/handlers/', views.LocalHandlersView.as_view(), name='local_handlers'), +] \ No newline at end of file diff --git a/ATestCServer/apps/sub_server/views.py b/ATestCServer/apps/sub_server/views.py new file mode 100644 index 00000000..9b6389ca --- /dev/null +++ b/ATestCServer/apps/sub_server/views.py @@ -0,0 +1,420 @@ +"""模块：子服务器 REST API 视图 +提供业务 CRUD 端点 + MSYNC 状态查询 + 服务间调用演示 + FormService ORM 管理。""" + +import logging +from django.conf import settings +from django.shortcuts import render +from rest_framework.views import APIView +from rest_framework.response import Response +from rest_framework.permissions import AllowAny, IsAuthenticated +from .service_auth import SubServerServiceAuthentication +from .models import BusinessUser, BusinessOrder, SyncLog +from .serializers import ( + BusinessUserSerializer, + BusinessOrderSerializer, + OrderCreateSerializer, + OGMRegisterSerializer, + SyncLogSerializer, +) + +logger = logging.getLogger(__name__) + + +def _ok(data=None, msg='success'): + return {'code': 200, 'msg': msg, 'data': data} + + +def _err(msg, code=400): + return {'code': code, 'msg': msg, 'data': None} + + +def _get_msync(): + """获取 MSYNC 客户端实例""" + return getattr(settings, 'MSYNC_CLIENT', None) + + +def _log_sync(event_type, detail, status=1): + """记录同步日志""" + try: + SyncLog.objects.create( + EventType=event_type, + EventDetail=detail, + EventStatus=status, + ) + except Exception: + pass + + +class HealthView(APIView): + authentication_classes = [] + permission_classes = [AllowAny] + + def get(self, request): + msync = _get_msync() + token = msync.AccessToken if msync else None + msync_status = { + 'Connected': msync.IsConnected if msync else False, + 'Authenticated': msync.IsAuthenticated if msync else False, + 'Registered': msync.IsRegistered if msync else False, + 'ServiceName': msync.ServiceName if msync else None, + 'ServiceDomain': msync.ServiceDomain if msync else None, + 'MasterURL': msync.MasterURL if msync else None, + 'AccessToken': 'exists' if token else 'none', + } + + # SDK 版本信息 + sdk_info = {'Name': 'gvsdsdk'} + try: + import gvsdsdk + sdk_info.update(gvsdsdk.get_version_info()) + except ImportError as e: + sdk_info['Error'] = str(e) + + # SAAS 数据库连接状态 + saas_db_status = {'Configured': 'saas' in settings.DATABASES} + try: + from django.db import connections + conn = connections['saas'] + conn.ensure_connection() + saas_db_status['Connected'] = True + # 尝试统计用户数验证连接可用 + from gvsdsdk.models import User + saas_db_status['UserCount'] = User.objects.using('saas').count() + except Exception as e: + saas_db_status['Connected'] = False + saas_db_status['Error'] = str(e) + + return Response(_ok({ + 'Server': 'ATestCServer', + 'Status': 'running', + 'MSync': msync_status, + 'SDK': sdk_info, + 'SAASDB': saas_db_status, + })) + + +class MSyncStatusView(APIView): + authentication_classes = [] + permission_classes = [AllowAny] + + def get(self, request): + msync = _get_msync() + if not msync: + return Response(_err('MSYNC 未初始化', 500), status=500) + + services = msync.DiscoverAll() + token = msync.AccessToken + return Response(_ok({ + 'ServiceName': msync.ServiceName, + 'ServiceDomain': msync.ServiceDomain, + 'MasterURL': msync.MasterURL, + 'Authenticated': msync.IsAuthenticated, + 'Registered': msync.IsRegistered, + 'Connected': msync.IsConnected, + 'AccessToken': 'exists' if token else 'none', + 'PeerServices': services, + })) + + +class UserListView(APIView): + + def get(self, request): + users = BusinessUser.objects.filter(UserStatus=1).order_by('-JoinTime') + return Response(_ok(BusinessUserSerializer(users, many=True).data)) + + def post(self, request): + username = request.data.get('username', '').strip() + nickname = request.data.get('nickname', '') + + if not username: + return Response(_err('用户名不能为空', 400), status=400) + + user, created = BusinessUser.objects.get_or_create( + Username=username, + defaults={'Nickname': nickname or username}, + ) + + if created: + msync = _get_msync() + if msync: + try: + ogm_result = msync.OGMRegister( + object_type='user', + local_id=str(user.UserID), + service_name=msync.ServiceName, + metadata={'username': username, 'nickname': nickname or username}, + ) + if ogm_result: + user.OGMMappingID = ogm_result.get('ObjectGlobalID', '') + user.save(update_fields=['OGMMappingID']) + _log_sync('ogm_register', { + 'object_type': 'user', + 'local_id': str(user.UserID), + 'global_id': user.OGMMappingID, + }) + except Exception as e: + logger.warning(f'OGM注册失败: {e}') + + status_code = 201 if created else 200 + return Response(_ok(BusinessUserSerializer(user).data), status=status_code) + + +class UserDetailView(APIView): + + def get(self, request, user_id): + try: + user = BusinessUser.objects.get(UserID=user_id) + except BusinessUser.DoesNotExist: + return Response(_err('用户不存在', 404), status=404) + return Response(_ok(BusinessUserSerializer(user).data)) + + +class OrderListView(APIView): + + def get(self, request): + orders = BusinessOrder.objects.all().order_by('-CreateTime')[:50] + return Response(_ok(BusinessOrderSerializer(orders, many=True).data)) + + def post(self, request): + ser = OrderCreateSerializer(data=request.data) + if not ser.is_valid(): + return Response(_err(ser.errors, 400), status=400) + + username = ser.validated_data['username'] + amount = ser.validated_data['amount'] + + try: + user = BusinessUser.objects.get(Username=username) + except BusinessUser.DoesNotExist: + return Response(_err('用户不存在', 404), status=404) + + import uuid + order = BusinessOrder.objects.create( + OrderCode=f'SUB-{uuid.uuid4().hex[:12].upper()}', + User=user, + Amount=amount, + OrderStatus=0, + ) + + msync = _get_msync() + if msync: + try: + ogm_result = msync.OGMRegister( + object_type='order', + local_id=str(order.OrderID), + service_name=msync.ServiceName, + metadata={ + 'order_code': order.OrderCode, + 'amount': str(amount), + 'user_id': str(user.UserID), + }, + ) + if ogm_result: + order.OGMMappingID = ogm_result.get('ObjectGlobalID', '') + order.save(update_fields=['OGMMappingID']) + _log_sync('ogm_register', { + 'object_type': 'order', + 'local_id': str(order.OrderID), + 'global_id': order.OGMMappingID, + }) + except Exception as e: + logger.warning(f'订单OGM注册失败: {e}') + + return Response(_ok(BusinessOrderSerializer(order).data), status=201) + + +class OrderDetailView(APIView): + + def get(self, request, order_id): + try: + order = BusinessOrder.objects.get(OrderID=order_id) + except BusinessOrder.DoesNotExist: + return Response(_err('订单不存在', 404), status=404) + return Response(_ok(BusinessOrderSerializer(order).data)) + + +class OGMRegisterView(APIView): + + def post(self, request): + ser = OGMRegisterSerializer(data=request.data) + if not ser.is_valid(): + return Response(_err(ser.errors, 400), status=400) + + msync = _get_msync() + if not msync: + return Response(_err('MSYNC 未初始化', 500), status=500) + + result = msync.OGMRegister( + object_type=ser.validated_data['object_type'], + local_id=ser.validated_data['local_id'], + service_name=msync.ServiceName, + metadata=ser.validated_data.get('metadata', {}), + ) + + if result: + _log_sync('ogm_register_manual', { + 'object_type': ser.validated_data['object_type'], + 'local_id': ser.validated_data['local_id'], + 'global_id': result.get('ObjectGlobalID', ''), + }) + return Response(_ok(result), status=201) + + return Response(_err('OGM注册失败', 500), status=500) + + +class OGMResolveView(APIView): + + def get(self, request, global_id): + msync = _get_msync() + if not msync: + return Response(_err('MSYNC 未初始化', 500), status=500) + + result = msync.OGMResolve(global_id) + if result: + return Response(_ok(result)) + return Response(_err('映射不存在', 404), status=404) + + +class RemoteCallDemoView(APIView): + + def get(self, request): + msync = _get_msync() + if not msync: + return Response(_err('MSYNC 未初始化', 500), status=500) + + services = msync.DiscoverAll() + results = [] + + for svc in services: + svc_name = svc.get('ServiceName', '') + if svc_name == msync.ServiceName: + continue + + target_domain = svc.get('ServiceDomain', '') + resp = msync.CallRemote( + service_name=svc_name, + method='GET', + path='/health/', + ) + + if resp is None: + results.append({ + 'service': svc_name, + 'domain': target_domain, + 'error': '远端调用失败', + }) + else: + # CallRemote 返回 requests.Response（或 dict 时兼容） + if isinstance(resp, dict): + body = resp + else: + try: + body = resp.json() + except ValueError: + body = {'code': resp.status_code, 'msg': resp.text[:200], 'data': None} + results.append({ + 'service': svc_name, + 'domain': target_domain, + 'code': body.get('code'), + 'msg': body.get('msg'), + 'data': body.get('data'), + }) + + return Response(_ok(results)) + + +class SyncLogView(APIView): + authentication_classes = [] + permission_classes = [AllowAny] + + def get(self, request): + logs = SyncLog.objects.all()[:50] + return Response(_ok(SyncLogSerializer(logs, many=True).data)) + + +def StatusPageView(request): + """连接状态仪表盘页面""" + return render(request, 'status.html', { + 'service_name': settings.MSYNC_SERVICE_NAME, + 'service_domain': settings.MSYNC_SERVICE_DOMAIN, + 'master_url': settings.MSYNC_MASTER_URL, + }) + + +class LocalExecuteView(APIView): + """本地事务执行端点——由主服务器 Saga 编排引擎调用（需主服务器 Service 令牌）""" + + authentication_classes = [SubServerServiceAuthentication] + permission_classes = [IsAuthenticated] + + def post(self, request): + from gvsdsdk.msync.local_tx import local_tx + + step_name = request.data.get('StepName', '') + payload = request.data.get('payload') or {} + idempotency_key = request.headers.get('Idempotency-Key') + + if not step_name: + return Response(_err('StepName 不能为空', 400), status=400) + + result = local_tx.ExecuteStep( + name=step_name, + payload=payload, + idempotency_key=idempotency_key, + ) + + if result['success']: + return Response(_ok(result['result'])) + return Response(_err(result['error'], 500), status=500) + + +class LocalCompensateView(APIView): + """本地事务补偿端点——由主服务器 Saga 编排引擎调用（需主服务器 Service 令牌）""" + + authentication_classes = [SubServerServiceAuthentication] + permission_classes = [IsAuthenticated] + + def post(self, request): + from gvsdsdk.msync.local_tx import local_tx + + step_name = request.data.get('StepName', '') + payload = request.data.get('payload') or {} + action_result = request.data.get('ActionResult') or {} + idempotency_key = request.headers.get('Idempotency-Key') + + if not step_name: + return Response(_err('StepName 不能为空', 400), status=400) + + result = local_tx.CompensateStep( + name=step_name, + payload=payload, + action_result=action_result, + idempotency_key=idempotency_key, + ) + + if result['success']: + return Response(_ok(result['result'])) + return Response(_err(result['error'], 500), status=500) + + +class LocalHandlersView(APIView): + """列出已注册的本地事务步骤处理器""" + + authentication_classes = [SubServerServiceAuthentication] + permission_classes = [IsAuthenticated] + + def get(self, request): + from gvsdsdk.msync.local_tx import local_tx + return Response(_ok(local_tx.ListHandlers())) + + +class LocalResetView(APIView): + """重置实验处理器状态（demo_flaky 重试计数清零），仅限本地回环使用""" + + authentication_classes = [] + permission_classes = [AllowAny] + + def post(self, request): + from .local_handlers import ResetExperimentState + ResetExperimentState() + return Response(_ok({'reset': True})) \ No newline at end of file diff --git a/ATestCServer/manage.py b/ATestCServer/manage.py new file mode 100644 index 00000000..6ef1a1d9 --- /dev/null +++ b/ATestCServer/manage.py @@ -0,0 +1,30 @@ +#!/usr/bin/env python +import os +import sys + +sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +def main(): + sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + os.environ['DJANGO_SETTINGS_MODULE'] = 'ATestCServer.settings' + os.environ['DB_NAME'] = 'atestc' + # 安全：禁止硬编码默认密钥/口令，缺失时显式报错（settings.py 亦有 + # ImproperlyConfigured 兜底）。MSYNC_SERVICE_SECRET 与 DB_PASSWORD + # 由部署环境（环境变量）提供。 + required_envs = ['DJANGO_SECRET_KEY', 'MSYNC_SERVICE_SECRET', 'DB_PASSWORD'] + missing = [name for name in required_envs if not os.environ.get(name)] + if missing: + raise SystemExit( + '缺少必需的环境变量: ' + ', '.join(missing) + + '（禁止使用默认值，请通过环境变量注入）' + ) + try: + from django.core.management import execute_from_command_line + except ImportError as exc: + raise ImportError( + "Couldn't import Django. Make sure it's installed and available." + ) from exc + execute_from_command_line(sys.argv) + + +if __name__ == '__main__': + main() \ No newline at end of file diff --git a/ATestCServer/msync_client/__init__.py b/ATestCServer/msync_client/__init__.py new file mode 100644 index 00000000..89220d60 --- /dev/null +++ b/ATestCServer/msync_client/__init__.py @@ -0,0 +1,9 @@ +"""MSYNC 客户端 + +模块入口，导出： + - MSyncSocketClient: Socket.IO 客户端（实时通道） +""" + +from .sio_client import MSyncSocketClient + +__all__ = ['MSyncSocketClient'] \ No newline at end of file diff --git a/ATestCServer/msync_client/sio_client.py b/ATestCServer/msync_client/sio_client.py new file mode 100644 index 00000000..99b878d1 --- /dev/null +++ b/ATestCServer/msync_client/sio_client.py @@ -0,0 +1,540 @@ +"""模块：MSYNC Socket.IO 客户端 + +替代 HTTP REST，通过 Socket.IO 与 SMANAGE 主服务器实时通信。 +所有方法名、事件字段全部使用 PascalCase。 + +优势： + - 连接复用，无需每次请求建立 TCP/TLS + - 内置心跳保活，替代应用层心跳线程 + - 服务端可主动推送事件（服务上下线通知） + - 断线自动重连 +""" + +import hashlib +import logging +import threading +import time +import socketio + +from gvsdsdk.msync.auth import GenerateChallenge, BuildAuthPayload, SignPayload + +logger = logging.getLogger(__name__) + + +class MSyncSocketClient: + """MSYNC Socket.IO 客户端——子服务器接入 SMANAGE 的实时通道""" + + def __init__(self, master_url, service_name, service_domain, + service_secret, scope=None, heartbeat_interval=30, + service_type=0, metadata=None): + self.MasterURL = master_url.rstrip('/') + self.ServiceName = service_name + self.ServiceDomain = service_domain + self.ServiceSecret = service_secret + self.Scope = scope or ['ogm:read', 'ogm:write', 'registry:read'] + self.HeartbeatInterval = heartbeat_interval + self.ServiceType = service_type + self.Metadata = metadata or {} + + self._sio = socketio.Client( + reconnection=True, + reconnection_attempts=0, + reconnection_delay=2, + reconnection_delay_max=30, + logger=True, + engineio_logger=True, + ) + + self._token_data = None + self._registered = False + self._running = False + self._heartbeat_thread = None + self._initial_connected = False + self._reconnect_lock = threading.Lock() + + self._SetupEvents() + + def _SetupEvents(self): + @self._sio.on('connect', namespace='/msync') + def on_connect(): + logger.info(f'[SIO] ← connect ns=/msync') + if self._initial_connected: + logger.info(f'[SIO] 重连 detected, 重新注册/认证') + t = threading.Thread(target=self._DoReconnect, daemon=True) + t.start() + + @self._sio.on('disconnect', namespace='/msync') + def on_disconnect(): + logger.warning(f'[SIO] ← disconnect ns=/msync') + self._registered = False + self._token_data = None + logger.info(f'[SIO] 状态已清理: registered=False, token_data=None') + + @self._sio.on('msync:service:online', namespace='/msync') + def on_service_online(data): + logger.info(f'[SIO] ← msync:service:online data={data}') + + @self._sio.on('msync:service:offline', namespace='/msync') + def on_service_offline(data): + logger.info(f'[SIO] ← msync:service:offline data={data}') + + @self._sio.on('msync:force:deregister', namespace='/msync') + def on_force_deregister(data): + logger.warning(f'[SIO] ← msync:force:deregister data={data}') + self._registered = False + self._token_data = None + # 延迟后自动重新注册 + t = threading.Thread(target=self._DoReconnect, daemon=True) + t.start() + + @self._sio.on('*') + def catch_all(event, data): + logger.info(f'[SIO] ← unhandled event="{event}" data={data}') + + # ---------- 连接管理 ---------- + + def Connect(self): + self._running = True + url = self.MasterURL + try: + logger.info(f'[SIO] → connect url={url} ns=/msync') + self._sio.connect(url, namespaces=['/msync'], wait=True, wait_timeout=15) + except Exception as e: + logger.warning(f'Socket.IO 连接失败: {e}，启动兜底重连线程') + self._StartBootstrap() + return + + self._OnReady() + + def _StartBootstrap(self): + """初次连接失败的兜底：循环重连，成功后走完整注册/认证流程 + + python-socketio 的 reconnection 仅在"已连接后断开"时生效， + 初次 connect 失败不会自动重连，需自行兜底。 + """ + + def bootstrap(): + while self._running and not self._sio.connected: + try: + logger.info(f'[SIO] 兜底重连尝试: {self.MasterURL}') + self._sio.connect( + self.MasterURL, + namespaces=['/msync'], + wait=True, + wait_timeout=10, + ) + break + except Exception as e: + logger.warning(f'[SIO] 兜底重连失败: {e}') + time.sleep(3) + if self._sio.connected: + self._OnReady() + + threading.Thread(target=bootstrap, daemon=True, name='msync-bootstrap').start() + + def _OnReady(self): + """连接建立后的统一就绪流程：注册/认证/心跳/ORM 中转""" + logger.info(f'[SIO] Socket 已连接, 开始注册/认证...') + for attempt in range(1, 4): + reg_ok = self._DoRegister() + auth_ok = self._DoAuth() + if reg_ok and auth_ok: + break + logger.warning(f'[SIO] 首次注册/认证失败 第 {attempt}/3 次 (reg={reg_ok}, auth={auth_ok})') + if attempt < 3: + time.sleep(2) + self._DoHeartbeat() + self._initial_connected = True + + self._StartHeartbeat() + + # 注册 ORM 中转处理器 + try: + from gvsdsdk.msync.orm_handler import ORMRelayHandler + self._orm_handler = ORMRelayHandler() + self._orm_handler.Register(self) + except Exception as e: + logger.warning(f'[ORM] ORM 中转处理器注册失败: {e}') + + logger.info(f'MSYNC Socket.IO 客户端已启动: {self.ServiceName}') + + def Disconnect(self): + self._running = False + try: + self._DoDeregister() + except Exception: + pass + try: + self._sio.disconnect() + except Exception: + pass + logger.info(f'MSYNC Socket.IO 客户端已断开: {self.ServiceName}') + + def _DoReconnect(self, max_retries=3): + """重连后重新注册/认证，失败时自动重试（内部持有互斥锁）""" + if not self._reconnect_lock.acquire(blocking=False): + logger.warning('[SIO] 重连已在进行中，跳过本次重连') + return + try: + for attempt in range(1, max_retries + 1): + logger.info(f'[SIO] 重连注册/认证 第 {attempt}/{max_retries} 次') + reg_ok = self._DoRegister() + auth_ok = self._DoAuth() + if reg_ok and auth_ok: + logger.info(f'[SIO] 重连注册/认证成功') + return + logger.warning(f'[SIO] 重连注册/认证失败 (reg={reg_ok}, auth={auth_ok})，{2}s 后重试') + if attempt < max_retries: + time.sleep(2) + logger.error(f'[SIO] 重连注册/认证 {max_retries} 次均失败，等待 Socket.IO 下次重连触发') + finally: + self._reconnect_lock.release() + + def _StartHeartbeat(self): + if self._heartbeat_thread and self._heartbeat_thread.is_alive(): + return + self._heartbeat_thread = threading.Thread( + target=self._HeartbeatLoop, + daemon=True, + name=f'msync-heartbeat-{self.ServiceName}', + ) + self._heartbeat_thread.start() + logger.info(f'心跳线程已启动, 间隔={self.HeartbeatInterval}s') + + def _HeartbeatLoop(self): + consecutive_failures = 0 + while self._running: + time.sleep(self.HeartbeatInterval) + try: + if not self._sio.connected: + continue + # 检查令牌是否即将过期，主动续期 + if not self.IsAuthenticated: + logger.info('[SIO] 令牌已过期或即将过期，主动续期') + try: + self.EnsureAuthenticated() + except Exception as e: + logger.warning(f'[SIO] 令牌续期失败: {e}，触发重连') + consecutive_failures += 1 + if consecutive_failures >= 3: + logger.warning('[SIO] 连续失败次数过多，强制重连') + self._ForceReconnect() + consecutive_failures = 0 + continue + # 发送心跳 + result = self._Call('msync:heartbeat', {}) + if result and result.get('code') not in (200, 0): + code = result.get('code') + msg = result.get('msg', '') + logger.warning(f'[SIO] 心跳异常: code={code} msg={msg}') + # 404 = 服务不存在/已禁用，需要立即重新注册 + if code == 404: + logger.warning('[SIO] 服务已被注销，立即重新注册') + try: + reg_ok = self._DoRegister() + auth_ok = self._DoAuth() + if reg_ok and auth_ok: + logger.info('[SIO] 重新注册/认证成功') + consecutive_failures = 0 + continue + except Exception as e: + logger.error(f'[SIO] 重新注册失败: {e}') + # 注册也失败，强制重连 + self._ForceReconnect() + consecutive_failures = 0 + continue + consecutive_failures += 1 + if consecutive_failures >= 3: + logger.warning('[SIO] 心跳连续失败次数过多，强制重连') + self._ForceReconnect() + consecutive_failures = 0 + elif result: + consecutive_failures = 0 + except Exception as e: + logger.warning(f'[SIO] 心跳异常: {e}') + consecutive_failures += 1 + if consecutive_failures >= 3: + self._ForceReconnect() + consecutive_failures = 0 + + def _Call(self, event, data, timeout=15): + if not self._sio.connected: + logger.warning(f'[SIO] ✗ 未连接，无法发送 event={event}') + return None + try: + logger.info(f'[SIO] → event="{event}" ns=/msync data_keys={list(data.keys()) if isinstance(data, dict) else type(data).__name__}') + result = self._sio.call(event, data, namespace='/msync', timeout=timeout) + logger.info(f'[SIO] ← ack event="{event}" result={result}') + return result + except socketio.exceptions.TimeoutError: + logger.error(f'[SIO] ✗ timeout event="{event}" ({timeout}s)') + return None + except Exception as e: + logger.error(f'[SIO] ✗ call failed event="{event}": {type(e).__name__}: {e}') + return None + + # ---------- 注册与认证 ---------- + + def _DoRegister(self): + payload = { + 'ServiceName': self.ServiceName, + 'ServiceDomain': self.ServiceDomain, + 'ServiceSecretHash': hashlib.sha256(self.ServiceSecret.encode()).hexdigest(), + 'ServiceSecret': self.ServiceSecret, + 'ServiceScope': self.Scope, + 'ServiceType': self.ServiceType, + 'ServiceMetadata': self.Metadata, + } + result = self._Call('msync:register', payload) + if result and result.get('code') == 200: + self._registered = True + logger.info(f'服务注册成功: {self.ServiceName}') + return True + if result and result.get('code') == 409: + logger.warning(f'服务已存在(409冲突)，可能已用不同密钥注册: {self.ServiceName}，尝试认证') + return True + logger.warning(f'服务注册失败: result={result}') + return False + + def _DoAuth(self): + challenge = GenerateChallenge() + payload = BuildAuthPayload(self.ServiceName, challenge) + signature = SignPayload(payload, self.ServiceSecret) + + # 调试：打印签名详情 + import json + canonical = json.dumps(payload, sort_keys=True, separators=(',', ':')) + logger.info(f'[AUTH] 客户端签名: service={self.ServiceName}, canonical={canonical}, sig={signature}') + + result = self._Call('msync:auth', { + 'ServiceName': self.ServiceName, + 'Payload': payload, + 'Signature': signature, + }) + if result and result.get('code') == 200 and result.get('data'): + token_data = result['data'] + self._token_data = { + 'AccessToken': token_data.get('AccessToken'), + 'RefreshToken': token_data.get('RefreshToken'), + 'Scope': token_data.get('Scope', []), + 'ExpiresAt': token_data.get('ExpiresAt'), + } + logger.info(f'服务认证成功: {self.ServiceName}') + return True + logger.warning(f'服务认证失败: result={result}') + return False + + def _DoDeregister(self): + self._Call('msync:deregister', {'ServiceName': self.ServiceName}) + self._registered = False + + def _DoHeartbeat(self): + result = self._Call('msync:heartbeat', {}) + if result and result.get('code') == 200: + logger.info(f'初始心跳成功: {self.ServiceName}') + else: + logger.warning(f'初始心跳失败: {result}') + + # ---------- 状态 ---------- + + @property + def AccessToken(self): + return self._token_data.get('AccessToken') if self._token_data else None + + @property + def RefreshToken(self): + return self._token_data.get('RefreshToken') if self._token_data else None + + @property + def IsAuthenticated(self): + if not self._token_data: + return False + expires = self._token_data.get('ExpiresAt', 0) + if expires and time.time() > expires - 60: + return False + return True + + @property + def IsRegistered(self): + return self._registered + + @property + def IsConnected(self): + return self._sio.connected + + def EnsureAuthenticated(self): + """确保认证有效，过期则续期，续期失败则断开重连""" + if not self.IsAuthenticated: + if self._token_data and self._token_data.get('RefreshToken'): + result = self._RefreshToken() + if result: + return result + logger.warning('[SIO] 令牌刷新失败，尝试完整认证') + auth_ok = self._DoAuth() + if auth_ok: + return self._token_data + # 认证彻底失败，断开连接触发 Socket.IO 自动重连 + logger.error('[SIO] 完整认证也失败，断开连接触发自动重连') + self._ForceReconnect() + return None + return self._token_data + + def _RefreshToken(self): + result = self._Call('msync:auth:refresh', { + 'RefreshToken': self._token_data.get('RefreshToken', ''), + }) + if result and result.get('code') == 200 and result.get('data'): + token_data = result['data'] + self._token_data = { + 'AccessToken': token_data.get('AccessToken'), + 'RefreshToken': token_data.get('RefreshToken'), + 'Scope': token_data.get('Scope', []), + 'ExpiresAt': token_data.get('ExpiresAt'), + } + logger.info(f'令牌刷新成功: {self.ServiceName}') + return self._token_data + logger.info('令牌刷新失败，回退到完整认证') + return None + + def _ForceReconnect(self): + """强制断开连接，让 Socket.IO 自动重连""" + try: + self._registered = False + self._token_data = None + self._sio.disconnect() + except Exception as e: + logger.warning(f'[SIO] 强制断开失败: {e}') + + def _Headers(self): + token = self.AccessToken + if not token: + self.EnsureAuthenticated() + token = self.AccessToken + return {'Authorization': f'Bearer {token}'} if token else {} + + # ---------- 服务发现 ---------- + + def DiscoverAll(self, active_only=True): + result = self._Call('msync:discover', {'ActiveOnly': active_only}) + if result and result.get('code') == 200: + return result.get('data', []) + return [] + + def Discover(self, service_name): + result = self._Call('msync:discover:one', {'ServiceName': service_name}) + if result and result.get('code') == 200: + return result.get('data') + return None + + # ---------- OGM ---------- + + def OGMRegister(self, object_type, local_id, service_name, + global_id=None, metadata=None): + body = { + 'ObjectType': object_type, + 'ObjectLocalID': str(local_id), + 'ServiceName': service_name, + } + if global_id: + body['ObjectGlobalID'] = global_id + if metadata: + body['Metadata'] = metadata + result = self._Call('msync:ogm:register', body) + if result and result.get('code') == 200: + return result.get('data') + return None + + def OGMResolve(self, global_id): + result = self._Call('msync:ogm:resolve', {'ObjectGlobalID': global_id}) + if result and result.get('code') == 200: + return result.get('data') + return None + + def OGMQuery(self, object_type=None, service_name=None, local_id=None): + body = {} + if object_type: + body['ObjectType'] = object_type + if service_name: + body['ServiceName'] = service_name + if local_id: + body['ObjectLocalID'] = local_id + result = self._Call('msync:ogm:query', body) + if result and result.get('code') == 200: + return result.get('data') + return None + + def OGMDeregister(self, global_id): + result = self._Call('msync:ogm:deregister', {'ObjectGlobalID': global_id}) + return result is not None and result.get('code') == 200 + + # ---------- 事务 ---------- + + def TxCreate(self, name, steps, **kwargs): + body = {'TransactionName': name, 'Steps': steps} + for key in ('TransactionDescription', 'TransactionType', 'InitiatorServiceName', + 'ReferenceOGMUUID', 'Payload', 'TimeoutSeconds'): + kwarg_key = key[0].lower() + key[1:] + if kwarg_key in kwargs: + body[key] = kwargs.pop(kwarg_key) + body.update(kwargs) + result = self._Call('msync:tx:create', body, timeout=30) + if result and result.get('code') == 200: + return result.get('data') + return None + + def TxExecute(self, transaction_code, access_token=None): + result = self._Call('msync:tx:execute', { + 'TransactionCode': transaction_code, + 'AccessToken': access_token, + }, timeout=60) + if result and result.get('code') == 200: + return result.get('data') + return None + + def TxDetail(self, transaction_code): + result = self._Call('msync:tx:detail', {'TransactionCode': transaction_code}) + if result and result.get('code') == 200: + return result.get('data') + return None + + # ---------- 远程调用 ---------- + + def CallRemote(self, service_name, method, path, data=None, json=None, **kwargs): + import requests + service_info = self.Discover(service_name) + if not service_info: + logger.error(f'目标服务未发现: {service_name}') + return None + domain = service_info.get('ServiceDomain', '') + url = f'{domain.rstrip("/")}{path}' + kwargs.setdefault('headers', {}).update(self._Headers()) + kwargs.setdefault('timeout', 30) + method_func = getattr(requests, method.lower(), requests.get) + return method_func(url, data=data, json=json, **kwargs) + + # ---------- 应用清单上报（远端 App 注入） ---------- + + def ReportAppManifests(self, manifests, service_domain=None): + """上报应用清单到主服务器（远端 App 注入 GERP 菜单机制） + + Args: + manifests: 清单 dict 列表，结构与主服务器 AppManifest 一致： + {AppCode, AppName, AppIcon, AppDesc, AppType, FrontendRoute, + FrontendType, LegacySrc, Menus: [{MenuName, MenuTitle, MenuLink, + MenuIcon, MenuSort, ParentMenuName}], ConfigKeys: []} + service_domain: 本服务对外域名（缺省用注册时 ServiceDomain， + 页面 iframe 加载 base） + + Returns: + (ok, result)：ok=True 时 result={Registered:[], Failed:[]}； + ok=False 时 result 为原始响应 dict + """ + body = { + 'ServiceName': self.ServiceName, + 'ServiceDomain': service_domain or self.ServiceDomain, + 'Manifests': manifests, + } + result = self._Call('msync:app:manifest:report', body, timeout=30) + if result and result.get('code') == 200: + return True, result.get('data', {}) + return False, result \ No newline at end of file diff --git a/ATestCServer/requirements.txt b/ATestCServer/requirements.txt new file mode 100644 index 00000000..2d631089 --- /dev/null +++ b/ATestCServer/requirements.txt @@ -0,0 +1,23 @@ +# ATestCServer 子服务器依赖 +# 安装顺序：先装 SDK，再装项目依赖 +# 1) pip install -e ../gvsdsdk +# 2) pip install -r requirements.txt + +# === GVS-DSDK（双端共享，必须装，**版本与主服务器严格一致**） === +gvsdsdk==1.0.0 + +# === Django 核心 === +django>=4.2,<5.0 +djangorestframework>=3.14,<4.0 + +# === 数据库（连接 SAAS 主库 MySQL） === +mysqlclient>=2.2 + +# === MSYNC Socket.IO 客户端（子服务器需要） === +python-socketio[client]>=5.0,<6.0 +websocket-client>=1.5 +requests>=2.28 + +# === 加密（与主服务器同版本） === +pyOpenSSL>=23.0 +pycryptodome>=3.18 \ No newline at end of file diff --git a/ATestCServer/start.ps1 b/ATestCServer/start.ps1 new file mode 100644 index 00000000..2807c470 --- /dev/null +++ b/ATestCServer/start.ps1 @@ -0,0 +1,35 @@ +# ATestCServer 启动脚本 +# HTTP 端口 25080，HTTPS 端口 25443 +# +# 用法: +# .\start.ps1 # 普通启动 +# .\start.ps1 -Migrate # 带数据库迁移 + +param( + [switch]$Migrate = $false +) + +$ErrorActionPreference = "Stop" +$ScriptDir = Split-Path -Parent $MyInvocation.MyCommand.Path +Set-Location $ScriptDir + +Write-Host "============================" -ForegroundColor Cyan +Write-Host " ATestCServer 子服务器启动" -ForegroundColor Cyan +Write-Host " HTTP : http://127.0.0.1:25080" -ForegroundColor Green +Write-Host " HTTPS : https://127.0.0.1:25443" -ForegroundColor Green +Write-Host " MSYNC → https://www.gvsds.com" -ForegroundColor Yellow +Write-Host " (父) → http://127.0.0.1:25080 (本地回环)" -ForegroundColor Yellow +Write-Host "============================" -ForegroundColor Cyan + +$env:PYTHONPATH = "$ScriptDir;$ScriptDir\..\gvsdsdk" + +if ($Migrate) { + Write-Host "[*] 执行数据库迁移..." -ForegroundColor Magenta + python manage.py makemigrations sub_server + python manage.py migrate + Write-Host "[√] 迁移完成" -ForegroundColor Green +} + +Write-Host "[*] 启动 Django 开发服务器..." -ForegroundColor Magenta + +python manage.py runserver 0.0.0.0:25080 \ No newline at end of file