python:Divide and Conquer Algorithm
____simple_html_dom__voku__html_wrapper____>
项目结构：
# 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.py
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Callable, Any
def 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 results
def 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.py
from dataclasses import dataclass
from typing import Dict
from 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.py
from typing import List, Optional
class 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.py
from typing import List, Dict
from DivideConquer.entity.order import JewelryOrder
from DivideConquer.constants import SHOP_LIST, QUARTER_MONTH_MAP
class 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.py
from typing import Optional
from DivideConquer.entity.role import JewelryRoleNode
class 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.py
from abc import ABC, abstractmethod
from 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.py
from typing import List
from DivideConquer.dac.base_dac import BaseDAC
from DivideConquer.entity.order import JewelryOrder, SaleSummary, build_single_summary
class 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.py
from typing import List
from DivideConquer.dac.base_dac import BaseDAC
from DivideConquer.entity.order import JewelryOrder
class 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.py
from typing import Optional
from typing import List
from DivideConquer.entity.role import JewelryRoleNode
from DivideConquer.entity.order import JewelryOrder
class 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.py
from DivideConquer.repository.role_repo import RoleRepository
from DivideConquer.dac.perm_dac import PermissionDAC
from DivideConquer.entity.order import JewelryOrder
from typing import List
class 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.py
from DivideConquer.repository.order_repo import OrderRepository
from DivideConquer.dac.stat_dac import SaleStatDAC
from DivideConquer.utils import parallel_batch_execute
from DivideConquer.constants import QUARTER_MONTH_MAP
from typing import Dict, Tuple
from DivideConquer.entity.order import SaleSummary
class 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.py
from DivideConquer.repository.order_repo import OrderRepository
from DivideConquer.dac.risk_dac import RiskOrderDAC
from DivideConquer.entity.order import JewelryOrder
from typing import List
class 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.py
from DivideConquer.repository.order_repo import OrderRepository
from DivideConquer.repository.role_repo import RoleRepository
from DivideConquer.service.perm_service import PermissionService
from DivideConquer.service.sale_stat_service import SaleStatService
from DivideConquer.service.risk_service import RiskScreenService
from DivideConquer.constants import RISK_AMOUNT_THRESHOLD
from DivideConquer.entity.order import JewelryOrder
from DivideConquer.entity.role import JewelryRoleNode
from typing import List, Dict
from DivideConquer.constants import *
from DivideConquer.utils import format_float
class 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}")
输出：
© 版权声明
文章版权归作者所有，未经允许请勿转载。
