|
|
@@ -0,0 +1,572 @@
|
|
|
+import argparse
|
|
|
+import asyncio
|
|
|
+import csv
|
|
|
+import hashlib
|
|
|
+import json
|
|
|
+import math
|
|
|
+import os
|
|
|
+import re
|
|
|
+import secrets
|
|
|
+import subprocess
|
|
|
+import time
|
|
|
+from datetime import UTC, datetime, timedelta
|
|
|
+from pathlib import Path
|
|
|
+from uuid import UUID, uuid4
|
|
|
+
|
|
|
+import httpx
|
|
|
+from dotenv import dotenv_values, set_key
|
|
|
+from filelock import FileLock
|
|
|
+from sqlalchemy import func, select
|
|
|
+from sqlalchemy.ext.asyncio import async_sessionmaker
|
|
|
+
|
|
|
+from zhifu.core.config import Settings
|
|
|
+from zhifu.core.database import create_engine
|
|
|
+from zhifu.modules.after_sales.models import BusinessEvent, ExchangeRule, Ticket
|
|
|
+from zhifu.modules.identity.access_models import Principal, StoreGrant
|
|
|
+from zhifu.modules.identity.auth import password_hasher
|
|
|
+from zhifu.modules.identity.models import Store, Tenant
|
|
|
+from zhifu.modules.integrations.models import Connector, ExternalOperation
|
|
|
+from zhifu.modules.storefront.models import CatalogProduct
|
|
|
+from zhifu.oms.database import oms_engine
|
|
|
+from zhifu.oms.models import Credential
|
|
|
+
|
|
|
+ROOT = Path(__file__).resolve().parents[1]
|
|
|
+DATA = ROOT / ".work" / "course-demo"
|
|
|
+TENANT = UUID("ac700000-2026-4000-8000-000000000001")
|
|
|
+STORE = UUID("ac700000-2026-4000-8000-000000000002")
|
|
|
+TOKEN_ENV = "ZHIFU_COURSE_OMS_TOKEN"
|
|
|
+CONNECTOR_URL = "http://oms:8100"
|
|
|
+CUSTOMER_PERMISSIONS = (
|
|
|
+ "orders:read", "tickets:read", "after_sales:create", "after_sales:confirm",
|
|
|
+ "after_sales:cancel",
|
|
|
+)
|
|
|
+
|
|
|
+
|
|
|
+def require(condition, message):
|
|
|
+ if not condition:
|
|
|
+ raise ValueError(message)
|
|
|
+
|
|
|
+
|
|
|
+def utc_now():
|
|
|
+ return datetime.now(UTC).isoformat()
|
|
|
+
|
|
|
+
|
|
|
+def read_json(path):
|
|
|
+ return json.loads(path.read_text(encoding="utf-8"))
|
|
|
+
|
|
|
+
|
|
|
+def protect_file(path):
|
|
|
+ # Windows 使用当前用户 SID 设置访问权限,凭据文件仅当前用户和 SYSTEM 可读取。
|
|
|
+ if os.name == "nt":
|
|
|
+ result = subprocess.run(
|
|
|
+ ["whoami", "/user", "/fo", "csv", "/nh"],
|
|
|
+ check=True, capture_output=True, text=True, timeout=10,
|
|
|
+ )
|
|
|
+ sid = next(csv.reader(result.stdout.strip().splitlines()))[1]
|
|
|
+ subprocess.run(
|
|
|
+ ["icacls", str(path), "/inheritance:r", "/grant:r",
|
|
|
+ f"*{sid}:(F)", "*S-1-5-18:(F)"],
|
|
|
+ check=True, capture_output=True, timeout=10,
|
|
|
+ )
|
|
|
+ else:
|
|
|
+ path.chmod(0o600)
|
|
|
+
|
|
|
+
|
|
|
+def write_json(path, payload, *, private=False):
|
|
|
+ path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
+ temporary = path.with_name(f".{path.name}.{uuid4().hex}.new")
|
|
|
+ # 同目录临时文件与原子替换保证中断后仍能读取完整 JSON;临时文件也使用受限权限。
|
|
|
+ descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
|
+ with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
|
|
|
+ if private:
|
|
|
+ protect_file(temporary)
|
|
|
+ json.dump(payload, stream, ensure_ascii=False, indent=2)
|
|
|
+ stream.write("\n")
|
|
|
+ stream.flush()
|
|
|
+ os.fsync(stream.fileno())
|
|
|
+ temporary.replace(path)
|
|
|
+
|
|
|
+
|
|
|
+def session_name(value):
|
|
|
+ require(
|
|
|
+ re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_-]{0,47}", value) is not None,
|
|
|
+ "课程名称须为 1 至 48 个英文字母、数字、下划线或连字符,并以字母或数字开头",
|
|
|
+ )
|
|
|
+ require(
|
|
|
+ value.upper() not in {"CON", "PRN", "AUX", "NUL"}
|
|
|
+ and re.fullmatch(r"(?:COM|LPT)[1-9]", value.upper()) is None,
|
|
|
+ "课程名称不能使用 Windows 保留文件名",
|
|
|
+ )
|
|
|
+ return value
|
|
|
+
|
|
|
+
|
|
|
+def validate_scope(value):
|
|
|
+ require(
|
|
|
+ value["tenant_id"] == str(TENANT) and value["store_id"] == str(STORE),
|
|
|
+ "课程数据仅允许使用固定开发商家和店铺",
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def validate_url(value):
|
|
|
+ url = httpx.URL(value)
|
|
|
+ require(
|
|
|
+ url.scheme in {"http", "https"} and url.host in {"127.0.0.1", "localhost", "::1"}
|
|
|
+ and not url.username and not url.password and not url.query and not url.fragment
|
|
|
+ and url.path in {"", "/"},
|
|
|
+ "课程接口地址必须为本机 HTTP 或 HTTPS 服务根地址",
|
|
|
+ )
|
|
|
+ return str(url).rstrip("/")
|
|
|
+
|
|
|
+
|
|
|
+def load_manifest(name=None):
|
|
|
+ path = DATA / "sessions" / f"{session_name(name)}.json" if name else DATA / "active.json"
|
|
|
+ manifest = read_json(path)
|
|
|
+ validate_scope(manifest)
|
|
|
+ session_name(manifest["session"])
|
|
|
+ require(manifest["schema_version"] == 1, "课程记录版本不受支持")
|
|
|
+ validate_url(manifest["api_url"])
|
|
|
+ validate_url(manifest["oms_url"])
|
|
|
+ require(manifest["order"]["item_id"] == "keyboard-01", "课程订单商品不匹配")
|
|
|
+ return manifest
|
|
|
+
|
|
|
+
|
|
|
+def save_manifest(manifest, *, activate=False):
|
|
|
+ write_json(DATA / "sessions" / f"{manifest['session']}.json", manifest)
|
|
|
+ active = DATA / "active.json"
|
|
|
+ if activate or (active.exists() and read_json(active)["session"] == manifest["session"]):
|
|
|
+ write_json(active, manifest)
|
|
|
+
|
|
|
+
|
|
|
+def credentials(accounts=None):
|
|
|
+ path = DATA / "credentials.json"
|
|
|
+ if path.exists():
|
|
|
+ value = read_json(path)
|
|
|
+ validate_scope(value)
|
|
|
+ require(value["schema_version"] == 1, "课程凭据版本不受支持")
|
|
|
+ protect_file(path)
|
|
|
+ return value
|
|
|
+ require(accounts is not None, "课程凭据不存在,请执行 prepare")
|
|
|
+ value = {
|
|
|
+ "schema_version": 1, "tenant_id": str(TENANT), "store_id": str(STORE),
|
|
|
+ "customer_password": accounts["password"],
|
|
|
+ "service": {
|
|
|
+ "principal_id": str(uuid4()), "username": "course-order-sync",
|
|
|
+ "password": secrets.token_urlsafe(32),
|
|
|
+ },
|
|
|
+ "oms_platform_token": secrets.token_urlsafe(32),
|
|
|
+ "oms_fulfillment_token": secrets.token_urlsafe(32),
|
|
|
+ }
|
|
|
+ # 先保存凭据,再创建数据库记录;后续显式重试继续使用相同身份与令牌。
|
|
|
+ write_json(path, value, private=True)
|
|
|
+ return value
|
|
|
+
|
|
|
+
|
|
|
+def prepare_files(args):
|
|
|
+ accounts = read_json(ROOT / ".work" / "frontend-accounts.json")
|
|
|
+ validate_scope(accounts)
|
|
|
+ name = session_name(args.session or datetime.now().strftime("%Y%m%d-%H%M%S-%f"))
|
|
|
+ path = DATA / "sessions" / f"{name}.json"
|
|
|
+ secret = credentials(accounts)
|
|
|
+ if path.exists():
|
|
|
+ manifest = load_manifest(name)
|
|
|
+ require(
|
|
|
+ manifest["api_url"] == args.api_url and manifest["oms_url"] == args.oms_url,
|
|
|
+ "已有课程接口地址与本次参数不同,请使用原有地址",
|
|
|
+ )
|
|
|
+ else:
|
|
|
+ manifest = {
|
|
|
+ "schema_version": 1, "session": name, "tenant_id": str(TENANT),
|
|
|
+ "store_id": str(STORE), "created_at": utc_now(), "status": "PREPARING",
|
|
|
+ "api_url": args.api_url, "oms_url": args.oms_url,
|
|
|
+ "customer": {"principal_id": str(uuid4()), "username": f"course-{name}"},
|
|
|
+ "order": {
|
|
|
+ "order_id": str(uuid4()), "item_id": "keyboard-01", "product_type": "keyboard",
|
|
|
+ "available_quantity": 1, "source_version": 1,
|
|
|
+ "signed_at": (datetime.now(UTC) - timedelta(days=2)).isoformat(),
|
|
|
+ },
|
|
|
+ "staff": {name: identifier for name, identifier in accounts["actors"].items()
|
|
|
+ if name in {"staff01", "supervisor01", "admin01"}},
|
|
|
+ }
|
|
|
+ save_manifest(manifest)
|
|
|
+ return manifest, secret
|
|
|
+
|
|
|
+
|
|
|
+async def ensure_principal(session, actor, password, kind, permissions):
|
|
|
+ identifier = UUID(actor["principal_id"]).bytes
|
|
|
+ existing = await session.get(Principal, identifier, with_for_update=True)
|
|
|
+ if existing is None:
|
|
|
+ collision = await session.scalar(select(Principal).where(
|
|
|
+ Principal.tenant_id == TENANT.bytes, Principal.username == actor["username"],
|
|
|
+ ))
|
|
|
+ require(collision is None, "课程账户名已经被其他身份使用")
|
|
|
+ session.add(Principal(
|
|
|
+ id=identifier, tenant_id=TENANT.bytes, username=actor["username"],
|
|
|
+ password_hash=await asyncio.to_thread(password_hasher.hash, password), kind=kind,
|
|
|
+ ))
|
|
|
+ await session.flush()
|
|
|
+ for permission in permissions:
|
|
|
+ session.add(StoreGrant(
|
|
|
+ tenant_id=TENANT.bytes, store_id=STORE.bytes,
|
|
|
+ principal_id=identifier, permission=permission,
|
|
|
+ ))
|
|
|
+ else:
|
|
|
+ require(
|
|
|
+ existing.tenant_id == TENANT.bytes and existing.username == actor["username"]
|
|
|
+ and existing.kind == kind and existing.status == "ACTIVE",
|
|
|
+ "已有课程账户身份或状态与课程配置不一致",
|
|
|
+ )
|
|
|
+ require(
|
|
|
+ await asyncio.to_thread(password_hasher.verify, password, existing.password_hash),
|
|
|
+ "已有课程账户密码与本地受保护凭据不一致",
|
|
|
+ )
|
|
|
+ granted = set((await session.scalars(select(StoreGrant.permission).where(
|
|
|
+ StoreGrant.tenant_id == TENANT.bytes, StoreGrant.store_id == STORE.bytes,
|
|
|
+ StoreGrant.principal_id == identifier,
|
|
|
+ ))).all())
|
|
|
+ require(granted == set(permissions), "已有课程账户授权与课程要求不一致")
|
|
|
+
|
|
|
+
|
|
|
+async def bootstrap(settings, manifest, secret):
|
|
|
+ engine = create_engine(settings)
|
|
|
+ oms = oms_engine()
|
|
|
+ try:
|
|
|
+ sessions = async_sessionmaker(engine, expire_on_commit=False)
|
|
|
+ async with sessions.begin() as session:
|
|
|
+ tenant = await session.get(Tenant, TENANT.bytes)
|
|
|
+ store = await session.get(Store, STORE.bytes, with_for_update=True)
|
|
|
+ require(tenant is not None and tenant.status == "ACTIVE", "开发商家未启用")
|
|
|
+ require(
|
|
|
+ store is not None and store.tenant_id == TENANT.bytes and store.status == "ACTIVE",
|
|
|
+ "开发店铺未启用或不属于指定商家",
|
|
|
+ )
|
|
|
+ product = await session.get(CatalogProduct, (TENANT.bytes, STORE.bytes, "keyboard-01"))
|
|
|
+ require(product is not None and product.status == "ACTIVE", "键盘商品未发布")
|
|
|
+ rule = await session.scalar(select(ExchangeRule).where(
|
|
|
+ ExchangeRule.tenant_id == TENANT.bytes, ExchangeRule.store_id == STORE.bytes,
|
|
|
+ ).order_by(ExchangeRule.version.desc()).limit(1))
|
|
|
+ require(
|
|
|
+ rule is not None and not rule.revoked and rule.window_days > 2
|
|
|
+ and "keyboard" in rule.product_types,
|
|
|
+ "当前规则不支持签收两天的键盘订单,请检查已发布规则",
|
|
|
+ )
|
|
|
+ manifest["rule"] = {"version": rule.version, "window_days": rule.window_days,
|
|
|
+ "citation": rule.citation}
|
|
|
+ manifest["product_name"] = product.name
|
|
|
+ manifest["existing_ticket_count"] = await session.scalar(
|
|
|
+ select(func.count()).select_from(Ticket).where(
|
|
|
+ Ticket.tenant_id == TENANT.bytes,
|
|
|
+ Ticket.order_id == UUID(manifest["order"]["order_id"]).bytes,
|
|
|
+ )
|
|
|
+ )
|
|
|
+ await ensure_principal(
|
|
|
+ session, manifest["customer"], secret["customer_password"],
|
|
|
+ "CUSTOMER", CUSTOMER_PERMISSIONS,
|
|
|
+ )
|
|
|
+ await ensure_principal(
|
|
|
+ session, secret["service"], secret["service"]["password"],
|
|
|
+ "SERVICE", ("orders:sync",),
|
|
|
+ )
|
|
|
+ connector = await session.get(Connector, (TENANT.bytes, STORE.bytes))
|
|
|
+ if connector is None:
|
|
|
+ session.add(Connector(
|
|
|
+ tenant_id=TENANT.bytes, store_id=STORE.bytes, source_system="oms",
|
|
|
+ base_url=CONNECTOR_URL, credential_env=TOKEN_ENV, enabled=True,
|
|
|
+ ))
|
|
|
+ session.add(BusinessEvent(
|
|
|
+ id=uuid4().bytes, tenant_id=TENANT.bytes,
|
|
|
+ actor_id=UUID(secret["service"]["principal_id"]).bytes,
|
|
|
+ aggregate_id=STORE.bytes, aggregate_version=1,
|
|
|
+ event_type="connector.configured", request_id=str(uuid4()),
|
|
|
+ payload={"credential_env": TOKEN_ENV, "source": "prepare_course_demo"},
|
|
|
+ occurred_at=datetime.now(UTC).replace(tzinfo=None),
|
|
|
+ ))
|
|
|
+ else:
|
|
|
+ require(
|
|
|
+ connector.source_system == "oms" and connector.base_url == CONNECTOR_URL
|
|
|
+ and connector.credential_env == TOKEN_ENV and connector.enabled,
|
|
|
+ "已有 OMS 连接器配置与课程环境不一致",
|
|
|
+ )
|
|
|
+ oms_sessions = async_sessionmaker(oms, expire_on_commit=False)
|
|
|
+ async with oms_sessions.begin() as session:
|
|
|
+ for name, can_fulfill in (
|
|
|
+ ("oms_platform_token", False), ("oms_fulfillment_token", True),
|
|
|
+ ):
|
|
|
+ token_hash = hashlib.sha256(secret[name].encode()).digest()
|
|
|
+ existing = await session.get(Credential, token_hash, with_for_update=True)
|
|
|
+ if existing is None:
|
|
|
+ session.add(Credential(
|
|
|
+ token_hash=token_hash, tenant_id=TENANT.bytes, store_id=STORE.bytes,
|
|
|
+ can_fulfill=can_fulfill, enabled=True,
|
|
|
+ ))
|
|
|
+ else:
|
|
|
+ require(
|
|
|
+ existing.tenant_id == TENANT.bytes and existing.store_id == STORE.bytes
|
|
|
+ and existing.can_fulfill == can_fulfill and existing.enabled,
|
|
|
+ "已有 OMS 凭据权限与课程配置不一致",
|
|
|
+ )
|
|
|
+ finally:
|
|
|
+ await engine.dispose()
|
|
|
+ await oms.dispose()
|
|
|
+
|
|
|
+
|
|
|
+def configure_runtime(secret):
|
|
|
+ path = ROOT / ".work" / "runtime.env"
|
|
|
+ require(path.is_file(), "完整项目的 .work/runtime.env 不存在")
|
|
|
+ existing = dotenv_values(path).get(TOKEN_ENV)
|
|
|
+ require(
|
|
|
+ existing is None or existing == secret["oms_platform_token"],
|
|
|
+ "运行配置已存在不同的课程 OMS 凭据",
|
|
|
+ )
|
|
|
+ if existing is None:
|
|
|
+ set_key(path, TOKEN_ENV, secret["oms_platform_token"], quote_mode="always")
|
|
|
+
|
|
|
+
|
|
|
+async def request(client, method, url, *, token=None, body=None, key=None, allow_missing=False):
|
|
|
+ headers = {}
|
|
|
+ if token:
|
|
|
+ headers["Authorization"] = f"Bearer {token}"
|
|
|
+ if key:
|
|
|
+ headers["Idempotency-Key"] = key
|
|
|
+ response = await client.request(method, url, headers=headers, json=body)
|
|
|
+ if allow_missing and response.status_code == 404:
|
|
|
+ return None
|
|
|
+ response.raise_for_status()
|
|
|
+ return response.json()
|
|
|
+
|
|
|
+
|
|
|
+async def login(client, manifest, actor, password):
|
|
|
+ result = await request(client, "POST", manifest["api_url"] + "/openapi/v1/auth/sessions", body={
|
|
|
+ "tenant_id": str(TENANT), "username": actor["username"], "password": password,
|
|
|
+ })
|
|
|
+ return result["access_token"], time.monotonic() + result["expires_in"] - 60
|
|
|
+
|
|
|
+
|
|
|
+def verify_snapshot(manifest, snapshot):
|
|
|
+ expected = manifest["order"]
|
|
|
+ require(
|
|
|
+ snapshot["tenant_id"] == str(TENANT) and snapshot["store_id"] == str(STORE)
|
|
|
+ and snapshot["subject_id"] == manifest["customer"]["principal_id"]
|
|
|
+ and snapshot["order_id"] == expected["order_id"]
|
|
|
+ and snapshot["item_id"] == "keyboard-01" and snapshot["product_type"] == "keyboard"
|
|
|
+ and datetime.fromisoformat(snapshot["signed_at"]) == datetime.fromisoformat(
|
|
|
+ expected["signed_at"]),
|
|
|
+ "OMS 订单身份、商品或签收时间与课程记录不一致",
|
|
|
+ )
|
|
|
+ require(
|
|
|
+ snapshot["source_version"] >= 1 and 0 <= snapshot["available_quantity"] <= 1,
|
|
|
+ "OMS 订单版本或数量不符合课程订单约束",
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+async def synchronize(client, manifest, secret, token):
|
|
|
+ snapshot = await request(
|
|
|
+ client, "GET", manifest["oms_url"] + "/orders/" + manifest["order"]["order_id"],
|
|
|
+ token=secret["oms_platform_token"],
|
|
|
+ )
|
|
|
+ verify_snapshot(manifest, snapshot)
|
|
|
+ body = {name: snapshot[name] for name in (
|
|
|
+ "store_id", "subject_id", "source_version", "item_id", "product_type",
|
|
|
+ "available_quantity", "signed_at",
|
|
|
+ )}
|
|
|
+ body["source_system"] = "oms"
|
|
|
+ # 每次可信观察使用新的幂等键,确保相同来源版本也更新 observed_at。
|
|
|
+ await request(
|
|
|
+ client, "PUT", manifest["api_url"] + "/internal/v1/orders/" + snapshot["order_id"],
|
|
|
+ token=token, body=body, key=f"course-sync-{uuid4().hex}",
|
|
|
+ )
|
|
|
+ manifest["last_sync_at"] = utc_now()
|
|
|
+ manifest["last_order_snapshot"] = snapshot
|
|
|
+ return snapshot
|
|
|
+
|
|
|
+
|
|
|
+async def prepare(settings, args):
|
|
|
+ async with asyncio.timeout(90):
|
|
|
+ manifest, secret = await asyncio.to_thread(prepare_files, args)
|
|
|
+ await bootstrap(settings, manifest, secret)
|
|
|
+ await asyncio.to_thread(configure_runtime, secret)
|
|
|
+ async with httpx.AsyncClient(timeout=10, trust_env=False) as client:
|
|
|
+ order_url = manifest["oms_url"] + "/orders/" + manifest["order"]["order_id"]
|
|
|
+ existing = await request(client, "GET", order_url,
|
|
|
+ token=secret["oms_platform_token"], allow_missing=True)
|
|
|
+ if existing is None:
|
|
|
+ body = {name: value for name, value in manifest["order"].items()
|
|
|
+ if name != "order_id"}
|
|
|
+ body["subject_id"] = manifest["customer"]["principal_id"]
|
|
|
+ await request(client, "PUT", order_url,
|
|
|
+ token=secret["oms_fulfillment_token"], body=body)
|
|
|
+ else:
|
|
|
+ verify_snapshot(manifest, existing)
|
|
|
+ service_token, _ = await login(
|
|
|
+ client, manifest, secret["service"], secret["service"]["password"],
|
|
|
+ )
|
|
|
+ snapshot = await synchronize(client, manifest, secret, service_token)
|
|
|
+ customer_token, _ = await login(
|
|
|
+ client, manifest, manifest["customer"], secret["customer_password"],
|
|
|
+ )
|
|
|
+ order = await request(
|
|
|
+ client, "GET", manifest["api_url"] + "/openapi/v1/orders/" + snapshot["order_id"],
|
|
|
+ token=customer_token,
|
|
|
+ )
|
|
|
+ require(
|
|
|
+ order["subject_id"] == manifest["customer"]["principal_id"]
|
|
|
+ and order["source_system"] == "oms"
|
|
|
+ and order["source_version"] == snapshot["source_version"]
|
|
|
+ and order["available_quantity"] == snapshot["available_quantity"],
|
|
|
+ "客户读取的订单与 OMS 当前订单不一致",
|
|
|
+ )
|
|
|
+ conversation = await request(
|
|
|
+ client, "POST", manifest["api_url"] + "/openapi/v1/conversations",
|
|
|
+ token=customer_token, body={"store_id": str(STORE)},
|
|
|
+ key=f"course-conversation-{manifest['customer']['principal_id']}",
|
|
|
+ )
|
|
|
+ manifest["conversation_id"] = conversation["conversation_id"]
|
|
|
+ unused = (
|
|
|
+ conversation["sequence"] == 0 and manifest["existing_ticket_count"] == 0
|
|
|
+ and snapshot["source_version"] == 1 and snapshot["available_quantity"] == 1
|
|
|
+ )
|
|
|
+ if manifest["status"] != "OMS_COMPLETED":
|
|
|
+ manifest["status"] = "READY" if unused else "IN_USE"
|
|
|
+ manifest["verification"] = {
|
|
|
+ "verified_at": utc_now(), "customer_order_read": True,
|
|
|
+ "conversation_sequence": conversation["sequence"],
|
|
|
+ "generation_model_calls": 0,
|
|
|
+ }
|
|
|
+ await asyncio.to_thread(save_manifest, manifest, activate=True)
|
|
|
+ return manifest
|
|
|
+
|
|
|
+
|
|
|
+async def sync_loop(args):
|
|
|
+ manifest = await asyncio.to_thread(load_manifest, args.session)
|
|
|
+ secret = await asyncio.to_thread(credentials)
|
|
|
+ duration = args.duration_seconds
|
|
|
+ deadline = time.monotonic() + duration
|
|
|
+ maximum = max(1, math.ceil(duration / args.interval_seconds))
|
|
|
+ token, expires = None, 0
|
|
|
+ count = 0
|
|
|
+ async with (
|
|
|
+ asyncio.timeout(duration + 35),
|
|
|
+ httpx.AsyncClient(timeout=10, trust_env=False) as client,
|
|
|
+ ):
|
|
|
+ for index in range(maximum):
|
|
|
+ if index and time.monotonic() >= deadline:
|
|
|
+ break
|
|
|
+ # 锁等待计入本次同步的 30 秒期限,短时间重叠的课程操作允许等待完成。
|
|
|
+ lock = FileLock(DATA / "course.lock", timeout=10, thread_local=False)
|
|
|
+ async with asyncio.timeout(30):
|
|
|
+ await asyncio.to_thread(lock.acquire)
|
|
|
+ try:
|
|
|
+ manifest = await asyncio.to_thread(load_manifest, manifest["session"])
|
|
|
+ if token is None or time.monotonic() >= expires:
|
|
|
+ token, expires = await login(
|
|
|
+ client, manifest, secret["service"], secret["service"]["password"],
|
|
|
+ )
|
|
|
+ await synchronize(client, manifest, secret, token)
|
|
|
+ await asyncio.to_thread(save_manifest, manifest)
|
|
|
+ finally:
|
|
|
+ await asyncio.to_thread(lock.release)
|
|
|
+ count += 1
|
|
|
+ print(json.dumps({
|
|
|
+ "session": manifest["session"], "sync_count": count,
|
|
|
+ "last_sync_at": manifest["last_sync_at"],
|
|
|
+ "source_version": manifest["last_order_snapshot"]["source_version"],
|
|
|
+ "available_quantity": manifest["last_order_snapshot"]["available_quantity"],
|
|
|
+ }, ensure_ascii=False), flush=True)
|
|
|
+ if index + 1 < maximum:
|
|
|
+ remaining = deadline - time.monotonic()
|
|
|
+ if remaining <= 0:
|
|
|
+ break
|
|
|
+ await asyncio.sleep(min(args.interval_seconds, remaining))
|
|
|
+ return {"session": manifest["session"], "sync_count": count, "generation_model_calls": 0}
|
|
|
+
|
|
|
+
|
|
|
+async def complete(settings, args):
|
|
|
+ async with asyncio.timeout(90):
|
|
|
+ manifest = await asyncio.to_thread(load_manifest, args.session)
|
|
|
+ secret = await asyncio.to_thread(credentials)
|
|
|
+ engine = create_engine(settings)
|
|
|
+ try:
|
|
|
+ sessions = async_sessionmaker(engine, expire_on_commit=False)
|
|
|
+ async with sessions() as session:
|
|
|
+ query = select(ExternalOperation).where(
|
|
|
+ ExternalOperation.tenant_id == TENANT.bytes,
|
|
|
+ ExternalOperation.store_id == STORE.bytes,
|
|
|
+ ExternalOperation.order_id == UUID(manifest["order"]["order_id"]).bytes,
|
|
|
+ ExternalOperation.status.in_(("ACCEPTED", "COMPLETED")),
|
|
|
+ )
|
|
|
+ if args.operation_id:
|
|
|
+ query = query.where(ExternalOperation.id == args.operation_id.bytes)
|
|
|
+ operations = (await session.scalars(query)).all()
|
|
|
+ require(
|
|
|
+ len(operations) == 1,
|
|
|
+ "需要唯一的本课程已受理履约操作,请检查审批与履约进度",
|
|
|
+ )
|
|
|
+ operation_id = str(UUID(bytes=operations[0].id))
|
|
|
+ ticket_id = str(UUID(bytes=operations[0].ticket_id))
|
|
|
+ finally:
|
|
|
+ await engine.dispose()
|
|
|
+ async with httpx.AsyncClient(timeout=10, trust_env=False) as client:
|
|
|
+ url = manifest["oms_url"] + "/exchanges/" + operation_id
|
|
|
+ existing = await request(client, "GET", url, token=secret["oms_fulfillment_token"])
|
|
|
+ require(
|
|
|
+ existing["order_id"] == manifest["order"]["order_id"]
|
|
|
+ and existing["status"] in {"ACCEPTED", "COMPLETED"},
|
|
|
+ "OMS 操作不属于本课程订单或当前状态不能完成",
|
|
|
+ )
|
|
|
+ result = await request(client, "POST", url + "/complete",
|
|
|
+ token=secret["oms_fulfillment_token"])
|
|
|
+ require(result["status"] == "COMPLETED", "OMS 未返回履约完成状态")
|
|
|
+ manifest["completion"] = {**result, "ticket_id": ticket_id, "completed_at": utc_now()}
|
|
|
+ manifest["status"] = "OMS_COMPLETED"
|
|
|
+ service_token, _ = await login(
|
|
|
+ client, manifest, secret["service"], secret["service"]["password"],
|
|
|
+ )
|
|
|
+ await synchronize(client, manifest, secret, service_token)
|
|
|
+ await asyncio.to_thread(save_manifest, manifest)
|
|
|
+ return {
|
|
|
+ "session": manifest["session"], **manifest["completion"], "generation_model_calls": 0,
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+async def locked_command(settings, args):
|
|
|
+ # 跨线程共享同一个锁状态;准备与完成的 90 秒期限包含最长 10 秒的锁等待。
|
|
|
+ lock = FileLock(DATA / "course.lock", timeout=10, thread_local=False)
|
|
|
+ async with asyncio.timeout(90):
|
|
|
+ await asyncio.to_thread(lock.acquire)
|
|
|
+ try:
|
|
|
+ if args.command == "prepare":
|
|
|
+ return await prepare(settings, args)
|
|
|
+ return await complete(settings, args)
|
|
|
+ finally:
|
|
|
+ await asyncio.to_thread(lock.release)
|
|
|
+
|
|
|
+
|
|
|
+def main():
|
|
|
+ parser = argparse.ArgumentParser(description="准备、同步和完成完整项目的课程演示订单")
|
|
|
+ commands = parser.add_subparsers(dest="command", required=True)
|
|
|
+ prepare_parser = commands.add_parser("prepare", help="创建本次课程客户、OMS 订单与空会话")
|
|
|
+ prepare_parser.add_argument("--session", type=session_name)
|
|
|
+ prepare_parser.add_argument("--api-url", type=validate_url, default="http://127.0.0.1:28000")
|
|
|
+ prepare_parser.add_argument("--oms-url", type=validate_url, default="http://127.0.0.1:28100")
|
|
|
+ sync_parser = commands.add_parser("sync", help="从 OMS 读取并同步当前课程订单")
|
|
|
+ sync_parser.add_argument("--session", type=session_name)
|
|
|
+ sync_parser.add_argument("--duration-seconds", type=int, default=0)
|
|
|
+ sync_parser.add_argument("--interval-seconds", type=int, default=60)
|
|
|
+ complete_parser = commands.add_parser("complete", help="使用仓储身份登记本课程履约完成")
|
|
|
+ complete_parser.add_argument("--session", type=session_name)
|
|
|
+ complete_parser.add_argument("--operation-id", type=UUID)
|
|
|
+ args = parser.parse_args()
|
|
|
+ settings = Settings()
|
|
|
+ require(settings.environment in {"development", "test"}, "课程工具仅允许 development 或 test")
|
|
|
+ require(Path.cwd().resolve() == ROOT, "请在完整项目根目录执行课程工具")
|
|
|
+ DATA.mkdir(parents=True, exist_ok=True)
|
|
|
+ if args.command == "sync":
|
|
|
+ require(0 <= args.duration_seconds <= 43200, "同步时长须为 0 至 43200 秒")
|
|
|
+ require(30 <= args.interval_seconds <= 240, "同步间隔须为 30 至 240 秒")
|
|
|
+ result = asyncio.run(sync_loop(args))
|
|
|
+ else:
|
|
|
+ result = asyncio.run(locked_command(settings, args))
|
|
|
+ print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ main()
|