mysql树形结构联结之尝试

思路

MYSQL版本（5.7）功能所限，需要设计一个存储函数来并联数据集合。已知：

1. 工单so下有工序，工序中信息是最终要提取的数据来源
2. 工单so含有字段父工单parent_shop_order，可据其追溯母工单
3. 假设’工单so’最多有四层

目标：

1. 根据卷号并联所有的工单so，并以唯一的id记录
2. 查询工序的信息，呈现的效果如下：

树形编号	一层	二层	三层	四层	本层	本层主要属性	本层工序1	本层工序2
26011101	A1	B1	C2	D2	D2	咕噜噜	滴沥沥	呜啦啦
26011102	A1	B1	C1	D2	C1	咕噜噜	滴沥沥	呜啦啦

需要设计的工单链条chain函数组成如下：

1. Parent_chain ：父节点工单号，向上一层追溯
2. Child so：子节点工单号（或许可以根据上下文工单来查询） 完全不需要，因为是分叉的树形，向下找会出现多个结果，数据处理不友好，可通过入库的标记（最终的成品）作为最子节点的特征向上追溯。
3. level：记根节点root so的层数为$0$，取本层的在树形结构的层数 会造成很多中间工单被查出来，结构臃肿

需要设计的工序线process_line函数如下：

1. 能够识别父工单里自己这一分支，
   1. 通过前缀相似来筛选：如11111分叉了11111A与11111B，11111A又分叉了11111AE，从11111AE向上查到11111A与11111AB，然后根据11111A与11111AE的前缀相似来筛选出来这道工序。
   2. 通过批次来追：上工序出料的批次与下工序进料的批次相同，可根据末工序逆向到初工序（这个效率绝对比方法1高）
2. 在追完层级与工序后进行排序，按照工单顺序作为父序，再按照工序排序作为子序，排列出来。
3. 格式化输出：根据关注的特性，来格式化各个工序的输出。整体模块框架先整理，以方便后期数据分析的易抓取性。

最终呈现的内容详细设计如下：

• 基本信息：钢种、销售订单、成品要求、工单、树形编号
• 工序信息：
  1. 工序名称：可以筛选想要呈现的核心工序，如轧机、退火等
  2. 生产结束时间：取年月日即可，按照时间降序
  3. 工序关键工艺：如轧制压下率、用辊需求、进出料厚度、退火温度、TV值等
  4. 其他：如硬度等

实现

父节点追溯函数(parent_chain)

CREATE FUNCTION `mes_prod`.`parent_chain`(so_present VARCHAR(50)) RETURNS varchar(50) CHARSET utf8mb4 COLLATE utf8mb4_bin
    READS SQL DATA
BEGIN
	DECLARE so_parent VARCHAR(50);
    SELECT so.parent_shop_order INTO so_parent
    FROM so 
    WHERE so.shop_order = so_present AND so.parent_shop_order != so_present
	LIMIT 1;
RETURN so_parent;
END

应用：查询本月入库卷的工单溯源信息：

SELECT
srso.sfc 钢卷号,
srso.batch 批次号,
srso.end_date_time_lt 生产结束时间,
srso.inspect_order 入库检验批号,
so.shop_order 终结工单,
-- srs.operation_desc 末工序,
-- srs.plan_next_station 流向,
srso.next_storage_location_desc 实际流向,
parent_chain(so.shop_order) 父工单,
parent_chain(parent_chain(so.shop_order)) 父父工单,
parent_chain(parent_chain(parent_chain(so.shop_order))) 父父父工单,
parent_chain(parent_chain(parent_chain(parent_chain(so.shop_order)))) 父父父父工单,
parent_chain(parent_chain(parent_chain(parent_chain(parent_chain(so.shop_order))))) 父父父父父工单,
parent_chain(parent_chain(parent_chain(parent_chain(parent_chain(parent_chain(so.shop_order)))))) 父父父父父父工单
FROM so
LEFT JOIN so_status_view ssv ON so.id = ssv.main_table_id
LEFT JOIN so_router_step srs ON srs.main_table_id = so.id AND srs.is_last_operation = '1' AND  srs.is_last_operation IS NOT NULL 
LEFT JOIN so_router_step_out srso ON srso.detail_table_id = srs.id 
WHERE ssv.so_status != '在制工单' AND (srso.next_storage_location_desc != '余材仓' AND srso.next_storage_location_desc != '半成品仓') AND srso.end_date_time_lt >= TIMESTAMP(DATE_FORMAT(CURDATE(), '%Y-%m-01'), '00:00:00')
-- AND srso.batch = '2601160012' -- 批次筛选
-- AND srso.sfc ='Y30125122302AE0102' -- 钢卷号筛选
AND parent_chain(so.shop_order) IS NULL -- 单工单筛选
ORDER BY srso.end_date_time_lt -- 按时间正序

