数据迁移与批量导入：TC 导入工具、XML 模板与增量同步策略

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

迁移策略选择：
┌─────────────────────────────────────────┐
│ 数据量                                  │
│ ├── < 1000 条 → Excel 导入              │
│ ├── 1000-10000 条 → XML 批量导入        │
│ ├── 10000-100000 条 → ITK 批量导入      │
│ └── > 100000 条 → 数据库直接导入        │
├─────────────────────────────────────────┤
│ 迁移频率                                │
│ ├── 一次性 → 批量导入工具               │
│ ├── 定期同步 → 增量同步方案             │
│ └── 实时同步 → 消息队列/事件驱动        │
├─────────────────────────────────────────┤
│ 数据复杂度                              │
│ ├── 简单属性 → 直接导入                 │
│ ├── 关联关系 → 分步导入（先父后子）     │
│ └── 文件数据 → FMS 导入                 │
└─────────────────────────────────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

<?xml version="1.0" encoding="UTF-8"?>
<TeamcenterDataModel>
  <Items>
    <Item>
      <item_id>MOTOR-001</item_id>
      <object_string>YE3-132M-4 三相异步电动机</object_string>
      <object_desc>7.5kW, 4极, 380V, IP55</object_desc>
      <object_type>Part</object_type>

      <!-- 业务属性 -->
      <POM_part_weight>45.5</POM_part_weight>
      <POM_material>铸铁</POM_material>
      <POM_specification>GB/T 1032</POM_specification>

      <!-- 分类 -->
      <Classification>
        <category>电机类</category>
        <subcategory>三相异步电机</subcategory>
      </Classification>
    </Item>

    <Item>
      <item_id>MOTOR-002</item_id>
      <object_string>YE3-160M-4 三相异步电动机</object_string>
      <object_desc>11kW, 4极, 380V, IP55</object_desc>
      <object_type>Part</object_type>
      <POM_part_weight>78.0</POM_part_weight>
      <POM_material>铸铁</POM_material>
    </Item>
  </Items>
</TeamcenterDataModel>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

<?xml version="1.0" encoding="UTF-8"?>
<ImportTemplate>
  <TargetClass>Item</TargetClass>

  <!-- 匹配键（用于判断是否更新） -->
  <MatchKey>
    <Property>item_id</Property>
  </MatchKey>

  <!-- 属性映射 -->
  <PropertyMappings>
    <PropertyMap>
      <SourceField>item_id</SourceField>
      <TargetProperty>item_id</TargetProperty>
      <Required>true</Required>
    </PropertyMap>
    <PropertyMap>
      <SourceField>object_string</SourceField>
      <TargetProperty>object_string</TargetProperty>
      <Required>true</Required>
    </PropertyMap>
    <PropertyMap>
      <SourceField>object_desc</SourceField>
      <TargetProperty>object_desc</TargetProperty>
      <Required>false</Required>
    </PropertyMap>
    <PropertyMap>
      <SourceField>weight</SourceField>
      <TargetProperty>POM_part_weight</TargetProperty>
      <Required>false</Required>
      <DataType>double</DataType>
    </PropertyMap>
    <PropertyMap>
      <SourceField>material</SourceField>
      <TargetProperty>POM_material</TargetProperty>
      <Required>false</Required>
    </PropertyMap>
  </PropertyMappings>

  <!-- 操作模式 -->
  <OperationMode>
    <!-- create: 仅创建, update: 仅更新, upsert: 创建或更新 -->
    <Mode>upsert</Mode>
  </OperationMode>

  <!-- 错误处理 -->
  <ErrorHandling>
    <OnError>continue</OnError>
    <MaxErrors>100</MaxErrors>
  </ErrorHandling>
</ImportTemplate>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

# 命令行导入
import_data_from_xml \
  -u=infodba \
  -p=infodba \
  -g=dba \
  -template=import_template.xml \
  -data=motor_data.xml \
  -log=import.log

# 常见参数：
# -u: 用户名
# -p: 密码
# -g: 组
# -template: 导入模板
# -data: 数据文件
# -log: 日志文件
# -dryrun: 试运行（不实际导入）
# -batchsize: 批量大小

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

安装 Excel 导入插件后：
1. 打开 Excel
2. 加载 Teamcenter 插件
3. 连接 Teamcenter 服务器
4. 导入模板（文件 → 导入模板）
5. 填写数据
6. 执行导入

优点：用户友好，适合业务人员
缺点：大数据量性能较差

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

CSV 格式：
item_id,object_string,object_desc,weight,material
MOTOR-001,YE3-132M-4,7.5kW 4极电机,45.5,铸铁
MOTOR-002,YE3-160M-4,11kW 4极电机,78.0,铸铁
MOTOR-003,YE3-180L-4,15kW 4极电机,120.0,铸铁

导入命令：
batch_import_csv \
  -u=infodba -p=infodba -g=dba \
  -file=motor_data.csv \
  -template=part_import.xml \
  -log=import.log

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

