certflow.services.sale_plan_service module

销售计划业务服务模块

提供销售计划的Excel导入、查询和分组等业务服务， 支持带格式读取单元格边框字体背景色等信息 封装完整的读取、清洗、校验、分组排序和持久化流程。

class certflow.services.sale_plan_service.SalePlanService(db_session)[源代码]

基类：object

销售计划服务

提供销售计划数据的完整业务流程管理，包括: - Excel文件导入(支持带格式和不带格式两种模式) - 数据清洗和校验 - 分组排序(先按业务分组，组内按产品排序) - 数据库持久化 - 查询和统计

参数:

db_session (Session)

session

SQLAlchemy数据库会话对象

Type:

sqlalchemy.orm.session.Session

excel_handler

Excel文件处理器实例

Type:

certflow.handlers.excel_handler.ExcelHandler

cleaner

数据清洗器实例

Type:

certflow.handlers.data_cleaner.DataCleaner

sorter

排序器实例

Type:

certflow.handlers.sorter.Sorter

save_handler

保存处理器实例

Type:

certflow.handlers.save_handler.SaveHandler

示例

>>> from sqlalchemy import create_engine
>>> from sqlalchemy.orm import sessionmaker
>>>
>>> engine = create_engine("sqlite:///certflow.db")
>>> Session = sessionmaker(bind=engine)
>>> session = Session()
>>>
>>> service = SalePlanService(session)
>>>
>>> # 导入Excel文件
>>> result = service.import_from_excel_with_config({
...     "file_path": "sales_plan.xlsx",
...     "sheet_name": "Sheet1",
...     "preserve_formatting": True
... })
>>> print(f"导入成功: {result['new_count']}条")
static read_excel(path, sheet_name=None, **kwargs)[源代码]

读取 Excel 工作簿（委托 ExcelHandler.read_excel）。

参数:

  • path (str)

  • sheet_name (str | None)

  • kwargs (Any)

返回类型:

Any

static clean_sale_plan(df)[源代码]

清洗销售计划数据框（委托 DataCleaner.clean_sale_plan）。

参数:

df (Any)

返回类型:

Any

static detect_header_row(df_raw, keywords, max_rows=20)[源代码]

自动检测表头行（委托 ExcelHandler.detect_header_row）。

参数:
返回类型:

int

static extract_year_from_filename(file_path)[源代码]

从工作簿文件名提取年份后两位（委托 Sorter._extract_year_from_filename）。

参数:

file_path (str)

返回类型:

str

static apply_range_filter(records, range_filter, selected_rows)[源代码]

B0-8 导入范围筛选。

  • selected_rows：勾选行模式，传入 0-based 数据行序号集合，仅保留这些行。

  • range_filter：按订单筛选模式，dict 含 plan_date/customer/project_name/plan_no 中若干非空字段，仅保留这些字段**全部精确匹配**的行（空值字段视为通配）。

  • 两者均未指定：返回原 records（整表导入）。

临时字段 _src_row 用后即清，不污染落库数据。

参数:
返回类型:

list[dict[str, Any]]

record_shipment(sale_plan_id, quantity, ship_date=None, operator='', note='')[源代码]

记录一次分批发货（B0-3 分批发货数量追踪）。

将本次发货追加到 shipment_batches JSON 列表，累加 shipped_quantity； 当累计已发货量达到订单总量 quantity 时，标记 shipping_status='已发货' （仅更新发货状态，不动生产状态），并记录发货日期。

参数:

  • sale_plan_id (int) -- 销售计划主键 id。

  • quantity (int) -- 本次发货数量（正整数）。

  • ship_date (str | None) -- 发货日期 YYYY-MM-DD，默认今天。

  • operator (str) -- 操作人。

  • note (str) -- 备注。

返回:

含 id / shipped_quantity / remaining_quantity / batches。

返回类型:

dict

抛出:

ValueError -- 记录不存在或发货数量非法。

session: Session
excel_handler: ExcelHandler
cleaner: DataCleaner
sorter: Sorter
save_handler: SaveHandler
shipped_handler: ShippedExportHandler
import_from_excel_with_config(config_params)[源代码]

使用自定义配置从Excel导入销售计划

根据配置参数选择带格式或不带格式的导入方式。

参数:

config_params (dict[str, Any]) -- 导入配置参数字典，包含以下字段: - file_path: Excel文件路径(必填) - sheet_name: 工作表名称或索引，默认为0 - header_row: 表头行号(可选) - skip_rows: 跳过的行数，默认为0 - column_mapping: 自定义列映射字典(可选) - preserve_formatting: 是否保留单元格格式，默认为False

返回:

导入结果字典

返回类型:

Dict[str, Any]

抛出:

Exception -- 导入失败时抛出异常

示例

>>> service = SalePlanService(session)
>>> result = service.import_from_excel_with_config({
...     "file_path": "sales.xlsx",
...     "sheet_name": "1月",
...     "header_row": 1,
...     "skip_rows": 0,
...     "preserve_formatting": True,
... })
>>> print(result["new_count"])
get_all_sale_plans()[源代码]

获取所有销售计划

返回:

按分组键排序的所有销售计划列表

返回类型:

List[SalePlan]

示例

>>> service = SalePlanService(session)
>>> plans = service.get_all_sale_plans()
>>> print(len(plans))
get_by_sort_group(sort_group)[源代码]

根据分组键获取销售计划

参数:

sort_group (str) -- 分组键值

返回:

指定分组的销售计划列表

返回类型:

List[SalePlan]

get_by_product_model(product_model)[源代码]

根据产品型号获取销售计划

参数:

