Excel字段提取（供应链库存数据）

功能概述

从供应链Excel报表的"整体库存"工作表中，自动提取指定的核心字段，生成精简的数据表格。适用于日常库存监控、数据分享和报表制作。

适用场景：

• 供应链数据分析师需要从完整报表中提取关键字段
• 跨部门分享精简数据（避免泄露完整数据结构）
• 生成标准化数据模板用于后续分析
• 定期数据提取任务自动化

数据来源：鱼跃品牌供应链全品类库存报表

---

一、字段提取清单

默认字段（2026-05-21版）

从"整体库存"工作表中提取以下14个字段（按列名精确匹配）：

| 序号 | 字段名 | 说明 | |------|--------|------| | 1 | SAP代码 | 产品唯一标识 | | 2 | SAP代码2 | 备用产品标识 | | 3 | 京东库存（截止前一日24点） | 京东平台实时库存 | | 4 | 京东在途+商家所有（含开单未提货及今日入库） | 京东在途及商家库存 | | 5 | 股份库存 | 股份公司库存 | | 6 | 京东近7天/7*30 | 京东近7天销量折算月销 | | 7 | 京东周转 | 京东库存周转率 | | 8 | 2026-5_排产数量 | 2026年5月计划排产数量 | | 9 | 2026-5_入库数量 | 2026年5月实际入库数量 | | 10 | 2026-5_待入库数量 | 2026年5月待入库数量 | | 11 | 2026-6_排产数量 | 2026年6月计划排产数量 | | 12 | 2026-5_京东销量 | 2026年5月京东平台销量 | | 13 | 2026-5_天猫销量 | 2026年5月天猫平台销量 | | 14 | 2026-5_渠道出货量 | 2026年5月渠道总出货量 |

字段更新说明

• 此清单基于历史工作记录固化，如需调整可修改技能配置
• 列名可能存在前导/后导空格，提取时使用精确匹配
• 新增字段需在技能配置中添加对应列名

---

二、完整工作流程

输入Excel文件 → 读取"整体库存"工作表 → 
验证字段存在性 → 提取指定字段 → 
生成新Excel文件 → 输出结果

输入要求：

• Excel文件（.xlsx格式）
• 必须包含名为"整体库存"的工作表
• 工作表应包含上述14个字段（列名需完全一致）

输出规则：

• 新文件命名：{原文件名}_指定字段摘取.xlsx
• 保存路径：与原文件相同目录
• 数据格式：保留所有行数据，不修改原始值
• 不包含索引列

---

三、核心脚本

3.1 Python提取脚本 (extract_fields.py)

import pandas as pd
import sys
import os

def extract_excel_fields(input_file, sheet_name='整体库存', output_suffix='_指定字段摘取.xlsx'):
    """
    从Excel文件中提取指定字段
    
    参数：
    ----------
    input_file : str
        输入Excel文件路径
    sheet_name : str, default='整体库存'
        要读取的工作表名称
    output_suffix : str, default='_指定字段摘取.xlsx'
        输出文件后缀
    
    返回：
    -------
    str : 输出文件路径
    """
    
    # 字段提取清单
    columns_to_extract = [
        'SAP代码',
        'SAP代码2',
        '京东库存（截止前一日24点）',
        '京东在途+商家所有（含开单未提货及今日入库）',
        '股份库存',
        '京东近7天/7*30',
        '京东周转',
        '2026-5_排产数量',
        '2026-5_入库数量',
        '2026-5_待入库数量',
        '2026-6_排产数量',
        '2026-5_京东销量',
        '2026-5_天猫销量',
        '2026-5_渠道出货量'
    ]
    
    # 验证输入文件
    if not os.path.exists(input_file):
        raise FileNotFoundError(f"输入文件不存在: {input_file}")
    
    # 生成输出路径
    base_name = os.path.splitext(os.path.basename(input_file))[0]
    output_file = os.path.join(os.path.dirname(input_file), f'{base_name}{output_suffix}')
    
    print(f'📂 读取文件: {input_file}')
    print(f'📊 工作表: {sheet_name}')
    print(f'🔍 提取字段数: {len(columns_to_extract)}')
    
    try:
        # 读取Excel文件
        df = pd.read_excel(input_file, sheet_name=sheet_name)
        print(f'📈 原始数据形状: {df.shape}')
        
        # 检查列是否存在
        missing_columns = [col for col in columns_to_extract if col not in df.columns]
        if missing_columns:
            print('❌ 错误: 以下字段在数据中不存在:')
            for col in missing_columns:
                print(f'  {repr(col)}')
            print('\n📋 实际列名:')
            for col in df.columns:
                print(f'  {repr(col)}')
            raise ValueError(f"缺失字段: {missing_columns}")
        
        # 提取指定列
        df_extracted = df[columns_to_extract].copy()
        print(f'✅ 提取后形状: {df_extracted.shape}')
        
        # 保存到新Excel文件
        df_extracted.to_excel(output_file, index=False)
        print(f'💾 数据已保存至: {output_file}')
        
        # 显示统计信息
        print('\n📊 数据统计:')
        print(f'  行数: {len(df_extracted)}')
        print(f'  列数: {len(df_extracted.columns)}')
        print(f'  文件大小: {os.path.getsize(output_file):,} bytes')
        
        return output_file
        
    except Exception as e:
        print(f'❌ 处理过程中出错: {e}')
        raise

