4. 数据采集与清洗

数据采集与清洗是越权采购排查的基石。原始数据通常分散在ERP、OA、SRM等多个异构系统中，存在缺失、重复、格式不统一等问题。本节将系统讲解如何通过Python实现自动化数据采集与清洗，确保后续分析的准确性和可追溯性。

4.1 数据采集：多源异构系统连接

越权采购分析需要整合采购订单、审批记录、用户权限、组织架构四类核心数据。Python通过sqlalchemy和pyodbc驱动实现统一连接。

from sqlalchemy import create_engine, text
import pandas as pd

# ERP系统（SQL Server）
erp_engine = create_engine(
    'mssql+pyodbc://user:pass@erp_host:1433/erp_db?driver=ODBC+Driver+17+for+SQL+Server',
    pool_size=5, pool_recycle=3600
)

# OA系统（MySQL）
oa_engine = create_engine(
    'mysql+pymysql://oa_user:oa_pass@oa_host:3306/oa_db',
    pool_size=3
)

# 执行查询
po_query = """
SELECT PO_NUMBER, CREATED_DATE, APPROVER_ID, AMOUNT, STATUS
FROM PURCHASE_ORDERS
WHERE CREATED_DATE >= '2024-01-01'
"""
df_po = pd.read_sql_query(text(po_query), erp_engine)

4.2 缺失值处理：分级策略与业务规则

缺失值不能简单删除或填充，必须结合业务含义分级处理。越权采购场景中，审批人ID缺失和金额缺失的处置逻辑完全不同。

# 分级缺失值处理
def handle_missing_values(df):
    # 1级：关键字段缺失 → 标记异常
    df['approver_missing'] = df['APPROVER_ID'].isna()
    
    # 2级：金额缺失 → 从明细表回填
    amount_missing = df['AMOUNT'].isna()
    if amount_missing.any():
        detail_amounts = df_detail.groupby('PO_NUMBER')['LINE_AMOUNT'].sum()
        df.loc[amount_missing, 'AMOUNT'] = df.loc[amount_missing, 'PO_NUMBER'].map(detail_amounts)
    
    # 3级：部门缺失 → 左连接补全
    df = df.merge(df_org[['EMP_ID', 'DEPT']], on='EMP_ID', how='left', suffixes=('', '_org'))
    df['DEPT'] = df['DEPT'].fillna(df['DEPT_org'])
    df.drop(columns=['DEPT_org'], inplace=True)
    
    # 记录缺失统计
    missing_stats = {
        'approver_missing': df['approver_missing'].sum(),
        'amount_filled': amount_missing.sum(),
        'dept_filled': df['DEPT'].isna().sum()  # 仍缺失的部门
    }
    return df, missing_stats

4.3 统一日期格式：时区归一化与业务周期对齐

不同系统输出的日期格式差异巨大：ERP使用2024-01-15 14:30:00，OA使用2024/01/15 14:30，部分旧系统甚至使用15-JAN-2024。统一日期格式是时间序列分析的前提。

import pandas as pd

def unify_date_format(df, date_columns):
    """
    统一日期格式为 YYYY-MM-DD HH:MM:SS，时区统一为 UTC+8
    """
    for col in date_columns:
        # 尝试多种常见格式
        df[col] = pd.to_datetime(df[col], errors='coerce', 
                                 format='mixed', dayfirst=False)
        
        # 时区处理：统一转为 UTC+8
        if df[col].dt.tz is None:
            df[col] = df[col].dt.tz_localize('Asia/Shanghai', ambiguous='infer')
        else:
            df[col] = df[col].dt.tz_convert('Asia/Shanghai')
        
        # 提取业务周期字段
        df[f'{col}_year'] = df[col].dt.year
        df[f'{col}_month'] = df[col].dt.month
        df[f'{col}_week'] = df[col].dt.isocalendar().week.astype(int)
        df[f'{col}_hour'] = df[col].dt.hour
    
    return df