product_model (str) -- 产品型号

返回:

指定产品型号的销售计划列表

返回类型:

List[SalePlan]

clear_all()[源代码]

清空所有销售计划数据

返回:

删除的记录数

返回类型:

int

get_statistics()[源代码]

获取销售计划统计信息

返回:

统计信息字典，包含:

  • total_records: 总记录数

  • group_count: 分组数

  • status_stats: 按生产状态分组统计

返回类型:

Dict[str, Any]

get_by_status(status)[源代码]

根据格式状态获取销售计划

参数:

status (str) -- 格式状态值(如"已开票"、"已发货"等)

返回:

指定格式状态的销售计划列表

返回类型:

List[SalePlan]

get_status_statistics()[源代码]

获取格式状态统计信息

返回:

各格式状态对应的记录数统计

返回类型:

Dict

get_exportable_plans()[源代码]

获取可导出的销售计划（排除已发货、外购未回、隐藏行）

对应 VBA 中导出合格证时的过滤逻辑： - 排除字体蓝色(已发货)的记录 - 排除背景橙棕色(外购未回)的记录 - 排除行高=0(隐藏/取消)的记录

返回:

可导出的销售计划列表

返回类型:

List[SalePlan]

find_by_unique_key(unique_key)[源代码]

根据唯一键查找已存在的销售计划

参数:

unique_key (str) -- 唯一键

返回:

已存在的 SalePlan 或 None

返回类型:

SalePlan | None

find_by_id(plan_id)[源代码]

按主键查询销售计划。

参数:

plan_id (int) -- 销售计划主键

返回:

命中的 SalePlan 或 None

返回类型:

SalePlan | None

find_by_production_order_no(production_order_no)[源代码]

根据生产令号查找已存在的销售计划

用于老数据模式降级去重：当唯一键匹配失败但生产令号有值时， 按生产令号查找已有记录，避免同一记录从不同来源重复导入。

参数:

production_order_no (str) -- 生产令号

返回:

已存在的 SalePlan 或 None

返回类型:

SalePlan | None

find_by_contract_and_model(contract_no, product_model)[源代码]

根据合同号 + 产品型号查找已存在的销售计划

用于跨来源补充场景的兜底匹配：当唯一键和生产令号都匹配失败时， 通过合同号 + 产品型号定位已有记录，实现缺失字段补充。

参数:

  • contract_no (str) -- 合同号

  • product_model (str) -- 产品型号

返回:

已存在的 SalePlan 或 None

返回类型:

SalePlan | None

add_sale_plan(sale_plan)[源代码]

添加销售计划到会话

参数:

sale_plan (SalePlan) -- SalePlan 对象

返回类型:

None

add_change(change)[源代码]

添加变更记录到会话

参数:

change (SalePlanChange) -- SalePlanChange 对象

返回类型:

None

flush_session()[源代码]

刷新会话

返回类型:

None

commit_session()[源代码]

提交事务

返回类型:

None

rollback_session()[源代码]

回滚事务

返回类型:

None

CERT_EDITABLE_FIELDS: tuple[str, ...] = ('cert_product_name', 'cert_product_model', 'cert_product_spec', 'certificate_remarks')
update_cert_fields(plan_id, fields)[源代码]

仅更新合格证打印字段（蓝图 §4.2.3：查询视图 cert-only 订正）

只写入 cert_product_name/model/spec 与 certificate_remarks， 绝不触碰合同字段（product_* 等）。这是合格证清单/打印的数据源， 订正即时生效于后续 create_certificates。

参数:

  • plan_id (int) -- 销售计划记录主键

  • fields (dict[str, Any]) -- 待写入字段字典；非合格证字段会被静默忽略

返回:

是否成功更新（id 不存在返回 False）

返回类型:

bool

count_sale_plans()[源代码]

统计销售计划总数

返回类型:

int

archive_to_shipped(sale_plan_ids)[源代码]

将选中的 SalePlan 归档到 sale_plans_shipped 表并从主表移除

三层模型的 Layer 1 → Layer 2: - 将记录完整拷贝到 sale_plans_shipped（设置 shipped_at / shipped_by="archive"） - 从 sale_plans 表删除对应记录 - 同时将 production_status 设为"已发货"（如果还不是）

参数:

sale_plan_ids (list[int]) -- 要归档的 SalePlan ID 列表

返回:

成功归档的记录数

返回类型:

int

export_and_delete(sale_plan_ids, output_dir)[源代码]

导出选中 SalePlan 的完整字段 Excel，归档到 shipped 表，并从主表删除

三层模型的 Layer 1 → Layer 3: - 生成完整字段 Excel 文件 - 将记录拷贝到 sale_plans_shipped（shipped_by="export_delete"） - 从 sale_plans 表删除

参数:

  • sale_plan_ids (list[int]) -- 要导出的 SalePlan ID 列表

  • output_dir (str) -- Excel 输出目录

返回:

N, "filepath": "..."}

返回类型:

{"count"

export_selected_records(records, columns, file_path)[源代码]

导出选中记录到 Excel（按列定义）

参数:

  • records (list[Any]) -- ORM 记录列表

  • columns (list[dict]) -- 列定义列表 [{"field": "...", "label": "..."}, ...]

  • file_path (str) -- 目标文件路径

返回:

导出记录数

返回类型:

int

export_query_result(records, columns, file_path)[源代码]

导出查询结果到 Excel（带 plan_date 格式化）

参数:

  • records (list[Any]) -- ORM 记录列表

  • columns (list[dict]) -- 列定义列表

  • file_path (str) -- 目标文件路径

返回:

导出记录数

返回类型:

int