Python字符串转时间戳的7种实战方案与避坑

from datetime import datetime

def safe_strptime(date_string, fmt, default=None):
    """带空格清理和异常兜底的strptime封装"""
    if not isinstance(date_string, str):
        return default
    # 移除首尾空格，但保留中间空格（如"May 12, 2023"中的空格）
    cleaned = date_string.strip()
    try:
        return datetime.strptime(cleaned, fmt)
    except ValueError as e:
        # 记录具体失败字符串和格式，用于后续分析
        print(f"strptime failed for '{date_string}' with format '{fmt}': {e}")
        return default

# 使用示例
result = safe_strptime("2023-05-12 ", "%Y-%m-%d")  # 返回 datetime(2023, 5, 12)
result = safe_strptime("2023/05/12", "%Y-%m-%d")    # 返回 None，不崩溃

from dateutil import parser

# 方案1：全局设置（影响所有parse调用）
parser.parserinfo(dayfirst=True)  # 优先将"01/02/03"解析为2003-02-01

# 方案2：单次调用指定（推荐）
dt = parser.parse("01/02/03", dayfirst=True, yearfirst=False)
# dt -> datetime(2003, 2, 1, 0, 0)

# 方案3：结合fuzzy参数容忍非关键字符
# "Order placed on May 12, 2023 at 14:30" -> 解析出May 12, 2023 14:30
dt = parser.parse("Order placed on May 12, 2023 at 14:30", fuzzy=True)

import pandas as pd

# 原始数据包含脏数据
df = pd.DataFrame({"time_str": ["2023-05-12", "invalid_date", "2023-05-13"]})

# 错误示范：errors='raise'（默认）会导致整个操作失败
# df["time_dt"] = pd.to_datetime(df["time_str"])  # 报错！

# 正确做法：errors='coerce'，无效值变NaT
df["time_dt"] = pd.to_datetime(df["time_str"], errors="coerce")
# 结果：[Timestamp('2023-05-12'), NaT, Timestamp('2023-05-13')]

# 后续可筛选出脏数据进行专项处理
dirty_mask = df["time_dt"].isna() & df["time_str"].notna()
print("脏数据行：", df[dirty_mask])

# 对10万条ISO格式数据测试
# infer_datetime_format=False（默认）：耗时 0.41s
# infer_datetime_format=True：耗时 0.28s
df["time_dt"] = pd.to_datetime(
    df["time_str"],
    infer_datetime_format=True,  # 告诉Pandas：别猜了，就是ISO格式！
    errors="coerce"
)

# dateutil方式（繁琐）
from dateutil import parser, tz
dt = parser.parse("2023-05-12T14:30:45Z")
dt_utc = dt.replace(tzinfo=tz.UTC)
dt_beijing = dt_utc.astimezone(tz.gettz("Asia/Shanghai"))

# arrow方式（一行解决）
import arrow
dt = arrow.get("2023-05-12T14:30:45Z").to("Asia/Shanghai")
# dt -> <Arrow [2023-05-12T22:30:45+08:00]>

# 更酷的是：支持相对时间自然语言
arrow.get("in 2 hours")  # 当前时间+2小时
arrow.get("last monday") # 上周一00:00

def safe_arrow_get(date_string, to_tz="UTC"):
    try:
        return arrow.get(date_string).to(to_tz)
    except Exception as e:
        print(f"Arrow parse failed for '{date_string}': {e}")
        return None

import dateparser

# 中文解析示例
cn_dates = ["昨天", "下周一", "2023年5月12日下午3点"]
for d in cn_dates:
    parsed = dateparser.parse(d, languages=["zh"])
    print(f"'{d}' -> {parsed}")

# 输出：
# '昨天' -> 2023-05-11 00:00:00
# '下周一' -> 2023-05-15 00:00:00
# '2023年5月12日下午3点' -> 2023-05-12 15:00:00

from datetime import datetime
settings = {
    "RELATIVE_BASE": datetime(2023, 1, 1),  # 所有"昨天"都相对于2023-01-01
    "RETURN_AS_TIMEZONE_AWARE": True,
}
dateparser.parse("昨天", settings=settings)

import re
from datetime import datetime

