"""带样式 Excel 导入处理器
提供销售计划 Excel 的带格式读取、样式解析和格式信息合并功能。
从 SalePlanService 中拆分出来，作为独立的 Handler 层工具。
功能:
- 读取带样式的 Excel（背景色、字体色、批注）
- 行隐藏状态检测
- 表头解析和行跳过
- 字体颜色 → 发货状态映射
- 背景颜色 → 产品状态映射
- 格式信息合并到数据记录
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
import pandas as pd
from loguru import logger
from certflow.config.settings import (
BG_COLOR_TO_STATUS,
FONT_COLOR_TO_SHIPPING,
SALES_PLAN_HEADER_ROW,
THEME_FONT_COLORS,
)
from certflow.handlers.excel_styler import ExcelStyler
[文档]
class StyledExcelImporter:
"""带样式 Excel 导入处理器
负责读取和解析 Excel 单元格的格式信息，
并将格式信息合并到数据记录中。
所有方法均为静态方法，无状态依赖。
"""
# ============================================================
# 主入口：读取带样式 Excel
# ============================================================
[文档]
@staticmethod
def read_styled_excel(
file_path: str,
header_row: int | None,
sheet_name: Any = 0,
skip_rows: int = 0,
) -> tuple[
pd.DataFrame,
pd.DataFrame | None,
pd.DataFrame | None,
pd.DataFrame | None,
pd.DataFrame | None,
list[bool],
]:
"""读取带样式的Excel并设置列名
Args:
file_path: Excel文件路径
header_row: 用户指定的表头行号，None则自动检测
sheet_name: 工作表名称或索引，默认为0
skip_rows: 表头之后额外跳过的数据行数（跳过已导入的行）
Returns:
tuple: (df_data, bg_colors, font_colors, comments, statuses, hidden_rows)
"""
# 读取带格式的数据
df_data, bg_colors, font_colors, comments, statuses = ExcelStyler.read_with_styles(
Path(file_path), sheet_name=sheet_name, include_styles=True, include_comments=True
)
# 确定标题行
resolved_header = SALES_PLAN_HEADER_ROW - 1 if header_row is None else header_row
# 读取行隐藏状态
hidden_rows = StyledExcelImporter._read_hidden_rows(file_path, sheet_name, len(df_data))
# 设置列名
df_data, bg_colors, font_colors, comments, statuses, hidden_rows = (
StyledExcelImporter._apply_header(
df_data, bg_colors, font_colors, comments, statuses, hidden_rows, resolved_header
)
)
# 跳过已导入的行
df_data, bg_colors, font_colors, comments, statuses, hidden_rows = (
StyledExcelImporter._skip_rows(
df_data, bg_colors, font_colors, comments, statuses, hidden_rows, skip_rows
)
)
return df_data, bg_colors, font_colors, comments, statuses, hidden_rows
# ============================================================
# 行隐藏状态
# ============================================================
@staticmethod
def _read_hidden_rows(file_path: str, sheet_name: Any, row_count: int) -> list[bool]:
"""读取工作表的行隐藏状态"""
from openpyxl import load_workbook
try:
wb = load_workbook(file_path, data_only=True)
ws = (
wb[sheet_name]
if isinstance(sheet_name, str)
else wb.worksheets[sheet_name]
if isinstance(sheet_name, int)
else wb.active
)
if ws is not None:
hidden = []
for r in range(1, row_count + 1):
row_dim = ws.row_dimensions.get(r)
hidden.append(row_dim.hidden if row_dim else False)
else:
hidden = [False] * row_count
wb.close()
return hidden
except Exception:
return [False] * row_count
# ============================================================
# 表头与行跳过
# ============================================================
@staticmethod
def _apply_header(
df_data: pd.DataFrame,
bg_colors: pd.DataFrame | None,
font_colors: pd.DataFrame | None,
comments: pd.DataFrame | None,
statuses: pd.DataFrame | None,
hidden_rows: list[bool],
resolved_header: int,
) -> tuple[
pd.DataFrame,
pd.DataFrame | None,
pd.DataFrame | None,
pd.DataFrame | None,
pd.DataFrame | None,
list[bool],
]:
"""用指定行作为表头并设置列名，同步裁剪样式数据"""
if resolved_header is None or resolved_header >= len(df_data):
return df_data, bg_colors, font_colors, comments, statuses, hidden_rows
headers_row = df_data.iloc[resolved_header]
headers = ["" if pd.isna(h) else str(h).strip() for h in headers_row]
# 清理重复的列名
unique_headers: list[str] = []
seen: set[str] = set()
for h in headers:
if h and h not in seen:
unique_headers.append(h)
seen.add(h)
elif h:
new_name = f"{h}_{len(seen)}"
unique_headers.append(new_name)
seen.add(new_name)
else:
new_name = f"col_{len(seen)}"
unique_headers.append(new_name)
seen.add(new_name)
df_data.columns = [str(col) for col in unique_headers]
df_data = df_data.iloc[resolved_header + 1 :].reset_index(drop=True)
for style_df in (bg_colors, font_colors, comments, statuses):
if style_df is not None and len(style_df) > resolved_header:
style_df.drop(style_df.index[: resolved_header + 1], inplace=True)
style_df.reset_index(drop=True, inplace=True)
hidden_rows = hidden_rows[resolved_header + 1 :]
return df_data, bg_colors, font_colors, comments, statuses, hidden_rows
@staticmethod
def _skip_rows(
df_data: pd.DataFrame,
bg_colors: pd.DataFrame | None,
font_colors: pd.DataFrame | None,
comments: pd.DataFrame | None,
statuses: pd.DataFrame | None,
hidden_rows: list[bool],
skip_rows: int,
) -> tuple[
pd.DataFrame,
pd.DataFrame | None,
pd.DataFrame | None,
pd.DataFrame | None,
pd.DataFrame | None,
list[bool],
]:
"""跳过指定行数（表头之后的数据行）"""
if skip_rows <= 0 or len(df_data) <= skip_rows:
return df_data, bg_colors, font_colors, comments, statuses, hidden_rows
logger.info(
f"跳过前 {skip_rows} 行已导入数据"
f"（共 {len(df_data)} 行，保留后 {len(df_data) - skip_rows} 行）"
)
df_data = df_data.iloc[skip_rows:].reset_index(drop=True)
for style_df in (bg_colors, font_colors, comments, statuses):
if style_df is not None and len(style_df) > skip_rows:
style_df.drop(style_df.index[:skip_rows], inplace=True)
style_df.reset_index(drop=True, inplace=True)
hidden_rows = hidden_rows[skip_rows:]
return df_data, bg_colors, font_colors, comments, statuses, hidden_rows
# ============================================================
# 格式信息合并
# ============================================================
# ============================================================
# 字体颜色解析 → 发货状态
# ============================================================
@staticmethod
def _find_exec_column(df_data: pd.DataFrame) -> int | None:
"""查找执行情况列索引"""
for i, col in enumerate(df_data.columns):
if col in ("执行情况", "execution_status", "发货状态"):
return i
return None
@staticmethod
def _check_red_font_exec_status(
font_colors: pd.DataFrame | None,
df_data: pd.DataFrame,
idx: int,
exec_col_idx: int | None,
detected_status: str,
) -> str:
"""检查红色字体执行状态"""
if exec_col_idx is None or font_colors is None or idx >= len(font_colors):
return detected_status
exec_font_color = font_colors.iloc[idx, exec_col_idx]
if pd.isna(exec_font_color):
return detected_status
exec_font_int = (
int(exec_font_color) if isinstance(exec_font_color, int | float) else exec_font_color
)
if exec_font_int != 255:
return detected_status
exec_text = str(
df_data.iloc[idx, exec_col_idx] if exec_col_idx < len(df_data.columns) else ""
)
exec_text = exec_text.strip() if exec_text != "nan" else ""
if "已发" in exec_text:
return f"{detected_status},红字已发" if detected_status else "红字已发"
if exec_text:
return f"{detected_status},红字未发" if detected_status else "红字未发"
return f"{detected_status},红字" if detected_status else "红字"
@staticmethod
def _check_hidden(hidden_rows: list[bool] | None, idx: int) -> bool:
"""检查行是否隐藏"""
if hidden_rows is not None and idx < len(hidden_rows):
return hidden_rows[idx]
return False
@staticmethod
def _parse_font_colors(
font_colors: pd.DataFrame,
df_data: pd.DataFrame,
idx: int,
row_font_colors: dict[str, Any],
detected_status: str,
) -> str:
"""解析字体颜色 → 发货状态
Args:
font_colors: 字体色DataFrame
df_data: 数据DataFrame
idx: 行索引
row_font_colors: 行字体颜色字典（输出参数）
detected_status: 已检测到的状态
Returns:
str: 更新后的状态字符串
"""
for col_idx in range(len(font_colors.columns)):
color_val = font_colors.iloc[idx, col_idx]
if pd.notna(color_val) and str(color_val) != "000000": # 跳过默认黑色
col_name = (
df_data.columns[col_idx] if col_idx < len(df_data.columns) else f"col_{col_idx}"
)
row_font_colors[col_name] = (
int(color_val) if isinstance(color_val, int | float) else str(color_val)
)
# 将颜色值转为匹配 key（支持 hex 字符串、整数和 THEME 颜色）
color_key = StyledExcelImporter._to_color_int(color_val)
if color_key is not None:
shipping_status = FONT_COLOR_TO_SHIPPING.get(color_key)
if shipping_status and not detected_status:
detected_status = shipping_status
# 也尝试 THEME 颜色映射
if (
not shipping_status
and isinstance(color_key, str)
and color_key.startswith("THEME_")
):
theme_status = THEME_FONT_COLORS.get(color_key)
if theme_status and not detected_status:
detected_status = theme_status
return detected_status
@staticmethod
def _parse_bg_colors(
bg_colors: pd.DataFrame,
df_data: pd.DataFrame,
idx: int,
row_bg_colors: dict[str, Any],
detected_status: str,
) -> str:
"""解析背景色 → 产品状态（外购未回/外购已回/自产）"""
for col_idx in range(len(bg_colors.columns)):
color_val = bg_colors.iloc[idx, col_idx]
if pd.notna(color_val) and str(color_val) != "FFFFFF": # 跳过默认白色
col_name = (
df_data.columns[col_idx] if col_idx < len(df_data.columns) else f"col_{col_idx}"
)
row_bg_colors[col_name] = (
int(color_val) if isinstance(color_val, int | float) else str(color_val)
)
color_key = StyledExcelImporter._to_color_int(color_val)
if color_key is not None:
bg_status = BG_COLOR_TO_STATUS.get(color_key)
if bg_status:
detected_status = (
f"{detected_status},{bg_status}" if detected_status else bg_status
)
return detected_status
@staticmethod
def _to_color_int(color_val: Any) -> int | str | None:
"""将颜色值统一转换为匹配 key（支持 hex 字符串、整数和 THEME 颜色）
例如: "0000FF" -> 255, "FF0000" -> 16711680, "THEME_1" -> "THEME_1", 16711680 -> 16711680
"""
if isinstance(color_val, int | float):
return int(color_val)
if isinstance(color_val, str):
if color_val.startswith("THEME_"):
return color_val
if color_val.startswith("INDEXED_"):
return None
try:
return int(color_val, 16)
except ValueError:
return None
return None
@staticmethod
def _parse_comments(
comments: pd.DataFrame | None,
df_data: pd.DataFrame,
idx: int,
) -> list[dict[str, Any]]:
"""解析批注
Returns:
list[dict[str, Any]]: 批注列表，每个元素包含 column 和 text
"""
row_comments: list[dict[str, Any]] = []
if comments is not None and idx < len(comments):
for col_idx in range(len(comments.columns)):
comment = comments.iloc[idx, col_idx]
if comment and str(comment) != "nan":
col_name = (
df_data.columns[col_idx]
if col_idx < len(df_data.columns)
else f"col_{col_idx}"
)
row_comments.append({"column": col_name, "text": str(comment)})
return row_comments