if __name__ == '__main__':
    # 使用示例
    if len(sys.argv) < 2:
        print("使用方法: python extract_fields.py <Excel文件路径>")
        print("示例: python extract_fields.py C:\\路径\\全品类库存.xlsx")
        sys.exit(1)
    
    input_file = sys.argv[1]
    try:
        output_file = extract_excel_fields(input_file)
        print(f'\n🎉 提取完成! 文件已保存: {output_file}')
    except Exception as e:
        sys.exit(1)

3.2 脚本安装与运行

安装依赖：

pip install pandas openpyxl

运行方式：

1. 命令行运行：

   python extract_fields.py "C:\路径\全品类库存-调整后.xlsx"
   

2. Python交互式运行：

   from extract_fields import extract_excel_fields
   output = extract_excel_fields("C:\\路径\\全品类库存.xlsx")
   

3. 批量处理：

   import glob
   for file in glob.glob("C:\\路径\\*.xlsx"):
       extract_excel_fields(file)
   

---

四、在WorkBuddy中使用

4.1 快速调用命令

当需要提取Excel字段时，可以直接告诉WorkBuddy：

"按上次的字段清单提取" 或 "提取供应链库存字段"

4.2 完整使用流程

1. 提供Excel文件：通过@file语法附加Excel文件
2. 指定指令："按上次的字段清单提取"
3. 等待处理：WorkBuddy自动读取文件、提取字段
4. 获取结果：下载生成的_指定字段摘取.xlsx文件

4.3 示例对话

用户：@"C:/路径/全品类库存.xlsx" 按上次的字段清单提取

WorkBuddy：
✅ 已按照字段清单提取数据
📊 结果：650行 × 14列
💾 文件：全品类库存_指定字段摘取.xlsx
📎 附件已交付

---

五、自定义与扩展

5.1 修改字段清单

编辑extract_fields.py中的columns_to_extract列表：

columns_to_extract = [
    # 添加或删除字段
    'SAP代码',
    '新增字段名',
    # ...
]

5.2 处理多个工作表

如需处理多个工作表，修改函数参数：

def extract_multiple_sheets(input_file, sheet_names=['整体库存', '其他表']):
    for sheet in sheet_names:
        extract_excel_fields(input_file, sheet_name=sheet)

5.3 添加数据清洗

在提取后添加数据处理逻辑：

# 去除空值行
df_extracted = df_extracted.dropna(subset=['SAP代码'])

# 数据类型转换
df_extracted['SAP代码'] = df_extracted['SAP代码'].astype(str)

# 添加计算列
df_extracted['库存总量'] = df_extracted['京东库存'] + df_extracted['股份库存']

---

六、错误处理与调试

6.1 常见错误及解决

| 错误现象 | 可能原因 | 解决方案 | |----------|----------|----------| | "文件不存在" | 路径错误或文件被占用 | 检查文件路径，确保文件已关闭 | | "工作表不存在" | 工作表名称不匹配 | 确认工作表名为"整体库存"（区分大小写） | | "字段不存在" | 列名不匹配（前导/后导空格） | 使用repr()查看精确列名，调整字段列表 | | 内存不足 | 文件过大 | 分块读取：pd.read_excel(..., chunksize=1000) | | 编码错误 | 文件包含特殊字符 | 指定编码：pd.read_excel(..., engine='openpyxl') |