<?xml version="1.0" encoding="UTF-8"?>
<BOMImport>
  <BOM>
    <ParentItem>ASSEMBLY-001</ParentItem>
    <ParentRevision>A</ParentRevision>
    <Children>
      <ChildItem>
        <item_id>PART-001</item_id>
        <revision>A</revision>
        <quantity>2</quantity>
        <find_number>10</find_number>
        <unit>EA</unit>
      </ChildItem>
      <ChildItem>
        <item_id>PART-002</item_id>
        <revision>A</revision>
        <quantity>1</quantity>
        <find_number>20</find_number>
        <unit>EA</unit>
      </ChildItem>
    </Children>
  </BOM>
</BOMImport>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

1. 确保父项和子项已存在
2. 导入 BOM 结构
3. 验证导入结果
4. 处理错误项

导入命令：
import_bom \
  -u=infodba -p=infodba -g=dba \
  -file=bom_data.xml \
  -log=bom_import.log

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109

#include <tc/tc.h>
#include <item/item.h>
#include <aom/aom.h>
#include <emh/emh.h>

typedef struct {
    char item_id[50];
    char object_string[200];
    char object_desc[500];
    double weight;
    char material[100];
} ItemData;

int import_items_from_csv(const char* csv_file) {
    int status = ITK_ok;
    int total = 0, success = 0, failed = 0;

    FILE* fp = fopen(csv_file, "r");
    if (!fp) {
        printf("无法打开文件: %s\n", csv_file);
        return ITK_error;
    }

    // 跳过表头
    char line[1024];
    fgets(line, sizeof(line), fp);

    while (fgets(line, sizeof(line), fp)) {
        ItemData data;
        memset(&data, 0, sizeof(data));

        // 解析 CSV 行
        sscanf(line, "%[^,],%[^,],%[^,],%lf,%s",
               data.item_id, data.object_string,
               data.object_desc, &data.weight,
               data.material);

        total++;

        // 创建或更新 Item
        status = import_single_item(&data);
        if (status == ITK_ok) {
            success++;
            printf("[OK] %s - %s\n", data.item_id, data.object_string);
        } else {
            failed++;
            printf("[FAIL] %s - %s (error: %d)\n",
                   data.item_id, data.object_string, status);
        }

        // 每 100 条提交一次事务
        if (total % 100 == 0) {
            EMH_ask_errors();
            printf("进度: %d 条，成功 %d，失败 %d\n",
                   total, success, failed);
        }
    }

    fclose(fp);

    printf("\n导入完成: 总计 %d，成功 %d，失败 %d\n",
           total, success, failed);

    return ITK_ok;
}