# 应用
date_cols = ['CREATED_DATE', 'APPROVED_DATE', 'DELIVERY_DATE']
df = unify_date_format(df, date_cols)

4.4 去重逻辑：基于业务语义的精确去重

越权采购场景中，去重不能仅依赖主键。同一笔采购可能因系统重试、接口超时、人工补录产生多条记录。需要设计业务主键 + 时间窗口的复合去重策略。

def dedup_purchase_orders(df, time_window_minutes=5):
    """
    复合去重：业务主键 + 时间窗口
    """
    # 步骤1：按业务主键去重，保留最早记录
    df_sorted = df.sort_values('CREATED_DATE')
    df_dedup = df_sorted.drop_duplicates(
        subset=['PO_NUMBER', 'APPROVER_ID', 'AMOUNT'],
        keep='first'
    )
    
    # 步骤2：时间窗口去重（同一PO在5分钟内多次提交）
    df_dedup['time_diff'] = df_dedup.groupby('PO_NUMBER')['CREATED_DATE'].diff()
    df_dedup['is_duplicate_window'] = (
        df_dedup['time_diff'].notna() & 
        (df_dedup['time_diff'].dt.total_seconds() < time_window_minutes * 60)
    )
    
    # 保留时间窗口内的第一条
    df_final = df_dedup[~df_dedup['is_duplicate_window']].copy()
    
    # 步骤3：标记疑似重复（供人工复核）
    df_final['duplicate_flag'] = '正常'
    df_final.loc[df_dedup['is_duplicate_window'], 'duplicate_flag'] = '疑似重复'
    
    # 统计
    stats = {
        '原始记录数': len(df),
        '业务主键去重后': len(df_dedup),
        '时间窗口去重后': len(df_final),
        '疑似重复数': df_final['duplicate_flag'].eq('疑似重复').sum()
    }
    return df_final, stats

4.5 数据清洗流水线：端到端执行

将上述步骤封装为可复用的清洗流水线，确保每次数据刷新时执行一致的清洗逻辑。

class DataCleaningPipeline:
    """采购数据清洗流水线"""
    
    def __init__(self, config):
        self.config = config
        self.stats = {}
    
    def run(self, raw_data_dict):
        # 1. 数据采集
        df_po = self._extract_erp(raw_data_dict['erp'])
        df_oa = self._extract_oa(raw_data_dict['oa'])
        df_ad = self._extract_ad(raw_data_dict['ad'])
        df_hr = self._extract_hr(raw_data_dict['hr'])
        
        # 2. 合并数据集
        df = self._merge_datasets(df_po, df_oa, df_ad, df_hr)
        
        # 3. 缺失值处理
        df, self.stats['missing'] = handle_missing_values(df)
        
        # 4. 统一日期格式
        df = unify_date_format(df, self.config['date_columns'])
        
        # 5. 去重
        df, self.stats['dedup'] = dedup_purchase_orders(df)
        
        # 6. 数据质量报告
        self._generate_quality_report(df)
        
        return df
    
    def _generate_quality_report(self, df):
        """生成数据质量报告"""
        report = {
            '总记录数': len(df),
            '缺失审批人': df['approver_missing'].sum(),
            '重复标记数': df['duplicate_flag'].eq('疑似重复').sum(),
            '日期范围': f"{df['CREATED_DATE'].min()} ~ {df['CREATED_DATE'].max()}",
            '涉及PO数': df['PO_NUMBER'].nunique()
        }
        self.stats['quality'] = report
        return report

4.6 数据质量监控看板

清洗完成后，需要建立持续的数据质量监控机制，确保每次数据刷新后质量可控。

通过上述数据采集与清洗流程，我们构建了高质量、可追溯的采购数据集，为后续的越权规则计算和异常检测奠定了坚实的数据基础。清洗后的数据将进入下一阶段的特征工程与规则引擎处理。