以下脚本面向“多源报表 CSV 与应用日志”的统一预处理，解决字段命名不统一、日期格式混杂、空值与重复记录等问题，并为后续建模、指标分析与可视化输出干净一致的数据。脚本具备以下能力：

• 自动发现并加载目录中的多份 CSV 报表与多种格式的应用日志（JSON 行、key=value、常见时间戳开头的文本日志）。
• 统一字段命名与数据类型，标准化日期时间（含混合格式、时区），规范缺失值与字符串。
• 按业务主键去重，保留最新记录。
• 输出清洗后的标准数据（Parquet/CSV）与基础指标聚合结果（按天/来源/事件）。
• 可配置字段映射、类型与缺失值策略。

使用说明简述：

• 依赖：pandas, numpy, python-dateutil
• 可选：pyarrow（写 Parquet 更快）
• 运行示例： python preprocess_pipeline.py --report-dir ./reports --log-dir ./logs --output-dir ./output --write-parquet

Python 脚本 preprocess_pipeline.py：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
多源 CSV 报表与应用日志预处理脚本
- 统一字段：依据可配置映射，将来源异构字段对齐到统一 schema
- 解析日期：支持混合格式与时区，统一到 UTC
- 处理缺失：标准化空值、按阈值丢弃列/行、可选简单填充
- 去重：按业务主键或近似主键策略去重
- 合并：多 CSV 与日志合并到统一事件级明细
- 导出：清洗明细 + 指标聚合
"""

import os
import re
import json
import argparse
from pathlib import Path
from typing import Dict, List, Optional, Tuple
import pandas as pd
import numpy as np
from dateutil import parser as du_parser
from datetime import timezone

# ----------------------------
# 配置：统一字段名、类型与映射
# 可根据实际数据源扩展
# ----------------------------
CANONICAL_COLUMNS = [
    "event_time", "event_date", "event_name", "user_id", "session_id", "source",
    "amount", "currency", "app_version", "device", "country", "id", "created_at"
]

# 字段同义词映射（lower 后匹配）
COLUMN_SYNONYMS = {
    "event_time": ["event_time", "time", "timestamp", "ts", "event_ts", "log_time"],
    "event_name": ["event_name", "event", "action", "eventtype", "name"],
    "user_id": ["user_id", "uid", "user", "userid", "member_id", "account_id"],
    "session_id": ["session_id", "sid", "session"],
    "source": ["source", "src", "channel", "platform", "origin"],
    "amount": ["amount", "revenue", "value", "price", "amt"],
    "currency": ["currency", "curr", "ccy"],
    "app_version": ["app_version", "appver", "version"],
    "device": ["device", "device_type", "ua_device"],
    "country": ["country", "country_code", "ctry"],
    "id": ["id", "event_id", "row_id", "uuid"],
    "created_at": ["created_at", "ingest_time", "createdtime", "created_ts", "server_time"]
}

# 推荐数据类型
TARGET_DTYPES = {
    "event_time": "datetime64[ns, UTC]",   # pandas >= 1.1 支持 tz-aware dtype
    "event_date": "date",                  # 后续由 event_time 派生（日期粒度）
    "event_name": "string",
    "user_id": "string",
    "session_id": "string",
    "source": "string",
    "amount": "float64",
    "currency": "string",
    "app_version": "string",
    "device": "string",
    "country": "string",
    "id": "string",
    "created_at": "datetime64[ns, UTC]"
}

# 识别为缺失的字符串
NA_VALUES = ["", "na", "n/a", "null", "none", "-", "nan", "NaN", "NAN", "Nil"]

# 日期格式尝试集合（尽量覆盖常见格式）
DATETIME_FORMATS = [
    "%Y-%m-%d %H:%M:%S%z",
    "%Y-%m-%d %H:%M:%S",
    "%Y/%m/%d %H:%M:%S",
    "%d/%m/%Y %H:%M:%S",
    "%m/%d/%Y %H:%M:%S",
    "%Y-%m-%d",
    "%d/%m/%Y",
    "%m/%d/%Y",
    "%Y-%m-%dT%H:%M:%S%z",
    "%Y-%m-%dT%H:%M:%S.%f%z",
    "%Y-%m-%dT%H:%M:%S.%fZ",
    "%Y-%m-%dT%H:%M:%SZ",
]

# 日志时间戳识别（示例：2024-01-01 12:00:00Z / 2024-01-01T12:00:00+08:00 / Jan 02 03:04:05）
LOG_TS_REGEXES = [
    re.compile(r"^\s*(?P<ts>\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}(?:[.,]\d+)?(?:Z|[+-]\d{2}:\d{2})?)\s+"),
    re.compile(r"^\s*(?P<ts>[A-Z][a-z]{2}\s+\d{1,2}\s+\d{2}:\d{2}:\d{2})\s+"),  # Syslog: Mon DD HH:MM:SS
]


# ----------------------------
# 工具函数
# ----------------------------
def normalize_colname(name: str) -> str:
    return re.sub(r"[^a-z0-9_]", "", name.strip().lower())

def infer_and_map_columns(cols: List[str], synonyms: Dict[str, List[str]]) -> Dict[str, str]:
    """
    根据同义词映射，自动将来源列名映射到规范列名。
    返回：source_col -> canonical_col
    """
    norm_cols = {normalize_colname(c): c for c in cols}
    mapping = {}
    for canonical, syns in synonyms.items():
        syns_norm = [normalize_colname(s) for s in syns]
        for syn in syns_norm:
            if syn in norm_cols:
                mapping[norm_cols[syn]] = canonical
                break
    return mapping

def parse_mixed_datetime(s: str) -> Optional[pd.Timestamp]:
    """
    解析混合日期时间字符串，返回 UTC tz-aware Timestamp；失败返回 None
    """
    if s is None:
        return None
    if isinstance(s, (pd.Timestamp, np.datetime64)):
        try:
            ts = pd.to_datetime(s, utc=True, errors="coerce")
            return ts if pd.notna(ts) else None
        except Exception:
            return None
    if not isinstance(s, str):
        try:
            ts = pd.to_datetime(s, utc=True, errors="coerce")
            return ts if pd.notna(ts) else None
        except Exception:
            return None
    s_clean = s.strip()
    if s_clean == "":
        return None

    # 先用指定格式尝试
    for fmt in DATETIME_FORMATS:
        try:
            dt = pd.to_datetime(s_clean, format=fmt, utc=True)
            return dt
        except Exception:
            pass
    # 再用 dateutil
    try:
        dt = du_parser.parse(s_clean)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)  # 无时区视为 UTC
        return pd.Timestamp(dt).tz_convert("UTC")
    except Exception:
        return None

def parse_datetime_series(sr: pd.Series) -> pd.Series:
    return sr.apply(parse_mixed_datetime)

def standardize_strings(sr: pd.Series, lower: bool = True, strip: bool = True) -> pd.Series:
    if sr.dtype == "O" or pd.api.types.is_string_dtype(sr):
        sr = sr.astype("string")
        if strip:
            sr = sr.str.strip()
        if lower:
            sr = sr.str.lower()
        # 将空字符串标准化为 <NA>
        sr = sr.replace("", pd.NA)
    return sr

def to_float_series(sr: pd.Series) -> pd.Series:
    # 清理千分位、货币符号等，转 float
    if sr.dtype == "O" or pd.api.types.is_string_dtype(sr):
        sr = sr.str.replace(r"[,\s]", "", regex=True)
        sr = sr.str.replace(r"[^\d\.\-eE]", "", regex=True)
    return pd.to_numeric(sr, errors="coerce").astype("float64")

def derive_event_date(df: pd.DataFrame) -> pd.DataFrame:
    if "event_time" in df.columns:
        df["event_date"] = df["event_time"].dt.date
    return df

def drop_sparse_columns(df: pd.DataFrame, missing_ratio_threshold: float = 0.98) -> pd.DataFrame:
    missing_ratio = df.isna().mean()
    keep_cols = missing_ratio[missing_ratio < missing_ratio_threshold].index.tolist()
    return df[keep_cols]

def deduplicate(df: pd.DataFrame) -> pd.DataFrame:
    """
    优先级：
    1) 若存在 'id'，按 id 去重
    2) 否则按 ['user_id','event_time','event_name','source'] 去重
    若存在 created_at，则保留最近（最大）的一条
    """
    subset = None
    if "id" in df.columns and df["id"].notna().any():
        subset = ["id"]
    else:
        cand = [c for c in ["user_id", "event_time", "event_name", "source"] if c in df.columns]
        subset = cand if len(cand) >= 2 else df.columns.tolist()  # 退化为整行去重

    if "created_at" in df.columns:
        df = df.sort_values("created_at", ascending=True)  # 保留最新：后续 drop_duplicates(keep='last')
        df = df.drop_duplicates(subset=subset, keep="last")
    else:
        df = df.drop_duplicates(subset=subset, keep="first")
    return df

def cast_to_target_dtypes(df: pd.DataFrame) -> pd.DataFrame:
    # 字符串标准化
    for col in ["event_name", "user_id", "session_id", "source", "currency", "app_version", "device", "country", "id"]:
        if col in df.columns:
            df[col] = standardize_strings(df[col], lower=True)
    # 金额
    if "amount" in df.columns:
        df["amount"] = to_float_series(df["amount"])
    # 日期时间
    for col in ["event_time", "created_at"]:
        if col in df.columns:
            df[col] = parse_datetime_series(df[col])
            # 显式设为 UTC tz-aware
            df[col] = df[col].dt.tz_convert("UTC")
    # 派生 event_date
    df = derive_event_date(df)
    return df

def read_csv_reports(report_dir: Path, na_values: List[str]) -> List[pd.DataFrame]:
    dfs = []
    for p in report_dir.rglob("*.csv"):
        try:
            df = pd.read_csv(
                p,
                dtype=str,
                na_values=na_values,
                keep_default_na=True,
                encoding="utf-8",
                on_bad_lines="skip"
            )
            if df.empty:
                continue
            # 原字段标准化与映射
            df.columns = [c.strip() for c in df.columns]
            col_map = infer_and_map_columns(df.columns.tolist(), COLUMN_SYNONYMS)
            df = df.rename(columns=col_map)
            df["source"] = df.get("source", pd.Series(["report"] * len(df), index=df.index))
            df["source"] = df["source"].fillna("report")
            df["__file"] = str(p)
            dfs.append(df)
        except Exception as e:
            print(f"[WARN] 读取 CSV 失败: {p} -> {e}")
    return dfs

def extract_kv_pairs(text: str) -> Dict[str, str]:
    """
    从文本中提取 key=value 对，支持 key="val with space"
    """
    pairs = {}
    for k, v in re.findall(r'(\b[a-zA-Z_][\w\-]*?)=(".*?"|\'.*?\'|\S+)', text):
        if v.startswith(("\"", "'")) and v.endswith(("\"", "'")):
            v = v[1:-1]
        pairs[k] = v
    return pairs

def parse_log_line(line: str) -> Optional[Dict]:
    # 优先尝试 JSON
    line = line.strip()
    if not line:
        return None
    try:
        obj = json.loads(line)
        if isinstance(obj, dict):
            return obj
    except Exception:
        pass

    # 尝试从常见日志格式提取时间戳
    ts_val = None
    for rgx in LOG_TS_REGEXES:
        m = rgx.match(line)
        if m:
            ts_val = m.group("ts")
            # 去掉时间戳前缀，保留剩余文本以便提取 kv
            line = line[m.end():]
            break

    kv = extract_kv_pairs(line)
    if ts_val and "timestamp" not in kv and "time" not in kv:
        kv["timestamp"] = ts_val
    return kv if kv else None

def read_logs(log_dir: Path) -> List[pd.DataFrame]:
    rows = []
    for p in list(log_dir.rglob("*.log")) + list(log_dir.rglob("*.jsonl")) + list(log_dir.rglob("*.txt")):
        try:
            with open(p, "r", encoding="utf-8", errors="ignore") as f:
                for line in f:
                    obj = parse_log_line(line)
                    if not obj:
                        continue
                    obj["__file"] = str(p)
                    rows.append(obj)
        except Exception as e:
            print(f"[WARN] 读取日志失败: {p} -> {e}")
    if not rows:
        return []
    df = pd.DataFrame(rows)
    if df.empty:
        return []
    # 映射列名
    df.columns = [c.strip() for c in df.columns]
    col_map = infer_and_map_columns(df.columns.tolist(), COLUMN_SYNONYMS)
    df = df.rename(columns=col_map)
    # 若日志中只有 timestamp，没有 event_time，进行补齐
    if "event_time" not in df.columns and "timestamp" in df.columns:
        df = df.rename(columns={"timestamp": "event_time"})
    # 若没有 source，则标记为 log
    if "source" not in df.columns:
        df["source"] = "log"
    return [df]

def harmonize_schema(dfs: List[pd.DataFrame]) -> pd.DataFrame:
    if not dfs:
        return pd.DataFrame(columns=CANONICAL_COLUMNS)
    # 统一列集合
    all_cols = set()
    for df in dfs:
        all_cols |= set(df.columns)
    all_cols |= set(CANONICAL_COLUMNS)
    all_cols = list(all_cols)
    # 对齐并合并
    aligned = []
    for df in dfs:
        missing = [c for c in all_cols if c not in df.columns]
        for c in missing:
            df[c] = pd.NA
        aligned.append(df[all_cols])
    merged = pd.concat(aligned, ignore_index=True)
    return merged

def clean_and_standardize(df: pd.DataFrame, drop_sparse_threshold: float, fill_amount_zero: bool) -> pd.DataFrame:
    # 标准化字符串空值
    for c in df.columns:
        if df[c].dtype == "O" or pd.api.types.is_string_dtype(df[c]):
            df[c] = df[c].replace(NA_VALUES, pd.NA)

    df = cast_to_target_dtypes(df)

    # 可选：金额缺失填 0
    if fill_amount_zero and "amount" in df.columns:
        df["amount"] = df["amount"].fillna(0.0)

    # 丢弃极度稀疏列
    df = drop_sparse_columns(df, missing_ratio_threshold=drop_sparse_threshold)

    # 关键字段最小约束：必须有 event_time
    if "event_time" in df.columns:
        df = df[~df["event_time"].isna()]

    # 去重
    df = deduplicate(df)

    # 排序
    sort_cols = [c for c in ["event_time", "created_at"] if c in df.columns]
    if sort_cols:
        df = df.sort_values(sort_cols).reset_index(drop=True)

    return df

def aggregate_metrics(df: pd.DataFrame) -> pd.DataFrame:
    """
    基础指标：按天/来源/事件
    - 事件量
    - 去重用户数
    - 金额汇总
    """
    needed = ["event_date", "source", "event_name", "user_id", "amount"]
    for c in needed:
        if c not in df.columns:
            df[c] = pd.NA
    grp = df.groupby(["event_date", "source", "event_name"], dropna=False)
    out = grp.agg(
        events=("event_name", "count"),
        users=("user_id", pd.Series.nunique),
        amount_sum=("amount", "sum")
    ).reset_index()
    # 处理空值显示
    out["amount_sum"] = out["amount_sum"].fillna(0.0)
    return out

def save_outputs(df_clean: pd.DataFrame, metrics: pd.DataFrame, output_dir: Path, write_parquet: bool) -> Tuple[Path, Path]:
    output_dir.mkdir(parents=True, exist_ok=True)
    clean_path = output_dir / ("clean_events.parquet" if write_parquet else "clean_events.csv")
    metrics_path = output_dir / "metrics_daily.csv"
    if write_parquet:
        try:
            df_clean.to_parquet(clean_path, index=False)
        except Exception as e:
            print(f"[WARN] 写入 Parquet 失败，回退为 CSV：{e}")
            clean_path = output_dir / "clean_events.csv"
            df_clean.to_csv(clean_path, index=False)
    else:
        df_clean.to_csv(clean_path, index=False)
    metrics.to_csv(metrics_path, index=False)
    return clean_path, metrics_path

def report_stats(raw_dfs_count: int, df_clean: pd.DataFrame):
    print("========== 处理统计 ==========")
    print(f"合并数据源数量: {raw_dfs_count}")
    print(f"清洗后记录数: {len(df_clean)}")
    if "user_id" in df_clean.columns:
        print(f"去重用户数: {df_clean['user_id'].nunique(dropna=True)}")
    if "event_date" in df_clean.columns:
        print(f"日期范围: {df_clean['event_date'].min()} ~ {df_clean['event_date'].max()}")
    print("=============================")

def main():
    parser = argparse.ArgumentParser(description="多源 CSV + 日志预处理管道")
    parser.add_argument("--report-dir", type=str, required=False, help="CSV 报表目录")
    parser.add_argument("--log-dir", type=str, required=False, help="日志目录（.log/.jsonl/.txt）")
    parser.add_argument("--output-dir", type=str, required=True, help="输出目录")
    parser.add_argument("--drop-sparse-threshold", type=float, default=0.98, help="丢弃稀疏列阈值（缺失率>=阈值将被丢弃）")
    parser.add_argument("--fill-amount-zero", action="store_true", help="将金额缺失填充为 0")
    parser.add_argument("--write-parquet", action="store_true", help="输出 Parquet（需安装 pyarrow 或 fastparquet）")

    args = parser.parse_args()
    report_dir = Path(args.report_dir) if args.report_dir else None
    log_dir = Path(args.log_dir) if args.log_dir else None
    output_dir = Path(args.output_dir)

    dfs: List[pd.DataFrame] = []

    if report_dir and report_dir.exists():
        dfs += read_csv_reports(report_dir, NA_VALUES)
    if log_dir and log_dir.exists():
        dfs += read_logs(log_dir)

    if not dfs:
        print("[ERROR] 未在指定目录中发现可用数据源。")
        return

    df_all = harmonize_schema(dfs)
    df_clean = clean_and_standardize(
        df_all,
        drop_sparse_threshold=args.drop_sparse_threshold,
        fill_amount_zero=args.fill_amount_zero
    )
    metrics = aggregate_metrics(df_clean)

    save_outputs(df_clean, metrics, output_dir, write_parquet=args.write_parquet)
    report_stats(len(dfs), df_clean)
    print(f"输出目录: {output_dir.resolve()}")

if __name__ == "__main__":
    pd.options.mode.chained_assignment = None  # 降低不必要的警告噪声
    main()

关键实现细节与策略说明：

• 字段对齐：通过 COLUMN_SYNONYMS 自动将来源字段映射到统一 schema，规避“命名不统一”问题。未识别到的字段保留原列名，便于后续扩展映射。
• 日期时间解析：parse_mixed_datetime 优先尝试常见格式，再回退到 dateutil 解析，并统一转换为 UTC。无时区视为 UTC，避免偏移误差。
• 缺失值与字符串标准化：将多种“空值表示”统一为缺失；对事件名、来源、用户 ID 等进行去空格、统一小写处理，利于分组与去重。
• 金额字段：去除千分位与符号后转 float；可选填充缺失为 0（--fill-amount-zero）。
• 去重策略：优先使用唯一 id；否则退化到 ['user_id','event_time','event_name','source'] 组合主键；如有 created_at，保留最新记录。
• 日志解析：优先 JSON 行；否则用正则提取时间戳与 key=value 对，最大化结构化信息保留。
• 指标聚合：输出每日/来源/事件的事件量、去重用户数与金额汇总，便于可视化与监控。
• 可扩展性：COLUMN_SYNONYMS 与 DATETIME_FORMATS 可根据实际数据源更新；如需复杂数据类型与业务校验，可在 cast_to_target_dtypes 与 clean_and_standardize 中扩展。

如需进一步集成特定业务规则（例如事件白名单、非法事件过滤、国家与货币映射、异常值截断/Winsorization），可在清洗阶段插入对应逻辑，以保证下游分析与建模的稳定性与一致性。