def parse_iot_timestamp(ts_string):
    """统一解析IoT设备时间字符串"""
    # 匹配无分隔符格式：20230512143045
    m1 = re.match(r"^(\d{4})(\d{2})(\d{2})(\d{2})(\d{2})(\d{2})$", ts_string)
    if m1:
        y, m, d, H, M, S = map(int, m1.groups())
        return datetime(y, m, d, H, M, S)
    
    # 匹配ISO格式：2023-05-12 14:30:45
    m2 = re.match(r"^(\d{4})-(\d{2})-(\d{2}) (\d{2}):(\d{2}):(\d{2})$", ts_string)
    if m2:
        y, m, d, H, M, S = map(int, m2.groups())
        return datetime(y, m, d, H, M, S)
    
    # 匹配带毫秒：2023-05-12T14:30:45.123Z
    m3 = re.match(r"^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2})\.(\d{3})Z$", ts_string)
    if m3:
        y, m, d, H, M, S, ms = map(int, m3.groups())
        return datetime(y, m, d, H, M, S, ms * 1000)
    
    raise ValueError(f"Unrecognized timestamp format: {ts_string}")

# 测试
print(parse_iot_timestamp("20230512143045"))      # 2023-05-12 14:30:45
print(parse_iot_timestamp("2023-05-12 14:30:45")) # 同上

import pandas as pd
from dateutil import parser
from datetime import datetime

def robust_datetime_parser(series):
    """
    生产级时间解析：三层容错
    Layer1: 快速ISO格式（80%数据）
    Layer2: dateutil智能解析（15%数据）
    Layer3: 手动正则兜底（5%数据）
    """
    # Layer1: 尝试ISO格式（最快）
    result = pd.to_datetime(
        series, 
        format="%Y-%m-%d %H:%M:%S", 
        errors="coerce",
        infer_datetime_format=True
    )
    
    # Layer2: 对NaT位置，用dateutil解析
    na_mask = result.isna() & series.notna()
    if na_mask.any():
        # 提取待解析的字符串
        to_parse = series[na_mask].copy()
        # dateutil解析结果
        parsed_list = []
        for s in to_parse:
            try:
                dt = parser.parse(str(s))
                parsed_list.append(dt)
            except:
                parsed_list.append(pd.NaT)
        
        # 替换Layer1的NaT
        result.loc[na_mask] = parsed_list
    
    # Layer3: 对剩余NaT，启动手动正则
    final_na_mask = result.isna() & series.notna()
    if final_na_mask.any():
        # 调用自定义正则解析器
        custom_parsed = series[final_na_mask].apply(
            lambda x: parse_iot_timestamp(str(x)) if isinstance(x, str) else pd.NaT
        )
        result.loc[final_na_mask] = custom_parsed
    
    return result

# 使用
df["trade_time"] = robust_datetime_parser(df["raw_time_field"])

"2023-11-11T00:00:00.123+08:00"
"2023-11-11 00:00:00"
"11/11/2023 00:00:00 AM"
"2023年11月11日 00:00:00"
"20231111000000"

import pandas as pd
import numpy as np
from datetime import datetime, timezone
import pytz
from dateutil import parser

# Step1: 加载数据（模拟）
# df = pd.read_csv("clickstream_20231111.csv", usecols=["event_time", "user_id", "action"])

# Step2: 定义时区对象（避免重复创建）
SHANGHAI_TZ = pytz.timezone("Asia/Shanghai")