工单筛选函数(process_line)

需要设计的工序线process_line函数如下：

1. 能够识别父工单里自己这一分支，
   1. 通过前缀相似来筛选：如11111分叉了11111A与11111B，11111A又分叉了11111AE，从11111AE向上查到11111A与11111AB，然后根据11111A与11111AE的前缀相似来筛选出来这道工序。
   2. 通过批次来追：上工序出料的批次与下工序进料的批次相同，可根据末工序逆向到初工序（这个效率绝对比方法1高）
2. 在追完层级与工序后进行排序，按照工单顺序作为父序，再按照工序排序作为子序，排列出来。
3. 格式化输出：根据关注的特性，来格式化各个工序的输出。整体模块框架先整理，以方便后期数据分析的易抓取性。

---

逻辑过于复杂，借助mysql很难实现，而且不利于后面分析，不如放弃，然后用python来实现报表？！（2026/1/25）

---

新思路

pymysql中用到了游标，这一章当时忽略了，开始补课。其实很简单，就是一个搬运小车，用它从sql里把数据搬出来，然后python把小车里的数据拿出来分析。

上面的表实际上是一个关系，表示了不同工单之间的连接关系，全部查询下来需要一分钟左右，可以定时执行将其放置到一个实表中。

着手利用python直接把结果输出到excel表格里，变相实现可视化。

1. 按日增量抓取入库物料，存放到文件中，一天一个，每月汇总成一个大的工作表（Sheet）
2. 根据父工单函数寻找关联关系，生成溯源码
3. 并联各个工序的数据、质量性能、销售订单数据
4. 输出为报表。

$$MySQL\xrightarrow[提取]{PyMySQL}Python\xrightarrow[输出]{openpyxl/xlwings/pandas} Excel$$

新实现

基本信息

最近两天入库的物料

SELECT 
o.zt210 钢卷号,
-- 宽度、厚度、重量、钢种等基本信息
qil.inspect_lot 检验批号,  -- 连接材料性能视图properties，桥接字段为`检验批号`
o.zt180 销售订单行,  --  连接成品要求视图requirement，桥接字段为`销售订单行`
LEFT(p.zmesgd, 12) 工单号,  -- 连接工序信息，桥接字段为末工单号
FROM mes_prod.material_out_data o
LEFT JOIN mes_prod.post p ON p.id = o.main_table_id
LEFT JOIN mes_prod.sl ON sl.storage_location = o.lgort
LEFT JOIN wms_prod.inspect_lot_source  ils ON ils.batch_number =  o.batch
LEFT JOIN wms_prod.qm_inspect_lot qil ON qil.id =  ils.main_table_id
WHERE LOCATE('213', o.matnr, 1) AND o.zt10 > DATE_SUB(CURDATE(), INTERVAL 2 DAY)

材料性能（properties）

副本任务：品质表过于细碎，需要整理成视图。经过艰苦卓绝的努力，终于建成视图properties，使用了MySQL的Json语法，速度比join提高了好几倍。可以显示主要工艺信息

检验批号	决策时间	工序硬度	硬度1	硬度2	硬度3	硬度4	硬度均值	抗拉强度_MPa	屈服	延伸率	晶粒度	中性盐雾检验结果	电阻率	磁导率	易锡焊

