"""数据库保存处理器模块
提供销售计划记录的数据转换和变更检测功能，
ORM 操作通过回调委托给 Service 层。
"""
from __future__ import annotations
import hashlib
import json
import re
from datetime import datetime
from typing import Any, Protocol
from loguru import logger
from openpyxl.styles.colors import COLOR_INDEX
from sqlalchemy.orm import Session
from certflow.config.settings import (
DUAL_FIELD_MAPPING,
ON_COLLISION,
REQUIRE_NUMBERED,
UPDATE_ON_DUPLICATE,
)
from certflow.handlers import IDGenerator
from certflow.models import SalePlan, SalePlanChange
from certflow.utils import extract_yymm_from_date
from certflow.utils.date_utils import normalize_plan_date
from certflow.utils.status_inference import StatusInference
# 6 位十六进制颜色（RRGGBB / BGR 字面量等）
_HEX6_RE = re.compile(r"^[0-9A-Fa-f]{6}$")
[文档]
def resolve_dual_fields(
record: dict[str, Any],
import_mode: str,
pairs: list[dict[str, str]],
) -> dict[str, str]:
"""按 import_mode 计算 cert_* 合格证字段值（蓝图 §4.2.1）。
Args:
record: 已映射字段的导入记录（可能含 cert_* 键，仅当 Excel 提供对应列）
import_mode: ``copy`` / ``override`` / ``lazy``
pairs: 双字段映射对列表，每项 ``{contract, cert}``
Returns:
dict: ``{cert_field: value}``，仅含 pairs 中声明的 cert 字段
规则：
- ``copy``：cert_* = 合同值（杜绝"改合同静默跟随"风险）
- ``override``：仅当源行含 cert 列（Excel 提供）时写 cert_*，否则留空回退
- ``lazy``（默认）：cert_* 留空，打印时回退合同值（保持现状）
"""
import_mode = (import_mode or "lazy").strip().lower()
def _clean(v: Any) -> str:
s = "" if v is None else str(v)
return "" if "【空白" in s else s.strip()
result: dict[str, str] = {}
for p in pairs or []:
contract = p.get("contract")
cert = p.get("cert")
if not contract or not cert:
continue
contract_val = _clean(record.get(contract, ""))
cert_src = _clean(record.get(cert, ""))
if import_mode == "copy":
result[cert] = contract_val
elif import_mode == "override":
result[cert] = cert_src # 仅当 Excel 提供 cert 列时非空
else: # lazy（默认）
result[cert] = ""
return result
def _indexed_to_bgr(idx: int) -> int | None:
"""openpyxl 索引色调色板 → BGR int（与其余分支统一 BGR 方向）。"""
if not (0 <= idx < len(COLOR_INDEX)):
return None
hexv = COLOR_INDEX[idx]
if not hexv:
return None
hexv = hexv[-6:] # 去掉可能的 alpha 前缀
return int(hexv[4:6] + hexv[2:4] + hexv[0:2], 16)
def _coerce_color_int(v: Any) -> int | None:
"""颜色值统一转为整数 key（BGR 格式，与 config.yaml 的 bg/font 映射对齐）
支持来源：
- int / float：直接取整（config 的 BGR 整数键）
- 6 位十六进制字符串 ``"RRGGBB"``：**按 BGR 方向反转**为 int
（openpyxl 存的是 RGB 字面量，config 键是 BGR，方向必须反转；
例如字体蓝 ``"0000FF"`` → 255<<16 = 16711680）
- ``"INDEXED_n"``：经 openpyxl 标准索引色调色板转为 BGR int
- ``"THEME_n"``：主题色（透明/默认，无业务信号）→ 返回 None
- 纯数字字符串：按 int 解析
旧实现仅 ``int()`` 数字，遇到 ``"INDEXED_27"`` / ``"0000FF"`` / ``"THEME_0"``
一律返回 None，导致背景色 / 字体色派生全部失效、回落错误默认。
"""
if v is None or isinstance(v, bool):
return None
if isinstance(v, int | float):
return int(v)
s = str(v).strip()
if not s:
return None
if _HEX6_RE.match(s):
s = s.upper()
# RRGGBB -> BGR int
return int(s[4:6] + s[2:4] + s[0:2], 16)
if s.upper().startswith("INDEXED_"):
try:
return _indexed_to_bgr(int(s.split("_", 1)[1]))
except (ValueError, IndexError):
return None
if s.upper().startswith("THEME_"):
# 主题色在 openpyxl 中无法可靠转为 BGR（且与白透明塌缩），视为无信号
return None
try:
return int(s)
except ValueError:
return None
def _comments_to_text(comments: Any) -> str:
"""将 _comments 列表（[{column, text}, ...]）拼为单行变更说明文本。
用于在 SalePlan.change_note 中保留单元格批注（变更说明）的可追溯原文。
"""
if not comments:
return ""
parts: list[str] = []
for c in comments:
if isinstance(c, dict):
text = str(c.get("text", "")).strip()
if not text:
continue
col = str(c.get("column", "")).strip()
parts.append(f"[{col}] {text}" if col else text)
elif c and str(c).strip():
parts.append(str(c).strip())
return "\n".join(parts)
def _refresh_comment_fields(existing: SalePlan, record: dict[str, Any]) -> None:
"""重导入时刷新 技术要求原文 与 批注（来自 Excel 最新单元格/批注）。
- technical_requirement_raw：仅当本次导入的技术要求非空时覆盖，避免清空首发快照。
- change_note：仅当本次记录携带批注时刷新（无批注行保持既有值）。
- comments：批注 JSON（[{column, text}, ...]），仅当本次记录携带批注时刷新；
无批注行保持既有值，避免清空首发导入时已落库的批注原文。
"""
raw = record.get("tech_requirements")
if raw not in (None, ""):
existing.technical_requirement_raw = str(raw)
logger.debug(f"刷新 technical_requirement_raw | id={existing.id} 长度={len(str(raw))}")
comments = record.get("_comments")
if comments:
existing.change_note = _comments_to_text(comments)
existing.comments = json.dumps(comments, ensure_ascii=False)
n = len(comments) if isinstance(comments, list) else "?"
logger.debug(f"刷新批注 | id={existing.id} 批注条数={n}")
[文档]
class SaveHandlerOrmDelegate(Protocol):
"""SaveHandler 的 ORM 委托协议
Service 层实现此协议，Handler 通过此接口间接操作数据库。
"""
[文档]
def find_by_unique_key(self, unique_key: str) -> SalePlan | None:
"""按唯一键查找销售计划记录。
Args:
unique_key: 记录唯一键（由 IDGenerator 生成）。
Returns:
SalePlan | None: 匹配的销售计划记录；未找到时返回 None。
"""
...
[文档]
def find_by_production_order_no(self, production_order_no: str) -> SalePlan | None:
"""按生产令号查找销售计划记录（老数据降级匹配）。
用于缺少唯一键的老数据模式，与生产令号精确匹配。
Args:
production_order_no: 生产令号。
Returns:
SalePlan | None: 匹配的销售计划记录；未找到时返回 None。
"""
...
[文档]
def find_by_contract_and_model(self, contract_no: str, product_model: str) -> SalePlan | None:
"""按合同号 + 产品型号兜底匹配（跨来源补充场景）。
当 unique_key 与生产令号均不匹配时，用合同号与型号跨来源
补充缺失字段。
Args:
contract_no: 合同号。
product_model: 产品型号。
Returns:
SalePlan | None: 匹配的销售计划记录；未找到时返回 None。
"""
...
[文档]
def add_sale_plan(self, sale_plan: SalePlan) -> None:
"""向会话新增一条销售计划记录（不自动提交）。
Args:
sale_plan: 待新增的 SalePlan 实例。
"""
...
[文档]
def add_change(self, change: SalePlanChange) -> None:
"""向会话新增一条变更记录（不自动提交）。
Args:
change: 待新增的 SalePlanChange 实例。
"""
...
[文档]
def flush_session(self) -> None:
"""刷新会话，将挂起的对象写入数据库（不提交事务）。"""
...
[文档]
def commit_session(self) -> None:
"""提交事务，持久化会话中所有挂起的变更。"""
...
[文档]
class SaveHandler:
"""数据库保存处理器
负责数据转换和变更检测，ORM 操作通过 orm_delegate 回调委托给 Service 层。
Examples:
>>> handler = SaveHandler(session, orm_delegate=service)
>>> imported, skipped = handler.save_styled_records(
... sorted_records, file_path, sheet_name, batch_id, "2601", 2601
... )
"""
def __init__(
self,
session: Session,
orm_delegate: SaveHandlerOrmDelegate | None = None,
):
"""初始化保存处理器
Args:
session: SQLAlchemy 数据库会话对象（用于只读查询）
orm_delegate: ORM 操作委托对象（Service 层实现）
"""
self.session = session
self._orm = orm_delegate
# 本批次已插入记录的 unique_key 集合，用于在兜底匹配时排除
# 同批次内刚插入的行（避免把不同 unique_key 但共享生产令号/
# 合同号+型号的行误判为"已存在"而漏存并产生虚假变更）。
self._batch_inserted_keys: set[str] = set()
def _find_existing(self, record: dict[str, Any], unique_key: str) -> SalePlan | None:
"""查找已有记录：三层降级匹配策略。
1. 按 unique_key 精确匹配
2. 按 production_order_no 匹配（老数据模式降级）
3. 按 contract_no + product_model 匹配（跨来源补充场景）
老数据模式下，同一条记录从不同来源导入时字段值可能有差异，
导致 unique_key 不匹配。通过 contract_no + product_model 兜底匹配，
实现缺失字段的跨来源补充。
"""
orm = self._orm
if orm is None:
return None
existing = orm.find_by_unique_key(unique_key)
if self._match_ok(existing) is not None:
return existing
# 兜底匹配只应命中「既往批次」记录；本批次内刚插入的行
# （unique_key 不同但共享生产令号 / 合同号+型号）应视为独立新记录，
# 否则会漏存数据并产生虚假「变更」。
def _is_prior_batch(record_: SalePlan) -> bool:
return record_.unique_key not in self._batch_inserted_keys
pno = str(record.get("production_order_no", "")).strip()
if pno and not pno.startswith(IDGenerator._get_empty_marker_prefix()):
cand = orm.find_by_production_order_no(pno)
if cand and _is_prior_batch(cand) and self._match_ok(cand) is not None:
logger.debug(f"按生产令号 {pno} 匹配到已有记录 (id={cand.id})")
return cand
# 第三级降级：按 contract_no + product_model 兜底匹配
contract_no = str(record.get("contract_no", "")).strip()
product_model = str(record.get("product_model", "")).strip()
if (
contract_no
and not contract_no.startswith(IDGenerator._get_empty_marker_prefix())
and product_model
and not product_model.startswith(IDGenerator._get_empty_marker_prefix())
):
cand = orm.find_by_contract_and_model(contract_no, product_model)
if cand and _is_prior_batch(cand) and self._match_ok(cand) is not None:
logger.debug(
f"按合同号 {contract_no} + 型号 {product_model} 匹配到已有记录 (id={cand.id})"
)
return cand
return None
@staticmethod
def _match_ok(cand: SalePlan | None) -> SalePlan | None:
"""require_numbered 门控（§9 落地待办①）
返回 cand 本身（命中可用）；若启用 require_numbered 且 cand 未编号
（无证书号），则视为新行不参与判重，返回 None。
"""
if cand is None:
return None
if REQUIRE_NUMBERED and not (cand.certificate_number or "").strip():
logger.debug(f"require_numbered: 命中行 id={cand.id} 未编号，视为新行不判重")
return None
return cand
def _handle_existing(
self,
existing: SalePlan,
record: dict[str, Any],
source_file: str,
source_sheet: str,
import_batch_id: str,
) -> int:
"""处理匹配到的已有记录：补充缺失字段 + 检测变更。
Returns:
产生的变更记录数
"""
orm = self._orm
filled_fields = self._fill_missing_fields(existing, record)
if filled_fields:
logger.info(f"记录 {existing.id} 补充缺失字段: {', '.join(filled_fields)}")
# B0-2 / B0-5：无论是否启用变更追踪，重复导入都刷新 技术要求原文 与 变更说明
_refresh_comment_fields(existing, record)
if not UPDATE_ON_DUPLICATE:
return []
# §9 落地待办②：变更签名短路——身份键命中时若内容签名一致则判定无变更，
# 跳过逐字段 diff（签名由 monitored_fields 摘要，与 _detect_changes 判定等价）。
new_sig = IDGenerator.generate_change_signature(record)
if existing.change_signature and existing.change_signature == new_sig:
changes = []
else:
changes = self._detect_changes(
existing, record, source_file, source_sheet, import_batch_id
)
existing.change_signature = new_sig
# #30 P0 颜色语义显式化：带格式重导入时刷新 供货类型/发货状态
self._refresh_styled_fields(existing, record)
# #30 P0 DN/PN 字典 DB 化：重复导入时刷新解析压力/标准号与标黄。
# 压力/标准号非空才覆盖，避免清空人工修正值；标黄按当前记录的解析信号
# 重新计算（不再 OR 既有 flag）——字典补全后即解除标黄，符合「重导刷新」语义。
if record.get("pressure_value"):
existing.pressure_value = record["pressure_value"]
if record.get("test_standard"):
existing.test_standard = record["test_standard"]
existing.flag = bool(
record.get("flag", False)
or record.get("spec_needs_manual", False)
or record.get("pn_needs_manual", False)
)
if changes and orm:
for change in changes:
orm.add_change(change)
orm.flush_session()
logger.debug(f"记录 {existing.id} 有 {len(changes)} 个字段变更")
# 返回本记录发生的字段变更列表（空列表=无变更），供上层分类统计。
return changes
def _refresh_styled_fields(self, existing: SalePlan, record: dict[str, Any]) -> None:
"""带格式重导入时刷新 供货类型/发货状态（仅覆盖可判定值，不降级、不还原手动值）。
仅在记录确为带格式导入（含颜色键）时生效；供货类型只覆盖已判定值
或既往为空者，绝不把既有的 自产/外购 降级为 待定。
发货状态(shipping_status) 溯源保护（#30 P0 A 方案）：
- 若既有来源为 UI 手动设置(source=='manual')，则跳过字体色覆盖，
避免下月重导入把人工修正值还原为颜色派生值；
- 否则覆盖，并写入 source='import'、set_at=now 的溯源信息；
- 单调守卫：已发货/已收款 等终态不被降级为 未发货；
- 当本次派生为 已发货 且 shipped_date 为空时，记录发货日期(导入日期)。
"""
if "_background_colors" not in record and "_font_colors" not in record:
return
supply, shipping_status = self._resolve_color_statuses(record)
if shipping_status:
self._apply_shipping_status(existing, shipping_status)
if supply and supply != "待定":
existing.outsource_type = supply
existing.supply_type = supply
elif not (existing.supply_type or "").strip():
# 仅当既往为空时才落“待定”占位，绝不覆盖已判定值
existing.outsource_type = supply
existing.supply_type = supply
@staticmethod
def _apply_shipping_status(existing: SalePlan, shipping_status: str) -> None:
"""应用派生发货状态并维护溯源（#30 P0 A 方案）
规则见 ``_refresh_styled_fields`` 文档。手动来源受保护，终态不被降级。
"""
# 1) 手动来源受保护：跳过字体色覆盖，避免还原人工修正
if (existing.shipping_status_source or "import") == "manual":
logger.debug(
f"记录 {existing.id} 发货状态为手动设置(source=manual)，"
f"跳过导入派生覆盖(派生值={shipping_status})"
)
return
# 2) 应用并写入导入溯源（带格式重导入以当前字体色派生为准，刷新终态）
if existing.shipping_status != shipping_status:
existing.shipping_status = shipping_status
existing.shipping_status_source = "import"
existing.shipping_status_set_at = datetime.now()
# 4) 首次判定为已发货时记录发货日期
if shipping_status == "已发货" and not (existing.shipped_date or "").strip():
existing.shipped_date = datetime.now().strftime("%Y-%m-%d")
@staticmethod
def _resolve_color_statuses(
record: dict[str, Any],
bg_map: dict[Any, str] | None = None,
font_map: dict[int, str] | None = None,
) -> tuple[str, str]:
"""由已捕获的颜色字典 + 执列文字派生 供货类型 与 发货状态（#30 P0）
颜色 JSON（font_colors / background_colors）保留为真相源，本列为派生物。
供货类型(supply_type) 派生优先级（对齐用户最终裁定）：
1. ``执`` 列文字（已随列映射进入 ``record["supply_type"]``，最强信号）：
``生产`` → 自产；供应商名等其它非空 → 外购；
2. 背景色（次要信号）：命中 外购未回/外购已回/外购 → 外购；
3. 2025+ 经 AO(销售订单号)+AP(生产令号) 确认真实 ERP 订单，但 AP 顺序分配
无法区分自产/外购 → 待定；
4. 其余无可判信号 → **待定**（绝不猜“自产”，透明/白 ≠ 自产）。
发货状态(shipping_status) 由字体色派生：蓝(16711680)→已发货、黑(0)→未发货、
红(255)→已收款。未捕获字体色时返回空，避免覆盖既有值。
Args:
record: 单条导入记录
bg_map: 背景色映射（BGR 整数键或 "INDEXED_n" 字符串键；默认读配置）
font_map: 字体色映射（默认读配置）
Returns:
(supply_type, shipping_status)
"""
from certflow.config.settings import (
BG_COLOR_TO_STATUS,
FONT_COLOR_TO_SHIPPING,
)
if bg_map is None:
bg_map = BG_COLOR_TO_STATUS or {}
if font_map is None:
font_map = FONT_COLOR_TO_SHIPPING or {}
bg = record.get("_background_colors") or {}
font = record.get("_font_colors") or {}
if not isinstance(bg, dict):
bg = {}
if not isinstance(font, dict):
font = {}
# 非带格式导入（record 不含任何颜色键）→ 返回空，避免覆盖既有有效值
if "_background_colors" not in record and "_font_colors" not in record:
return "", ""
supply = SaveHandler._resolve_supply_type(record, bg, bg_map)
shipping = SaveHandler._resolve_shipping_status(font, font_map)
return supply, shipping
@staticmethod
def _resolve_supply_type(
record: dict[str, Any], bg: dict[Any, Any], bg_map: dict[Any, str]
) -> str:
"""供货类型派生：执文字 > 背景色(直接映射) > 待定
背景色经 openpyxl 读取后可能为 ``"INDEXED_27"`` / ``"THEME_0"`` 等字符串，
故映射查询同时尝试整数 key 与原始字符串 key（config.yaml 两种都配）。
背景色映射 ``bg_map`` 已是显式状态（外购未回/外购已回/自产/待编号），
直接采用，不再折叠为 外购/待定。
"""
# 1) 执 列文字（最强信号）
zhi = str(record.get("supply_type", "") or "").strip()
# 导入管线会把空单元格填成 “【空白供货类型】” 等占位符，视为无信号
if zhi.startswith("【空白"):
zhi = ""
if zhi == "生产":
return "自产"
if zhi: # 供应商名等 → 外购
return "外购"
# 2) 背景色（次要信号）：仅命中明确的外部采购状态才派生
# 外购未回/外购已回/外购 → 采用该显式状态；
# 其余背景色（透明/待编号/特殊订单/订单取消等）不参与供货类型判定，
# 交由下方默认分支兜底。
for v in bg.values():
vi = _coerce_color_int(v)
s = bg_map.get(vi) if vi is not None else None
if s is None:
s = bg_map.get(str(v)) # 支持字符串 key（如 "INDEXED_27"）
if s in ("外购未回", "外购已回", "外购"):
return s
# 3) / 4) 无强信号 → 默认 自产（带格式导入的白色/透明背景即视为自产）
return "自产"
@staticmethod
def _resolve_shipping_status(font: dict[Any, Any], font_map: dict[int, str]) -> str:
"""字体色字典 → 发货状态（蓝→已发货、红→红字、黑→未发货）
字体色字典为空（非带格式导入）时按既定默认返回 未发货
（调用方在既无背景色也无字体色时已先行返回空，避免覆盖既有有效值）。
"""
for v in font.values():
vi = _coerce_color_int(v)
if vi is None:
continue
s = font_map.get(vi)
if s == "已发货":
return "已发货"
if s == "红字":
return "红字"
if s == "已收款":
return "已收款"
return "未发货"
[文档]
def save_styled_records( # noqa: C901
self,
sorted_records: list[dict[str, Any]],
file_path: str,
sheet_name: str,
import_batch_id: str,
year_month_prefix: str,
year_month_int: int,
conflict_resolutions: dict[str, str] | None = None,
collision_policy: str | None = None,
) -> dict[str, Any]:
"""保存带格式的分组记录到数据库
Args:
sorted_records: 已分组排序的记录列表
file_path: 源文件路径
sheet_name: 工作表名称
import_batch_id: 导入批次ID
year_month_prefix: 年月前缀字符串，如 "2601"
year_month_int: 年月前缀整数，如 2601
conflict_resolutions: db_conflict 决议（unique_key -> 动作）
collision_policy: 批次内撞键处理策略（UI 选择，覆盖 ON_COLLISION 配置）：
isolate（隔离，默认）/ first（取首行）/ discard（丢弃全部重复）
Returns:
dict: {'new_count': int, 'duplicate_count': int, 'changes_count': int, ...}
- new_count: 新增记录数
- duplicate_count: 跳过（已存在且内容无变更）记录数；
注意「变更」行不计入此处，避免与 changes_count 重叠
- changes_count: 已存在但内容发生字段变更的记录数
- change_categories / changed_fields: 变更按字段归类统计
"""
orm = self._orm
imported_count = 0
skipped_count = 0
changes_count = 0
change_categories: dict[str, int] = {}
changed_fields: dict[str, int] = {}
quarantined_count = 0
collision_quarantined = 0 # 批次内撞键（unique_key 重复）隔离数
collision_dropped = 0 # 批次内撞键按 first/discard 直接丢弃的行数
resolution_quarantined = 0 # 冲突决议选「隔离」的隔离数
merged_count = 0 # 批内撞键按 on_collision=merge_* 合并的行数
import_date = datetime.now()
seen_keys: set[str] = set() # 同批次内存去重
self._seen_objects: dict[str, Any] = {} # 同批次首个插入的 ORM 对象（供 merge 更新数量）
self._batch_inserted_keys = set() # 同批次已插入的 unique_key
self._collision_policy = collision_policy # 显式撞键策略（UI 选择，覆盖 ON_COLLISION）
# 已隔离未处理行的记录内容哈希集合（幂等：避免重导时重复隔离撞键行）
existing_q_keys = self._load_existing_quarantine_hashes(file_path, sheet_name)
for original_idx, record in enumerate(sorted_records):
record = IDGenerator.normalize_spec(record, self.session)
record = IDGenerator.normalize_pn(record, self.session)
unique_key = IDGenerator.generate_unique_key(record)
# 1. 同批次内存去重
if unique_key in seen_keys:
# 撞键：按 collision_policy（UI 显式选择）或 ON_COLLISION（配置）处理
action = self._apply_batch_collision(
record,
unique_key,
existing_q_keys,
file_path,
sheet_name,
import_batch_id,
self._collision_policy,
)
if action == "merged":
merged_count += 1
elif action == "quarantine":
collision_quarantined += 1
else: # "drop"（取首行 / 丢弃全部重复）
collision_dropped += 1
continue
seen_keys.add(unique_key)
# 丢弃全部重复：首行也不入库（整组判废，不进主表/待处理表）
if self._collision_policy == "discard":
continue
# 生成序号字段
sort_group, original_order, sort_order = self._generate_order_fields(
record=record,
original_idx=original_idx,
year_month_prefix=year_month_prefix,
year_month_int=year_month_int,
)
# 2. 数据库去重（unique_key + 生产令号降级）
existing = self._find_existing(record, unique_key)
if existing:
action = (conflict_resolutions or {}).get(unique_key, "skip")
decision, delta, new_key = self._resolve_existing(
existing, record, unique_key, file_path, sheet_name, import_batch_id, action
)
if decision == "updated":
changes_count += delta
continue
if decision == "isolate":
resolution_quarantined += 1
continue
if decision == "new":
unique_key = new_key # 以新键插入，existing 保留
else:
# 与入库同清洗管线已在上游完成；此处仅判定内容是否变化。
# 注意：变更行不得计入「跳过」，否则「跳过」与「变更」重叠。
rec_changes = self._handle_existing(
existing, record, file_path, sheet_name, import_batch_id
)
if rec_changes:
changes_count += 1
self._accumulate_change_categories(
rec_changes, change_categories, changed_fields
)
else:
skipped_count += 1
continue
# 创建新记录
sale_plan = self._create_sale_plan(
record=record,
unique_key=unique_key,
original_order=original_order,
sort_order=sort_order,
sort_group=sort_group,
file_path=file_path,
sheet_name=sheet_name,
import_batch_id=import_batch_id,
import_date=import_date,
)
if orm:
orm.add_sale_plan(sale_plan)
self._seen_objects[unique_key] = sale_plan
self._batch_inserted_keys.add(unique_key)
imported_count += 1
if imported_count % 100 == 0 and orm:
orm.commit_session()
logger.info(f"已提交 {imported_count} 条记录")
if orm:
orm.commit_session()
_cat_str = (
", ".join(f"{k}:{v}" for k, v in change_categories.items()) if change_categories else ""
)
logger.info(
f"保存完成: 新增 {imported_count} 条, 跳过(无变更) {skipped_count} 条, "
f"变更 {changes_count} 条"
+ (f" [分类: {_cat_str}]" if _cat_str else "")
+ (
f", 批内撞键合并 {merged_count} 条(on_collision={ON_COLLISION})"
if merged_count
else ""
)
)
quarantined_count = collision_quarantined + resolution_quarantined
return {
"new_count": imported_count,
"duplicate_count": skipped_count,
"changes_count": changes_count,
"change_categories": change_categories,
"changed_fields": changed_fields,
"merged_count": merged_count,
"quarantined_count": quarantined_count,
"collision_quarantined": collision_quarantined,
"collision_dropped": collision_dropped,
"resolution_quarantined": resolution_quarantined,
}
[文档]
def save_records_with_stats( # noqa: C901
self,
records: list[dict[str, Any]],
source_file: str,
source_sheet: str,
import_batch_id: str,
year_month_prefix: str,
year_month_int: int,
conflict_resolutions: dict[str, str] | None = None,
collision_policy: str | None = None,
) -> dict[str, Any]:
"""保存记录到数据库，并返回统计信息（不带格式）
Args:
records: 待保存的记录字典列表
source_file: 源文件路径
source_sheet: 源工作表名称
import_batch_id: 导入批次ID
year_month_prefix: 年月前缀字符串
year_month_int: 年月前缀整数
Returns:
Dict[str, Any]: 保存统计信息
"""
from collections import defaultdict
orm = self._orm
new_count = 0
duplicate_count = 0
quarantined_count = 0
collision_quarantined = 0 # 批次内撞键（unique_key 重复）隔离数
collision_dropped = 0 # 批次内撞键按 first/discard 直接丢弃的行数
resolution_quarantined = 0 # 冲突决议选「隔离」的隔离数
merged_count = 0 # 批内撞键按 on_collision=merge_* 合并的行数
changes_count = 0
change_categories: dict[str, int] = {}
changed_fields: dict[str, int] = {}
duplicates = []
group_stats = defaultdict(int)
import_date = datetime.now()
seen_keys: set[str] = set()
self._seen_objects: dict[str, Any] = {} # 同批次首个插入的 ORM 对象（供 merge 更新数量）
self._batch_inserted_keys = set()
self._collision_policy = collision_policy # 显式撞键策略（UI 选择，覆盖 ON_COLLISION）
existing_q_keys = self._load_existing_quarantine_hashes(source_file, source_sheet)
for original_idx, record in enumerate(records):
record = IDGenerator.normalize_spec(record, self.session)
record = IDGenerator.normalize_pn(record, self.session)
unique_key = IDGenerator.generate_unique_key(record)
# 1. 同批次内存去重
if unique_key in seen_keys:
# 撞键：按 collision_policy（UI 显式选择）或 ON_COLLISION（配置）处理
action = self._apply_batch_collision(
record,
unique_key,
existing_q_keys,
source_file,
source_sheet,
import_batch_id,
self._collision_policy,
)
if action == "merged":
merged_count += 1
elif action == "quarantine":
collision_quarantined += 1
else: # "drop"（取首行 / 丢弃全部重复）
collision_dropped += 1
continue
seen_keys.add(unique_key)
# 丢弃全部重复：首行也不入库（整组判废，不进主表/待处理表）
if self._collision_policy == "discard":
continue
# 生成序号字段
sort_group, original_order, sort_order = self._generate_order_fields(
record=record,
original_idx=original_idx,
year_month_prefix=year_month_prefix,
year_month_int=year_month_int,
)
existing = self._find_existing(record, unique_key)
if existing:
action = (conflict_resolutions or {}).get(unique_key, "skip")
decision, delta, new_key = self._resolve_existing(
existing, record, unique_key, source_file, source_sheet, import_batch_id, action
)
if decision == "updated":
changes_count += delta
continue
if decision == "isolate":
resolution_quarantined += 1
continue
if decision == "new":
unique_key = new_key # 以新键插入，existing 保留
else:
# 变更行不计入「跳过」，避免「跳过」与「变更」重叠统计。
rec_changes = self._handle_existing(
existing, record, source_file, source_sheet, import_batch_id
)
if rec_changes:
changes_count += 1
self._accumulate_change_categories(
rec_changes, change_categories, changed_fields
)
else:
duplicate_count += 1
duplicate_info = {
"product_model": record.get("product_model", "N/A"),
"order_no": record.get("order_no", record.get("sales_order_no", "N/A")),
"reason": f"已存在 (ID: {existing.id})",
}
duplicates.append(duplicate_info)
continue
# 统计分组
group_key = record.get("_group_key", "")
if group_key:
group_stats[group_key] += 1
# 创建新记录
sale_plan = self._create_sale_plan(
record=record,
unique_key=unique_key,
original_order=original_order,
sort_order=sort_order,
sort_group=sort_group,
file_path=source_file,
sheet_name=source_sheet,
import_batch_id=import_batch_id,
import_date=import_date,
)
if orm:
orm.add_sale_plan(sale_plan)
self._seen_objects[unique_key] = sale_plan
self._batch_inserted_keys.add(unique_key)
new_count += 1
if orm:
orm.commit_session()
_cat_str = (
", ".join(f"{k}:{v}" for k, v in change_categories.items()) if change_categories else ""
)
logger.info(
f"成功保存 {new_count} 条记录，跳过(无变更) {duplicate_count} 条重复"
+ (f" [变更分类: {_cat_str}]" if _cat_str else "")
+ (
f", 批内撞键合并 {merged_count} 条(on_collision={ON_COLLISION})"
if merged_count
else ""
)
)
quarantined_count = collision_quarantined + resolution_quarantined
return {
"new_count": new_count,
"duplicate_count": duplicate_count,
"changes_count": changes_count,
"change_categories": change_categories,
"changed_fields": changed_fields,
"merged_count": merged_count,
"duplicates": duplicates,
"group_stats": dict(group_stats),
"quarantined_count": quarantined_count,
"collision_quarantined": collision_quarantined,
"collision_dropped": collision_dropped,
"resolution_quarantined": resolution_quarantined,
}
# ============================================================
# 撞键隔离（坑1 修复：撞键行进待处理表而非合并/覆盖）
# ============================================================
@staticmethod
def _record_content_hash(record: dict[str, Any]) -> str:
"""记录内容哈希（幂等隔离用）。同文件重导时内容一致 → 命中跳过。"""
try:
s = json.dumps(record, sort_keys=True, ensure_ascii=False, default=str)
except Exception:
s = json.dumps(record, sort_keys=True, default=str)
return hashlib.sha256(s.encode("utf-8")).hexdigest()
def _load_existing_quarantine_hashes(self, source_file: str, source_sheet: str) -> set[str]:
"""加载该 (文件, 表) 已隔离未处理行的记录内容哈希集合，供撞键隔离幂等判定。
以「记录内容」而非 unique_key 作幂等键：同组内多条撞键行（如分批发货、
数量不同）内容各异，都应各自隔离；重导同一文件时内容一致则命中跳过。
"""
hashes: set[str] = set()
if self.session is None:
return hashes
try:
from sqlalchemy import select
from certflow.models.quarantine_sale_plan import QuarantineSalePlan
stmt = select(QuarantineSalePlan).where(
QuarantineSalePlan.source_file == source_file,
QuarantineSalePlan.source_sheet == source_sheet,
QuarantineSalePlan.resolved == False, # noqa: E712
)
for q in self.session.execute(stmt).scalars().all():
try:
hashes.add(self._record_content_hash(json.loads(q.raw_record_json or "{}")))
except Exception:
continue
except Exception as e:
logger.warning(f"加载已隔离哈希失败(忽略): {e}")
return hashes
# ============================================================
# 冲突决议（B0-8：同 unique_key 但字段值不同的 5 选项处理）
# ============================================================
# 受保护、不被覆盖/合并写入的系统字段
_RESOLUTION_PROTECTED = {
"id",
"unique_key",
"created_at",
"updated_at",
"import_date",
"import_batch_id",
"source_file",
"source_sheet",
"original_order",
"sort_order",
"sort_group",
"change_signature",
"source_row_index",
}
# 变更分类（导入结果按字段归类，便于查询「哪些记录因何变更」）
CHANGE_CATEGORIES: dict[str, set[str]] = {
"发货日期": {"plan_date", "planned_delivery_date", "contract_delivery_date"},
"数量": {"quantity"},
"供货方式": {"supply_type"},
"产品信息": {"product_name", "product_model", "product_spec"},
"客户项目": {"customer", "project_name"},
}
@staticmethod
def _accumulate_change_categories(
changes: list[Any], category_counter: dict[str, int], field_counter: dict[str, int]
) -> None:
"""把一批 SalePlanChange 累加进分类计数与字段计数（一条记录可命中多类）。"""
for ch in changes:
field_name = getattr(ch, "field_name", None)
if not field_name:
continue
field_counter[field_name] = field_counter.get(field_name, 0) + 1
for cat, fields in SaveHandler.CHANGE_CATEGORIES.items():
if field_name in fields:
category_counter[cat] = category_counter.get(cat, 0) + 1
break
def _apply_record_to_existing(
self, existing: Any, record: dict[str, Any], overwrite: bool
) -> None:
"""把 incoming record 的字段写回 existing（覆盖/合并）。
- overwrite=True：record 中所有非空字段均写回（含空串会清空现有值）。
- overwrite=False（merge）：仅写回 record 中**非空**字段，空字段保留现有值。
系统字段（id/unique_key/时间戳等）受保护不被改写。
"""
for key, val in record.items():
if key.startswith("_") or key in self._RESOLUTION_PROTECTED:
continue
if not hasattr(existing, key):
continue
if val is None:
continue
sval = str(val).strip() if isinstance(val, str) else val
if not overwrite and (isinstance(sval, str) and sval == ""):
continue # merge：空值保留现有
setattr(existing, key, sval)
# B0-2 / B0-5：覆盖/合并决议时同步刷新 技术要求原文 与 变更说明
_refresh_comment_fields(existing, record)
existing.updated_at = datetime.now()
self.session.commit()
def _resolve_existing(
self,
existing: Any,
record: dict[str, Any],
unique_key: str,
source_file: str,
source_sheet: str,
import_batch_id: str,
action: str | None,
) -> tuple[str, int, str | None]:
"""处理「已存在记录（同 unique_key）」的冲突决议。
Returns:
(decision, change_delta, new_unique_key)
- decision:
"skip" 调用方按默认跳过（计 duplicate）
"updated" 已就地覆盖/合并 existing，调用方 continue（计 changes）
"isolate" 已将 incoming 隔离到待处理表，调用方 continue（计 quarantined）
"new" 调用方应以 new_unique_key 作为新记录插入（existing 不动）
- change_delta: updated 时为 1，其余 0
- new_unique_key: decision=="new" 时返回新键，否则 None
"""
action = (action or "skip").lower()
if action in ("", "skip"):
return ("skip", 0, None)
if action == "overwrite":
self._apply_record_to_existing(existing, record, overwrite=True)
return ("updated", 1, None)
if action == "merge":
self._apply_record_to_existing(existing, record, overwrite=False)
return ("updated", 1, None)
if action == "isolate":
q_keys = self._load_existing_quarantine_hashes(source_file, source_sheet)
self._route_collision_to_quarantine(
record, unique_key, source_file, source_sheet, import_batch_id, q_keys
)
return ("isolate", 0, None)
if action == "new":
from uuid import uuid4
new_key = f"{unique_key}_new_{uuid4().hex[:6]}"
return ("new", 0, new_key)
return ("skip", 0, None)
def _apply_batch_collision(
self,
record: dict[str, Any],
unique_key: str,
existing_q_keys: set[str],
file_path: str,
sheet_name: str,
import_batch_id: str,
collision_policy: str | None = None,
) -> str:
"""批内撞键(unique_key 重复)的处理。
显式 ``collision_policy``（UI 选择）优先于全局 ``ON_COLLISION`` 配置。
返回 "merged" / "quarantine" / "drop"，由调用方计数：
- merge_max / merge_sum：同键数量不同 → 合并（取大/求和）到首个插入的
ORM 对象(self._seen_objects[unique_key])，不隔离、不新增行；
- first（取首行）/ discard（丢弃全部重复）：后续撞键行直接丢弃（不隔离、不入库）；
首行是否入库由调用方在首次出现时按 discard 跳过决定；
- 其他（默认 isolate / quarantine）：隔离到待处理表。
"""
policy = collision_policy or ON_COLLISION
if policy in ("merge_max", "merge_sum") and unique_key in self._seen_objects:
kept = self._seen_objects[unique_key]
try:
cur = int(record.get("quantity") or 0)
except (TypeError, ValueError):
cur = 0
try:
base = int(getattr(kept, "quantity", 0) or 0)
except (TypeError, ValueError):
base = 0
merged = max(base, cur) if policy == "merge_max" else base + cur
kept.quantity = merged
logger.debug(
f"批内撞键合并(policy={policy}): {unique_key} 数量 {base} -> {merged} (本行 {cur})"
)
return "merged"
if policy in ("first", "discard"):
# 取首行 / 丢弃全部重复：后续撞键行直接丢弃（不隔离、不入库）
logger.debug(f"批内撞键丢弃(policy={policy}): {unique_key}")
return "drop"
self._route_collision_to_quarantine(
record, unique_key, file_path, sheet_name, import_batch_id, existing_q_keys
)
return "quarantine"
def _route_collision_to_quarantine(
self,
record: dict[str, Any],
unique_key: str,
source_file: str,
source_sheet: str,
import_batch_id: str,
existing_q_hashes: set[str],
) -> None:
"""撞键行隔离到 sale_plans_quarantine（而非合并/覆盖），供 UI 人工处理。
幂等：若该记录内容哈希已存在于未处理隔离行中，则不重复写入。
对应需求：空白库首导保留全量（804 进主表 + 13 进待处理），不动主表 UNIQUE 约束。
"""
content_hash = self._record_content_hash(record)
if content_hash in existing_q_hashes:
logger.debug(f"撞键行已隔离，跳过重复隔离: {unique_key}")
return
try:
from certflow.models.quarantine_sale_plan import QuarantineSalePlan
q = QuarantineSalePlan.from_record(
record,
meta={
"source_file": source_file,
"source_sheet": source_sheet,
"import_batch_id": import_batch_id,
"reason": "导入撞键(unique_key 重复，待人工合并/改源后重导)",
},
)
self.session.add(q)
existing_q_hashes.add(content_hash)
logger.info(f"撞键行已隔离到待处理表: {unique_key}")
except Exception as e:
logger.error(f"撞键行隔离失败(忽略，继续): {e}")
def _generate_order_fields(
self,
record: dict[str, Any],
original_idx: int,
year_month_prefix: str,
year_month_int: int,
) -> tuple[str, int, int]:
"""生成排序相关字段
Returns:
tuple: (sort_group, original_order, sort_order)
"""
# original_order: 原始导入顺序（年月 + 原始行号）
# year_month_int 可能为 None（如测试或缺失年月前缀时），缺省按 0 处理避免 None*int 崩溃。
ym = year_month_int or 0
original_order = ym * 100000 + (original_idx + 1)
# sort_group: 分组标识（年月-分组前缀）
# 使用 _group_prefix（如 G001）而非 _group_key（如 2026-01-01_客户A_项目X）
group_prefix = record.get("_group_prefix", "G000")
group_number = group_prefix[1:] if group_prefix.startswith("G") else group_prefix
sort_group = f"{year_month_prefix}-{group_prefix}"
# sort_order: 年月索引(3位) + 分组号(3位) + 组内序号(3位)
group_seq = record.get("group_seq", f"{original_idx + 1:03d}")
group_number_int = int(group_number) if group_number.isdigit() else 0
group_seq_int = int(group_seq) if group_seq.isdigit() else 0
# 计算年月索引 (2000年1月 = 1)
year = 2000 + (ym // 100)
month = ym % 100
month_index = (year - 2000) * 12 + month # 1-1200
sort_order = month_index * 1000000 + group_number_int * 1000 + group_seq_int
return sort_group, original_order, sort_order
@staticmethod
def _resolve_production_status(
record: dict[str, Any],
) -> tuple[str, str | None, bool]:
"""从记录中解析 production_status、execution_date 和 execution_inferred
优先保留已有的 production_status（非默认值时），否则从 execution_status 推断。
Args:
record: 原始数据行
Returns:
(production_status, execution_date, execution_inferred) 三元组
"""
raw_execution_status = record.get("execution_status", "")
existing_production_status = record.get("production_status", "")
execution_date: str | None = None
inferred_status = "待生产"
if raw_execution_status:
execution_date = StatusInference.extract_date(raw_execution_status)
inferred_status = StatusInference.infer(raw_execution_status)
if existing_production_status and existing_production_status != "待生产":
return existing_production_status, execution_date, False
return inferred_status, execution_date, bool(raw_execution_status)
def _create_sale_plan(
self,
record: dict[str, Any],
unique_key: str,
original_order: int,
sort_order: int,
sort_group: str,
file_path: str,
sheet_name: str,
import_batch_id: str,
import_date: datetime,
) -> SalePlan:
"""创建 SalePlan 实例"""
def _safe_get(val: str) -> str:
"""清理【空白】占位符
Args:
val: 输入字符串值
Returns:
清理后的字符串，如果包含【空白】占位符则返回空字符串
"""
return "" if "【空白" in val else val
def _safe_float(val: Any, default: float = 0.0) -> float:
"""安全转换为浮点数
Args:
val: 需要转换的值
default: 转换失败时的默认值
Returns:
转换后的浮点数，如果转换失败则返回默认值
"""
if val is None:
return default
if isinstance(val, (int | float)):
return float(val)
try:
# 清理【空白】占位符
str_val = str(val).strip()
if "【空白" in str_val:
return default
return float(str_val)
except (ValueError, TypeError):
return default
def _format_plan_no(record: dict[str, Any]) -> str:
"""格式化计划单号
规则:
- 纯数字且长度 <= 4: 补零到4位，如 123 -> 0123
- 纯数字且长度 > 4: 截取右边4位，如 54321 -> 4321
- 非纯数字或无 YYMM: 返回原值（清理【空白】）
Args:
record: 记录数据
Returns:
格式化后的计划单号
"""
plan_no_raw = str(record.get("plan_no", ""))
plan_date = record.get("plan_date")
yymm = extract_yymm_from_date(plan_date)
if plan_no_raw and plan_no_raw.isdigit() and yymm:
length = len(plan_no_raw)
formatted_no = plan_no_raw.zfill(4) if length <= 4 else plan_no_raw[-4:]
return f"{yymm}_{formatted_no}"
return _safe_get(plan_no_raw)
# 解析数量和日期
quantity = self._parse_quantity(record.get("quantity"))
plan_date = normalize_plan_date(record.get("plan_date", ""))
planned_delivery_date = normalize_plan_date(record.get("planned_delivery_date", ""))
contract_delivery_date = normalize_plan_date(record.get("contract_delivery_date", ""))
# 状态推断：从 execution_status 推断 production_status
raw_execution_status = record.get("execution_status", "")
production_status, execution_date, execution_inferred = self._resolve_production_status(
record
)
# 双字段映射（蓝图 §4.2.1）：按 import_mode 决定 cert_* 合格证字段写入
_dual_pairs = DUAL_FIELD_MAPPING.get("pairs", []) or []
_import_mode = (DUAL_FIELD_MAPPING.get("import_mode", "lazy") or "lazy").strip().lower()
_cert_fields = resolve_dual_fields(record, _import_mode, _dual_pairs)
# #30 P0 颜色语义显式化：由已捕获的颜色字典 + 执列文字派生 供货类型/发货状态
supply, shipping_status = self._resolve_color_statuses(record)
from certflow.services.cert_numbering_policy import derive_needs_numbering
readable_record = dict(record)
readable_record["product_spec"] = record.get(
"product_spec_norm", record.get("product_spec", "")
)
_raw_len = len(str(record.get("tech_requirements", "") or ""))
_cmt = record.get("_comments")
_cmt_len = len(_cmt) if isinstance(_cmt, list) else ("?" if _cmt else 0)
logger.debug(
f"新建 SalePlan | 捕获 technical_requirement_raw 长度={_raw_len} "
f"change_note 批注条数={_cmt_len}"
)
return SalePlan(
unique_key=unique_key,
unique_key_readable=IDGenerator.generate_unique_key_readable(readable_record),
original_order=original_order,
sort_order=sort_order,
sort_group=sort_group,
source_file=file_path,
source_sheet=sheet_name,
import_batch_id=import_batch_id,
import_date=import_date,
change_signature=IDGenerator.generate_change_signature(readable_record),
source_row_index=int(record.get("source_row_index", 0) or 0),
contract_no=_safe_get(str(record.get("contract_no", ""))),
sales_order_no=_safe_get(str(record.get("sales_order_no", ""))),
production_order_no=_safe_get(str(record.get("production_order_no", ""))),
plan_no=_format_plan_no(record),
plan_date=plan_date,
customer=_safe_get(str(record.get("customer", ""))),
project_name=_safe_get(str(record.get("project_name", ""))),
project_unit=_safe_get(str(record.get("project_unit", ""))),
product_name=_safe_get(str(record.get("product_name", ""))),
product_model=_safe_get(str(record.get("product_model", ""))),
product_spec=_safe_get(str(record.get("product_spec", ""))),
spec_norm=_safe_get(
str(record.get("product_spec_norm", record.get("product_spec", "")))
),
# 双字段映射（蓝图 §4.2.1）：cert_* 由 import_mode 决定（lazy 默认留空）
cert_product_name=_cert_fields.get("cert_product_name", ""),
cert_product_model=_cert_fields.get("cert_product_model", ""),
cert_product_spec=_cert_fields.get("cert_product_spec", ""),
# #30 P0 DN/PN 字典 DB 化：由 product_model 经 PNService 解析得到的公称压力/标准号
pressure_value=_safe_get(str(record.get("pressure_value", ""))),
test_standard=_safe_get(str(record.get("test_standard", ""))),
quantity=quantity,
weight=_safe_float(record.get("weight")),
category=record.get("category", ""),
unit_price=_safe_float(record.get("unit_price")),
total_price=_safe_float(record.get("total_price")),
business_dept=record.get("business_dept", ""),
affiliated_dept=record.get("affiliated_dept", ""),
payment_method=record.get("payment_method", ""),
supply_type=supply,
tech_requirements=record.get("tech_requirements", ""),
technical_requirement_raw=str(record.get("tech_requirements", "") or ""),
change_note=_comments_to_text(record.get("_comments", [])),
shipped_quantity=0,
shipment_batches="[]",
equipment_code=record.get("equipment_code", ""),
product_code=record.get("product_code", ""),
production_status=production_status,
execution_status=raw_execution_status,
execution_date=execution_date,
execution_inferred=execution_inferred,
sales_plan_remarks=record.get("sales_plan_remarks", ""),
order_no=record.get("order_no", record.get("sales_order_no", "")),
specification=record.get("specification", record.get("product_spec", "")),
material=record.get("material", ""),
format_status=record.get("_format_status", ""),
font_colors=json.dumps(record.get("_font_colors", {}), ensure_ascii=False),
background_colors=json.dumps(record.get("_background_colors", {}), ensure_ascii=False),
comments=json.dumps(record.get("_comments", []), ensure_ascii=False),
needs_numbering=derive_needs_numbering(record),
is_hidden=record.get("_hidden", False),
outsource_type=supply,
shipping_status=shipping_status,
# #30 P0 DN/PN 字典 DB 化：口径待复核(spec_needs_manual) 或 标准号缺失(pn_needs_manual)
# → 标黄放行（非 Stop）。原 record["flag"] 也保留（兼容其他来源）。
flag=bool(
record.get("flag", False)
or record.get("spec_needs_manual", False)
or record.get("pn_needs_manual", False)
),
planned_delivery_date=planned_delivery_date,
contract_delivery_date=contract_delivery_date,
)
# 当已有记录缺失、新数据有值时，需要补充填充的业务字段
# 这些字段不在 MONITORED_FIELDS 中，但在跨来源去重时应当被补全
_FILLABLE_FIELDS: list[str] = [
"production_order_no",
"contract_no",
"sales_order_no",
"supply_type",
"category",
"business_dept",
"affiliated_dept",
"payment_method",
"tech_requirements",
"equipment_code",
"product_code",
"material",
"order_no",
"specification",
"unit_price",
"total_price",
"weight",
]
@classmethod
def _fill_missing_fields(
cls,
existing: SalePlan,
record: dict[str, Any],
) -> list[str]:
"""补充已有记录中缺失的字段值（仅填充，不覆盖已有值）
当从两个不同来源（如"销售计划"和"本地计划"）导入同一条记录时，
其中一个来源可能缺少某些字段。此方法用新数据中的非空值补充
已有记录中的空字段，且只填充不覆盖。
Args:
existing: 已存在的销售计划记录
record: 新的记录数据
Returns:
被填充的字段名列表（用于日志记录）
"""
filled: list[str] = []
for field in cls._FILLABLE_FIELDS:
if field not in record:
continue
new_raw = record.get(field)
# 将 None 和空字符串等同处理
new_val = str(new_raw).strip() if new_raw is not None else ""
if not new_val:
continue
# 跳过空白占位符
if "【空白" in new_val:
continue
old_val = str(getattr(existing, field, None) or "").strip()
if not old_val:
try:
setattr(existing, field, new_raw)
except (TypeError, ValueError):
setattr(existing, field, new_val)
filled.append(field)
return filled
def _detect_changes(
self,
existing: SalePlan,
record: dict[str, Any],
source_file: str,
source_sheet: str,
import_batch_id: str,
) -> list[SalePlanChange]:
"""检测记录变更
Args:
existing: 已存在的销售计划记录
record: 新的记录数据
source_file: 数据来源文件
source_sheet: 数据来源工作表
import_batch_id: 导入批次ID
Returns:
变更记录列表
"""
from certflow.config.settings import MONITORED_FIELDS
from certflow.utils.date_utils import normalize_plan_date
# 需要统一格式化的日期字段列表
date_fields = ["plan_date", "planned_delivery_date", "contract_delivery_date"]
changes = []
for field in MONITORED_FIELDS:
old_val = str(getattr(existing, field, "") or "")
# supply_type 为派生字段（'执'文本+背景色经 _resolve_color_statuses 派生），
# 落库存的是派生值；比对须用「派生值 vs 派生值」，否则原始'执'文本与派生值
# （生产→自产、供应商名→外购）必然不等，产生假阳性变更。
# 仅带格式导入（含颜色键）可重新派生；非带格式导入无颜色真相源，跳过比对
# （与 _create_sale_plan 中 supply_type 留空的行为一致，避免假阳性/误覆盖）。
if field == "supply_type" and (
"_background_colors" in record or "_font_colors" in record
):
new_val = SaveHandler._resolve_color_statuses(record)[0]
elif field == "supply_type":
continue
else:
new_val = str(record.get(field, "") or "")
# 对所有日期字段统一格式化为 YYYY-MM-DD
if field in date_fields:
old_val = normalize_plan_date(old_val) or ""
new_val = normalize_plan_date(new_val) or ""
old_normalized = self._normalize_for_comparison(old_val)
new_normalized = self._normalize_for_comparison(new_val)
if old_normalized != new_normalized:
# supply_type 特殊保护：如果新值为空（或空白占位符），保留已有有效值不覆盖
if field == "supply_type" and not new_normalized and old_normalized:
continue # 跳过：不能把已有的有效供货方式覆盖为空
# 数量合并保护：分批发货的结转行数量(余数)小于原始订单量时，
# 保留较大的原始订单总量，避免被后期余数覆盖成"1台"。
if field == "quantity":
old_q = self._parse_quantity(old_val)
new_q = self._parse_quantity(new_val)
if new_q <= old_q:
continue # 余数不大于原量 → 保留原值，不记变更
effective_new_val = new_val
else:
effective_new_val = new_val
changes.append(
SalePlanChange(
sale_plan_id=existing.id,
unique_key=existing.unique_key,
field_name=field,
old_value=old_val,
new_value=effective_new_val,
source_file=source_file,
source_sheet=source_sheet,
import_batch_id=import_batch_id,
comment_text=str(record.get("_comments", "")),
)
)
setattr(existing, field, effective_new_val)
return changes
@staticmethod
def _normalize_for_comparison(value: str) -> str:
"""规范化值用于比较
Args:
value: 需要规范化的字符串值
Returns:
规范化后的字符串值
"""
import re
if not value:
return ""
if re.match(r"^【空白.*】$", value):
return ""
return value.strip()
@staticmethod
def _parse_quantity(quantity_val: Any) -> int:
"""解析数量字段为整数
Args:
quantity_val: 需要解析的数量值
Returns:
解析后的整数数量值
"""
quantity_str = str(quantity_val if quantity_val else "1")
try:
return (
int(float(quantity_str))
if quantity_str.replace(".", "").replace("-", "").isdigit()
else 1
)
except (ValueError, TypeError):
return 1
@staticmethod
def _parse_plan_date(plan_date_raw: Any) -> str | None:
"""解析计划日期，返回 YYYY-MM-DD 格式的日期字符串
委托给全局工具方法 normalize_plan_date 处理。
"""
from certflow.utils.date_utils import normalize_plan_date
return normalize_plan_date(plan_date_raw)