fix(benteler): add onboarding department retry script

README.md

@@ -38,6 +38,10 @@ mvn clean package -DskipTests
 java -jar mjava-{module}/target/mjava-{module}.jar --spring.profiles.active=prod
 ```
 
+## 运维脚本
+
+- 本特勒入职账户已创建但主部门同步失败时，使用 [`mjava-benteler/scripts/retry-onboarding-departments.js`](./mjava-benteler/scripts/retry-onboarding-departments.js) 做幂等补偿；执行条件和安全边界见同目录 README。
+
 ## 文档与设计规范
 
 > 本仓库保留源码、构建配置和源码级说明；客户事项、目录设计和开发规范统一在文档中心管理：

mjava-benteler/scripts/README.md

@@ -0,0 +1,51 @@
+# 本特勒临时运维脚本
+
+## 入职主部门补偿
+
+`retry-onboarding-departments.js` 用于以下场景：入职账户已在 EIAM 创建，但因宜搭入职单中的 iDaaS 部门错误，主部门未正确同步。
+
+脚本执行流程：
+
+1. 查询宜搭入职流程的最新人员子表和 iDaaS 部门 ID；
+2. 按手机号精确查询 EIAM 账户，并校验同一账户不存在目标部门冲突；
+3. 幂等调用 `setUserPrimaryOrganizationalUnit`；
+4. 仅在部门更新成功后，将宜搭人员明细、成功/失败统计及失败说明回写；
+5. 重新查询流程数据完成验收。
+
+### 前置条件
+
+- 在已登录 OpenYida 的本特勒资料目录执行命令；
+- 本机能免密 SSH 到 `root@120.55.113.155`；
+- 服务器 `/home/server/benteler/application-prod.yml` 保持可用，脚本只在远端进程内读取现有 EIAM 配置，不输出凭据。
+
+### 使用方式
+
+默认只预览，不修改 EIAM 或宜搭：
+
+```bash
+cd '/Users/malk/Desktop/Tech/1. malk/后端/2 one/本特勒/iDaaS 同步'
+node /Users/malk/server/cur/mjava-ai/mjava-benteler/scripts/retry-onboarding-departments.js --scan-failed
+```
+
+确认预览结果后，执行所有当前失败明细：
+
+```bash
+node /Users/malk/server/cur/mjava-ai/mjava-benteler/scripts/retry-onboarding-departments.js --execute --scan-failed
+```
+
+也可以指定一个或多个流程实例，适合精确补偿或重复验收：
+
+```bash
+node /Users/malk/server/cur/mjava-ai/mjava-benteler/scripts/retry-onboarding-departments.js \
+  --process-id <processInstanceId> \
+  --process-id <processInstanceId>
+```
+
+在上述命令中增加 `--execute` 才会真正更新。若账户已属于目标主部门，EIAM 更新会跳过，但宜搭结果仍会重新验收和回写，因此脚本可安全重试。
+
+### 安全边界
+
+- 子表恰好返回 50 行时停止，避免误用被截断的数据覆盖宜搭子表；
+- 同一账户出现多个目标主部门时停止，不执行任何 EIAM 更新；
+- EIAM 更新失败时，对应宜搭明细保持或改为失败，不伪造成功；
+- 输出仅展示遮蔽后的账户标识，不输出手机号、EIAM/Yida 凭据。

mjava-benteler/scripts/retry-onboarding-departments.js

@@ -0,0 +1,370 @@
+const { createHash } = require('crypto');
+const { spawnSync } = require('child_process');
+
+// prd 临时补偿任务：入职账户已创建、但主部门同步失败时，重试设置 EIAM 主部门并回写宜搭。
+
+const APP_TYPE = 'APP_MQP6UV1H5S5BDOU68TQD';
+const FORM_UUID = 'FORM-3794DBD2C83E42C5863853A80A3C0ACB5K6Q';
+const HOST = '120.55.113.155';
+const TABLE_FIELD = 'tableField_mrcw2nqc';
+const PHONE_FIELD = 'numberField_mrcw2nqe';
+const NAME_FIELD = 'textField_mrcw2nqz';
+const ORG_FIELD = 'selectField_mrmyvzrt_id';
+const STATUS_FIELD = 'selectField_mrcxjcq8';
+const TOTAL_FIELD = 'numberField_mrcxjcq9';
+const SUCCESS_FIELD = 'numberField_mrcxjcqe';
+const FAILED_FIELD = 'numberField_mrcxjcqf';
+const FAILURE_MESSAGE_FIELD = 'textareaField_mrn3kh8m';
+
+const REMOTE_PYTHON = String.raw`
+import json
+import sys
+import urllib.parse
+import urllib.request
+from pathlib import Path
+from typing import Any, Dict, List, Set, Tuple
+
+import yaml
+
+CONFIG_PATH = Path('/home/server/benteler/application-prod.yml')
+
+def request_json(request: urllib.request.Request) -> Tuple[Dict[str, Any], int]:
+    with urllib.request.urlopen(request, timeout=30) as response:
+        body = response.read().decode('utf-8')
+        return json.loads(body), response.getcode()
+
+def get_access_token(config: Dict[str, Any]) -> str:
+    eiam = config['eiam']
+    url = '{}/v2/{}/{}/oauth2/token'.format(
+        eiam['baseUrl'], eiam['instanceId'], eiam['applicationId'])
+    body = urllib.parse.urlencode({
+        'grant_type': 'client_credentials',
+        'client_id': eiam['clientId'],
+        'client_secret': eiam['clientSecret'],
+    }).encode('utf-8')
+    payload, _ = request_json(urllib.request.Request(url, data=body, method='POST'))
+    token = payload.get('access_token') or payload.get('accessToken')
+    if not token:
+        raise RuntimeError('EIAM token response did not contain access token')
+    return token
+
+def list_users(config: Dict[str, Any], token: str) -> Tuple[List[Dict[str, Any]], str]:
+    eiam = config['eiam']
+    base = '{}/v2/{}/{}/users'.format(
+        eiam['baseUrl'], eiam['instanceId'], eiam['applicationId'])
+    users = []
+    page_number = 1
+    while True:
+        query = urllib.parse.urlencode({'pageNumber': page_number, 'pageSize': 100})
+        request = urllib.request.Request(
+            '{}?{}'.format(base, query),
+            headers={'Authorization': 'Bearer {}'.format(token)},
+            method='GET')
+        payload, _ = request_json(request)
+        page = payload.get('data') or []
+        users.extend(item for item in page if isinstance(item, dict))
+        total = int(payload.get('totalCount') or len(users))
+        if not page or len(users) >= total:
+            return users, base
+        page_number += 1
+
+def identifiers(user: Dict[str, Any]) -> Set[str]:
+    values = (user.get('username'), user.get('phoneNumber'), user.get('userExternalId'))
+    return set(str(value) for value in values if value not in (None, ''))
+
+def get_user(base: str, token: str, user_id: str) -> Dict[str, Any]:
+    request = urllib.request.Request(
+        '{}/{}'.format(base, urllib.parse.quote(str(user_id), safe='')),
+        headers={'Authorization': 'Bearer {}'.format(token)},
+        method='GET')
+    payload, _ = request_json(request)
+    return payload
+
+def set_primary_org(base: str, token: str, user_id: str, org_id: str) -> int:
+    url = '{}/{}/actions/setUserPrimaryOrganizationalUnit'.format(
+        base, urllib.parse.quote(str(user_id), safe=''))
+    body = json.dumps({'organizationalUnitId': org_id}).encode('utf-8')
+    request = urllib.request.Request(
+        url,
+        data=body,
+        headers={
+            'Authorization': 'Bearer {}'.format(token),
+            'Content-Type': 'application/json',
+        },
+        method='POST')
+    with urllib.request.urlopen(request, timeout=30) as response:
+        response.read()
+        return response.getcode()
+
+def main() -> None:
+    tasks = json.loads(sys.stdin.readline())
+    config = yaml.safe_load(CONFIG_PATH.read_text(encoding='utf-8'))
+    token = get_access_token(config)
+    users, base = list_users(config, token)
+    results = []
+    for task in tasks:
+        matched = next((user for user in users if task['username'] in identifiers(user)), None)
+        if not matched:
+            results.append({'key': task['key'], 'success': False, 'message': 'EIAM account not found'})
+            continue
+        detail = get_user(base, token, matched.get('userId'))
+        current_org = detail.get('primaryOrganizationalUnitId')
+        if current_org == task['organizationalUnitId']:
+            results.append({'key': task['key'], 'success': True, 'changed': False,
+                            'userId': matched.get('userId'), 'httpStatus': None})
+            continue
+        try:
+            status = set_primary_org(base, token, matched.get('userId'), task['organizationalUnitId'])
+            results.append({'key': task['key'], 'success': True, 'changed': True,
+                            'userId': matched.get('userId'), 'httpStatus': status})
+        except Exception as error:
+            results.append({'key': task['key'], 'success': False,
+                            'message': '{}: {}'.format(type(error).__name__, str(error))[:240]})
+    print(json.dumps({'results': results}, ensure_ascii=False))
+
+if __name__ == '__main__':
+    main()
+`;
+
+/**
+ * 解析命令行参数。
+ * @returns {{execute: boolean, scanFailed: boolean, processIds: string[]}} 参数
+ */
+function parseArgs() {
+  const args = process.argv.slice(2);
+  const parsed = { execute: false, scanFailed: false, processIds: [] };
+  for (let index = 0; index < args.length; index += 1) {
+    if (args[index] === '--execute') parsed.execute = true;
+    else if (args[index] === '--scan-failed') parsed.scanFailed = true;
+    else if (args[index] === '--process-id' && args[index + 1]) parsed.processIds.push(args[++index]);
+    else throw new Error(`unknown or incomplete argument: ${args[index]}`);
+  }
+  if (!parsed.scanFailed && parsed.processIds.length === 0) parsed.scanFailed = true;
+  return parsed;
+}
+
+/**
+ * 从 OpenYida CLI 输出解析 JSON。
+ * @param {string} output CLI 输出
+ * @returns {Record<string, unknown>} JSON 响应
+ */
+function parsePayload(output) {
+  const start = output.indexOf('{\n  "content"');
+  if (start < 0) throw new Error('cannot locate OpenYida JSON response');
+  return JSON.parse(output.slice(start));
+}
+
+/**
+ * 分页查询全部入职流程。
+ * @returns {Array<Record<string, unknown>>} 流程实例
+ */
+function queryProcesses() {
+  const instances = [];
+  for (let page = 1; ; page += 1) {
+    const result = spawnSync('openyida', [
+      'data', 'query', 'process', APP_TYPE, FORM_UUID,
+      '--page', String(page), '--size', '100',
+    ], { encoding: 'utf8' });
+    if (result.status !== 0) throw new Error(result.stderr || result.stdout);
+    const content = parsePayload(result.stdout).content || {};
+    instances.push(...(content.data || []));
+    if (instances.length >= Number(content.totalCount || instances.length)) return instances;
+  }
+}
+
+/**
+ * 将字段值归一化为单个字符串。
+ * @param {unknown} value 字段值
+ * @returns {string} 单值
+ */
+function firstText(value) {
+  if (Array.isArray(value)) return value.length ? firstText(value[0]) : '';
+  return value === null || value === undefined ? '' : String(value).trim();
+}
+
+/**
+ * 移除查询响应中不可回写的 _value 派生字段。
+ * @param {unknown} value 原始值
+ * @returns {unknown} 可回写值
+ */
+function stripDerived(value) {
+  if (Array.isArray(value)) return value.map(stripDerived);
+  if (!value || typeof value !== 'object') return value;
+  return Object.fromEntries(Object.entries(value)
+    .filter(([key]) => !key.endsWith('_value'))
+    .map(([key, item]) => [key, stripDerived(item)]));
+}
+
+/**
+ * 为去重任务生成不包含人员标识的键。
+ * @param {string} username EIAM 账户名
+ * @param {string} organizationalUnitId 目标部门 ID
+ * @returns {string} 任务键
+ */
+function taskKey(username, organizationalUnitId) {
+  return createHash('sha256').update(`${username}\0${organizationalUnitId}`).digest('hex').slice(0, 16);
+}
+
+/**
+ * 构建待执行的人员部门修复明细。
+ * @param {Array<Record<string, unknown>>} instances 流程实例
+ * @param {{scanFailed: boolean, processIds: string[]}} options 筛选参数
+ * @returns {Array<Record<string, unknown>>} 修复明细
+ */
+function buildCandidates(instances, options) {
+  const requested = new Set(options.processIds);
+  const found = new Set();
+  const candidates = [];
+  for (const instance of instances) {
+    if (requested.size && !requested.has(instance.processInstanceId)) continue;
+    found.add(instance.processInstanceId);
+    const data = instance.data || {};
+    const rows = Array.isArray(data[TABLE_FIELD]) ? data[TABLE_FIELD] : [];
+    if (rows.length === 50) throw new Error(`${instance.processInstanceId}: subtable may be truncated at 50 rows`);
+    rows.forEach((row, rowIndex) => {
+      const status = firstText(row[STATUS_FIELD] || row[`${STATUS_FIELD}_id`]);
+      if (options.scanFailed && status !== '失败') return;
+      const username = firstText(row[PHONE_FIELD] ?? row[`${PHONE_FIELD}_value`]);
+      const organizationalUnitId = firstText(data[ORG_FIELD]);
+      if (!username || !organizationalUnitId.startsWith('ou_')) {
+        throw new Error(`${instance.processInstanceId} row ${rowIndex + 1}: missing account or valid department ID`);
+      }
+      candidates.push({ instance, rowIndex, username, organizationalUnitId,
+        key: taskKey(username, organizationalUnitId) });
+    });
+  }
+  const missing = [...requested].filter((id) => !found.has(id));
+  if (missing.length) throw new Error(`process instances not found: ${missing.join(', ')}`);
+  return candidates;
+}
+
+/**
+ * 阻止同一账户在同一批次被设置为多个主部门。
+ * @param {Array<Record<string, unknown>>} candidates 修复明细
+ * @returns {void}
+ */
+function assertNoConflicts(candidates) {
+  const byUser = new Map();
+  for (const item of candidates) {
+    if (!byUser.has(item.username)) byUser.set(item.username, new Set());
+    byUser.get(item.username).add(item.organizationalUnitId);
+  }
+  const conflicts = [...byUser.entries()].filter(([, orgs]) => orgs.size > 1);
+  if (conflicts.length) {
+    throw new Error(`department conflicts: ${conflicts.map(([user, orgs]) =>
+      `${mask(user)} => ${[...orgs].join(',')}`).join('; ')}`);
+  }
+}
+
+/**
+ * 遮蔽账户标识。
+ * @param {string} value 账户标识
+ * @returns {string} 遮蔽值
+ */
+function mask(value) {
+  return value.length >= 4 ? `***${value.slice(-4)}` : '***';
+}
+
+/**
+ * 通过 SSH 在服务器内调用 EIAM，凭据只在远端进程内使用。
+ * @param {Array<Record<string, unknown>>} tasks 去重后任务
+ * @returns {Array<Record<string, unknown>>} EIAM 执行结果
+ */
+function runRemoteTasks(tasks) {
+  const encoded = Buffer.from(REMOTE_PYTHON, 'utf8').toString('base64');
+  const command = `python3 -c "import base64;exec(base64.b64decode('${encoded}'))"`;
+  const result = spawnSync('ssh', [
+    '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=8', '-p', '22', `root@${HOST}`, command,
+  ], { encoding: 'utf8', input: `${JSON.stringify(tasks)}\n`, maxBuffer: 10 * 1024 * 1024 });
+  if (result.status !== 0) throw new Error(result.stderr || result.stdout || `ssh exited ${result.status}`);
+  return JSON.parse(result.stdout).results || [];
+}
+
+/**
+ * 将 EIAM 执行结果回写到流程实例。
+ * @param {Array<Record<string, unknown>>} candidates 修复明细
+ * @param {Array<Record<string, unknown>>} remoteResults EIAM 结果
+ * @returns {Array<Record<string, unknown>>} 回写摘要
+ */
+function writeBack(candidates, remoteResults) {
+  const resultByKey = new Map(remoteResults.map((item) => [item.key, item]));
+  const byProcess = new Map();
+  for (const candidate of candidates) {
+    if (!byProcess.has(candidate.instance.processInstanceId)) byProcess.set(candidate.instance.processInstanceId, []);
+    byProcess.get(candidate.instance.processInstanceId).push(candidate);
+  }
+  const summaries = [];
+  for (const [processId, items] of byProcess) {
+    summaries.push(updateOneProcess(processId, items, resultByKey));
+  }
+  return summaries;
+}
+
+/**
+ * 回写单个流程的全量子表和统计。
+ * @param {string} processId 流程实例 ID
+ * @param {Array<Record<string, unknown>>} items 当前流程修复明细
+ * @param {Map<string, Record<string, unknown>>} resultByKey EIAM 结果
+ * @returns {Record<string, unknown>} 回写摘要
+ */
+function updateOneProcess(processId, items, resultByKey) {
+  const data = items[0].instance.data || {};
+  const rows = stripDerived(data[TABLE_FIELD] || []);
+  const errors = [];
+  for (const item of items) {
+    const remote = resultByKey.get(item.key);
+    const success = Boolean(remote?.success);
+    rows[item.rowIndex][STATUS_FIELD] = success ? '成功' : '失败';
+    rows[item.rowIndex][`${STATUS_FIELD}_id`] = success ? '成功' : '失败';
+    if (!success) errors.push(`${firstText(rows[item.rowIndex][NAME_FIELD]) || `第 ${item.rowIndex + 1} 行`} - ${remote?.message || '部门更新失败'}`);
+  }
+  const successCount = rows.filter((row) => firstText(row[STATUS_FIELD]) === '成功').length;
+  const failedCount = rows.filter((row) => firstText(row[STATUS_FIELD]) === '失败').length;
+  const patch = { [TABLE_FIELD]: rows, [TOTAL_FIELD]: rows.length,
+    [SUCCESS_FIELD]: successCount, [FAILED_FIELD]: failedCount,
+    [FAILURE_MESSAGE_FIELD]: failedCount === 0 ? '' : errors.join('\n') || data[FAILURE_MESSAGE_FIELD] || '' };
+  const result = spawnSync('openyida', ['data', 'update', 'process', APP_TYPE,
+    '--process-inst-id', processId, '--form-uuid', FORM_UUID,
+    '--data-json', JSON.stringify(patch)], { encoding: 'utf8' });
+  if (result.status !== 0) throw new Error(result.stderr || result.stdout);
+  return { processInstanceId: processId, total: rows.length, success: successCount, failed: failedCount };
+}
+
+/**
+ * 验证目标流程回写结果。
+ * @param {string[]} processIds 目标流程 ID
+ * @returns {Array<Record<string, unknown>>} 验证摘要
+ */
+function verify(processIds) {
+  const wanted = new Set(processIds);
+  return queryProcesses().filter((instance) => wanted.has(instance.processInstanceId)).map((instance) => {
+    const data = instance.data || {};
+    const rows = Array.isArray(data[TABLE_FIELD]) ? data[TABLE_FIELD] : [];
+    return { processInstanceId: instance.processInstanceId,
+      total: Number(data[TOTAL_FIELD] ?? data[`${TOTAL_FIELD}_value`] ?? -1),
+      success: Number(data[SUCCESS_FIELD] ?? data[`${SUCCESS_FIELD}_value`] ?? -1),
+      failed: Number(data[FAILED_FIELD] ?? data[`${FAILED_FIELD}_value`] ?? -1),
+      detailStatuses: rows.map((row) => firstText(row[STATUS_FIELD] || row[`${STATUS_FIELD}_id`])),
+      failureMessageEmpty: !data[FAILURE_MESSAGE_FIELD] };
+  });
+}
+
+const options = parseArgs();
+const instances = queryProcesses();
+const candidates = buildCandidates(instances, options);
+assertNoConflicts(candidates);
+const uniqueTasks = [...new Map(candidates.map((item) => [item.key, {
+  key: item.key, username: item.username, organizationalUnitId: item.organizationalUnitId,
+}])).values()];
+process.stdout.write(`${JSON.stringify({ mode: options.execute ? 'execute' : 'dry-run',
+  processes: new Set(candidates.map((item) => item.instance.processInstanceId)).size,
+  rows: candidates.length, uniqueUsers: uniqueTasks.length,
+  candidates: candidates.map((item) => ({ processInstanceId: item.instance.processInstanceId,
+    row: item.rowIndex + 1, account: mask(item.username), organizationalUnitId: item.organizationalUnitId })) }, null, 2)}\n`);
+if (options.execute && uniqueTasks.length) {
+  const remoteResults = runRemoteTasks(uniqueTasks);
+  const writeback = writeBack(candidates, remoteResults);
+  const verification = verify([...new Set(candidates.map((item) => item.instance.processInstanceId))]);
+  process.stdout.write(`${JSON.stringify({ remoteResults, writeback, verification }, null, 2)}\n`);
+  if (remoteResults.some((item) => !item.success)) process.exitCode = 2;
+}