# encoding: utf-8# 版权所有 2026 ©涂聚文有限公司™ ®# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎# 描述：Divide and Conquer Algorithm# 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/18 22:45# User : geovindu# Product : PyCharm# Project : PyAlgorithms# File : constants.py# 业务常量统一管理，便于运维修改RISK_AMOUNT_THRESHOLD = 100000.0 # 高客单风险阈值 10万SHOP_LIST = ["福田店", "南山店"]QUARTER_MONTH_MAP = {1: [1, 2, 3], 2: [4, 5, 6], 3: [7, 8, 9], 4: [10, 11, 12]}CATEGORY_LIST = ["黄金", "钻石", "彩宝", "银饰"]# 权限标识常量PERM_DIAMOND_ADJUST = "钻石调价"PERM_EXPORT_ALL_SHOP = "全门店数据导出"PERM_SHOP_ADJUST = "本店调价"PERM_MODIFY_ROLE = "修改角色权限"PERM_VIEW_SELF_SALE = "查看个人业绩"PERM_VIEW_STOCK = "库存盘点"# encoding: utf-8# 版权所有 2026 ©涂聚文有限公司™ ®# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎# 描述：Divide and Conquer Algorithm# 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/18 22:46# User : geovindu# Product : PyCharm# Project : PyAlgorithms# File : utils.pyfrom concurrent.futures import ThreadPoolExecutor, as_completedfrom typing import List, Callable, Anydef parallel_batch_execute(task_list: List[Callable[[], Any]], max_workers: int = 4) -> List[Any]: """ 通用多线程并行执行工具：用于门店/月度分治并发计算 :param task_list: :param max_workers: :return: """ results = [] with ThreadPoolExecutor(max_workers=max_workers) as executor: future_map = {executor.submit(task): task for task in task_list} for future in as_completed(future_map): res = future.result() results.append(res) return resultsdef format_float(val: float) -> str: """ 金额格式化输出 :param val: :return: """ return f"{val:.2f}"# encoding: utf-8# 版权所有 2026 ©涂聚文有限公司™ ®# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎# 描述：Divide and Conquer Algorithm# 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/18 22:46# User : geovindu# Product : PyCharm# Project : PyAlgorithms# File : order.pyfrom dataclasses import dataclassfrom typing import Dictfrom DivideConquer.constants import CATEGORY_LIST@dataclass(frozen=False)class JewelryOrder: """ 珠宝订单实体：纯数据载体，无业务逻辑 """ order_id: str shop_name: str sale_month: int quarter: int category: str sale_amount: float profit: float seller_id: str@dataclass(frozen=False)class SaleSummary: """ 销售统计汇总实体，重载加法实现分治合并 """ total_sales: float = 0.0 total_profit: float = 0.0 gold: float = 0.0 diamond: float = 0.0 gem: float = 0.0 silver: float = 0.0 def __add__(self, other: "SaleSummary") -> "SaleSummary": return SaleSummary( total_sales=self.total_sales + other.total_sales, total_profit=self.total_profit + other.total_profit, gold=self.gold + other.gold, diamond=self.diamond + other.diamond, gem=self.gem + other.gem, silver=self.silver + other.silver ) def to_dict(self) -> Dict[str, float]: """ 转字典用于报表输出扩展 :return: """ return { "总销售额": self.total_sales, "总毛利": self.total_profit, "黄金": self.gold, "钻石": self.diamond, "彩宝": self.gem, "银饰": self.silver }def build_single_summary(order: JewelryOrder) -> SaleSummary: """ 单订单转换为汇总实体（最小子问题构造器） :param order: :return: """ s = SaleSummary() s.total_sales = order.sale_amount s.total_profit = order.profit if order.category == "黄金": s.gold = order.sale_amount elif order.category == "钻石": s.diamond = order.sale_amount elif order.category == "彩宝": s.gem = order.sale_amount elif order.category == "银饰": s.silver = order.sale_amount return s# encoding: utf-8# 版权所有 2026 ©涂聚文有限公司™ ®# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎# 描述：Divide and Conquer Algorithm# 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/18 22:48# User : geovindu# Product : PyCharm# Project : PyAlgorithms# File : role.pyfrom typing import List, Optionalclass JewelryRoleNode: """ 权限二叉树节点实体：承载角色与权限集合 """ def __init__(self, role_name: str, permissions: List[str]): self.role_name = role_name self.permissions: List[str] = permissions self.left: Optional[JewelryRoleNode] = None self.right: Optional[JewelryRoleNode] = None# encoding: utf-8# 版权所有 2026 ©涂聚文有限公司™ ®# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎# 描述：Divide and Conquer Algorithm# 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/18 22:49# User : geovindu# Product : PyCharm# Project : PyAlgorithms# File : order_repo.pyfrom typing import List, Dictfrom DivideConquer.entity.order import JewelryOrderfrom DivideConquer.constants import SHOP_LIST, QUARTER_MONTH_MAPclass OrderRepository: """ 订单仓储：统一管理订单数据增查，隔离上层与原始列表 """ def __init__(self): self._order_data: List[JewelryOrder] = [] def add(self, order: JewelryOrder) -> None: """ :param order: :return: """ self._order_data.append(order) def batch_add(self, orders: List[JewelryOrder]) -> None: """ :param orders: :return: """ self._order_data.extend(orders) def get_all(self) -> List[JewelryOrder]: """ :return: """ return self._order_data.copy() def split_by_shop(self) -> Dict[str, List[JewelryOrder]]: """ Divide：按门店拆分订单数据集 :return: """ split_map = {shop: [] for shop in SHOP_LIST} for o in self._order_data: if o.shop_name in split_map: split_map[o.shop_name].append(o) return split_map def split_by_month(self) -> Dict[int, List[JewelryOrder]]: """ Divide：按月份拆分订单数据集 :return: """ split_map = {m: [] for m in range(1, 13)} for o in self._order_data: split_map[o.sale_month].append(o) return split_map# encoding: utf-8# 版权所有 2026 ©涂聚文有限公司™ ®# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎# 描述：Divide and Conquer Algorithm# 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/18 22:50# User : geovindu# Product : PyCharm# Project : PyAlgorithms# File : role_repo.pyfrom typing import Optionalfrom DivideConquer.entity.role import JewelryRoleNodeclass RoleRepository: """ 权限树仓储：构造、存储全局权限树根节点 """ def __init__(self): self._root: Optional[JewelryRoleNode] = None def set_root(self, root_node: JewelryRoleNode) -> None: """ :param root_node: :return: """ self._root = root_node def get_root(self) -> Optional[JewelryRoleNode]: """ :return: """ return self._root# encoding: utf-8# 版权所有 2026 ©涂聚文有限公司™ ®# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎# 描述：Divide and Conquer Algorithm# 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/18 22:51# User : geovindu# Product : PyCharm# Project : PyAlgorithms# File : base_dac.pyfrom abc import ABC, abstractmethodfrom typing import TypeVar, Generic, List# 泛型适配不同分治输入输出T_INPUT = TypeVar("T_INPUT")T_OUTPUT = TypeVar("T_OUTPUT")class BaseDAC(ABC, Generic[T_INPUT, T_OUTPUT]): """ 分治算法标准抽象基类：统一 divide-conquer-combine 流程 """ @abstractmethod def conquer(self, data: T_INPUT) -> T_OUTPUT: """ 最小子问题求解 :param data: :return: """ pass @abstractmethod def divide(self, data: List[T_INPUT]) -> tuple[List[T_INPUT], List[T_INPUT]]: """ 二分拆分数据集 :param data: :return: """ pass @abstractmethod def combine(self, left_res: T_OUTPUT, right_res: T_OUTPUT) -> T_OUTPUT: """ 合并左右子问题结果 :param left_res: :param right_res: :return: """ pass def dac_recursive(self, data: List[T_INPUT], l: int, r: int) -> T_OUTPUT: """ 标准递归分治入口 :param data: :param l: :param r: :return: """ if l == r: return self.conquer(data[l]) mid = (l + r) // 2 left_result = self.dac_recursive(data, l, mid) right_result = self.dac_recursive(data, mid + 1, r) return self.combine(left_result, right_result)# encoding: utf-8# 版权所有 2026 ©涂聚文有限公司™ ®# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎# 描述：Divide and Conquer Algorithm# 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/18 22:52# User : geovindu# Product : PyCharm# Project : PyAlgorithms# File : stat_dac.pyfrom typing import Listfrom DivideConquer.dac.base_dac import BaseDACfrom DivideConquer.entity.order import JewelryOrder, SaleSummary, build_single_summaryclass SaleStatDAC(BaseDAC[JewelryOrder, SaleSummary]): """ 销售业绩统计专用分治实现 """ def conquer(self, data: JewelryOrder) -> SaleSummary: """ :param data: :return: """ return build_single_summary(data) def divide(self, data: List[JewelryOrder]) -> tuple[List[JewelryOrder], List[JewelryOrder]]: """ :param data: :return: """ mid = len(data) // 2 return data[:mid], data[mid:] def combine(self, left_res: SaleSummary, right_res: SaleSummary) -> SaleSummary: """ :param left_res: :param right_res: :return: """ return left_res + right_res def calc_total(self, data_list: List[JewelryOrder]) -> SaleSummary: """ :param data_list: :return: """ if not data_list: return SaleSummary() return self.dac_recursive(data_list, 0, len(data_list)-1)# encoding: utf-8# 版权所有 2026 ©涂聚文有限公司™ ®# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎# 描述：Divide and Conquer Algorithm# 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/18 22:54# User : geovindu# Product : PyCharm# Project : PyAlgorithms# File : risk_dac.pyfrom typing import Listfrom DivideConquer.dac.base_dac import BaseDACfrom DivideConquer.entity.order import JewelryOrderclass RiskOrderDAC(BaseDAC[JewelryOrder, List[JewelryOrder]]): """ 风控高客单检索分治 异常高客单订单检索分治 """ def __init__(self, risk_threshold: float): self.threshold = risk_threshold def conquer(self, data: JewelryOrder) -> List[JewelryOrder]: if data.sale_amount > self.threshold: return [data] return [] def divide(self, data: List[JewelryOrder]) -> tuple[List[JewelryOrder], List[JewelryOrder]]: mid = len(data) // 2 return data[:mid], data[mid:] def combine(self, left_res: List[JewelryOrder], right_res: List[JewelryOrder]) -> List[JewelryOrder]: return left_res + right_res def scan_risk_orders(self, data_list: List[JewelryOrder]) -> List[JewelryOrder]: if not data_list: return [] return self.dac_recursive(data_list, 0, len(data_list)-1)# encoding: utf-8# 版权所有 2026 ©涂聚文有限公司™ ®# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎# 描述：Divide and Conquer Algorithm# 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/18 22:55# User : geovindu# Product : PyCharm# Project : PyAlgorithms# File : perm_dac.pyfrom typing import Optionalfrom typing import Listfrom DivideConquer.entity.role import JewelryRoleNodefrom DivideConquer.entity.order import JewelryOrderclass PermissionDAC: """ 树形权限二分递归分治校验（无通用列表，独立实现） 权限树分治校验 """ def check_perm(self, root: Optional[JewelryRoleNode], target_perm: str) -> bool: """ :param root: :param target_perm: :return: """ if root is None: return False if target_perm in root.permissions: return True # Divide 拆分左右子树 left_ok = self.check_perm(root.left, target_perm) right_ok = self.check_perm(root.right, target_perm) # Combine：任一子树匹配即有权限 return left_ok or right_ok def filter_order_by_perm(self, orders: List[JewelryOrder], role_perms: List[str], seller_id: str) -> List[JewelryOrder]: """ 分治过滤当前角色可见订单（数据权限隔离） :param orders: :param role_perms: :param seller_id: :return: """ def dac_filter(l: int, r: int) -> List[JewelryOrder]: if l == r: order = orders[l] # 导购仅查看本人订单；店长/区域/集团查看全部 if "查看个人业绩" in role_perms and order.seller_id != seller_id: return [] return [order] mid = (l + r) // 2 left = dac_filter(l, mid) right = dac_filter(mid+1, r) return left + right if not orders: return [] return dac_filter(0, len(orders)-1)# encoding: utf-8# 版权所有 2026 ©涂聚文有限公司™ ®# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎# 描述：Divide and Conquer Algorithm# 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/18 22:57# User : geovindu# Product : PyCharm# Project : PyAlgorithms# File : perm_service.pyfrom DivideConquer.repository.role_repo import RoleRepositoryfrom DivideConquer.dac.perm_dac import PermissionDACfrom DivideConquer.entity.order import JewelryOrderfrom typing import Listclass PermissionService: """ 纯行业业务逻辑，依赖 dac / 仓储 """ def __init__(self, role_repo: RoleRepository): self._repo = role_repo self._dac = PermissionDAC() def has_permission(self, perm_key: str) -> bool: """ :param perm_key: :return: """ root = self._repo.get_root() return self._dac.check_perm(root, perm_key) def get_visible_orders(self, all_orders: List[JewelryOrder], login_seller_id: str) -> List[JewelryOrder]: """ 根据登录角色权限分治过滤可查看订单 :param all_orders: :param login_seller_id: :return: """ root = self._repo.get_root() if not root: return [] return self._dac.filter_order_by_perm(all_orders, root.permissions, login_seller_id)# encoding: utf-8# 版权所有 2026 ©涂聚文有限公司™ ®# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎# 描述：Divide and Conquer Algorithm# 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/18 22:58# User : geovindu# Product : PyCharm# Project : PyAlgorithms# File : sale_stat_service.pyfrom DivideConquer.repository.order_repo import OrderRepositoryfrom DivideConquer.dac.stat_dac import SaleStatDACfrom DivideConquer.utils import parallel_batch_executefrom DivideConquer.constants import QUARTER_MONTH_MAPfrom typing import Dict, Tuplefrom DivideConquer.entity.order import SaleSummaryclass SaleStatService: """ """ def __init__(self, order_repo: OrderRepository): self._repo = order_repo self._dac = SaleStatDAC() def stat_by_shop_parallel(self) -> Dict[str, SaleSummary]: """ 门店分治统计：多线程并行计算各门店业绩 :return: """ shop_split = self._repo.split_by_shop() task_list = [] shop_name_list = [] for shop, ords in shop_split.items(): shop_name_list.append(shop) task_list.append(lambda o=ords: self._dac.calc_total(o)) # 并行执行分治计算 stat_results = parallel_batch_execute(task_list) shop_result = dict(zip(shop_name_list, stat_results)) # 合并集团总业绩 total = SaleSummary() for s in stat_results: total += s shop_result["集团全部门店合计"] = total return shop_result def stat_by_time(self) -> Dict[str, object]: """ 按月分治，合并季度、年度报表 :return: """ month_split = self._repo.split_by_month() month_summary: Dict[int, SaleSummary] = {} # 月度分治统计 for m, ords in month_split.items(): month_summary[m] = self._dac.calc_total(ords) # 合并季度 quarter_summary: Dict[int, SaleSummary] = {1:SaleSummary(),2:SaleSummary(),3:SaleSummary(),4:SaleSummary()} year_total = SaleSummary() for q, month_list in QUARTER_MONTH_MAP.items(): for m in month_list: quarter_summary[q] += month_summary[m] for m_sum in month_summary.values(): year_total += m_sum # 封装返回 month_dict = {f"{m}月": month_summary[m] for m in range(1,13)} return { "月度明细": month_dict, "季度汇总": quarter_summary, "年度总报表": year_total }# encoding: utf-8# 版权所有 2026 ©涂聚文有限公司™ ®# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎# 描述：Divide and Conquer Algorithm# 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/18 22:59# User : geovindu# Product : PyCharm# Project : PyAlgorithms# File : risk_service.pyfrom DivideConquer.repository.order_repo import OrderRepositoryfrom DivideConquer.dac.risk_dac import RiskOrderDACfrom DivideConquer.entity.order import JewelryOrderfrom typing import Listclass RiskScreenService: """ """ def __init__(self, order_repo: OrderRepository, risk_threshold: float): self._repo = order_repo self._dac = RiskOrderDAC(risk_threshold) def scan_high_amount_orders(self) -> List[JewelryOrder]: """ :return: """ all_ords = self._repo.get_all() return self._dac.scan_risk_orders(all_ords) def modify_threshold(self, new_val: float) -> None: """ :param new_val: :return: """ self._dac.threshold = new_val
# encoding: utf-8# 版权所有 2026 ©涂聚文有限公司™ ®# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎# 描述：Divide and Conquer Algorithm# 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/18 23:00# User : geovindu# Product : PyCharm# Project : PyAlgorithms# File : DivideConquerBll.pyfrom DivideConquer.repository.order_repo import OrderRepositoryfrom DivideConquer.repository.role_repo import RoleRepositoryfrom DivideConquer.service.perm_service import PermissionServicefrom DivideConquer.service.sale_stat_service import SaleStatServicefrom DivideConquer.service.risk_service import RiskScreenServicefrom DivideConquer.constants import RISK_AMOUNT_THRESHOLDfrom DivideConquer.entity.order import JewelryOrderfrom DivideConquer.entity.role import JewelryRoleNodefrom typing import List, Dictfrom DivideConquer.constants import *from DivideConquer.utils import format_floatclass DivideConquerBll(object): """ """ def __init__(self): # 初始化仓储 self._order_repo = OrderRepository() self._role_repo = RoleRepository() # 初始化业务服务（依赖注入） self.perm_service = PermissionService(self._role_repo) self.sale_stat_service = SaleStatService(self._order_repo) self.risk_service = RiskScreenService(self._order_repo, RISK_AMOUNT_THRESHOLD) # 订单仓储代理方法 def add_order(self, order: JewelryOrder) -> None: """ :param order: :return: """ self._order_repo.add(order) def batch_add_orders(self, orders: List[JewelryOrder]) -> None: """ :param orders: :return: """ self._order_repo.batch_add(orders) # 权限树初始化 def build_permission_tree(self, root_node: JewelryRoleNode) -> None: """ :param root_node: :return: """ self._role_repo.set_root(root_node) # 对外业务接口 def check_permission(self, perm_key: str) -> bool: """ :param perm_key: :return: """ return self.perm_service.has_permission(perm_key) def get_user_visible_orders(self, seller_id: str) -> List[JewelryOrder]: """ :param seller_id: :return: """ all_ords = self._order_repo.get_all() return self.perm_service.get_visible_orders(all_ords, seller_id) def query_shop_sales_stat(self) -> Dict[str, object]: """ :return: """ return self.sale_stat_service.stat_by_shop_parallel() def query_time_sales_stat(self) -> Dict[str, object]: """ :return: """ return self.sale_stat_service.stat_by_time() def query_risk_orders(self) -> List[JewelryOrder]: """ :return: """ return self.risk_service.scan_high_amount_orders() def Demo(self): """ :return: """ # 2. 模拟构造测试订单 mock_orders = [ JewelryOrder("O001", "福田店", 2, 1, "黄金", 6280, 1300, "S001"), JewelryOrder("O002", "福田店", 3, 1, "钻石", 128600, 48000, "S001"), JewelryOrder("O003", "南山店", 5, 2, "彩宝", 8900, 3600, "S002"), JewelryOrder("O004", "南山店", 6, 2, "钻石", 96000, 39000, "S002"), JewelryOrder("O005", "福田店", 9, 3, "银饰", 499, 220, "S003"), JewelryOrder("O006", "南山店", 11, 4, "钻石", 156000, 62000, "S004"), JewelryOrder("O007", "福田店", 12, 4, "黄金", 8600, 1800, "S003"), ] self.batch_add_orders(mock_orders) # 3. 构建多层权限树 guide = JewelryRoleNode("门店导购", [PERM_VIEW_SELF_SALE]) keeper = JewelryRoleNode("仓管", [PERM_VIEW_STOCK]) shop_mgr = JewelryRoleNode("门店店长", [PERM_SHOP_ADJUST]) shop_mgr.left = guide shop_mgr.right = keeper area_mgr = JewelryRoleNode("华南区域经理", [PERM_DIAMOND_ADJUST]) area_mgr.left = shop_mgr group_admin = JewelryRoleNode("集团管理员", [PERM_EXPORT_ALL_SHOP, PERM_MODIFY_ROLE]) group_admin.left = area_mgr self.build_permission_tree(group_admin) # ========== 业务1：并行分治门店业绩统计 ========== print("======= 【并行分治-门店业绩汇总】 =======") shop_stat = self.query_shop_sales_stat() for shop, summary in shop_stat.items(): print(f"\n【{shop}】") data = summary.to_dict() for k, v in data.items(): print(f"{k}: {format_float(v)}") # ========== 业务2：时间维度分治（月/季/年） ========== print("\n======= 【时间分治-年度经营报表】 =======") time_stat = self.query_time_sales_stat() year_sum = time_stat["年度总报表"] print(f"全年总销售额：{format_float(year_sum.total_sales)}，全年毛利：{format_float(year_sum.total_profit)}") print("\n季度汇总：") for q, s in time_stat["季度汇总"].items(): print(f"Q{q} 销售总额：{format_float(s.total_sales)}") # ========== 业务3：风控分治高客单订单检索 ========== print("\n======= 【风控分治-超10万异常订单】 =======") risk_list = self.query_risk_orders() if risk_list: for item in risk_list: print(f"订单{item.order_id} | {item.shop_name} | {item.category} | {format_float(item.sale_amount)}元") else: print("无风险订单") # ========== 业务4：树形分治权限校验 ========== print("\n======= 【树形分治权限校验】 =======") print(f"区域经理是否可钻石调价：{self.check_permission(PERM_DIAMOND_ADJUST)}") print(f"店长是否可导出全门店数据：{self.check_permission(PERM_EXPORT_ALL_SHOP)}") print(f"集团是否可修改角色权限：{self.check_permission(PERM_MODIFY_ROLE)}") # ========== 业务5：分治数据权限过滤（导购仅看自己订单） ========== print("\n======= 【分治数据权限过滤-导购S001可见订单】 =======") user_ords = self.get_user_visible_orders("S001") for o in user_ords: print(f"{o.order_id} | 开单人:{o.seller_id} | {o.sale_amount}")