6.2 调试步骤

1. 查看列名：

   import pandas as pd
   df = pd.read_excel('文件.xlsx', nrows=1)
   for col in df.columns:
       print(repr(col))
   

2. 验证字段存在性：

   required = ['SAP代码', '京东库存']
   missing = [col for col in required if col not in df.columns]
   if missing:
       print(f"缺失字段: {missing}")
   

3. 测试小样本：

   # 只读取前100行测试
   df_sample = pd.read_excel('文件.xlsx', nrows=100)
   

---

七、自动化与集成

7.1 WorkBuddy自动化任务

创建定期提取任务：

# 自动化配置
schedule: "daily 09:00"
task: "提取供应链日报数据"
inputs:
  - "C:\日报\供应链全品类.xlsx"
outputs:
  - "C:\日报\提取结果\供应链日报_提取.xlsx"

7.2 与数据管道集成

# 数据管道示例
def data_pipeline():
    # 1. 提取字段
    raw_file = extract_excel_fields("原始数据.xlsx")
    
    # 2. 数据转换
    df = pd.read_excel(raw_file)
    df_processed = transform_data(df)
    
    # 3. 加载到数据库
    load_to_database(df_processed)
    
    # 4. 生成报告
    generate_report(df_processed)

7.3 版本控制

• 将字段清单保存为配置文件（JSON/YAML）
• 记录提取历史和时间戳
• 支持回滚到之前的字段版本

---

八、性能优化

8.1 大文件处理

# 分块读取（适用于超过10万行）
chunk_size = 10000
chunks = pd.read_excel('大文件.xlsx', chunksize=chunk_size)
results = []
for chunk in chunks:
    extracted = chunk[columns_to_extract]
    results.append(extracted)
df_final = pd.concat(results, ignore_index=True)

8.2 内存优化

# 只读取需要的列
usecols = columns_to_extract + ['必要辅助列']
df = pd.read_excel('文件.xlsx', usecols=usecols)

8.3 并行处理

from concurrent.futures import ThreadPoolExecutor

def process_multiple_files(file_list):
    with ThreadPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(extract_excel_fields, file_list))
    return results

---

九、技能依赖

必需工具

• Python 3.8+：运行提取脚本
• pandas：数据处理核心库
• openpyxl：Excel文件读写
• WorkBuddy：自动化执行环境

可选工具

• PyArrow：加速大文件读取
• DuckDB：内存数据库，用于复杂查询
• Streamlit：构建数据提取Web界面

---

十、质量保证

10.1 验证规则

• 提取后行数应与原始数据一致
• 关键字段（SAP代码）不应有空值
• 数值字段应进行范围检查（如库存不为负）
• 输出文件格式应符合Excel标准

10.2 测试用例

def test_extraction():
    # 测试正常文件
    test_file = "测试数据.xlsx"
    output = extract_excel_fields(test_file)
    assert os.path.exists(output)
    
    # 验证字段完整性
    df = pd.read_excel(output)
    assert 'SAP代码' in df.columns
    assert len(df.columns) == 14
    
    print("✅ 所有测试通过")

10.3 监控指标

• 提取成功率（%）
• 平均处理时间（秒）
• 数据质量评分
• 错误类型分布

---

十一、交付物规范

11.1 文件命名

{原文件名}_指定字段摘取.xlsx

示例：全品类库存-调整后_指定字段摘取.xlsx

11.2 文件结构

输出文件/
├── 数据表（14个指定字段）
├── 元数据（可选）
│   ├── 提取时间
│   ├── 字段清单版本
│   └── 数据统计
└── 质量报告（可选）

11.3 文档要求

• 每次提取应记录字段清单版本
• 重大字段变更需更新技能文档
• 保留历史配置便于追溯

---

十二、更新记录

| 版本 | 日期 | 变更说明 | |------|------|----------| | v1.0 | 2026-05-21 | 初始版本，基于历史工作记录创建 | | | | 固化14个核心字段清单 | | | | 添加完整脚本和文档 |

---

技能来源：基于鱼跃品牌供应链数据分析工作历史记录 适用场景：医疗器械、电商供应链、库存管理 数据安全：仅提取指定字段，不包含原始完整数据 维护责任：数据团队定期更新字段清单