旺店通到MySQL数据集成技术方案：基于轻易云平台的SSO架构与高效ETL实践

┌─────────────────────────────────────────────────────────────────┐
│                    统一认证层 (SSO Layer)                        │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────────┐ │
│  │  企业微信/钉钉  │  │   OA系统    │  │    旺店通/金蝶/用友      │ │
│  │  OAuth2.0   │  │   SAML2.0   │  │      API Key            │ │
│  └──────┬──────┘  └──────┬──────┘  └───────────┬─────────────┘ │
│         └─────────────────┼─────────────────────┘               │
│                           ▼                                     │
│                    ┌─────────────┐                              │
│                    │  CAS Server │                              │
│                    │  (轻易云MC) │                              │
│                    └──────┬──────┘                              │
└───────────────────────────┼─────────────────────────────────────┘
                            ▼
┌─────────────────────────────────────────────────────────────────┐
│                    数据集成引擎 (DataHub)                        │
│  ┌─────────────┐    ┌─────────────┐    ┌─────────────────────┐│
│  │  请求调度者   │───▶│  数据加工工厂  │───▶│    写入调度者        ││
│  │ (API适配层) │    │ (ETL转换)   │    │   (目标系统写入)     ││
│  └─────────────┘    └─────────────┘    └─────────────────────┘│
│         │                   │                      │            │
│         ▼                   ▼                      ▼            │
│  ┌─────────────┐    ┌─────────────┐    ┌─────────────────────┐│
│  │ 旺店通API    │    │ 数据清洗/转换 │    │    MySQL API        ││
│  │ wdt.stockout │    │ 字段映射/脚本 │    │   batchexecute     ││
│  └─────────────┘    └─────────────┘    └─────────────────────┘│
└─────────────────────────────────────────────────────────────────┘

数据流处理流程：

旺店通源数据 → API调用 → 分页处理 → 数据清洗 → ETL转换 → 批量写入 → MySQL
     │            │          │          │          │          │
     ▼            ▼          ▼          ▼          ▼          ▼
┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐
│ 入库成本 │  │ POST请求 │  │ 200条/ │  │ 字段映射 │  │ REPLACE │  │ 目标表  │
│ 明细数据 │  │ 分页查询 │  │ 页     │  │ 格式转换 │  │ INTO   │  │ 存储   │
└─────────┘  └─────────┘  └─────────┘  └─────────┘  └─────────┘  └─────────┘

{
  "api": "statistic.StockinCollect.queryCostWithDetail",
  "effect": "QUERY",
  "method": "POST",
  "number": "stockin_no",
  "id": "stockin_id",
  "name": "stockout_no",
  "idCheck": true,
  "request": [
    {
      "field": "params",
      "label": "查询参数",
      "type": "object",
      "children": [
        {"field": "start_time", "label": "开始时间", "type": "string", "value":"{{LAST_SYNC_TIME|datetime}}"},
        {"field": "end_time", "label": "结束时间", "type": "string", "value":"{{CURRENT_TIME|datetime}}"},
        {"field": "warehouse_no", "label": "仓库编号", "type": "string"},
        {"field": "time_type", "label": "时间查询类型", "type":"string", "describe":"1：创建时间\n2：审核时间\n默认创建时间"},
        {"field": "stockin_no", "label": "入库单号", "type": "string", "describe":"多个入库单号之间英文逗号分隔"}
      ]
    },
    {
      "field": "pager",
      "label": "分页参数",
      "type": "object",
      "value": "50",
      "children": [
        {"field": "page_size", "label": "分页大小", "type": "string", "value": "200"},
        {"field": "page_no", "label": "页号", "type": "string"}
      ]
    }
  ],
  "autoFillResponse": true,
  "beatFlat": ["detail_list"],
  "delay": 300
}