def parse_event_time(event_str):
    """高鲁棒性事件时间解析器"""
    if pd.isna(event_str):
        return pd.NaT
    
    s = str(event_str).strip()
    
    # 规则1: ISO 8601带时区（最常见，最快）
    if re.match(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}[+-]\d{2}:\d{2}$", s):
        try:
            # 使用pandas内置解析（对ISO优化）
            return pd.to_datetime(s, utc=True).dt.tz_convert(SHANGHAI_TZ)
        except:
            pass
    
    # 规则2: ISO无时区（补上海时区）
    if re.match(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$", s):
        try:
            dt = datetime.strptime(s, "%Y-%m-%d %H:%M:%S")
            return SHANGHAI_TZ.localize(dt)
        except:
            pass
    
    # 规则3: 美式日期（MM/DD/YYYY）
    if re.match(r"^\d{1,2}/\d{1,2}/\d{4} \d{1,2}:\d{2}:\d{2} [AP]M$", s):
        try:
            return parser.parse(s, dayfirst=False).astimezone(SHANGHAI_TZ)
        except:
            pass
    
    # 规则4: 中文日期（调用dateparser）
    if "年" in s and "月" in s and "日" in s:
        try:
            return parser.parse(s, languages=["zh"]).astimezone(SHANGHAI_TZ)
        except:
            pass
    
    # 规则5: 无分隔符（YYYYMMDDHHMMSS）
    if re.match(r"^\d{14}$", s):
        try:
            y, m, d, H, M, S = int(s[:4]), int(s[4:6]), int(s[6:8]), int(s[8:10]), int(s[10:12]), int(s[12:14])
            dt = datetime(y, m, d, H, M, S)
            return SHANGHAI_TZ.localize(dt)
        except:
            pass
    
    # 终极兜底：返回NaT，记录日志
    print(f"[WARN] Unparseable event_time: '{s}'")
    return pd.NaT

# Step3: 应用解析（向量化加速）
# 注意：不要直接df["event_time"].apply(parse_event_time)，太慢！
# 改用chunk处理 + 并行（需安装joblib）
from joblib import Parallel, delayed

def process_chunk(chunk_series):
    return chunk_series.apply(parse_event_time)

# 分块并行处理（10万条/块）
chunk_size = 100000
chunks = [df["event_time"][i:i+chunk_size] for i in range(0, len(df), chunk_size)]
results = Parallel(n_jobs=4)(delayed(process_chunk)(chunk) for chunk in chunks)
df["event_time_parsed"] = pd.concat(results, ignore_index=True)

# Step4: 验证与修复
# 检查解析率
parsed_rate = df["event_time_parsed"].notna().mean()
print(f"解析成功率: {parsed_rate:.4f}")

# 对失败行，抽样分析原因
failed_samples = df[df["event_time_parsed"].isna()]["event_time"].head(10)
print("失败样本示例:", failed_samples.tolist())

# Step5: 设置为索引（为时间序列分析准备）
df = df.set_index("event_time_parsed")
df = df.sort_index()  # 确保时间有序

# 错误代码！
dt = pd.to_datetime("2023-11-11T00:00:00+00:00")
# 结果：Naive datetime（无时区），被Pandas默认视为本地时区（服务器为UTC+8）
# 所以00:00 UTC被当成00:00 CST → 实际是前一天16:00 UTC

# 正确代码
dt = pd.to_datetime("2023-11-11T00:00:00+00:00", utc=True)  # 强制转为UTC
dt_sh = dt.dt.tz_convert("Asia/Shanghai")  # 再转上海时区 → 2023-11-11 08:00:00+08:00

assert df["event_time_parsed"].dt.tz is not None, "Time column is timezone-naive!"
assert str(df["event_time_parsed"].dt.tz) == "Asia/Shanghai", "Timezone mismatch!"

# 检查前10条数据的真实内容
for i, s in enumerate(df["time_str"].head(10)):
    if isinstance(s, str):
        print(f"Row {i}: {repr(s)}")

def get_format_fingerprint(s):
    """为时间字符串生成格式标识"""
    s = str(s).strip()
    if re.match(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}", s):
        return "iso_basic"
    elif re.match(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", s):
        return "iso_space"
    elif re.match(r"^\d{1,2}/\d{1,2}/\d{4}", s):
        return "us_date"
    elif "年" in s and "月" in s:
        return "chinese"
    else:
        return "unknown"

# 统计格式分布
df["format_type"] = df["time_str"].apply(get_format_fingerprint)
print(df["format_type"].value_counts())

# 生成标准测试集
test_dates = pd.date_range("2023-01-01", periods=10, freq="D")
# 转为各种格式字符串
iso_strs = test_dates.strftime("%Y-%m-%d %H:%M:%S").tolist()
us_strs = test_dates.strftime("%m/%d/%Y %I:%M:%S %p").tolist()

# 测试你的解析器
for s in iso_strs + us_strs:
    parsed = my_parser(s)
    assert abs((parsed - pd.to_datetime(s)).total_seconds