成品要求（requirement）

将销售订单与航班号连接后，生成视图requirement

客户代码	销售订单行	硬度区间	硬度测试力	抗拉强度	屈服强度	延伸率	晶粒度	电阻率	磁导率	表面	粗糙度	光泽度	锡焊性

工序信息

选取最近两天入库的工单号，末工序分卷不影响整体工序信息，故可将其分离开来，先抓取工单的顺序，后面再拼接起来（通过钢卷的工单号与工单顺序中的末工单号一致来拼接）。

六层连接测试

import pymysql
import pandas as pd

# 数据库配置
db_config = {
***
}

# 连接到数据库
conn = pymysql.connect(**db_config)
cursor = conn.cursor(pymysql.cursors.DictCursor)

# 获取源工单（末工单）
cursor.execute("""
SELECT LEFT(p.zmesgd, 12) as work_order
FROM mes_prod.material_out_data o
LEFT JOIN mes_prod.post p ON p.id = o.main_table_id
WHERE LOCATE('213', o.matnr, 1) -- AND  o.zt10 > DATE_SUB(CURDATE(), INTERVAL 2 DAY)
GROUP BY work_order
HAVING work_order = 'S20251113013'
LIMIT 500
""")

source_orders = [row['work_order'] for row in cursor.fetchall()]
all_data = []
work_order_hierarchy = {}

# 处理每个源工单
for source_order in source_orders:
    # 构建工单层级链
    chain = [source_order]
    current_order = source_order
    
    while True:
        cursor.execute("SELECT parent_chain(shop_order) as parent FROM so WHERE shop_order = %s", (current_order,))
        parent_result = cursor.fetchone()
        
        if not parent_result or not parent_result['parent'] or parent_result['parent'] == 'NULL':
            break
            
        chain.append(parent_result['parent'])
        current_order = parent_result['parent']
        
        if len(chain) > 10:
            break
    
    # 分配层级（最顶层为1）
    for i, work_order in enumerate(reversed(chain)):
        level = i + 1
        is_last_order = (work_order == source_order)
        
        # 查询工序
        cursor.execute("""
            SELECT step.sequence, step.operation_desc 
            FROM so 
            LEFT JOIN so_router_step step ON step.main_table_id = so.id 
            WHERE so.shop_order = %s AND step.status!= 'NEW' AND step.status!='REL'
            ORDER BY step.sequence
        """, (work_order,))
        
        processes = cursor.fetchall()
        
        for process in processes:
            converted_seq = f"{int(process['sequence']) // 10:02d}"
            hierarchy_code = f"{level}{converted_seq}"
            
            all_data.append({
                '源工单': source_order,
                '当前工单': work_order,
                '层级': level,
                '是否末工单': '是' if is_last_order else '否',
                '工序描述': process['operation_desc'],
                '层级编码': hierarchy_code,
                '编码数值': int(hierarchy_code)
            })
        
        # 存储层级信息
        if source_order not in work_order_hierarchy:
            work_order_hierarchy[source_order] = []
        work_order_hierarchy[source_order].append({
            '工单号': work_order,
            '层级': level,
            '是否末工单': is_last_order
        })

# 创建主DataFrame
df = pd.DataFrame(all_data)
df = df.sort_values('编码数值').reset_index(drop=True)

# 创建层级信息DataFrame
hierarchy_data = []
for source_order, chain_info in work_order_hierarchy.items():
    for item in sorted(chain_info, key=lambda x: x['层级']):
        hierarchy_data.append({
            '源工单': source_order,
            '工单号': item['工单号'],
            '层级': item['层级'],
            '是否末工单': '是' if item['是否末工单'] else '否'
        })

hierarchy_df = pd.DataFrame(hierarchy_data)

# 关闭连接
cursor.close()
conn.close()

# 输出统计信息
print(f"处理完成！共{len(df['源工单'].unique())}个工单链")
print(f"总记录数: {len(df)}")
print(f"层级分布: {hierarchy_df['层级'].value_counts().to_dict()}")

最终程序

...

做出后，发现