int import_single_item(ItemData* data) {
    tag_t item_tag = NULLTAG;
    tag_t rev_tag = NULLTAG;
    int status = ITK_ok;

    // 检查是否已存在
    status = ITEM_find_item(data->item_id, &item_tag);

    if (status != ITK_ok || item_tag == NULLTAG) {
        // 创建新 Item
        status = ITEM_create_item("Item", data->item_id, &item_tag);
        if (status != ITK_ok) return status;

        // 创建 Revision
        status = ITEM_create_rev(item_tag, "A", &rev_tag);
        if (status != ITK_ok) return status;
    } else {
        // 获取 Revision
        status = ITEM_find_rev(item_tag, "A", &rev_tag);
        if (status != ITK_ok) return status;
    }

    // 设置属性
    status = AOM_set_string_property(rev_tag, "object_string", data->object_string);
    if (status != ITK_ok) return status;

    status = AOM_set_string_property(rev_tag, "object_desc", data->object_desc);
    if (status != ITK_ok) return status;

    status = AOM_set_double_property(rev_tag, "POM_part_weight", data->weight);
    if (status != ITK_ok) return status;

    status = AOM_set_string_property(rev_tag, "POM_material", data->material);
    if (status != ITK_ok) return status;

    // 保存
    status = AOM_save(rev_tag);
    if (status != ITK_ok) return status;

    AOM_refresh(rev_tag, true);

    return ITK_ok;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

// 批量提交优化
void set_batch_mode(int enable) {
    if (enable) {
        // 关闭自动提交
        TC_set_autocommit(false);
        // 增大事务缓冲区
        TC_set_buffer_size(1000);
    } else {
        TC_set_autocommit(true);
    }
}

// 并行导入（多线程）
void parallel_import(const char* csv_file, int thread_count) {
    // 读取所有数据
    ItemData* data = load_all_items(csv_file);
    int total = count_items(data);

    // 分片
    int chunk_size = total / thread_count;

    // 启动线程
    pthread_t* threads = malloc(thread_count * sizeof(pthread_t));
    for (int i = 0; i < thread_count; i++) {
        int start = i * chunk_size;
        int end = (i == thread_count - 1) ? total : (i + 1) * chunk_size;
        pthread_create(&threads[i], NULL, import_chunk,
                       create_chunk_args(data, start, end));
    }

    // 等待完成
    for (int i = 0; i < thread_count; i++) {
        pthread_join(threads[i], NULL);
    }

    free(threads);
    free(data);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

同步流程：
1. 记录上次同步时间
   last_sync_time = "2026-06-11 23:00:00"

2. 查询增量数据
   SELECT * FROM source_table
   WHERE update_time > last_sync_time
   ORDER BY update_time ASC;

3. 逐条同步到 TC
   for each record:
     if exists in TC:
       update TC record
     else:
       create TC record

4. 更新同步时间
   last_sync_time = NOW()

5. 记录同步日志
   sync_log.write(sync_time, total, success, failed)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

#!/usr/bin/env python3
"""Teamcenter 增量同步脚本"""

import subprocess
import logging
from datetime import datetime
import json

logging.basicConfig(
    filename='sync.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

def get_last_sync_time():
    """获取上次同步时间"""
    try:
        with open('last_sync_time.txt', 'r') as f:
            return f.read().strip()
    except FileNotFoundError:
        return '2000-01-01 00:00:00'

def update_sync_time():
    """更新同步时间"""
    now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    with open('last_sync_time.txt', 'w') as f:
        f.write(now)

def query_incremental_data(last_sync_time):
    """从源系统查询增量数据"""
    sql = f"""
    SELECT item_id, name, description, weight, material, update_time
    FROM material_master
    WHERE update_time > '{last_sync_time}'
    ORDER BY update_time ASC
    """
    # 执行查询，返回数据列表
    # ...
    return []

def sync_item_to_tc(item_data):
    """同步单条数据到 TC"""
    # 调用 ITK 或 XML 导入
    # ...
    pass

def main():
    last_sync_time = get_last_sync_time()
    logging.info(f"开始增量同步，上次同步时间: {last_sync_time}")

    data = query_incremental_data(last_sync_time)
    logging.info(f"获取到 {len(data)} 条增量数据")

    success = 0
    failed = 0

    for item in data:
        try:
            sync_item_to_tc(item)
            success += 1
        except Exception as e:
            failed += 1
            logging.error(f"同步失败: {item['item_id']} - {e}")

    update_sync_time()

    logging.info(
        f"同步完成: 总计 {len(data)} 条, "
        f"成功 {success} 条, 失败 {failed} 条"
    )

if __name__ == '__main__':
    main()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# crontab 配置（每天凌晨 2 点执行）
0 2 * * * cd /opt/tc/sync && python3 incremental_sync.py >> /var/log/tc_sync.log 2>&1

# 或使用 systemd timer
# /etc/systemd/system/tc-sync.timer
[Unit]
Description=Teamcenter Incremental Sync Timer

[Timer]
OnCalendar=daily
Persistent=true

[Install]
WantedBy=timers.target

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

def validate_item(item):
    """校验单条数据"""
    errors = []
    warnings = []

    # 必填字段检查
    if not item.get('item_id'):
        errors.append("item_id 不能为空")
    if not item.get('object_string'):
        errors.append("名称不能为空")

    # 格式检查
    if item.get('item_id'):
        if len(item['item_id']) > 50:
            errors.append("item_id 长度不能超过 50")
        if not re.match(r'^[A-Z0-9-]+$', item['item_id']):
            errors.append("item_id 格式不正确（仅允许大写字母、数字、横线）")

    # 范围检查
    if item.get('weight'):
        if item['weight'] < 0:
            errors.append("重量不能为负数")
        if item['weight'] > 100000:
            warnings.append("重量超过 100 吨，请确认")

    # 关联检查
    if item.get('parent_id'):
        if not item_exists(item['parent_id']):
            errors.append(f"父项 {item['parent_id']} 不存在")

    return errors, warnings

def validate_csv(csv_file):
    """校验整个 CSV 文件"""
    results = []
    with open(csv_file, 'r') as f:
        reader = csv.DictReader(f)
        for row in reader:
            errors, warnings = validate_item(row)
            results.append({
                'row': reader.line_num,
                'item_id': row.get('item_id'),
                'errors': errors,
                'warnings': warnings
            })
    return results

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

数据比对报告模板：
┌──────────────────────────────────────────────┐
│ 数据迁移质量报告                              │
├──────────────────────────────────────────────┤
│ 源系统数据量：10,000 条                      │
│ TC 导入成功：9,850 条 (98.5%)                │
│ TC 导入失败：150 条 (1.5%)                   │
├──────────────────────────────────────────────┤
│ 失败原因分布：                                │
│ ├── 必填字段缺失：80 条                      │
│ ├── 格式不正确：40 条                        │
│ ├── 关联数据不存在：20 条                    │
│ └── 其他：10 条                              │
├──────────────────────────────────────────────┤
│ 数据一致性检查：                              │
│ ├── 属性一致性：99.2%                        │
│ ├── 关联一致性：98.8%                        │
│ └── 文件一致性：99.5%                        │
└──────────────────────────────────────────────┘