项目结构：
# 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_ok=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: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(stream=sys.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( filename=log_path, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-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( code=ResultCode.PARAM_ERROR.value[0], msg="物料总体积必须大于0", use_package_list=[], remain_material_volume=total_mat_vol, remain_package_stock={}, total_use_package_count=0, search_state_count=0, full_load_flag=False ) if len(spec_list) == 0: return PackResult( code=ResultCode.NO_AVAILABLE_PACKAGE.value[0], msg=ResultCode.NO_AVAILABLE_PACKAGE.value[1], use_package_list=[], remain_material_volume=total_mat_vol, remain_package_stock={}, total_use_package_count=0, search_state_count=0, full_load_flag=False ) # 初始化包装库存字典 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( code=ResultCode.NO_ANY_LOAD_PLAN.value[0], msg="未找到可行包装组合方案", use_package_list=[], remain_material_volume=total_mat_vol, remain_package_stock=init_stock.copy(), total_use_package_count=0, search_state_count=visited_count, full_load_flag=False ) 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_code=spec_code, volume=spec_volume_map[spec_code], use_count=use_cnt )) total_use = sum([x.use_count for x in use_package_list]) return PackResult( code=ResultCode.SUCCESS.value[0], msg="配载方案计算成功", use_package_list=use_package_list, remain_material_volume=abs(remain_material), remain_package_stock=remain_stock, total_use_package_count=total_use, search_state_count=visited_count, full_load_flag=full_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( code=ResultCode.PARAM_ERROR.value[0], msg="物料总体积必须大于0", use_package_list=[], remain_material_volume=total_mat_vol, remain_package_stock={}, total_use_package_count=0, search_state_count=0, full_load_flag=False ) if len(spec_list) == 0: return PackResult( code=ResultCode.NO_AVAILABLE_PACKAGE.value[0], msg=ResultCode.NO_AVAILABLE_PACKAGE.value[1], use_package_list=[], remain_material_volume=total_mat_vol, remain_package_stock={}, total_use_package_count=0, search_state_count=0, full_load_flag=False ) 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( code=ResultCode.NO_ANY_LOAD_PLAN.value[0], msg="未找到可行包装组合方案", use_package_list=[], remain_material_volume=total_mat_vol, remain_package_stock=init_stock.copy(), total_use_package_count=0, search_state_count=visited_count, full_load_flag=False ) 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_code=spec_code, volume=spec_volume_map[spec_code], use_count=use_cnt )) total_use = sum([x.use_count for x in use_package_list]) return PackResult( code=ResultCode.SUCCESS.value[0], msg="配载方案计算成功", use_package_list=use_package_list, remain_material_volume=abs(remain_material), remain_package_stock=remain_stock, total_use_package_count=total_use, search_state_count=visited_count, full_load_flag=full_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( code=ResultCode.PARAM_ERROR.value[0], msg="物料总体积必须大于0", use_package_list=[], remain_material_volume=total_mat_vol, remain_package_stock={}, total_use_package_count=0, search_state_count=0, full_load_flag=False ) if len(spec_list) == 0: return PackResult( code=ResultCode.NO_AVAILABLE_PACKAGE.value[0], msg=ResultCode.NO_AVAILABLE_PACKAGE.value[1], use_package_list=[], remain_material_volume=total_mat_vol, remain_package_stock={}, total_use_package_count=0, search_state_count=0, full_load_flag=False ) 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(f"A*搜索达到最大状态限制{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( code=ResultCode.NO_ANY_LOAD_PLAN.value[0], msg="未找到可行包装组合方案", use_package_list=[], remain_material_volume=total_mat_vol, remain_package_stock=init_stock.copy(), total_use_package_count=0, search_state_count=visited_count, full_load_flag=False ) 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_code=spec_code, volume=spec_volume_map[spec_code], use_count=use_cnt )) total_use = sum([x.use_count for x in use_package_list]) return PackResult( code=ResultCode.SUCCESS.value[0], msg="配载方案计算成功", use_package_list=use_package_list, remain_material_volume=abs(remain_material), remain_package_stock=remain_stock, total_use_package_count=total_use, search_state_count=visited_count, full_load_flag=full_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_workers=AppConfig.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(timeout=30) 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(timeout=30) result_list.append(res) except Exception as e: logger.error(f"批量任务异常 {str(e)}") return result_list def shutdown(self): """ 释放线程池，程序退出调用 :return: """ self.executor.shutdown(wait=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 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_code=code, volume=vol, stock_qty=stock)) 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_volume=material_volume, package_spec_list=package_list, strategy_type=strategy_map[opt], allow_over_load=False ) service.calc_single_load_plan(cond) elif opt == "0": self.logger.info("程序正常退出") break else: print("无效选项！") def Demo(self): """ :return: """ self.menu()输出：
