DataOps-platform/tasks/task_execute_instructions.md

Cursor 自动任务执行指令

重要：请立即执行以下任务！

生成时间: 2026-01-16 14:24:07

待执行任务数量: 2

任务完成后的操作

完成每个任务后，请更新 tasks/pending_tasks.json 中对应任务的 status 为 completed， 并填写 code_name（代码文件名）和 code_path（代码路径）。

调度脚本会自动将完成的任务同步到数据库。

任务约束要求

重要约束：完成脚本创建后，不需要生成任务总结文件。

• 不要创建任何 summary、report、总结类的文档文件
• 不要生成 task_summary.md、execution_report.md 等总结文件
• 只需创建任务要求的功能脚本文件
• 只需更新 tasks/pending_tasks.json 中的任务状态

---

任务 1: 产品库存表原始数据导入

• 任务ID: 37
• 创建时间: 2026-01-16 12:41:09
• 创建者: cursor

任务描述

Task: 产品库存表原始数据导入

Source Tables

product_inventory_table_raw_data

DDL

CREATE TABLE product_inventory_table_raw_data (
    updated_at timestamp COMMENT '更新时间',
    created_at timestamp COMMENT '创建时间',
    is_active boolean COMMENT '是否激活',
    turnover_rate numeric(5, 2) COMMENT '周转率',
    outbound_quantity_30d integer COMMENT '30天出库数量',
    inbound_quantity_30d integer COMMENT '30天入库数量',
    last_outbound_date date COMMENT '最近出库日期',
    last_inbound_date date COMMENT '最近入库日期',
    stock_status varchar(50) COMMENT '库存状态',
    selling_price numeric(10, 2) COMMENT '销售价格',
    unit_cost numeric(10, 2) COMMENT '单位成本',
    max_stock integer COMMENT '最大库存',
    safety_stock integer COMMENT '安全库存',
    current_stock integer COMMENT '当前库存',
    warehouse varchar(100) COMMENT '仓库',
    supplier varchar(200) COMMENT '供应商',
    brand varchar(100) COMMENT '品牌',
    category varchar(100) COMMENT '类别',
    product_name varchar(200) COMMENT '产品名称',
    sku varchar(50) COMMENT 'SKU',
    id serial COMMENT '主键'
);
COMMENT ON TABLE product_inventory_table_raw_data IS '产品库存表-原始数据';

Target Tables

test_product_inventory

Data Source

• Type: postgresql
• Host: 192.168.3.143
• Port: 5678
• Database: dataops
• Schema: dags

DDL

CREATE TABLE test_product_inventory (
    updated_at timestamp COMMENT '更新时间',
    created_at timestamp COMMENT '创建时间',
    is_active boolean COMMENT '是否启用',
    turnover_rate numeric(5, 2) COMMENT '周转率',
    outbound_quantity_30d integer COMMENT '30天出库数量',
    inbound_quantity_30d integer COMMENT '30天入库数量',
    last_outbound_date date COMMENT '最近出库日期',
    last_inbound_date date COMMENT '最近入库日期',
    stock_status varchar(50) COMMENT '库存状态',
    selling_price numeric(10, 2) COMMENT '销售价格',
    unit_cost numeric(10, 2) COMMENT '单位成本',
    max_stock integer COMMENT '最大库存',
    safety_stock integer COMMENT '安全库存',
    current_stock integer COMMENT '当前库存',
    warehouse varchar(100) COMMENT '仓库',
    supplier varchar(200) COMMENT '供应商',
    brand varchar(100) COMMENT '品牌',
    category varchar(100) COMMENT '类别',
    product_name varchar(200) COMMENT '产品名称',
    sku varchar(50) COMMENT 'SKU',
    id serial COMMENT 'ID',
    create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '数据创建时间'
);
COMMENT ON TABLE test_product_inventory IS '产品库存表';

Update Mode

• Mode: Append (追加模式)
• Description: 新数据将追加到目标表，不删除现有数据

Request Content

把产品库存表的原始数据导入到数据资源的产品库存表中。

Implementation Steps

1. Extract data from source tables as specified in the DDL
2. Apply transformation logic according to the rule:
   • Rule: 把产品库存表的原始数据导入到数据资源的产品库存表中。
3. Generate Python program to implement the data transformation logic
4. Write transformed data to target table using append mode

---

任务 2: DF_DO202601160001

• 任务ID: 38
• 创建时间: 2026-01-16 12:43:52
• 创建者: system

任务描述

Task: DF_DO202601160001

Source Tables

test_product_inventory

Data Source

• Type: postgresql
• Host: 192.168.3.143
• Port: 5678
• Database: dataops
• Schema: dags

DDL

CREATE TABLE test_product_inventory (
    updated_at timestamp COMMENT '更新时间',
    created_at timestamp COMMENT '创建时间',
    is_active boolean COMMENT '是否启用',
    turnover_rate numeric(5, 2) COMMENT '周转率',
    outbound_quantity_30d integer COMMENT '30天出库数量',
    inbound_quantity_30d integer COMMENT '30天入库数量',
    last_outbound_date date COMMENT '最近出库日期',
    last_inbound_date date COMMENT '最近入库日期',
    stock_status varchar(50) COMMENT '库存状态',
    selling_price numeric(10, 2) COMMENT '销售价格',
    unit_cost numeric(10, 2) COMMENT '单位成本',
    max_stock integer COMMENT '最大库存',
    safety_stock integer COMMENT '安全库存',
    current_stock integer COMMENT '当前库存',
    warehouse varchar(100) COMMENT '仓库',
    supplier varchar(200) COMMENT '供应商',
    brand varchar(100) COMMENT '品牌',
    category varchar(100) COMMENT '类别',
    product_name varchar(200) COMMENT '产品名称',
    sku varchar(50) COMMENT 'SKU',
    id serial COMMENT 'ID'
);
COMMENT ON TABLE test_product_inventory IS '产品库存表';

Target Tables

warehouse_inventory_summary

DDL

CREATE TABLE warehouse_inventory_summary (
    id BIGINT PRIMARY KEY COMMENT '主键ID',
    create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '数据创建时间'
);
COMMENT ON TABLE warehouse_inventory_summary IS '仓库库存汇总表';

Update Mode

• Mode: Full Refresh (全量更新)
• Description: 目标表将被清空后重新写入数据

Request Content

1.从标签为数据资源的产品库存表中提取字段：仓库编号、库存数量；2.按照仓库进行分组，对库存数量进行求和计算；3.无特殊过滤条件；4.最终输出数据格式包含字段：仓库编号、总库存数量

Implementation Steps

1. Extract data from source tables as specified in the DDL
2. Apply transformation logic according to the rule:
   • Rule: 1.从标签为数据资源的产品库存表中提取字段：仓库编号、库存数量；2.按照仓库进行分组，对库存数量进行求和计算；3.无特殊过滤条件；4.最终输出数据格式包含字段：仓库编号、总库存数量
3. Generate Python program to implement the data transformation logic
4. Write transformed data to target table using full mode

---