import sqlite3 import re from pathlib import Path import sys # 把当前目录加入 sys.path sys.path.insert(0, str(Path(__file__).parent)) from config import OUTPUT_DIR DB_PATH = OUTPUT_DIR / 'erp_data.db' class MaterialReconciliationService: def __init__(self): self.db_path = DB_PATH def get_conn(self): return sqlite3.connect(self.db_path) def get_data_period(self): """获取当前数据库中发料记录的时间周期，用于前端展示对账月份""" import datetime import calendar now = datetime.datetime.now() first_day = datetime.date(now.year, now.month, 1).strftime('%Y-%m-%d') last_day = datetime.date(now.year, now.month, calendar.monthrange(now.year, now.month)[1]).strftime('%Y-%m-%d') return {"start": first_day, "end": last_day} def step1_extract_and_match_work_orders(self): """ 第一步：匹配生产工单。 对于有生产工单的，直接赋值。 对于没有生产工单的，从备注中通过正则提取潜在的工单号（如 SFC... 或 连续数字）。 并将其更新到 inferred_production_order_no 字段中。 """ print("[开始] 正在连接数据库并准备清洗工单数据...") conn = self.get_conn() cursor = conn.cursor() # 确保表有 inferred_production_order_no 字段 try: cursor.execute("ALTER TABLE issue_receipt_details ADD COLUMN inferred_production_order_no TEXT;") print("[信息] 已在数据库中新建 inferred_production_order_no 字段。") except sqlite3.OperationalError: pass # 已经存在 print("[执行] 正在读取所有发料单明细...") # 获取所有发料单 cursor.execute("SELECT id, production_order_no, work_orders_remark, detailed_remark, production_order_remark FROM issue_receipt_details") rows = cursor.fetchall() print(f"[执行] 共读取到 {len(rows)} 条发料记录，开始进行正则比对...") # 预先加载 abnormal_report 中的有效工单号作为验证库 cursor.execute("SELECT DISTINCT work_orders_number FROM abnormal_report") valid_sfcs = {r[0].upper() for r in cursor.fetchall() if r[0]} updates = [] matched_count = 0 for row in rows: row_id = row[0] prod_no = row[1] work_remark = row[2] or '' detail_remark = row[3] or '' prod_remark = row[4] or '' inferred_sfc = '' # 如果系统本身已经有工单号 if prod_no and prod_no.strip(): inferred_sfc = prod_no.strip() else: # 合并所有备注信息 combined_remarks = f"{work_remark} {detail_remark} {prod_remark}" if combined_remarks.strip(): # 优先匹配完整的 SFC 格式 (例如 SFC018845) sfc_match = re.search(r'(SFC\d+)', combined_remarks, re.IGNORECASE) if sfc_match: inferred_sfc = sfc_match.group(1).upper() else: # 尝试匹配纯数字格式 (长度大于等于4位的连续数字) num_match = re.search(r'(\d{4,})', combined_remarks) if num_match: candidate_num = num_match.group(1) # 尝试拼凑 SFC 前缀并校验是否存在于有效库中 candidate_sfc = f"SFC0{candidate_num}" if candidate_sfc in valid_sfcs: inferred_sfc = candidate_sfc else: candidate_sfc2 = f"SFC{candidate_num}" if candidate_sfc2 in valid_sfcs: inferred_sfc = candidate_sfc2 else: inferred_sfc = candidate_sfc # 哪怕没在库里也提取出来，作为参考 if inferred_sfc: matched_count += 1 updates.append((inferred_sfc, row_id)) print(f"[信息] 匹配完成。有 {matched_count} 条数据成功匹配到工单号。正在将结果写回数据库...") # 批量更新数据库 cursor.executemany( "UPDATE issue_receipt_details SET inferred_production_order_no = ? WHERE id = ?", updates ) conn.commit() conn.close() print(f"[完成] 数据库更新完毕！本次清洗共处理了 {len(updates)} 条数据。") return len(updates) def step2_get_unmatched_materials(self, start_date=None, end_date=None): """ 第二步：整理出没有生产工单的物料明细表 """ conn = self.get_conn() conn.row_factory = sqlite3.Row cursor = conn.cursor() where_clause = "WHERE (inferred_production_order_no = '' OR inferred_production_order_no IS NULL)" params = [] if start_date and end_date: where_clause += " AND execution_time >= ? AND execution_time <= ?" params.extend([f"{start_date} 00:00:00", f"{end_date} 23:59:59"]) sql = f""" SELECT work_orders_number as issue_receipt_no, material_code, material_name, material_specification, issue_number, work_orders_remark, detailed_remark, production_order_remark, execution_time, executor_user_name, warehouse_name FROM issue_receipt_details {where_clause} ORDER BY execution_time DESC """ cursor.execute(sql, params) rows = cursor.fetchall() conn.close() return [dict(r) for r in rows] def step3_bom_reconciliation(self, start_date=None, end_date=None): """ 第三步：BOM 校准物料，比对发料与BOM差异 返回按工单分组的差异列表。 """ conn = self.get_conn() conn.row_factory = sqlite3.Row cursor = conn.cursor() where_actual = "WHERE inferred_production_order_no != '' AND inferred_production_order_no IS NOT NULL" params_actual = [] if start_date and end_date: where_actual += " AND execution_time >= ? AND execution_time <= ?" params_actual.extend([f"{start_date} 00:00:00", f"{end_date} 23:59:59"]) sql_actual = f""" SELECT inferred_production_order_no as sfc, material_code, material_name, SUM(issue_number) as total_actual_issue FROM issue_receipt_details {where_actual} GROUP BY inferred_production_order_no, material_code """ where_bom = "" params_bom = [] if start_date and end_date: where_bom = "WHERE order_date >= ? AND order_date <= ?" params_bom.extend([f"{start_date} 00:00:00", f"{end_date} 23:59:59"]) sql_bom = f""" SELECT work_orders_number as sfc, material_code, material_name, theoretical_issue_qty as bom_qty FROM abnormal_report {where_bom} """ cursor.execute(sql_actual, params_actual) actuals = cursor.fetchall() cursor.execute(sql_bom, params_bom) boms = cursor.fetchall() conn.close() reconciliation = {} for r in boms: key = (r['sfc'], r['material_code']) reconciliation[key] = { 'sfc': r['sfc'], 'material_code': r['material_code'], 'material_name': r['material_name'], 'bom_qty': r['bom_qty'] or 0, 'actual_qty': 0, 'diff_qty': 0 - (r['bom_qty'] or 0), 'status': '未发料' } for r in actuals: key = (r['sfc'], r['material_code']) if key not in reconciliation: reconciliation[key] = { 'sfc': r['sfc'], 'material_code': r['material_code'], 'material_name': r['material_name'], 'bom_qty': 0, 'actual_qty': 0, 'diff_qty': 0, 'status': 'BOM外发料' } reconciliation[key]['actual_qty'] += r['total_actual_issue'] reconciliation[key]['diff_qty'] = round(reconciliation[key]['actual_qty'] - reconciliation[key]['bom_qty'], 4) if reconciliation[key]['diff_qty'] > 0 and reconciliation[key]['bom_qty'] > 0: reconciliation[key]['status'] = '超领发料' elif reconciliation[key]['diff_qty'] < 0: reconciliation[key]['status'] = '少领发料' elif reconciliation[key]['diff_qty'] == 0 and reconciliation[key]['bom_qty'] > 0: reconciliation[key]['status'] = '发料正常' return list(reconciliation.values()) if __name__ == '__main__': service = MaterialReconciliationService() print("Executing Step 1: Matching work orders...") updated = service.step1_extract_and_match_work_orders() print(f"Updated {updated} issue receipts with inferred production orders.") unmatched = service.step2_get_unmatched_materials() print(f"Found {len(unmatched)} unmatched materials.") recon = service.step3_bom_reconciliation() abnormal = [r for r in recon if r['status'] in ['超领发料', 'BOM外发料']] print(f"Found {len(abnormal)} abnormal BOM issues.")