certflow.handlers.data_cleaner 源代码
"""数据清洗处理器模块
提供销售计划数据的清洗和校验功能，包括字符串清理、
数量字段规范化、订单号去特殊字符、统一空值处理等.
"""
from __future__ import annotations
import re
from datetime import datetime
from typing import Any
import pandas as pd
from loguru import logger
from certflow.handlers.id_generator import IDGenerator
from certflow.utils.date_utils import normalize_plan_date
[文档]
class DataCleaner:
"""数据清洗器
提供销售计划数据的清洗和校验功能.
负责处理原始导入数据中的格式问题、空值处理和字段规范化.
Attributes:
DATE_FIELDS: 日期字段列表，定义需要规范化的日期字段名称
"""
# 定义日期字段列表
DATE_FIELDS = ["plan_date", "planned_delivery_date", "contract_delivery_date"]
@staticmethod
def _normalize_date(value: Any) -> str:
"""规范化日期格式，只保留年月日
将各种格式的日期输入统一转换为YYYY-MM-DD格式.
委托给全局工具方法 normalize_plan_date 处理.
Args:
value: 原始日期值，可以是datetime对象、Timestamp对象或字符串
Returns:
str: 格式化为YYYY-MM-DD的日期字符串，转换失败时返回空字符串
Examples:
>>> DataCleaner._normalize_date("2024-01-15 14:30:00")
"2024-01-15"
>>> DataCleaner._normalize_date("2024/01/15")
"2024-01-15"
>>> DataCleaner._normalize_date(None)
""
"""
if value is None or value == "":
return ""
try:
# _fill_empty_date 会给空日期加 "*" 后缀，先去掉
value_str = str(value).rstrip("*") if isinstance(value, str) else value
result = normalize_plan_date(value_str)
if result is None:
return str(value) if value else ""
return result
except Exception as e:
logger.warning(f"日期格式转换失败: {value}, 错误: {e}")
return str(value) if value else ""
[文档]
@staticmethod
def clean_sale_plan(df: pd.DataFrame) -> pd.DataFrame:
"""清洗销售计划数据
对原始导入的 DataFrame 执行标准化清洗：去除字符串首尾空格、
规范化日期字段、清洗订单号特殊字符、数值/文本字段规范化，
并对销售计划日期填充当月默认值。
Args:
df: 原始销售计划 DataFrame，列应为字符串或数值类型
Returns:
pd.DataFrame: 清洗后的 DataFrame（字段已规范化，空值填充为【空白】标记）
Examples:
>>> import pandas as pd
>>> df = pd.DataFrame({
... "product_model": [" 阀门A ", ""],
... "order_no": ["PO-001", "PO/002"],
... "quantity": ["10", "5"],
... })
>>> cleaned = DataCleaner.clean_sale_plan(df)
>>> cleaned["product_model"][0]
"阀门A"
"""
df_clean = df.copy()
# 1. 清理字符串列的空格
string_columns = df_clean.select_dtypes(include=["object"]).columns
for col in string_columns:
df_clean[col] = df_clean[col].astype(str).str.strip()
# 2. 填充空计划日期
if "plan_date" in df_clean.columns:
df_clean["plan_date"] = df_clean["plan_date"].apply(DataCleaner._fill_empty_date)
# 3. 处理日期字段
for col in DataCleaner.DATE_FIELDS:
if col in df_clean.columns:
df_clean[col] = df_clean[col].apply(DataCleaner._normalize_date)
# 4. 清洗订单号
if "order_no" in df_clean.columns:
df_clean["order_no"] = df_clean["order_no"].apply(
lambda x: re.sub(r"[^\w\-]", "", str(x)) if pd.notna(x) else ""
)
# 5. 处理数值和文本字段
df_clean = DataCleaner._normalize_fields(df_clean)
# §9 落地待办③：注入源表原始行号（血缘追溯，仅作追溯，不用于去重）。
# 必须放在 _normalize_fields 之后，否则会被当成文本列填 【空白】 标记。
# 1-based 文件内序号（清洗后），随 to_dict("records") 自动透传到 save_handler。
df_clean = df_clean.reset_index(drop=True)
df_clean["source_row_index"] = range(1, len(df_clean) + 1)
# 统计
logger.info(
f"数据清洗完成: {len(df_clean)} 条记录, "
f"原始记录数: {len(df)}, 清洗掉 {len(df) - len(df_clean)} 条记录"
)
return df_clean
@staticmethod
def _fill_empty_date(x: Any) -> str:
"""空日期填充为当月，带*标记
Args:
x: 原始日期值（可能为 NaN、空字符串或日期字符串）
Returns:
str: 填充后的日期字符串（当月 YYYY-MM-DD 加 "*" 标记），否则原样返回
"""
if pd.isna(x) or str(x).strip() == "":
return datetime.now().strftime("%Y-%m-%d") + "*"
return x
@staticmethod
def _normalize_fields(df_clean: pd.DataFrame) -> pd.DataFrame:
"""识别数值/文本字段并规范化"""
numeric_keywords = ["quantity", "price", "total", "weight", "unit", "amount"]
text_fields = []
for col in df_clean.columns:
is_numeric = any(k in col.lower() for k in numeric_keywords) or col in [
"quantity",
"unit_price",
"total_price",
"weight",
]
if is_numeric:
df_clean[col] = pd.to_numeric(df_clean[col], errors="coerce")
else:
text_fields.append(col)
for col in text_fields:
empty_marker = IDGenerator._get_empty_marker(col)
df_clean[col] = df_clean[col].fillna(empty_marker)
df_clean[col] = df_clean[col].replace(["", "nan", "None", "NaN"], empty_marker)
df_clean[col] = df_clean[col].astype(str).str.strip()
if "quantity" in df_clean.columns:
df_clean["quantity"] = df_clean["quantity"].fillna(1).astype(int)
return df_clean
[文档]
@staticmethod
def validate_data(row: dict[str, Any]) -> tuple[bool, str]:
"""校验单条数据
检查数据行中产品型号是否为空、数量是否大于0.
Args:
row: 单条数据字典，包含product_model和quantity等字段
Returns:
Tuple[bool, str]: 校验结果元组
- 第一个元素: 是否通过校验，True表示通过，False表示不通过
- 第二个元素: 校验结果描述信息，通过时返回"校验通过"
Examples:
>>> row = {"product_model": "阀门A", "quantity": 10}
>>> is_valid, message = DataCleaner.validate_data(row)
>>> print(is_valid)
True
>>>
>>> row = {"product_model": "", "quantity": 5}
>>> is_valid, message = DataCleaner.validate_data(row)
>>> print(message)
"产品型号不能为空"
"""
# 注意: 此时空值已经被规范化为【空白产品型号】格式
product_model = row.get("product_model", "")
if not product_model or product_model == IDGenerator._get_empty_marker("product_model"):
return False, "产品型号不能为空"
quantity = row.get("quantity", 1)
try:
quantity = int(quantity)
except (ValueError, TypeError):
return False, "数量格式错误"
if quantity <= 0:
return False, "数量必须大于0"
return True, "校验通过"