CREATE TABLE stockincollect_querycostwithdetail (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    stockin_no VARCHAR(50) COMMENT '入库单号',
    stockin_id VARCHAR(50) COMMENT '入库单ID',
    src_order_id VARCHAR(50) COMMENT '源订单ID',
    warehouse_no VARCHAR(50) COMMENT '仓库编号',
    warehouse_name VARCHAR(100) COMMENT '仓库名称',
    goods_count INT COMMENT '货品数量',
    checked_goods_total_cost DECIMAL(18,4) COMMENT '实际成本总额',
    planned_goods_total_cost DECIMAL(18,4) COMMENT '计划成本总额',
    created DATETIME COMMENT '创建时间',
    check_time DATETIME COMMENT '审核时间',
    modified DATETIME COMMENT '修改时间',
    detail_list_rec_id VARCHAR(50) COMMENT '明细记录ID',
    detail_list_goods_no VARCHAR(50) COMMENT '货品编号',
    detail_list_goods_name VARCHAR(200) COMMENT '货品名称',
    detail_list_spec_no VARCHAR(50) COMMENT '规格编号',
    detail_list_checked_cost_price DECIMAL(18,4) COMMENT '实际成本价',
    detail_list_total_checked_cost_price DECIMAL(18,4) COMMENT '实际成本总价',
    detail_list_planned_cost DECIMAL(18,4) COMMENT '计划成本',
    detail_list_total_planned_cost DECIMAL(18,4) COMMENT '计划成本总价',
    detail_list_num INT COMMENT '数量',
    detail_list_batch_no VARCHAR(50) COMMENT '批次号',
    detail_list_production_date DATE COMMENT '生产日期',
    detail_list_expire_date DATE COMMENT '有效期至',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '记录创建时间',
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '记录更新时间',
    INDEX idx_stockin_no (stockin_no),
    INDEX idx_warehouse_no (warehouse_no),
    INDEX idx_created (created),
    INDEX idx_check_time (check_time)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='旺店通入库瞬时成本明细表';

{
  "condition": [
    [
      {"field": "warehouse_no", "logic": "notnull", "value": ""},
      {"field": "goods_count", "logic": "gt", "value": 0},
      {"field": "status", "logic": "eq", "value": "Confirmed"}
    ]
  ],
  "transform": [
    {
      "field": "created",
      "type": "datetime",
      "format": "yyyy-MM-dd HH:mm:ss",
      "source_format": "yyyy-MM-dd'T'HH:mm:ss'Z'"
    },
    {
      "field": "checked_goods_total_cost",
      "type": "decimal",
      "precision": 18,
      "scale": 4
    }
  ]
}

{
  "api": "batchexecute",
  "effect": "EXECUTE",
  "method": "POST",
  "idCheck": true,
  "request": [
    {
      "field": "main_sql",
      "label": "主语句",
      "type": "string",
      "describe": "SQL首次执行的语句，将会获取主ID",
      "value": "REPLACE INTO stockincollect_querycostwithdetail(stockin_no, stockin_id, src_order_id, warehouse_no, warehouse_name, goods_count, checked_goods_total_cost, planned_goods_total_cost, created, check_time, modified, detail_list_rec_id, detail_list_stockin_id, detail_list_num, detail_list_goods_name, detail_list_goods_no, detail_list_spec_no, detail_list_checked_cost_price, detail_list_total_checked_cost_price, detail_list_planned_cost, detail_list_total_planned_cost, detail_list_position_no, detail_list_batch_no, detail_list_production_date, detail_list_expire_date) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
    },
    {
      "field": "limit",
      "label": "限制",
      "type": "string",
      "describe": "每次获取的查询量",
      "value": "1000"
    }
  ]
}

[mysqld]
# InnoDB缓冲池大小，建议设置为物理内存的50-70%
innodb_buffer_pool_size = 4G

# 日志文件大小，增大可减少磁盘I/O
innodb_log_file_size = 512M
innodb_log_buffer_size = 64M

# 事务提交策略，设为2可提升写入性能（每秒刷盘一次）
innodb_flush_log_at_trx_commit = 2

# 批量插入缓冲区大小
bulk_insert_buffer_size = 256M

# 允许的最大数据包大小
max_allowed_packet = 256M

# 连接数配置
max_connections = 500
innodb_read_io_threads = 8
innodb_write_io_threads = 8

// SSO插件示例代码结构
public class WdtCasAuthenticationFilter implements AuthenticationFilter {
    
    @Override
    public boolean authenticate(HttpServletRequest request, HttpServletResponse response) {
        // 1. 获取CAS票据
        String ticket = request.getParameter("ticket");
        
        // 2. 验证票据有效性
        CasValidationResult result = validateTicket(ticket, casServerUrl);
        
        // 3. 映射用户身份
        if (result.isValid()) {
            String username = result.getUsername();
            String employeeId = mapToEmployeeId(username);
            request.setAttribute("USER_ID", employeeId);
            return true;
        }
        return false;
    }
    
    private String mapToEmployeeId(String username) {
        // 调用轻易云智能映射引擎
        return IdentityMappingEngine.getInstance().map(username);
    }
}

{
  "source_crontab": "0 */5 * * * *",
  "target_crontab": "0 */5 * * * *",
  "retry_policy": {
    "max_retries": 3,
    "retry_interval": 60,
    "exponential_backoff": true
  },
  "alert_config": {
    "on_failure": true,
    "on_timeout": true,
    "notification_channels": ["email", "sms", "webhook"]
  }
}