项目结构# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 20:08 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : app_config.py 全局应用配置 import os class AppConfig: # 日志配置 LOG_SAVE_PATH ./logs LOG_FILE_NAME jewelry_packer.log LOG_CONSOLE_ENABLE True LOG_FILE_ENABLE True # 并发设置 MAX_WORKER_THREAD 8 # 配载搜索最大状态数防止极端情况无限搜索 MAX_SEARCH_STATE 15000 staticmethod def init_log_dir(): 初始化日志文件夹 :return: if not os.path.exists(AppConfig.LOG_SAVE_PATH): os.makedirs(AppConfig.LOG_SAVE_PATH, exist_okTrue) # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 20:08 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : enums.py 系统枚举定义 from enum import Enum class LoadStrategyType(Enum): 配载算法策略类型 BFS BFS # 最优方案最少包装数量主推 DFS DFS # 深度优先可行方案不一定最优 ASTAR ASTAR # A*启发搜索 class ResultCode(Enum): 业务返回码 SUCCESS 0, 成功 PARAM_ERROR 400, 参数非法 NO_AVAILABLE_PACKAGE 401, 无可用包装规格 SEARCH_OVER_LIMIT 501, 搜索状态超限未找到方案 NO_ANY_LOAD_PLAN 502, 无法生成配载方案 # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 20:08 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : models.py 数据实体模型 DTO from dataclasses import dataclass from typing import Dict, List, Optional dataclass class PackageSpec: 包装规格实体 spec_code: str # 包装编码BOX-500 volume: float # 包装容积 cm³ stock_qty: int # 当前可用库存数量 dataclass class PackQueryCondition: 配载查询入参 total_material_volume: float # 首饰配件总体积 cm³ package_spec_list: List[PackageSpec] # 可用包装规格列表 strategy_type: str # 算法策略 allow_over_load: bool False # 是否允许包装内超容默认false dataclass class SingleUsePackage: 方案内使用的包装记录 spec_code: str volume: float use_count: int dataclass class PackResult: 配载最终输出结果 code: int msg: str use_package_list: List[SingleUsePackage] # 本次选用包装清单 remain_material_volume: float # 剩余未装入物料体积 remain_package_stock: Dict[str, int] # 各规格剩余库存 {spec_code:剩余数量} total_use_package_count: int # 一共使用多少个包装 search_state_count: int # 算法搜索遍历状态数量 full_load_flag: bool # True物料全部装完无剩余 # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 20:09 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : logger.py 统一日志组件 满足控制台与日志文件输出格式完全一致线程安全 import logging import sys from BreadthFirst.config.app_config import AppConfig import os import threading from typing import Optional from logging.handlers import RotatingFileHandler class AppLogger: _instance: Optional[logging.Logger] None # 【修复】使用线程锁而非原生object _lock threading.Lock() classmethod def get_logger(cls) - logging.Logger: 单例获取日志实例线程安全 :return: if cls._instance is None: with cls._lock: if cls._instance is None: cls._init_logger() return cls._instance classmethod def _init_logger(cls): :return: AppConfig.init_log_dir() logger logging.getLogger(JewelryPacker) logger.setLevel(logging.INFO) logger.propagate False format_str %(asctime)s [%(levelname)s] %(message)s formatter logging.Formatter(format_str, datefmt%Y-%m-%d %H:%M:%S) # 控制台输出 if AppConfig.LOG_CONSOLE_ENABLE: console_handler logging.StreamHandler(streamsys.stdout) console_handler.setFormatter(formatter) logger.addHandler(console_handler) # 文件滚动日志 if AppConfig.LOG_FILE_ENABLE: log_path os.path.join(AppConfig.LOG_SAVE_PATH, AppConfig.LOG_FILE_NAME) file_handler RotatingFileHandler( filenamelog_path, maxBytes10 * 1024 * 1024, backupCount5, encodingutf-8 ) file_handler.setFormatter(formatter) logger.addHandler(file_handler) cls._instance logger # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 20:10 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : strategy_base.py 配载策略抽象基类 所有算法统一实现该接口方便无缝切换策略符合开闭原则 from abc import ABC, abstractmethod from BreadthFirst.core.models import PackQueryCondition, PackResult class BasePackStrategy(ABC): 包装配载策略抽象父类 abstractmethod def execute(self, condition: PackQueryCondition) - PackResult: 执行配载计算 :param condition: 查询条件 :return: 配载结果对象 pass # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 21:01 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : strategy_bfs.py BFS配载最优策略 目标使用最少数量礼盒装载物料等价分水问题最短路径搜索 from collections import deque from typing import Tuple, Dict from BreadthFirst.core.strategy_base import BasePackStrategy from BreadthFirst.core.models import PackQueryCondition, PackResult, SingleUsePackage from BreadthFirst.core.enums import ResultCode from BreadthFirst.config.app_config import AppConfig from BreadthFirst.core.logger import AppLogger logger AppLogger.get_logger() class BFSPackStrategy(BasePackStrategy): def execute(self, condition: PackQueryCondition) - PackResult: BFS搜索最优包装组合方案 total_mat_vol condition.total_material_volume spec_list condition.package_spec_list allow_over condition.allow_over_load max_state AppConfig.MAX_SEARCH_STATE # 1. 参数校验 if total_mat_vol 0: return PackResult( codeResultCode.PARAM_ERROR.value[0], msg物料总体积必须大于0, use_package_list[], remain_material_volumetotal_mat_vol, remain_package_stock{}, total_use_package_count0, search_state_count0, full_load_flagFalse ) if len(spec_list) 0: return PackResult( codeResultCode.NO_AVAILABLE_PACKAGE.value[0], msgResultCode.NO_AVAILABLE_PACKAGE.value[1], use_package_list[], remain_material_volumetotal_mat_vol, remain_package_stock{}, total_use_package_count0, search_state_count0, full_load_flagFalse ) # 初始化包装库存字典 init_stock: Dict[str, int] {} spec_volume_map: Dict[str, float] {} for item in spec_list: init_stock[item.spec_code] item.stock_qty spec_volume_map[item.spec_code] item.volume # 状态定义(当前已装载体积, 当前各包装消耗字典) start_load_vol 0.0 start_used: Dict[str, int] {code: 0 for code in init_stock.keys()} visited set() visited_count 1 # 队列元素已装载体积已使用包装计数字典 queue deque() queue.append((start_load_vol, start_used)) visited.add((start_load_vol, tuple(start_used.values()))) best_solution None while queue: load_vol, used_dict queue.popleft() # 判断当前方案是否满足装载需求 if load_vol total_mat_vol: best_solution (load_vol, used_dict) break if visited_count max_state: logger.warning(f搜索达到最大状态限制{max_state}终止搜索) break # 遍历所有包装规格尝试放入一个礼盒 for spec_code, stock_num in init_stock.items(): used_num used_dict[spec_code] if used_num stock_num: continue # 该规格库存耗尽不能继续选 pack_vol spec_volume_map[spec_code] new_load_vol load_vol pack_vol # 不允许超载并且新容积超出则跳过 if not allow_over and new_load_vol total_mat_vol: continue new_used used_dict.copy() new_used[spec_code] 1 state_key (new_load_vol, tuple(new_used.values())) if state_key not in visited: visited.add(state_key) visited_count 1 queue.append((new_load_vol, new_used)) # 2. 方案结果组装 if best_solution is None: return PackResult( codeResultCode.NO_ANY_LOAD_PLAN.value[0], msg未找到可行包装组合方案, use_package_list[], remain_material_volumetotal_mat_vol, remain_package_stockinit_stock.copy(), total_use_package_count0, search_state_countvisited_count, full_load_flagFalse ) final_load_vol, final_used best_solution remain_material round(total_mat_vol - final_load_vol, 2) full_flag remain_material 0.001 use_package_list [] remain_stock {} for spec_code, total_stock in init_stock.items(): use_cnt final_used[spec_code] remain_cnt total_stock - use_cnt remain_stock[spec_code] remain_cnt if use_cnt 0: use_package_list.append(SingleUsePackage( spec_codespec_code, volumespec_volume_map[spec_code], use_countuse_cnt )) total_use sum([x.use_count for x in use_package_list]) return PackResult( codeResultCode.SUCCESS.value[0], msg配载方案计算成功, use_package_listuse_package_list, remain_material_volumeabs(remain_material), remain_package_stockremain_stock, total_use_package_counttotal_use, search_state_countvisited_count, full_load_flagfull_flag ) # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 21:02 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : strategy_dfs.py DFS配载策略备选 不保证最少包装数量仅作方案对比 from BreadthFirst.core.strategy_base import BasePackStrategy from BreadthFirst.core.models import PackQueryCondition, PackResult, SingleUsePackage from BreadthFirst.core.enums import ResultCode from BreadthFirst.config.app_config import AppConfig from BreadthFirst.core.logger import AppLogger logger AppLogger.get_logger() class DFSPackStrategy(BasePackStrategy): def execute(self, condition: PackQueryCondition) - PackResult: :param condition: :return: total_mat_vol condition.total_material_volume spec_list condition.package_spec_list allow_over condition.allow_over_load max_state AppConfig.MAX_SEARCH_STATE if total_mat_vol 0: return PackResult( codeResultCode.PARAM_ERROR.value[0], msg物料总体积必须大于0, use_package_list[], remain_material_volumetotal_mat_vol, remain_package_stock{}, total_use_package_count0, search_state_count0, full_load_flagFalse ) if len(spec_list) 0: return PackResult( codeResultCode.NO_AVAILABLE_PACKAGE.value[0], msgResultCode.NO_AVAILABLE_PACKAGE.value[1], use_package_list[], remain_material_volumetotal_mat_vol, remain_package_stock{}, total_use_package_count0, search_state_count0, full_load_flagFalse ) init_stock {} spec_volume_map {} for item in spec_list: init_stock[item.spec_code] item.stock_qty spec_volume_map[item.spec_code] item.volume visited set() visited_count 1 best_solution None def dfs(load_vol, used_dict): nonlocal visited_count, best_solution if best_solution is not None: return True if load_vol total_mat_vol: best_solution (load_vol, used_dict.copy()) return True state_key (load_vol, tuple(used_dict.values())) if state_key in visited: return False if visited_count max_state: return False visited.add(state_key) visited_count 1 for spec_code, stock_num in init_stock.items(): used_num used_dict[spec_code] if used_num stock_num: continue pack_v spec_volume_map[spec_code] new_load load_vol pack_v if not allow_over and new_load total_mat_vol: continue used_dict[spec_code] 1 if dfs(new_load, used_dict): return True used_dict[spec_code] - 1 return False start_used {k:0 for k in init_stock.keys()} dfs(0.0, start_used) if best_solution is None: return PackResult( codeResultCode.NO_ANY_LOAD_PLAN.value[0], msg未找到可行包装组合方案, use_package_list[], remain_material_volumetotal_mat_vol, remain_package_stockinit_stock.copy(), total_use_package_count0, search_state_countvisited_count, full_load_flagFalse ) final_load_vol, final_used best_solution remain_material round(total_mat_vol - final_load_vol, 2) full_flag remain_material 0.001 use_package_list [] remain_stock {} for spec_code, total_stock in init_stock.items(): use_cnt final_used[spec_code] remain_cnt total_stock - use_cnt remain_stock[spec_code] remain_cnt if use_cnt 0: use_package_list.append(SingleUsePackage( spec_codespec_code, volumespec_volume_map[spec_code], use_countuse_cnt )) total_use sum([x.use_count for x in use_package_list]) return PackResult( codeResultCode.SUCCESS.value[0], msg配载方案计算成功, use_package_listuse_package_list, remain_material_volumeabs(remain_material), remain_package_stockremain_stock, total_use_package_counttotal_use, search_state_countvisited_count, full_load_flagfull_flag ) # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 21:27 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : strategy_astar.py A*启发式配载策略 import heapq from BreadthFirst.core.strategy_base import BasePackStrategy from BreadthFirst.core.models import PackQueryCondition, PackResult, SingleUsePackage from BreadthFirst.core.enums import ResultCode from BreadthFirst.config.app_config import AppConfig from BreadthFirst.core.logger import AppLogger logger AppLogger.get_logger() class AStarPackStrategy(BasePackStrategy): def execute(self, condition: PackQueryCondition) - PackResult: :param condition: :return: total_mat_vol condition.total_material_volume spec_list condition.package_spec_list allow_over condition.allow_over_load max_state AppConfig.MAX_SEARCH_STATE if total_mat_vol 0: return PackResult( codeResultCode.PARAM_ERROR.value[0], msg物料总体积必须大于0, use_package_list[], remain_material_volumetotal_mat_vol, remain_package_stock{}, total_use_package_count0, search_state_count0, full_load_flagFalse ) if len(spec_list) 0: return PackResult( codeResultCode.NO_AVAILABLE_PACKAGE.value[0], msgResultCode.NO_AVAILABLE_PACKAGE.value[1], use_package_list[], remain_material_volumetotal_mat_vol, remain_package_stock{}, total_use_package_count0, search_state_count0, full_load_flagFalse ) init_stock {} spec_volume_map {} for item in spec_list: init_stock[item.spec_code] item.stock_qty spec_volume_map[item.spec_code] item.volume visited set() visited_count 1 heap [] start_used {k:0 for k in init_stock.keys()} # f代价g已使用包装数量已装载体积使用记录 heapq.heappush(heap, (0, 0, 0.0, start_used)) best_solution None while heap: f, g, load_vol, used_dict heapq.heappop(heap) state_key (load_vol, tuple(used_dict.values())) if state_key in visited: continue visited.add(state_key) visited_count 1 if load_vol total_mat_vol: best_solution (load_vol, used_dict) break if visited_count max_state: logger.warning(fA*搜索达到最大状态限制{max_state}) break for spec_code, stock_num in init_stock.items(): used_num used_dict[spec_code] if used_num stock_num: continue pv spec_volume_map[spec_code] new_load load_vol pv if not allow_over and new_load total_mat_vol: continue new_g g 1 # 启发函数剩余体积 h max(0, total_mat_vol - new_load) new_f new_g h new_used used_dict.copy() new_used[spec_code] 1 heapq.heappush(heap, (new_f, new_g, new_load, new_used)) if best_solution is None: return PackResult( codeResultCode.NO_ANY_LOAD_PLAN.value[0], msg未找到可行包装组合方案, use_package_list[], remain_material_volumetotal_mat_vol, remain_package_stockinit_stock.copy(), total_use_package_count0, search_state_countvisited_count, full_load_flagFalse ) final_load_vol, final_used best_solution remain_material round(total_mat_vol - final_load_vol, 2) full_flag remain_material 0.001 use_package_list [] remain_stock {} for spec_code, total_stock in init_stock.items(): use_cnt final_used[spec_code] remain_cnt total_stock - use_cnt remain_stock[spec_code] remain_cnt if use_cnt 0: use_package_list.append(SingleUsePackage( spec_codespec_code, volumespec_volume_map[spec_code], use_countuse_cnt )) total_use sum([x.use_count for x in use_package_list]) return PackResult( codeResultCode.SUCCESS.value[0], msg配载方案计算成功, use_package_listuse_package_list, remain_material_volumeabs(remain_material), remain_package_stockremain_stock, total_use_package_counttotal_use, search_state_countvisited_count, full_load_flagfull_flag ) # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 21:28 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : thread_safe_pool.py 线程安全执行池支持高并发调用配载算法 解决多线程竞争、资源复用 import concurrent.futures from threading import Lock from typing import List from BreadthFirst.core.models import PackQueryCondition, PackResult from BreadthFirst.core.strategy_base import BasePackStrategy from BreadthFirst.config.app_config import AppConfig from BreadthFirst.core.logger import AppLogger logger AppLogger.get_logger() class SafePackThreadPool: _instance None _lock Lock() def __init__(self): self.executor concurrent.futures.ThreadPoolExecutor(max_workersAppConfig.MAX_WORKER_THREAD) classmethod def get_instance(cls): 单例线程池 :return: if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance SafePackThreadPool() return cls._instance def submit_task(self, strategy: BasePackStrategy, condition: PackQueryCondition) - PackResult: 提交单次配载任务同步等待结果 :param strategy: :param condition: :return: future self.executor.submit(strategy.execute, condition) try: result future.result(timeout30) return result except Exception as e: logger.error(f配载任务执行异常: {str(e)}) raise e def batch_submit(self, strategy: BasePackStrategy, condition_list: List[PackQueryCondition]) - List[PackResult]: 批量并发执行多条配载请求 :param strategy: :param condition_list: :return: task_futures [] for cond in condition_list: task_futures.append(self.executor.submit(strategy.execute, cond)) result_list [] for ft in concurrent.futures.as_completed(task_futures): try: res ft.result(timeout30) result_list.append(res) except Exception as e: logger.error(f批量任务异常 {str(e)}) return result_list def shutdown(self): 释放线程池程序退出调用 :return: self.executor.shutdown(waitTrue) # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 21:29 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : package_service.py 业务服务层 对外统一入口封装策略创建、参数组装、结果格式化 职责解耦接口与底层算法 from typing import List from BreadthFirst.core.models import PackQueryCondition, PackResult, PackageSpec from BreadthFirst.core.enums import LoadStrategyType from BreadthFirst.core.strategy_base import BasePackStrategy from BreadthFirst.core.strategy_bfs import BFSPackStrategy from BreadthFirst.core.strategy_dfs import DFSPackStrategy from BreadthFirst.core.strategy_astar import AStarPackStrategy from BreadthFirst.concurrency.thread_safe_pool import SafePackThreadPool from BreadthFirst.core.logger import AppLogger logger AppLogger.get_logger() class JewelryPackageService: 珠宝包装配载业务服务 def _create_strategy(self, strategy_type: str) - BasePackStrategy: 根据策略标识实例化对应算法对象 :param strategy_type: :return: match strategy_type: case LoadStrategyType.BFS.value: return BFSPackStrategy() case LoadStrategyType.DFS.value: return DFSPackStrategy() case LoadStrategyType.ASTAR.value: return AStarPackStrategy() case _: raise ValueError(f不支持的配载策略类型:{strategy_type}) def calc_single_load_plan(self, condition: PackQueryCondition) - PackResult: 计算单个订单配载方案同步 :param condition: 配载条件 :return: 配载结果 strategy self._create_strategy(condition.strategy_type) thread_pool SafePackThreadPool.get_instance() result thread_pool.submit_task(strategy, condition) self._print_result_log(result) return result def batch_calc_load_plan(self, condition_list: List[PackQueryCondition]) - List[PackResult]: 批量并发计算多个配载方案 :param condition_list: :return: if not condition_list: return [] strategy self._create_strategy(condition_list[0].strategy_type) thread_pool SafePackThreadPool.get_instance() result_list thread_pool.batch_submit(strategy, condition_list) for res in result_list: self._print_result_log(res) return result_list def _print_result_log(self, result: PackResult): 格式化输出结果日志控制台日志文件统一打印 :param result: :return: logger.info(配载结果) logger.info(f返回码:{result.code} 消息:{result.msg}) logger.info(f搜索遍历状态总数{result.search_state_count}) logger.info(f总共使用包装数量{result.total_use_package_count}) logger.info(f物料是否完全装满{result.full_load_flag}) logger.info(f剩余未装入物料体积{result.remain_material_volume:.2f} cm³) logger.info(本次选用包装清单) for item in result.use_package_list: logger.info(f 编码:{item.spec_code} 单盒容积:{item.volume}cm³ 使用数量:{item.use_count}) logger.info(包装库存剩余情况) for code, num in result.remain_package_stock.items(): logger.info(f {code} 剩余库存{num}) logger.info()调用# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 21:30 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : BreadthFirstBll.py 程序入口 main 提供控制台交互式菜单模拟珠宝业务输入物料体积、包装规格库存 from BreadthFirst.core.models import PackQueryCondition, PackageSpec from BreadthFirst.core.enums import LoadStrategyType from BreadthFirst.service.package_service import JewelryPackageService from BreadthFirst.core.logger import AppLogger class BreadthFirstBll(object): logger AppLogger.get_logger() def input_float(self,msg: str) - float: :param msg: :return: while True: try: val float(input(msg)) if val 0: print(数值必须大于0) continue return val except ValueError: print(输入格式错误请输入数字) def input_int(self,msg: str) - int: :param msg: :return: while True: try: val int(input(msg)) if val 0: print(不能为负数) continue return val except ValueError: print(输入格式错误请输入整数) def menu(self,): :return: service JewelryPackageService() package_list: list[PackageSpec] [] material_volume 0.0 while True: print(\n珠宝首饰包装智能配载系统) print(1. 新建包装规格) print(2. 设置待装载首饰物料总体积(cm³)) print(3. 执行BFS最优配载计算推荐最少礼盒数量) print(4. 执行DFS配载计算) print(5. 执行A*配载计算) print(0. 退出程序) opt input(请输入选项).strip() if opt 1: while True: code input(输入包装编码(例 BOX-500)).strip() vol self.input_float(包装容积 cm³) stock self.input_int(当前可用库存数量) package_list.append(PackageSpec(spec_codecode, volumevol, stock_qtystock)) again input(继续新增包装y/n:).strip().lower() if again ! y: break self.logger.info(f已录入包装规格列表{package_list}) elif opt 2: material_volume self.input_float(首饰配件总容积 cm³) self.logger.info(f设置物料总体积{material_volume} cm³) elif opt in [3, 4, 5]: if len(package_list) 0 or material_volume 0: print(请先录入包装规格和物料体积) continue strategy_map { 3: LoadStrategyType.BFS.value, 4: LoadStrategyType.DFS.value, 5: LoadStrategyType.ASTAR.value } cond PackQueryCondition( total_material_volumematerial_volume, package_spec_listpackage_list, strategy_typestrategy_map[opt], allow_over_loadFalse ) service.calc_single_load_plan(cond) elif opt 0: self.logger.info(程序正常退出) break else: print(无效选项) def Demo(self): :return: self.menu()输出
项目结构# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 20:08 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : app_config.py 全局应用配置 import os class AppConfig: # 日志配置 LOG_SAVE_PATH ./logs LOG_FILE_NAME jewelry_packer.log LOG_CONSOLE_ENABLE True LOG_FILE_ENABLE True # 并发设置 MAX_WORKER_THREAD 8 # 配载搜索最大状态数防止极端情况无限搜索 MAX_SEARCH_STATE 15000 staticmethod def init_log_dir(): 初始化日志文件夹 :return: if not os.path.exists(AppConfig.LOG_SAVE_PATH): os.makedirs(AppConfig.LOG_SAVE_PATH, exist_okTrue) # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 20:08 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : enums.py 系统枚举定义 from enum import Enum class LoadStrategyType(Enum): 配载算法策略类型 BFS BFS # 最优方案最少包装数量主推 DFS DFS # 深度优先可行方案不一定最优 ASTAR ASTAR # A*启发搜索 class ResultCode(Enum): 业务返回码 SUCCESS 0, 成功 PARAM_ERROR 400, 参数非法 NO_AVAILABLE_PACKAGE 401, 无可用包装规格 SEARCH_OVER_LIMIT 501, 搜索状态超限未找到方案 NO_ANY_LOAD_PLAN 502, 无法生成配载方案 # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 20:08 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : models.py 数据实体模型 DTO from dataclasses import dataclass from typing import Dict, List, Optional dataclass class PackageSpec: 包装规格实体 spec_code: str # 包装编码BOX-500 volume: float # 包装容积 cm³ stock_qty: int # 当前可用库存数量 dataclass class PackQueryCondition: 配载查询入参 total_material_volume: float # 首饰配件总体积 cm³ package_spec_list: List[PackageSpec] # 可用包装规格列表 strategy_type: str # 算法策略 allow_over_load: bool False # 是否允许包装内超容默认false dataclass class SingleUsePackage: 方案内使用的包装记录 spec_code: str volume: float use_count: int dataclass class PackResult: 配载最终输出结果 code: int msg: str use_package_list: List[SingleUsePackage] # 本次选用包装清单 remain_material_volume: float # 剩余未装入物料体积 remain_package_stock: Dict[str, int] # 各规格剩余库存 {spec_code:剩余数量} total_use_package_count: int # 一共使用多少个包装 search_state_count: int # 算法搜索遍历状态数量 full_load_flag: bool # True物料全部装完无剩余 # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 20:09 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : logger.py 统一日志组件 满足控制台与日志文件输出格式完全一致线程安全 import logging import sys from BreadthFirst.config.app_config import AppConfig import os import threading from typing import Optional from logging.handlers import RotatingFileHandler class AppLogger: _instance: Optional[logging.Logger] None # 【修复】使用线程锁而非原生object _lock threading.Lock() classmethod def get_logger(cls) - logging.Logger: 单例获取日志实例线程安全 :return: if cls._instance is None: with cls._lock: if cls._instance is None: cls._init_logger() return cls._instance classmethod def _init_logger(cls): :return: AppConfig.init_log_dir() logger logging.getLogger(JewelryPacker) logger.setLevel(logging.INFO) logger.propagate False format_str %(asctime)s [%(levelname)s] %(message)s formatter logging.Formatter(format_str, datefmt%Y-%m-%d %H:%M:%S) # 控制台输出 if AppConfig.LOG_CONSOLE_ENABLE: console_handler logging.StreamHandler(streamsys.stdout) console_handler.setFormatter(formatter) logger.addHandler(console_handler) # 文件滚动日志 if AppConfig.LOG_FILE_ENABLE: log_path os.path.join(AppConfig.LOG_SAVE_PATH, AppConfig.LOG_FILE_NAME) file_handler RotatingFileHandler( filenamelog_path, maxBytes10 * 1024 * 1024, backupCount5, encodingutf-8 ) file_handler.setFormatter(formatter) logger.addHandler(file_handler) cls._instance logger # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 20:10 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : strategy_base.py 配载策略抽象基类 所有算法统一实现该接口方便无缝切换策略符合开闭原则 from abc import ABC, abstractmethod from BreadthFirst.core.models import PackQueryCondition, PackResult class BasePackStrategy(ABC): 包装配载策略抽象父类 abstractmethod def execute(self, condition: PackQueryCondition) - PackResult: 执行配载计算 :param condition: 查询条件 :return: 配载结果对象 pass # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 21:01 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : strategy_bfs.py BFS配载最优策略 目标使用最少数量礼盒装载物料等价分水问题最短路径搜索 from collections import deque from typing import Tuple, Dict from BreadthFirst.core.strategy_base import BasePackStrategy from BreadthFirst.core.models import PackQueryCondition, PackResult, SingleUsePackage from BreadthFirst.core.enums import ResultCode from BreadthFirst.config.app_config import AppConfig from BreadthFirst.core.logger import AppLogger logger AppLogger.get_logger() class BFSPackStrategy(BasePackStrategy): def execute(self, condition: PackQueryCondition) - PackResult: BFS搜索最优包装组合方案 total_mat_vol condition.total_material_volume spec_list condition.package_spec_list allow_over condition.allow_over_load max_state AppConfig.MAX_SEARCH_STATE # 1. 参数校验 if total_mat_vol 0: return PackResult( codeResultCode.PARAM_ERROR.value[0], msg物料总体积必须大于0, use_package_list[], remain_material_volumetotal_mat_vol, remain_package_stock{}, total_use_package_count0, search_state_count0, full_load_flagFalse ) if len(spec_list) 0: return PackResult( codeResultCode.NO_AVAILABLE_PACKAGE.value[0], msgResultCode.NO_AVAILABLE_PACKAGE.value[1], use_package_list[], remain_material_volumetotal_mat_vol, remain_package_stock{}, total_use_package_count0, search_state_count0, full_load_flagFalse ) # 初始化包装库存字典 init_stock: Dict[str, int] {} spec_volume_map: Dict[str, float] {} for item in spec_list: init_stock[item.spec_code] item.stock_qty spec_volume_map[item.spec_code] item.volume # 状态定义(当前已装载体积, 当前各包装消耗字典) start_load_vol 0.0 start_used: Dict[str, int] {code: 0 for code in init_stock.keys()} visited set() visited_count 1 # 队列元素已装载体积已使用包装计数字典 queue deque() queue.append((start_load_vol, start_used)) visited.add((start_load_vol, tuple(start_used.values()))) best_solution None while queue: load_vol, used_dict queue.popleft() # 判断当前方案是否满足装载需求 if load_vol total_mat_vol: best_solution (load_vol, used_dict) break if visited_count max_state: logger.warning(f搜索达到最大状态限制{max_state}终止搜索) break # 遍历所有包装规格尝试放入一个礼盒 for spec_code, stock_num in init_stock.items(): used_num used_dict[spec_code] if used_num stock_num: continue # 该规格库存耗尽不能继续选 pack_vol spec_volume_map[spec_code] new_load_vol load_vol pack_vol # 不允许超载并且新容积超出则跳过 if not allow_over and new_load_vol total_mat_vol: continue new_used used_dict.copy() new_used[spec_code] 1 state_key (new_load_vol, tuple(new_used.values())) if state_key not in visited: visited.add(state_key) visited_count 1 queue.append((new_load_vol, new_used)) # 2. 方案结果组装 if best_solution is None: return PackResult( codeResultCode.NO_ANY_LOAD_PLAN.value[0], msg未找到可行包装组合方案, use_package_list[], remain_material_volumetotal_mat_vol, remain_package_stockinit_stock.copy(), total_use_package_count0, search_state_countvisited_count, full_load_flagFalse ) final_load_vol, final_used best_solution remain_material round(total_mat_vol - final_load_vol, 2) full_flag remain_material 0.001 use_package_list [] remain_stock {} for spec_code, total_stock in init_stock.items(): use_cnt final_used[spec_code] remain_cnt total_stock - use_cnt remain_stock[spec_code] remain_cnt if use_cnt 0: use_package_list.append(SingleUsePackage( spec_codespec_code, volumespec_volume_map[spec_code], use_countuse_cnt )) total_use sum([x.use_count for x in use_package_list]) return PackResult( codeResultCode.SUCCESS.value[0], msg配载方案计算成功, use_package_listuse_package_list, remain_material_volumeabs(remain_material), remain_package_stockremain_stock, total_use_package_counttotal_use, search_state_countvisited_count, full_load_flagfull_flag ) # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 21:02 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : strategy_dfs.py DFS配载策略备选 不保证最少包装数量仅作方案对比 from BreadthFirst.core.strategy_base import BasePackStrategy from BreadthFirst.core.models import PackQueryCondition, PackResult, SingleUsePackage from BreadthFirst.core.enums import ResultCode from BreadthFirst.config.app_config import AppConfig from BreadthFirst.core.logger import AppLogger logger AppLogger.get_logger() class DFSPackStrategy(BasePackStrategy): def execute(self, condition: PackQueryCondition) - PackResult: :param condition: :return: total_mat_vol condition.total_material_volume spec_list condition.package_spec_list allow_over condition.allow_over_load max_state AppConfig.MAX_SEARCH_STATE if total_mat_vol 0: return PackResult( codeResultCode.PARAM_ERROR.value[0], msg物料总体积必须大于0, use_package_list[], remain_material_volumetotal_mat_vol, remain_package_stock{}, total_use_package_count0, search_state_count0, full_load_flagFalse ) if len(spec_list) 0: return PackResult( codeResultCode.NO_AVAILABLE_PACKAGE.value[0], msgResultCode.NO_AVAILABLE_PACKAGE.value[1], use_package_list[], remain_material_volumetotal_mat_vol, remain_package_stock{}, total_use_package_count0, search_state_count0, full_load_flagFalse ) init_stock {} spec_volume_map {} for item in spec_list: init_stock[item.spec_code] item.stock_qty spec_volume_map[item.spec_code] item.volume visited set() visited_count 1 best_solution None def dfs(load_vol, used_dict): nonlocal visited_count, best_solution if best_solution is not None: return True if load_vol total_mat_vol: best_solution (load_vol, used_dict.copy()) return True state_key (load_vol, tuple(used_dict.values())) if state_key in visited: return False if visited_count max_state: return False visited.add(state_key) visited_count 1 for spec_code, stock_num in init_stock.items(): used_num used_dict[spec_code] if used_num stock_num: continue pack_v spec_volume_map[spec_code] new_load load_vol pack_v if not allow_over and new_load total_mat_vol: continue used_dict[spec_code] 1 if dfs(new_load, used_dict): return True used_dict[spec_code] - 1 return False start_used {k:0 for k in init_stock.keys()} dfs(0.0, start_used) if best_solution is None: return PackResult( codeResultCode.NO_ANY_LOAD_PLAN.value[0], msg未找到可行包装组合方案, use_package_list[], remain_material_volumetotal_mat_vol, remain_package_stockinit_stock.copy(), total_use_package_count0, search_state_countvisited_count, full_load_flagFalse ) final_load_vol, final_used best_solution remain_material round(total_mat_vol - final_load_vol, 2) full_flag remain_material 0.001 use_package_list [] remain_stock {} for spec_code, total_stock in init_stock.items(): use_cnt final_used[spec_code] remain_cnt total_stock - use_cnt remain_stock[spec_code] remain_cnt if use_cnt 0: use_package_list.append(SingleUsePackage( spec_codespec_code, volumespec_volume_map[spec_code], use_countuse_cnt )) total_use sum([x.use_count for x in use_package_list]) return PackResult( codeResultCode.SUCCESS.value[0], msg配载方案计算成功, use_package_listuse_package_list, remain_material_volumeabs(remain_material), remain_package_stockremain_stock, total_use_package_counttotal_use, search_state_countvisited_count, full_load_flagfull_flag ) # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 21:27 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : strategy_astar.py A*启发式配载策略 import heapq from BreadthFirst.core.strategy_base import BasePackStrategy from BreadthFirst.core.models import PackQueryCondition, PackResult, SingleUsePackage from BreadthFirst.core.enums import ResultCode from BreadthFirst.config.app_config import AppConfig from BreadthFirst.core.logger import AppLogger logger AppLogger.get_logger() class AStarPackStrategy(BasePackStrategy): def execute(self, condition: PackQueryCondition) - PackResult: :param condition: :return: total_mat_vol condition.total_material_volume spec_list condition.package_spec_list allow_over condition.allow_over_load max_state AppConfig.MAX_SEARCH_STATE if total_mat_vol 0: return PackResult( codeResultCode.PARAM_ERROR.value[0], msg物料总体积必须大于0, use_package_list[], remain_material_volumetotal_mat_vol, remain_package_stock{}, total_use_package_count0, search_state_count0, full_load_flagFalse ) if len(spec_list) 0: return PackResult( codeResultCode.NO_AVAILABLE_PACKAGE.value[0], msgResultCode.NO_AVAILABLE_PACKAGE.value[1], use_package_list[], remain_material_volumetotal_mat_vol, remain_package_stock{}, total_use_package_count0, search_state_count0, full_load_flagFalse ) init_stock {} spec_volume_map {} for item in spec_list: init_stock[item.spec_code] item.stock_qty spec_volume_map[item.spec_code] item.volume visited set() visited_count 1 heap [] start_used {k:0 for k in init_stock.keys()} # f代价g已使用包装数量已装载体积使用记录 heapq.heappush(heap, (0, 0, 0.0, start_used)) best_solution None while heap: f, g, load_vol, used_dict heapq.heappop(heap) state_key (load_vol, tuple(used_dict.values())) if state_key in visited: continue visited.add(state_key) visited_count 1 if load_vol total_mat_vol: best_solution (load_vol, used_dict) break if visited_count max_state: logger.warning(fA*搜索达到最大状态限制{max_state}) break for spec_code, stock_num in init_stock.items(): used_num used_dict[spec_code] if used_num stock_num: continue pv spec_volume_map[spec_code] new_load load_vol pv if not allow_over and new_load total_mat_vol: continue new_g g 1 # 启发函数剩余体积 h max(0, total_mat_vol - new_load) new_f new_g h new_used used_dict.copy() new_used[spec_code] 1 heapq.heappush(heap, (new_f, new_g, new_load, new_used)) if best_solution is None: return PackResult( codeResultCode.NO_ANY_LOAD_PLAN.value[0], msg未找到可行包装组合方案, use_package_list[], remain_material_volumetotal_mat_vol, remain_package_stockinit_stock.copy(), total_use_package_count0, search_state_countvisited_count, full_load_flagFalse ) final_load_vol, final_used best_solution remain_material round(total_mat_vol - final_load_vol, 2) full_flag remain_material 0.001 use_package_list [] remain_stock {} for spec_code, total_stock in init_stock.items(): use_cnt final_used[spec_code] remain_cnt total_stock - use_cnt remain_stock[spec_code] remain_cnt if use_cnt 0: use_package_list.append(SingleUsePackage( spec_codespec_code, volumespec_volume_map[spec_code], use_countuse_cnt )) total_use sum([x.use_count for x in use_package_list]) return PackResult( codeResultCode.SUCCESS.value[0], msg配载方案计算成功, use_package_listuse_package_list, remain_material_volumeabs(remain_material), remain_package_stockremain_stock, total_use_package_counttotal_use, search_state_countvisited_count, full_load_flagfull_flag ) # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 21:28 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : thread_safe_pool.py 线程安全执行池支持高并发调用配载算法 解决多线程竞争、资源复用 import concurrent.futures from threading import Lock from typing import List from BreadthFirst.core.models import PackQueryCondition, PackResult from BreadthFirst.core.strategy_base import BasePackStrategy from BreadthFirst.config.app_config import AppConfig from BreadthFirst.core.logger import AppLogger logger AppLogger.get_logger() class SafePackThreadPool: _instance None _lock Lock() def __init__(self): self.executor concurrent.futures.ThreadPoolExecutor(max_workersAppConfig.MAX_WORKER_THREAD) classmethod def get_instance(cls): 单例线程池 :return: if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance SafePackThreadPool() return cls._instance def submit_task(self, strategy: BasePackStrategy, condition: PackQueryCondition) - PackResult: 提交单次配载任务同步等待结果 :param strategy: :param condition: :return: future self.executor.submit(strategy.execute, condition) try: result future.result(timeout30) return result except Exception as e: logger.error(f配载任务执行异常: {str(e)}) raise e def batch_submit(self, strategy: BasePackStrategy, condition_list: List[PackQueryCondition]) - List[PackResult]: 批量并发执行多条配载请求 :param strategy: :param condition_list: :return: task_futures [] for cond in condition_list: task_futures.append(self.executor.submit(strategy.execute, cond)) result_list [] for ft in concurrent.futures.as_completed(task_futures): try: res ft.result(timeout30) result_list.append(res) except Exception as e: logger.error(f批量任务异常 {str(e)}) return result_list def shutdown(self): 释放线程池程序退出调用 :return: self.executor.shutdown(waitTrue) # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 21:29 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : package_service.py 业务服务层 对外统一入口封装策略创建、参数组装、结果格式化 职责解耦接口与底层算法 from typing import List from BreadthFirst.core.models import PackQueryCondition, PackResult, PackageSpec from BreadthFirst.core.enums import LoadStrategyType from BreadthFirst.core.strategy_base import BasePackStrategy from BreadthFirst.core.strategy_bfs import BFSPackStrategy from BreadthFirst.core.strategy_dfs import DFSPackStrategy from BreadthFirst.core.strategy_astar import AStarPackStrategy from BreadthFirst.concurrency.thread_safe_pool import SafePackThreadPool from BreadthFirst.core.logger import AppLogger logger AppLogger.get_logger() class JewelryPackageService: 珠宝包装配载业务服务 def _create_strategy(self, strategy_type: str) - BasePackStrategy: 根据策略标识实例化对应算法对象 :param strategy_type: :return: match strategy_type: case LoadStrategyType.BFS.value: return BFSPackStrategy() case LoadStrategyType.DFS.value: return DFSPackStrategy() case LoadStrategyType.ASTAR.value: return AStarPackStrategy() case _: raise ValueError(f不支持的配载策略类型:{strategy_type}) def calc_single_load_plan(self, condition: PackQueryCondition) - PackResult: 计算单个订单配载方案同步 :param condition: 配载条件 :return: 配载结果 strategy self._create_strategy(condition.strategy_type) thread_pool SafePackThreadPool.get_instance() result thread_pool.submit_task(strategy, condition) self._print_result_log(result) return result def batch_calc_load_plan(self, condition_list: List[PackQueryCondition]) - List[PackResult]: 批量并发计算多个配载方案 :param condition_list: :return: if not condition_list: return [] strategy self._create_strategy(condition_list[0].strategy_type) thread_pool SafePackThreadPool.get_instance() result_list thread_pool.batch_submit(strategy, condition_list) for res in result_list: self._print_result_log(res) return result_list def _print_result_log(self, result: PackResult): 格式化输出结果日志控制台日志文件统一打印 :param result: :return: logger.info(配载结果) logger.info(f返回码:{result.code} 消息:{result.msg}) logger.info(f搜索遍历状态总数{result.search_state_count}) logger.info(f总共使用包装数量{result.total_use_package_count}) logger.info(f物料是否完全装满{result.full_load_flag}) logger.info(f剩余未装入物料体积{result.remain_material_volume:.2f} cm³) logger.info(本次选用包装清单) for item in result.use_package_list: logger.info(f 编码:{item.spec_code} 单盒容积:{item.volume}cm³ 使用数量:{item.use_count}) logger.info(包装库存剩余情况) for code, num in result.remain_package_stock.items(): logger.info(f {code} 剩余库存{num}) logger.info()调用# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述广度优先搜索 Breadth First Search Algorithm 深度优先遍历 Depth First Search # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/26 21:30 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : BreadthFirstBll.py 程序入口 main 提供控制台交互式菜单模拟珠宝业务输入物料体积、包装规格库存 from BreadthFirst.core.models import PackQueryCondition, PackageSpec from BreadthFirst.core.enums import LoadStrategyType from BreadthFirst.service.package_service import JewelryPackageService from BreadthFirst.core.logger import AppLogger class BreadthFirstBll(object): logger AppLogger.get_logger() def input_float(self,msg: str) - float: :param msg: :return: while True: try: val float(input(msg)) if val 0: print(数值必须大于0) continue return val except ValueError: print(输入格式错误请输入数字) def input_int(self,msg: str) - int: :param msg: :return: while True: try: val int(input(msg)) if val 0: print(不能为负数) continue return val except ValueError: print(输入格式错误请输入整数) def menu(self,): :return: service JewelryPackageService() package_list: list[PackageSpec] [] material_volume 0.0 while True: print(\n珠宝首饰包装智能配载系统) print(1. 新建包装规格) print(2. 设置待装载首饰物料总体积(cm³)) print(3. 执行BFS最优配载计算推荐最少礼盒数量) print(4. 执行DFS配载计算) print(5. 执行A*配载计算) print(0. 退出程序) opt input(请输入选项).strip() if opt 1: while True: code input(输入包装编码(例 BOX-500)).strip() vol self.input_float(包装容积 cm³) stock self.input_int(当前可用库存数量) package_list.append(PackageSpec(spec_codecode, volumevol, stock_qtystock)) again input(继续新增包装y/n:).strip().lower() if again ! y: break self.logger.info(f已录入包装规格列表{package_list}) elif opt 2: material_volume self.input_float(首饰配件总容积 cm³) self.logger.info(f设置物料总体积{material_volume} cm³) elif opt in [3, 4, 5]: if len(package_list) 0 or material_volume 0: print(请先录入包装规格和物料体积) continue strategy_map { 3: LoadStrategyType.BFS.value, 4: LoadStrategyType.DFS.value, 5: LoadStrategyType.ASTAR.value } cond PackQueryCondition( total_material_volumematerial_volume, package_spec_listpackage_list, strategy_typestrategy_map[opt], allow_over_loadFalse ) service.calc_single_load_plan(cond) elif opt 0: self.logger.info(程序正常退出) break else: print(无效选项) def Demo(self): :return: self.menu()输出
