Data Processing (数据处理)

name = name.strip()

text = text.encode("utf-8").decode("utf-8")

sql = f"SELECT * FROM users WHERE id = {uid}"

",".join(ids)

clean = re.sub(r"\s+", " ", text)

from decimal import Decimal
price = Decimal("0.1") + Decimal("0.2")

def to_int(x: str | int) -> int | None:
    try:
        return int(x)
    except Exception:
        return None

math.isclose(a, b)

from datetime import datetime, timezone
now = datetime.now(timezone.utc)

dt = datetime.strptime("2025-12-04 16:00:00", "%Y-%m-%d %H:%M:%S")

datetime.fromtimestamp(ts, tz=timezone.utc)

dt.date() == date.today()

results = [x*x for x in nums if x > 0]

items = map_filter(source, pick)

all_items = [*a, *b]

sorted(users, key=lambda u: u["age"])

uid = data.get("id")

d = d1 | d2

clean = {k: v for k, v in data.items() if v is not None}

new = {new_key: data[old_key]}

utils.deep_get(user, "profile.name")

re.compile(r"\d+")

PATTERN = re.compile(r"foo.*bar")

code = re.search(r"code=(\w+)", url).group(1)

parts = re.split(r"[,; ]+", text)

{x: v for x, v in d.items() if v not in ("", None, [], {})}

clean = {k: v.strip() if isinstance(v, str) else v for k,v in d.items()}

dt.strftime("%Y-%m-%d %H:%M:%S")

def safe_str(x) -> str:
    return x if isinstance(x, str) else str(x)

def auto_type(val):
    if val.isdigit():
        return int(val)
    try:
        return float(val)
    except:
        return val

import orjson

data = orjson.loads(json_bytes)
json_bytes = orjson.dumps(data)

orjson.dumps(data, option=orjson.OPT_SERIALIZE_DATACLASS)

with open("a.txt") as f:
    lines = f.readlines()

res = requests.get(url, timeout=10)
res.raise_for_status()
data = res.json()

utils/
    __init__.py
    string.py        # 字符串处理
    date.py          # 日期时间处理
    json.py          # orjson 包装
    data_clean.py    # 数据清洗
    data_convert.py  # 类型转换
    validators.py    # 校验器
    dict.py          # 字典工具
    list.py          # 列表工具
    regex.py         # 正则工具

def strip_string(s: str | None) -> str | None:
    """移除前后空白（安全模式）"""
    return s.strip() if isinstance(s, str) else s


def ensure_str(value) -> str:
    """强制转换为 str"""
    return value if isinstance(value, str) else str(value)


def join_list(items: list[str], sep: str = ",") -> str:
    """安全 join"""
    return sep.join(item.strip() for item in items if item)


def substring(s: str, start: int, end: int | None = None) -> str:
    return s[start:end]

from datetime import datetime, date, timezone


def to_datetime(s: str) -> datetime | None:
    """字符串转 datetime（自动识别常见格式）"""
    patterns = [
        "%Y-%m-%d %H:%M:%S",
        "%Y-%m-%d",
        "%Y/%m/%d %H:%M:%S",
        "%Y年%m月%d日 %H:%M:%S",
    ]
    for p in patterns:
        try:
            return datetime.strptime(s, p).replace(tzinfo=timezone.utc)
        except Exception:
            pass
    return None


def datetime_to_string(dt: datetime, fmt: str = "%Y-%m-%d %H:%M:%S") -> str:
    return dt.strftime(fmt)


def timestamp_to_datetime(ts: int | float) -> datetime:
    return datetime.fromtimestamp(ts, tz=timezone.utc)


def is_today(dt: datetime) -> bool:
    return dt.date() == date.today()

from decimal import Decimal, InvalidOperation


def safe_int(value) -> int | None:
    try:
        return int(value)
    except Exception:
        return None


def safe_decimal(value) -> Decimal | None:
    try:
        return Decimal(str(value))
    except InvalidOperation:
        return None

from typing import TypeVar, Callable, Iterable

T = TypeVar("T")
R = TypeVar("R")


def map_filter(iterable: Iterable[T], func: Callable[[T], R]) -> list[R]:
    """map + filter None 的组合"""
    return [x for x in map(func, iterable) if x is not None]


def flatten(list_of_lists: list[list[T]]) -> list[T]:
    """拍平列表"""
    return [item for sub in list_of_lists for item in sub]


def unique(items: list[T]) -> list[T]:
    """保持顺序的去重"""
    seen = set()
    result = []
    for x in items:
        if x not in seen:
            seen.add(x)
            result.append(x)
    return result

def dict_pick(d: dict, keys: list[str]) -> dict:
    """提取指定字段"""
    return {k: d[k] for k in keys if k in d}


def dict_rename(d: dict, mapping: dict[str, str]) -> dict:
    """批量重命名字段"""
    return {mapping.get(k, k): v for k, v in d.items()}


def deep_get(d: dict, path: str, default=None):
    """嵌套字典取值 user.profile.name"""
    try:
        for key in path.split("."):
            d = d[key]
        return d
    except Exception:
        return default


def clean_dict(d: dict) -> dict:
    """移除值为空的项"""
    return {k: v for k, v in d.items() if v not in ("", None, [], {})}

import re


def regex_search(pattern: str, text: str) -> str | None:
    m = re.search(pattern, text)
    return m.group(1) if m else None


def split_by_regex(pattern: str, text: str) -> list[str]:
    return re.split(pattern, text)


def extract_after_equal(s: str) -> str | None:
    m = re.search(r"=([^&]+)", s)
    return m.group(1) if m else None

import orjson


def json_loads(data: str | bytes):
    return orjson.loads(data)


def json_dumps(obj) -> str:
    return orjson.dumps(obj).decode("utf-8")

def trim_dict_strings(d: dict) -> dict:
    """移除 dict 中所有字符串的前后空白"""
    return {k: (v.strip() if isinstance(v, str) else v) for k, v in d.items()}


def normalize_datetime_dict(d: dict) -> dict:
    """dict 中 datetime 转字符串"""
    result = {}
    for k, v in d.items():
        if isinstance(v, datetime):
            result[k] = v.strftime("%Y-%m-%d %H:%M:%S")
        elif isinstance(v, date):
            result[k] = v.strftime("%Y-%m-%d")
        else:
            result[k] = v
    return result

def auto_type(value: str):
    """自动类型识别：int/float/bool"""
    value = value.strip()
    if value.isdigit():
        return int(value)
    try:
        return float(value)
    except Exception:
        return value

from pathlib import Path

def get_script_dir() -> Path:
    """获取当前脚本目录"""
    return Path(__file__).resolve().parent