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