"""销售计划业务服务模块
提供销售计划的Excel导入、查询和分组等业务服务，
支持带格式读取单元格边框字体背景色等信息
封装完整的读取、清洗、校验、分组排序和持久化流程。
"""
from __future__ import annotations
import json
import re
from datetime import datetime
from pathlib import Path
from typing import Any
import pandas as pd
from loguru import logger
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session
from certflow.config.settings import (
COLUMN_ALIASES,
GROUP_INDEX_FIELD,
GROUPING_KEYS,
GROUPING_PREFIX_FORMAT,
GROUPING_SEQ_FORMAT,
GROUPING_USE_GLOBAL_PREFIX,
MONITORED_FIELDS,
REQUIRED_FIELDS,
SORTING_FIELD_NAMES,
STYLED_IMPORT_COLUMNS,
)
from certflow.handlers import DataCleaner, ExcelHandler, ShippedExportHandler, Sorter
from certflow.handlers.id_generator import IDGenerator
from certflow.handlers.save_handler import SaveHandler
from certflow.handlers.styled_excel_importer import StyledExcelImporter
from certflow.models import SalePlan, SalePlanChange
[文档]
class SalePlanService:
"""销售计划服务
提供销售计划数据的完整业务流程管理，包括:
- Excel文件导入(支持带格式和不带格式两种模式)
- 数据清洗和校验
- 分组排序(先按业务分组，组内按产品排序)
- 数据库持久化
- 查询和统计
Attributes:
session: SQLAlchemy数据库会话对象
excel_handler: Excel文件处理器实例
cleaner: 数据清洗器实例
sorter: 排序器实例
save_handler: 保存处理器实例
Examples:
>>> 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']}条")
"""
# ============================================================
# 处理器门面（L3 收口 L4，避免 Controller 直连 Handler）
# ============================================================
[文档]
@staticmethod
def read_excel(path: str, sheet_name: str | None = None, **kwargs: Any) -> Any:
"""读取 Excel 工作簿（委托 ``ExcelHandler.read_excel``）。"""
return ExcelHandler().read_excel(path, sheet_name=sheet_name, **kwargs)
[文档]
@staticmethod
def clean_sale_plan(df: Any) -> Any:
"""清洗销售计划数据框（委托 ``DataCleaner.clean_sale_plan``）。"""
return DataCleaner().clean_sale_plan(df)
# ============================================================
# 辅助方法
# ============================================================
@staticmethod
def _get_year_month_prefix(file_path: str, sheet_name: Any) -> str:
"""从文件路径和工作表名称提取年月前缀
用于生成全局唯一的排序分组键，避免不同工作表的分组冲突。
Args:
file_path: Excel文件路径，如 "D:/links/Hard/2026年销售计划.xlsx"
sheet_name: 工作表名称，如 "1月" 或 "Sheet1"
Returns:
str: 4位年月前缀，如 "2601"（2026年1月），提取失败返回 "0000"
Examples:
>>> prefix = SalePlanService._get_year_month_prefix(
... "D:/links/Hard/2026年销售计划.xlsx", "1月"
... )
>>> print(prefix) # 输出: "2601"
"""
from certflow.handlers.sorter import Sorter
year = Sorter._extract_year_from_filename(file_path)
month = Sorter._extract_month_from_sheetname(str(sheet_name))
if year is not None and month is not None:
return f"{year:02d}{month:02d}"
return "0000"
@staticmethod
def _normalize_for_comparison(value: str) -> str:
"""规范化值用于比较，将【空白xxx】视为空值
Args:
value: 待规范化的值
Returns:
str: 规范化后的值，【空白xxx】格式视为空字符串
"""
import re
if not value:
return ""
if re.match(r"^【空白.*】$", value):
return ""
return value.strip()
[文档]
@staticmethod
def apply_range_filter(
records: list[dict[str, Any]],
range_filter: dict[str, str] | None,
selected_rows: list[int] | None,
) -> list[dict[str, Any]]:
"""B0-8 导入范围筛选。
- selected_rows：勾选行模式，传入 0-based 数据行序号集合，仅保留这些行。
- range_filter：按订单筛选模式，dict 含 plan_date/customer/project_name/plan_no
中若干非空字段，仅保留这些字段**全部精确匹配**的行（空值字段视为通配）。
- 两者均未指定：返回原 records（整表导入）。
临时字段 _src_row 用后即清，不污染落库数据。
"""
if not records:
return records
for i, rec in enumerate(records):
rec["_src_row"] = i
if selected_rows:
wanted = set(int(x) for x in selected_rows)
records = [r for r in records if r.get("_src_row") in wanted]
elif range_filter:
flt = {
k: str(v).strip() for k, v in (range_filter or {}).items() if v not in (None, "")
}
if flt:
kept = []
for r in records:
if all(str(r.get(k, "") or "").strip() == v for k, v in flt.items()):
kept.append(r)
records = kept
for r in records:
r.pop("_src_row", None)
return records
@staticmethod
def _apply_time_scope_filter(
records: list[dict[str, Any]],
scope: str = "all",
date_range_enabled: bool = False,
date_start: str | None = None,
date_end: str | None = None,
) -> list[dict[str, Any]]:
"""导入范围（按月份三层层次）行级筛选，与视图层同源语义。
- scope == "all_months"：多表由调用方逐表导入，本方法不在此行级过滤。
- date_range_enabled（current_month / single_month）：仅留 计划日期 落在 [起,止] 的行。
- 否则：返回全量（即该月欠交整表，含历史积压）。
"""
if scope == "all_months" or not date_range_enabled:
return records
if not date_start or not date_end:
return records
from certflow.utils.date_utils import normalize_plan_date
kept = []
for r in records:
r.pop("_src_row", None)
pd = normalize_plan_date(r.get("plan_date"))
if pd and date_start <= pd <= date_end:
kept.append(r)
return kept
@staticmethod
def _resolve_month_sheets(file_path: str) -> list[str]:
"""从工作簿中匹配全部 1-12 月欠交表（按工作表名称含月份号），按顺序返回。"""
try:
names = pd.ExcelFile(file_path).sheet_names
except Exception: # noqa: BLE001
names = []
out: list[str] = []
seen: set[str] = set()
for m in range(1, 13):
for s in names:
if s in seen:
continue
if re.search(rf"{m}\s*月", s) or re.search(rf"[-\\/]0?{m}(?![0-9])", s):
out.append(s)
seen.add(s)
break
return out
@staticmethod
def _resolve_single_month_sheet(file_path: str, month: int) -> str | None:
"""从工作簿中解析指定月份(1-12)的欠交表工作表名，未命中返回 None。"""
try:
names = pd.ExcelFile(file_path).sheet_names
except Exception: # noqa: BLE001
names = []
for s in names:
if re.search(rf"{month}\s*月", s) or re.search(rf"[-\\/]0?{month}(?![0-9])", s):
return s
return None
def _apply_incremental_group_seq(
self,
records: list[dict[str, Any]],
year_month_prefix: str,
grouping_keys: list[str] | None = None,
) -> list[dict[str, Any]]:
"""重构要点③：group_seq 增量续号安全。
``group_and_sort`` 只对本批导入集合从 001 重排。部分导入（按订单/按行）
时，同组的新行会与库内同组既有行（同 group_number、seq 001）撞
``sort_order``，进而让批次1 排序编号乱掉。
本方法在 ``group_and_sort`` 之后、落库之前，把同组的**新行**
（unique_key 不在库中）续号到库内同组末尾（max seq + 1），并复用该组
既有前缀（G00n），使折算出的 sort_order 不撞号。
- 既有行（unique_key 命中）由 ``_handle_existing`` 保留其库内 sort_order，
本方法不改动它们的临时 group_seq（落库时被丢弃）。
- 全新组（库内无同组行）：沿用 ``group_and_sort`` 的 001.. 编号，不干预。
- grouping_keys 为空（未配置分组）或 records 为空：直接返回，保持原行为。
"""
if not records:
return records
keys = grouping_keys or list(GROUPING_KEYS)
if not keys:
return records
for r in records:
r["_group_key"] = "_".join(str(r.get(k, "")) for k in keys)
groups: dict[str, list[dict[str, Any]]] = {}
for r in records:
groups.setdefault(r["_group_key"], []).append(r)
continued_groups = 0
continued_rows = 0
for _gk, recs in groups.items():
filters = {k: recs[0].get(k, "") for k in keys}
query = self.session.query(SalePlan).filter_by(**filters)
if year_month_prefix:
query = query.filter(SalePlan.sort_group.like(f"{year_month_prefix}-%"))
existing_rows = query.all()
if not existing_rows:
continue # 全新组：保留 group_and_sort 编号
self._continue_group_seq(recs, existing_rows)
continued_groups += 1
continued_rows += len(recs)
if continued_groups:
logger.info(
f"group_seq 增量续号 | 前缀={year_month_prefix} 续号组数={continued_groups} "
f"续号行数={continued_rows} 分组键={keys}"
)
return records
def _continue_group_seq(
self, recs: list[dict[str, Any]], existing_rows: list[SalePlan]
) -> None:
"""把一组内的新行续号到库内同组末尾（max seq + 1），复用既有前缀。
既有行（unique_key 命中库内）不动；新行从 max(seq)+1 起，前缀沿用该组
既有 ``sort_group`` 的 G00n。直接改写 recs 的 group_seq/_group_prefix/full_seq。
"""
sample_sg = existing_rows[0].sort_group or ""
existing_prefix = (
sample_sg.split("-")[-1] if "-" in sample_sg else GROUPING_PREFIX_FORMAT.format(0)
)
max_seq = 0
for e in existing_rows:
seq = (e.sort_order or 0) % 1000
if seq > max_seq:
max_seq = seq
existing_uks = {e.unique_key for e in existing_rows}
next_seq = max_seq
for r in recs:
uk = IDGenerator.generate_unique_key(r)
if uk in existing_uks:
continue # 既有行：保留其库内 sort_order
next_seq += 1
r["group_seq"] = GROUPING_SEQ_FORMAT.format(next_seq)
r["_group_prefix"] = existing_prefix
r["full_seq"] = f"{existing_prefix}-{r['group_seq']}"
try:
r["_group_index"] = int(existing_prefix[1:]) if existing_prefix[1:].isdigit() else 0
except (ValueError, IndexError):
r["_group_index"] = 0
logger.debug(
f"_continue_group_seq | 前缀={existing_prefix} max_seq={max_seq} "
f"新行续号至={next_seq} (组行数={len(recs)})"
)
[文档]
def record_shipment(
self,
sale_plan_id: int,
quantity: int,
ship_date: str | None = None,
operator: str = "",
note: str = "",
) -> dict[str, Any]:
"""记录一次分批发货（B0-3 分批发货数量追踪）。
将本次发货追加到 ``shipment_batches`` JSON 列表，累加 ``shipped_quantity``；
当累计已发货量达到订单总量 ``quantity`` 时，标记 ``shipping_status='已发货'``
（仅更新发货状态，不动生产状态），并记录发货日期。
Args:
sale_plan_id: 销售计划主键 id。
quantity: 本次发货数量（正整数）。
ship_date: 发货日期 ``YYYY-MM-DD``，默认今天。
operator: 操作人。
note: 备注。
Returns:
dict: 含 id / shipped_quantity / remaining_quantity / batches。
Raises:
ValueError: 记录不存在或发货数量非法。
"""
from sqlalchemy import select
plan = self.session.execute(
select(SalePlan).where(SalePlan.id == sale_plan_id)
).scalar_one_or_none()
if plan is None:
raise ValueError(f"未找到销售计划 id={sale_plan_id}")
qty = int(quantity)
if qty <= 0:
raise ValueError("发货数量必须为正整数")
logger.info(f"分批发货记录 | id={sale_plan_id} 本次发货量={qty}")
# 解析既有批次明细
try:
batches = json.loads(plan.shipment_batches or "[]")
except (ValueError, TypeError):
batches = []
if not isinstance(batches, list):
batches = []
batch_no = len(batches) + 1
ship_date = ship_date or datetime.now().strftime("%Y-%m-%d")
batches.append(
{
"batch_no": batch_no,
"quantity": qty,
"ship_date": ship_date,
"operator": operator or "",
"note": note or "",
}
)
plan.shipment_batches = json.dumps(batches, ensure_ascii=False)
# 累加上已发货量并封顶到订单总量，避免超发态
plan.shipped_quantity = (plan.shipped_quantity or 0) + qty
if plan.quantity and plan.shipped_quantity >= plan.quantity:
plan.shipped_quantity = plan.quantity
if (plan.shipping_status or "") != "已发货":
plan.shipping_status = "已发货"
plan.shipping_status_source = "manual"
plan.shipping_status_set_at = datetime.now()
if not (plan.shipped_date or "").strip():
plan.shipped_date = ship_date
logger.info(
f"分批发货完成 | id={plan.id} 已发={plan.shipped_quantity}/{plan.quantity} "
f"剩余={plan.remaining_quantity} 状态={plan.shipping_status} 累计批次数={len(batches)}"
)
self.session.commit()
return {
"id": plan.id,
"shipped_quantity": plan.shipped_quantity,
"remaining_quantity": plan.remaining_quantity,
"batches": batches,
}
@staticmethod
def _detect_changes(
existing: SalePlan,
record: dict[str, Any],
source_file: str,
source_sheet: str,
import_batch_id: str,
) -> list[SalePlanChange]:
"""对比已有记录和新记录，返回变更记录列表
注意：空值规范化（如【空白字段名】）不会被记录为变更。
Args:
existing: 数据库中已有的 SalePlan 记录
record: 新导入的记录字典
source_file: 来源文件路径
source_sheet: 来源工作表名称
import_batch_id: 导入批次ID
Returns:
List[SalePlanChange]: 变更记录列表，同时会直接更新 existing 的对应字段
"""
changes: list[SalePlanChange] = []
for field in MONITORED_FIELDS:
old_val = str(getattr(existing, field, "") or "")
new_val = str(record.get(field, "") or "")
old_normalized = SalePlanService._normalize_for_comparison(old_val)
new_normalized = SalePlanService._normalize_for_comparison(new_val)
if old_normalized != new_normalized:
changes.append(
SalePlanChange(
sale_plan_id=existing.id,
unique_key=existing.unique_key,
field_name=field,
old_value=old_val,
new_value=new_val,
source_file=source_file,
source_sheet=source_sheet,
import_batch_id=import_batch_id,
comment_text=str(record.get("_comments", "")),
)
)
# 同时更新 SalePlan 的当前值
setattr(existing, field, new_val)
return changes
# ============================================================
# 初始化和公共方法
# ============================================================
def __init__(self, db_session: Session) -> None:
"""初始化销售计划服务
Args:
db_session: SQLAlchemy数据库会话对象
"""
self.session: Session = db_session
self.excel_handler: ExcelHandler = ExcelHandler()
self.cleaner: DataCleaner = DataCleaner()
self.sorter: Sorter = Sorter()
self.save_handler: SaveHandler = SaveHandler(db_session, orm_delegate=self)
self.shipped_handler: ShippedExportHandler = ShippedExportHandler(db_session)
[文档]
def import_from_excel_with_config(self, config_params: dict[str, Any]) -> dict[str, Any]:
"""使用自定义配置从Excel导入销售计划
根据配置参数选择带格式或不带格式的导入方式。
Args:
config_params: 导入配置参数字典，包含以下字段:
- file_path: Excel文件路径(必填)
- sheet_name: 工作表名称或索引，默认为0
- header_row: 表头行号(可选)
- skip_rows: 跳过的行数，默认为0
- column_mapping: 自定义列映射字典(可选)
- preserve_formatting: 是否保留单元格格式，默认为False
Returns:
Dict[str, Any]: 导入结果字典
Raises:
Exception: 导入失败时抛出异常
Examples:
>>> 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"])
"""
try:
file_path = config_params["file_path"]
sheet_name = config_params.get("sheet_name", 0)
header_row = config_params.get("header_row")
skip_rows = config_params.get("skip_rows", 0)
custom_mapping = config_params.get("column_mapping", {})
preserve_formatting = config_params.get("preserve_formatting", True)
# B0-8：导入范围筛选 + 冲突决议
range_filter = config_params.get("range_filter")
selected_rows = config_params.get("selected_rows")
conflict_resolutions = config_params.get("conflict_resolutions")
# 盲区3 修复：批次内撞键处理策略（UI 选择，覆盖 ON_COLLISION 配置）
collision_policy = config_params.get("collision_policy")
# 导入范围（按月份三层层次）
scope = config_params.get("scope", "current_month")
scope_month = config_params.get("scope_month")
# 指定月份表：以 scope_month 解析实际工作表名（不依赖 UI 当前选择）
if scope == "single_month" and scope_month:
resolved = self._resolve_single_month_sheet(file_path, scope_month)
if resolved:
sheet_name = resolved
date_range = config_params.get("date_range") or {}
date_range_enabled = bool(date_range.get("enabled"))
date_range_start = date_range.get("start")
date_range_end = date_range.get("end")
dry_run = bool(config_params.get("dry_run", False))
logger.info(f"开始导入销售计划: {file_path}")
logger.debug(f"保留格式: {preserve_formatting}")
# 全部月份表（大批量）：依次导入 1-12 月中存在的全部欠交表
if scope == "all_months":
return self._import_all_months(
file_path=file_path,
header_row=header_row,
skip_rows=skip_rows,
custom_mapping=custom_mapping,
range_filter=range_filter,
selected_rows=selected_rows,
conflict_resolutions=conflict_resolutions,
collision_policy=collision_policy,
preserve_formatting=preserve_formatting,
)
if preserve_formatting:
return self._import_with_formatting(
file_path=file_path,
sheet_name=sheet_name,
header_row=header_row,
skip_rows=skip_rows,
custom_mapping=custom_mapping,
range_filter=range_filter,
selected_rows=selected_rows,
conflict_resolutions=conflict_resolutions,
collision_policy=collision_policy,
scope=scope,
date_range_enabled=date_range_enabled,
date_range_start=date_range_start,
date_range_end=date_range_end,
dry_run=dry_run,
)
return self._import_without_formatting(
file_path=file_path,
sheet_name=sheet_name,
header_row=header_row,
skip_rows=skip_rows,
custom_mapping=custom_mapping,
range_filter=range_filter,
selected_rows=selected_rows,
conflict_resolutions=conflict_resolutions,
collision_policy=collision_policy,
scope=scope,
date_range_enabled=date_range_enabled,
date_range_start=date_range_start,
date_range_end=date_range_end,
dry_run=dry_run,
)
except Exception as e:
logger.error(f"导入销售计划失败: {e}")
raise
def _import_all_months(
self,
file_path: str,
header_row: int | None,
skip_rows: int,
custom_mapping: dict[str, Any],
range_filter: dict[str, str] | None,
selected_rows: list[int] | None,
conflict_resolutions: dict[str, str] | None,
preserve_formatting: bool,
collision_policy: str | None = None,
) -> dict[str, Any]:
"""全部月份表（大批量）：依次导入 1-12 月中存在的全部欠交表。
每表独立提交一次（便于空库快速填充，单表失败不影响已提交表）；跨表重复行
由 save_handler 的 DB 冲突检测 + 默认策略（建议在 Tab③ 设为「跳过」）处理，
从而排除跨月结转重复。
"""
sheets = self._resolve_month_sheets(file_path)
if not sheets:
raise ValueError("未在工作簿中找到任何月份欠交表（按工作表名称匹配 1-12 月）")
agg: dict[str, Any] = {
"scope": "all_months",
"sheets": len(sheets),
"total": 0,
"new_count": 0,
"duplicate_count": 0,
"db_total_count": 0,
"groups": 0,
"records": [],
"duplicates": [],
"gated_count": 0,
"quarantined_count": 0,
"per_sheet": [],
}
try:
for sh in sheets:
if preserve_formatting:
res = self._import_with_formatting(
file_path=file_path,
sheet_name=sh,
header_row=header_row,
skip_rows=skip_rows,
custom_mapping=custom_mapping,
range_filter=range_filter,
selected_rows=selected_rows,
conflict_resolutions=conflict_resolutions,
collision_policy=collision_policy,
scope="single",
)
else:
res = self._import_without_formatting(
file_path=file_path,
sheet_name=sh,
header_row=header_row,
skip_rows=skip_rows,
custom_mapping=custom_mapping,
range_filter=range_filter,
selected_rows=selected_rows,
conflict_resolutions=conflict_resolutions,
collision_policy=collision_policy,
scope="single",
)
agg["total"] += res.get("total", 0)
agg["new_count"] += res.get("new_count", 0)
agg["duplicate_count"] += res.get("duplicate_count", res.get("skipped_count", 0))
agg["gated_count"] += res.get("gated_count", 0)
agg["quarantined_count"] += res.get("quarantined_count", 0)
agg["groups"] += res.get("groups", 0)
agg["records"].extend(res.get("records", [])[:2])
agg["per_sheet"].append(
{
"sheet": sh,
"new": res.get("new_count", 0),
"dup": res.get("duplicate_count", res.get("skipped_count", 0)),
}
)
agg["db_total_count"] = self.count_sale_plans()
logger.info(
f"全部月份表导入完成: 共 {len(sheets)} 个表, 新增 {agg['new_count']}, "
f"重复 {agg['duplicate_count']}"
)
return agg
except Exception as e:
self.rollback_session()
logger.error(f"全部月份表导入失败（已提交的表保留）: {e}")
raise
# ============================================================
# B0-4：导入期占位目录（配置驱动，默认 OFF）
# ============================================================
@staticmethod
def _maybe_create_placeholder_dirs(records: list[dict]) -> int:
"""B0-4：导入开关 ON 时，按订单组预建 VBA 风格占位项目文件夹。
开关 `import.create_placeholder_dirs`（默认 False）→ 导入只写 SQLite，
绝不碰磁盘目录，直接返回 0。仅当开关为 True 时，对每个订单组
（计划日期 + 订货单位 + 项目名称，去重）调用 ReportOutputService 的惰性
makedirs 建一个空项目资料文件夹，命名统一沿用 paths.report_archive.folder。
Args:
records: 已分组排序后的销售计划记录列表
Returns:
int: 实际创建的占位文件夹数量（去重后）
"""
from certflow.config import settings as app_settings
if not app_settings.CREATE_PLACEHOLDER_DIRS:
return 0
from certflow.services.report_output_service import ReportOutputService
svc = ReportOutputService()
seen: set[tuple[str, str, str]] = set()
created = 0
for rec in records:
key = (
str(rec.get("plan_date", "")),
(rec.get("customer") or "").strip(),
(rec.get("project_name") or "").strip(),
)
if key in seen:
continue
seen.add(key)
svc.create_placeholder_folder(
plan_date=rec.get("plan_date"),
customer=rec.get("customer"),
project=rec.get("project_name"),
)
created += 1
logger.info(f"B0-4 占位目录预建完成: {created} 个订单组文件夹（根: {svc.root}）")
return created
# ============================================================
# 导入方法（不带格式）
# ============================================================
def _import_without_formatting(
self,
file_path: str,
sheet_name: Any,
header_row: int | None,
skip_rows: int,
custom_mapping: dict[str, Any],
range_filter: dict[str, str] | None = None,
selected_rows: list[int] | None = None,
conflict_resolutions: dict[str, str] | None = None,
collision_policy: str | None = None,
scope: str = "all",
date_range_enabled: bool = False,
date_range_start: str | None = None,
date_range_end: str | None = None,
dry_run: bool = False,
) -> dict[str, Any]:
"""原有的导入逻辑(不带格式)
使用pandas读取Excel，不保留单元格样式信息。
Args:
file_path: Excel文件路径
sheet_name: 工作表名称或索引
header_row: 表头行号
skip_rows: 跳过的行数
custom_mapping: 自定义列映射
Returns:
Dict[str, Any]: 导入结果字典
"""
try:
df = self.excel_handler.read_excel(
file_path=Path(file_path),
sheet_name=sheet_name,
header_row=header_row,
skiprows=skip_rows,
)
if custom_mapping:
existing_cols = [col for col in custom_mapping if col in df.columns]
df = df[existing_cols]
df = df.rename(columns=custom_mapping)
logger.info(f"使用自定义列映射: {custom_mapping}")
else:
alias_mapping = COLUMN_ALIASES
df = self.excel_handler.validate_columns(
df, required_columns={}, use_aliases=True, alias_mapping=alias_mapping
)
df_cleaned = self.cleaner.clean_sale_plan(df)
required_fields = REQUIRED_FIELDS
missing_fields = [f for f in required_fields if f not in df_cleaned.columns]
if missing_fields:
raise ValueError(f"缺少必需列: {', '.join(missing_fields)}")
records = df_cleaned.to_dict("records")
# B0-8：导入范围筛选（按订单4字段 / 勾选行），在门控前生效
records = self.apply_range_filter(records, range_filter, selected_rows)
# 导入范围筛选（按月份 / 计划日期范围），在门控前生效
records = self._apply_time_scope_filter(
records, scope, date_range_enabled, date_range_start, date_range_end
)
total_count = len(records)
# #30 P1 导入门控：缺失要货单号的行按 mode 处理（skip/isolate/warn）
from certflow.services.import_gate_service import ImportGateService
gate_result = ImportGateService.apply(records)
records = gate_result["kept"]
# isolate 模式：被门控行持久化到隔离表（回收站），不进 sale_plans
gate_quarantined = 0
if gate_result["applied"] and gate_result["mode"] == "isolate" and not dry_run:
gate_quarantined = ImportGateService.persist_isolated(
gate_result["gated"],
meta={
"source_file": str(file_path),
"source_sheet": str(sheet_name) if sheet_name is not None else None,
},
session=self.session,
)
# 获取年月前缀（用于生成全局唯一的标识）
year_month_prefix = self._get_year_month_prefix(file_path, sheet_name)
year_month_int = int(year_month_prefix) if year_month_prefix.isdigit() else 0
# 使用分组排序方法
sorted_records = self.sorter.group_and_sort(
data=records,
group_keys=GROUPING_KEYS,
sort_keys=SORTING_FIELD_NAMES,
keep_original_order=True,
generate_seq=True,
use_global_prefix=GROUPING_USE_GLOBAL_PREFIX,
prefix_format=GROUPING_PREFIX_FORMAT,
seq_format=GROUPING_SEQ_FORMAT,
seq_field="group_seq",
group_key_field="_group_key",
group_index_field=GROUP_INDEX_FIELD,
source_file=file_path,
source_sheet=str(sheet_name),
)
# 获取分组信息
groups = self.sorter.get_all_groups(
sorted_records, seq_field="group_seq", group_key_field="_group_key"
)
logger.info(f"分组完成: {len(groups)} 个分组")
# 重构要点③：部分导入时同组新行续号到库内末尾，避免 sort_order 撞号
if range_filter or selected_rows:
self._apply_incremental_group_seq(sorted_records, year_month_prefix)
# dry_run：仅制备记录用于诊断（如签名比对），不落库、不提交
if dry_run:
return {
"dry_run": True,
"total": len(sorted_records),
"sorted_records": sorted_records,
"groups": groups,
}
import_batch_id = (
f"{Path(file_path).stem}_{sheet_name}_"
f"{datetime.now().strftime('%Y%m%d_%H%M%S_%f')[:-3]}"
)
save_result = self.save_handler.save_records_with_stats(
records=sorted_records,
source_file=file_path,
source_sheet=str(sheet_name),
import_batch_id=import_batch_id,
year_month_prefix=year_month_prefix,
year_month_int=year_month_int,
conflict_resolutions=conflict_resolutions,
collision_policy=collision_policy,
)
imported_count = save_result["new_count"]
duplicate_count = save_result["duplicate_count"]
duplicates = save_result["duplicates"]
# 隔离分源计数（撞键 / 冲突决议），与门控隔离分开统计
collision_quarantined = save_result.get("collision_quarantined", 0)
collision_dropped = save_result.get("collision_dropped", 0)
resolution_quarantined = save_result.get("resolution_quarantined", 0)
self.commit_session()
db_total_count = self.count_sale_plans()
logger.info(
f"导入完成: 总{total_count}条, 新增{imported_count}条, "
f"重复{duplicate_count}条, 分组数: {len(groups)}"
)
return {
"total": total_count,
"new_count": imported_count,
"duplicate_count": duplicate_count,
"db_total_count": db_total_count,
"groups": len(groups),
"records": sorted_records[:10],
"duplicates": duplicates[:50],
"grouped_data": groups,
"group_stats": {idx: info["count"] for idx, info in groups.items()},
# B0-4：导入期占位目录预建数量（开关 OFF 时为 0）
"placeholder_dirs": self._maybe_create_placeholder_dirs(sorted_records),
# #30 P1 导入门控统计
"gated_count": len(gate_result["gated"]),
"gate_mode": gate_result["mode"],
"gated_records": gate_result["gated"][:50],
"quarantined_count": gate_quarantined
+ collision_quarantined
+ resolution_quarantined,
"gate_quarantined": gate_quarantined,
"collision_quarantined": collision_quarantined,
"collision_dropped": collision_dropped,
"resolution_quarantined": resolution_quarantined,
}
except Exception as e:
self.rollback_session()
logger.error(f"导入失败，已回滚: {e}")
raise
# ============================================================
# 导入方法（带格式）
# ============================================================
def _import_with_formatting( # noqa: C901
self,
file_path: str,
sheet_name: Any,
header_row: int | None,
skip_rows: int,
custom_mapping: dict[str, Any],
range_filter: dict[str, str] | None = None,
selected_rows: list[int] | None = None,
conflict_resolutions: dict[str, str] | None = None,
collision_policy: str | None = None,
scope: str = "all",
date_range_enabled: bool = False,
date_range_start: str | None = None,
date_range_end: str | None = None,
dry_run: bool = False,
) -> dict[str, Any]:
"""带格式读取的导入逻辑
使用openpyxl读取Excel，保留单元格背景色、字体色、批注等格式信息。
Args:
file_path: Excel文件路径
sheet_name: 工作表名称或索引
header_row: 表头行号
skip_rows: 跳过的行数
custom_mapping: 自定义列映射
Returns:
Dict[str, Any]: 导入结果字典，包含格式状态统计等信息
"""
logger.info(f"开始带格式导入: {file_path}")
try:
# 1. 读取带格式的数据并设置列名
df_data, bg_colors, font_colors, comments, statuses, hidden_rows = (
StyledExcelImporter.read_styled_excel(file_path, header_row, sheet_name, skip_rows)
)
# 2. 清洗数据
df_data = self.excel_handler.clean_dataframe(df_data)
# 3. 处理列映射（先于清洗，确保后续清洗使用英文字段名）
if custom_mapping:
existing_cols = [col for col in custom_mapping if col in df_data.columns]
df_data = df_data[existing_cols]
df_data = df_data.rename(columns=custom_mapping)
logger.info(f"使用自定义列映射: {custom_mapping}")
else:
alias_mapping = COLUMN_ALIASES
df_data = self.excel_handler.validate_columns(
df_data, required_columns={}, use_aliases=True, alias_mapping=alias_mapping
)
# 4. 数据清洗（列名已映射为英文，清洗逻辑可正常工作）
df_data = self.cleaner.clean_sale_plan(df_data)
# 5. 保留配置的列
if STYLED_IMPORT_COLUMNS:
existing_cols = [col for col in STYLED_IMPORT_COLUMNS if col in df_data.columns]
if existing_cols:
df_data = df_data[existing_cols]
logger.info(f"保留的列: {existing_cols}")
if len(df_data) == 0:
raise ValueError("清洗后无有效数据")
# 6. 转换为字典并合并格式信息
records = StyledExcelImporter.merge_format_info(
df_data, bg_colors, font_colors, comments, statuses, hidden_rows
)
# B0-8：导入范围筛选（按订单4字段 / 勾选行），在门控前生效
records = self.apply_range_filter(records, range_filter, selected_rows)
# 导入范围筛选（按月份 / 计划日期范围），在门控前生效
records = self._apply_time_scope_filter(
records, scope, date_range_enabled, date_range_start, date_range_end
)
# #30 P1 导入门控：缺失要货单号的行按 mode 处理（skip/isolate/warn）
from certflow.services.import_gate_service import ImportGateService
gate_result = ImportGateService.apply(records)
original_total = len(records)
records = gate_result["kept"]
# isolate 模式：被门控行持久化到隔离表（回收站），不进 sale_plans
gate_quarantined = 0
if gate_result["applied"] and gate_result["mode"] == "isolate" and not dry_run:
gate_quarantined = ImportGateService.persist_isolated(
gate_result["gated"],
meta={
"source_file": str(file_path),
"source_sheet": str(sheet_name) if sheet_name is not None else None,
},
session=self.session,
)
# 7. 获取年月前缀
year_month_prefix = self._get_year_month_prefix(file_path, sheet_name)
year_month_int = int(year_month_prefix) if year_month_prefix.isdigit() else 0
# 8. 分组排序
sorted_records = self.sorter.group_and_sort(
data=records,
group_keys=GROUPING_KEYS,
sort_keys=SORTING_FIELD_NAMES,
keep_original_order=True,
generate_seq=True,
use_global_prefix=GROUPING_USE_GLOBAL_PREFIX,
prefix_format=GROUPING_PREFIX_FORMAT,
seq_format=GROUPING_SEQ_FORMAT,
seq_field="group_seq",
group_key_field="_group_key",
group_index_field=GROUP_INDEX_FIELD,
source_file=file_path,
source_sheet=str(sheet_name),
)
# 获取分组信息
groups = self.sorter.get_all_groups(
sorted_records, seq_field="group_seq", group_key_field="_group_key"
)
logger.info(f"分组完成: {len(groups)} 个分组")
# 重构要点③：部分导入时同组新行续号到库内末尾，避免 sort_order 撞号
if range_filter or selected_rows:
self._apply_incremental_group_seq(sorted_records, year_month_prefix)
# dry_run：仅制备记录用于诊断（如签名比对），不落库、不提交
if dry_run:
return {
"dry_run": True,
"total": len(sorted_records),
"sorted_records": sorted_records,
"groups": groups,
}
# 9. 保存到数据库
save_result = self.save_handler.save_styled_records(
sorted_records=sorted_records,
file_path=file_path,
sheet_name=str(sheet_name),
import_batch_id=f"batch_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
year_month_prefix=year_month_prefix,
year_month_int=year_month_int,
conflict_resolutions=conflict_resolutions,
collision_policy=collision_policy,
)
imported_count = save_result["new_count"]
skipped_count = save_result["duplicate_count"]
changes_count = save_result["changes_count"]
# 隔离分源计数（撞键 / 冲突决议），与门控隔离分开统计
collision_quarantined = save_result.get("collision_quarantined", 0)
collision_dropped = save_result.get("collision_dropped", 0)
resolution_quarantined = save_result.get("resolution_quarantined", 0)
quarantined_count = gate_quarantined + collision_quarantined + resolution_quarantined
status_stats = {}
for record in sorted_records:
status = record.get("_format_status")
if status:
status_stats[status] = status_stats.get(status, 0) + 1
logger.info(
f"导入完成: 新增{imported_count}条, 跳过{skipped_count}条, "
f"变更{changes_count}条, 分组数: {len(groups)}"
)
# 计数恒等式（#30 P1 修复）：total 必须等于 original_total，
# 且 original_total == new + skipped + changes + collision_q
# + collision_dropped + resolution_q + gated + unprocessed
# 旧实现误将 total 赋为 imported_count，导致 total 与 original_total 不一致。
unprocessed = original_total - (
imported_count
+ skipped_count
+ changes_count
+ collision_quarantined
+ collision_dropped
+ resolution_quarantined
+ len(gate_result["gated"])
)
return {
"success": True,
"total": original_total,
"new_count": imported_count,
"skipped_count": skipped_count,
"changes_count": changes_count,
# 兼容别名：部分调用方（对话框/旧测试）仍读取这些键名
"skipped": skipped_count,
"duplicate_count": skipped_count,
"changes": changes_count,
"groups": len(groups),
"status_stats": status_stats,
"records": sorted_records[:20],
"grouped_data": groups,
"group_stats": {idx: info["count"] for idx, info in groups.items()},
# #30 P1 导入门控统计
"original_total": original_total,
"gated_count": len(gate_result["gated"]),
"gate_mode": gate_result["mode"],
"gated_records": gate_result["gated"][:50],
"quarantined_count": quarantined_count,
"gate_quarantined": gate_quarantined,
"collision_quarantined": collision_quarantined,
"collision_dropped": collision_dropped,
"resolution_quarantined": resolution_quarantined,
# 显式暴露未处理差值，便于 UI/测试校验恒等式
"unprocessed": unprocessed,
# B0-4：导入期占位目录预建数量（开关 OFF 时为 0）
"placeholder_dirs": self._maybe_create_placeholder_dirs(sorted_records),
}
except Exception as e:
logger.error(f"导入失败: {e}")
import traceback
traceback.print_exc()
self.rollback_session()
return {"success": False, "error": str(e)}
# ============================================================
# 查询方法
# ============================================================
[文档]
def get_all_sale_plans(self) -> list[SalePlan]:
"""获取所有销售计划
Returns:
List[SalePlan]: 按分组键排序的所有销售计划列表
Examples:
>>> service = SalePlanService(session)
>>> plans = service.get_all_sale_plans()
>>> print(len(plans))
"""
return self.session.query(SalePlan).order_by(SalePlan.sort_group).all()
[文档]
def get_by_sort_group(self, sort_group: str) -> list[SalePlan]:
"""根据分组键获取销售计划
Args:
sort_group: 分组键值
Returns:
List[SalePlan]: 指定分组的销售计划列表
"""
return self.session.query(SalePlan).filter(SalePlan.sort_group == sort_group).all()
[文档]
def get_by_product_model(self, product_model: str) -> list[SalePlan]:
"""根据产品型号获取销售计划
Args:
product_model: 产品型号
Returns:
List[SalePlan]: 指定产品型号的销售计划列表
"""
return self.session.query(SalePlan).filter(SalePlan.product_model == product_model).all()
[文档]
def clear_all(self) -> int:
"""清空所有销售计划数据
Returns:
int: 删除的记录数
"""
count = self.session.query(SalePlan).delete()
self.session.commit()
logger.info(f"清空销售计划: 删除 {count} 条记录")
return count
[文档]
def get_statistics(self) -> dict[str, Any]:
"""获取销售计划统计信息
Returns:
Dict[str, Any]: 统计信息字典，包含:
- total_records: 总记录数
- group_count: 分组数
- status_stats: 按生产状态分组统计
"""
from sqlalchemy import func
total = self.session.query(SalePlan).count()
status_stats = (
self.session.query(SalePlan.production_status, func.count(SalePlan.id))
.group_by(SalePlan.production_status)
.all()
)
group_count = self.session.query(SalePlan.sort_group).distinct().count()
return {
"total_records": total,
"group_count": group_count,
"status_stats": dict(status_stats),
}
def _save_records_to_db_with_stats(
self, records: list[dict[str, Any]], source_file: str
) -> dict[str, Any]:
"""保存记录到数据库并返回统计信息
供查询/导入集成测试使用的轻量入口：基于既有 ``save_handler``
完成去重与持久化，并返回 ``new_count``/``duplicate_count``。
Args:
records: 待保存的记录字典列表（字段与 SalePlan 模型对应）
source_file: 源文件路径（用于生成导入批次标识）
Returns:
Dict[str, Any]: 保存统计信息，包含:
- new_count: 新增记录数
- duplicate_count: 重复记录数
"""
from datetime import datetime
source_sheet = "Sheet1"
import_batch_id = (
f"{Path(source_file).stem}_{source_sheet}_"
f"{datetime.now().strftime('%Y%m%d_%H%M%S_%f')[:-3]}"
)
prefix = self._get_year_month_prefix(source_file, source_sheet)
year_month_int = int(prefix) if prefix.isdigit() else 0
result = self.save_handler.save_records_with_stats(
records=records,
source_file=source_file,
source_sheet=source_sheet,
import_batch_id=import_batch_id,
year_month_prefix=prefix,
year_month_int=year_month_int,
)
return {
"new_count": result["new_count"],
"duplicate_count": result["duplicate_count"],
}
[文档]
def get_by_status(self, status: str) -> list[SalePlan]:
"""根据格式状态获取销售计划
Args:
status: 格式状态值(如"已开票"、"已发货"等)
Returns:
List[SalePlan]: 指定格式状态的销售计划列表
"""
return self.session.query(SalePlan).filter(SalePlan.format_status == status).all()
[文档]
def get_status_statistics(self) -> dict[str, int]:
"""获取格式状态统计信息
Returns:
Dict: 各格式状态对应的记录数统计
"""
from sqlalchemy import func
stats = (
self.session.query(SalePlan.format_status, func.count(SalePlan.id))
.group_by(SalePlan.format_status)
.all()
)
return {status or "未标记": count for status, count in stats if status}
[文档]
def get_exportable_plans(self) -> list[SalePlan]:
"""获取可导出的销售计划（排除已发货、外购未回、隐藏行）
对应 VBA 中导出合格证时的过滤逻辑：
- 排除字体蓝色(已发货)的记录
- 排除背景橙棕色(外购未回)的记录
- 排除行高=0(隐藏/取消)的记录
Returns:
List[SalePlan]: 可导出的销售计划列表
"""
return (
self.session.query(SalePlan)
.filter(
~SalePlan.format_status.contains("已发货"),
~SalePlan.format_status.contains("外购未回"),
SalePlan.is_hidden == False, # noqa: E712
)
.order_by(SalePlan.sort_group)
.all()
)
# ============================================================
# 持久化方法（从 SaveHandler 上移的 ORM 操作）
# ============================================================
[文档]
def find_by_unique_key(self, unique_key: str) -> SalePlan | None:
"""根据唯一键查找已存在的销售计划
Args:
unique_key: 唯一键
Returns:
已存在的 SalePlan 或 None
"""
return self.session.query(SalePlan).filter(SalePlan.unique_key == unique_key).first()
[文档]
def find_by_id(self, plan_id: int) -> SalePlan | None:
"""按主键查询销售计划。
Args:
plan_id: 销售计划主键
Returns:
命中的 SalePlan 或 None
"""
return self.session.get(SalePlan, plan_id)
[文档]
def find_by_production_order_no(self, production_order_no: str) -> SalePlan | None:
"""根据生产令号查找已存在的销售计划
用于老数据模式降级去重：当唯一键匹配失败但生产令号有值时，
按生产令号查找已有记录，避免同一记录从不同来源重复导入。
Args:
production_order_no: 生产令号
Returns:
已存在的 SalePlan 或 None
"""
return (
self.session.query(SalePlan)
.filter(SalePlan.production_order_no == production_order_no)
.first()
)
[文档]
def find_by_contract_and_model(self, contract_no: str, product_model: str) -> SalePlan | None:
"""根据合同号 + 产品型号查找已存在的销售计划
用于跨来源补充场景的兜底匹配：当唯一键和生产令号都匹配失败时，
通过合同号 + 产品型号定位已有记录，实现缺失字段补充。
Args:
contract_no: 合同号
product_model: 产品型号
Returns:
已存在的 SalePlan 或 None
"""
return (
self.session.query(SalePlan)
.filter(
SalePlan.contract_no == contract_no,
SalePlan.product_model == product_model,
)
.first()
)
[文档]
def add_sale_plan(self, sale_plan: SalePlan) -> None:
"""添加销售计划到会话
Args:
sale_plan: SalePlan 对象
"""
self.session.add(sale_plan)
[文档]
def add_change(self, change: SalePlanChange) -> None:
"""添加变更记录到会话
Args:
change: SalePlanChange 对象
"""
self.session.add(change)
[文档]
def flush_session(self) -> None:
"""刷新会话"""
self.session.flush()
[文档]
def commit_session(self) -> None:
"""提交事务"""
self.session.commit()
[文档]
def rollback_session(self) -> None:
"""回滚事务"""
self.session.rollback()
# 允许手工订正的合格证字段（仅 cert_* + 合格证备注，不动合同字段）
CERT_EDITABLE_FIELDS: tuple[str, ...] = (
"cert_product_name",
"cert_product_model",
"cert_product_spec",
"certificate_remarks",
)
[文档]
def update_cert_fields(self, plan_id: int, fields: dict[str, Any]) -> bool:
"""仅更新合格证打印字段（蓝图 §4.2.3：查询视图 cert-only 订正）
只写入 ``cert_product_name/model/spec`` 与 ``certificate_remarks``，
绝不触碰合同字段（``product_*`` 等）。这是合格证清单/打印的数据源，
订正即时生效于后续 ``create_certificates``。
Args:
plan_id: 销售计划记录主键
fields: 待写入字段字典；非合格证字段会被静默忽略
Returns:
bool: 是否成功更新（id 不存在返回 False）
"""
if not fields:
return False
allowed = {k: v for k, v in fields.items() if k in self.CERT_EDITABLE_FIELDS}
if not allowed:
return False
plan = self.session.get(SalePlan, plan_id)
if plan is None:
logger.warning(f"update_cert_fields 跳过：id={plan_id} 不存在")
return False
try:
for key, value in allowed.items():
setattr(plan, key, value)
self.session.commit()
logger.info(f"订正合格证字段 | id={plan_id} | 字段={sorted(allowed.keys())}")
return True
except SQLAlchemyError as e:
self.session.rollback()
logger.error(f"订正合格证字段失败，已回滚: {e}")
raise
[文档]
def count_sale_plans(self) -> int:
"""统计销售计划总数"""
return self.session.query(SalePlan).count()
# ============================================================
# 已发货归档 / 导出+删除（三层数据生命周期）
# ============================================================
[文档]
def archive_to_shipped(self, sale_plan_ids: list[int]) -> int:
"""将选中的 SalePlan 归档到 sale_plans_shipped 表并从主表移除
三层模型的 Layer 1 → Layer 2:
- 将记录完整拷贝到 sale_plans_shipped（设置 shipped_at / shipped_by="archive"）
- 从 sale_plans 表删除对应记录
- 同时将 production_status 设为"已发货"（如果还不是）
Args:
sale_plan_ids: 要归档的 SalePlan ID 列表
Returns:
成功归档的记录数
"""
if not sale_plan_ids:
return 0
plans = self.session.query(SalePlan).filter(SalePlan.id.in_(sale_plan_ids)).all()
if not plans:
return 0
return self.shipped_handler.batch_archive(plans, shipped_by="archive")
[文档]
def export_and_delete(self, sale_plan_ids: list[int], output_dir: str) -> dict[str, Any]:
"""导出选中 SalePlan 的完整字段 Excel，归档到 shipped 表，并从主表删除
三层模型的 Layer 1 → Layer 3:
- 生成完整字段 Excel 文件
- 将记录拷贝到 sale_plans_shipped（shipped_by="export_delete"）
- 从 sale_plans 表删除
Args:
sale_plan_ids: 要导出的 SalePlan ID 列表
output_dir: Excel 输出目录
Returns:
{"count": N, "filepath": "..."}
"""
if not sale_plan_ids:
return {"count": 0, "filepath": ""}
plans = self.session.query(SalePlan).filter(SalePlan.id.in_(sale_plan_ids)).all()
if not plans:
return {"count": 0, "filepath": ""}
# 先导出 Excel（在删除前）
result = self.shipped_handler.export_full_excel(plans, output_dir)
# 再归档 + 删除
self.shipped_handler.batch_archive(plans, shipped_by="export_delete")
return result
# ============================================================
# 通用导出方法（委托 Handler）
# ============================================================
[文档]
def export_selected_records(
self, records: list[Any], columns: list[dict], file_path: str
) -> int:
"""导出选中记录到 Excel（按列定义）
Args:
records: ORM 记录列表
columns: 列定义列表 [{"field": "...", "label": "..."}, ...]
file_path: 目标文件路径
Returns:
int: 导出记录数
"""
return self.shipped_handler.export_selected_to_excel(records, columns, file_path)
[文档]
def export_query_result(self, records: list[Any], columns: list[dict], file_path: str) -> int:
"""导出查询结果到 Excel（带 plan_date 格式化）
Args:
records: ORM 记录列表
columns: 列定义列表
file_path: 目标文件路径
Returns:
int: 导出记录数
"""
return self.shipped_handler.export_query_result_to_excel(records, columns, file_path)