Files
Neo-ZQYY/apps/etl/connectors/feiqiu/tasks/dws/dws_helpers.py

73 lines
1.9 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
"""
DWS 层公共辅助函数
从多个 DWS 子类中提取的纯函数,不依赖实例状态。
"""
from __future__ import annotations
from datetime import date, datetime
from decimal import Decimal
from typing import Any, Optional
def mask_mobile(mobile: Optional[str]) -> Optional[str]:
"""手机号脱敏138****1234
短于 7 位的号码原样返回,避免截断后信息丢失。
"""
if not mobile or len(mobile) < 7:
return mobile
return mobile[:3] + "****" + mobile[-4:]
def calc_days_since(stat_date: date, last_date: Optional[date]) -> Optional[int]:
"""计算距离目标日期的天数
若 last_date 为 datetime 则自动取 .date()。
"""
if not last_date:
return None
if isinstance(last_date, datetime):
last_date = last_date.date()
return (stat_date - last_date).days
def parse_id_list(value: Any) -> set[int]:
"""解析逗号分隔的 ID 列表字符串为 int 集合
支持 str / list / tuple / set 输入;非法值静默跳过。
"""
if not value:
return set()
if isinstance(value, str):
items = [v.strip() for v in value.split(",") if v.strip()]
return {int(v) for v in items if v.isdigit()}
if isinstance(value, (list, tuple, set)):
result: set[int] = set()
for item in value:
if item is None:
continue
try:
result.add(int(item))
except (ValueError, TypeError):
continue
return result
return set()
def safe_division(
numerator: Any,
denominator: Any,
default: Decimal = Decimal("0"),
) -> Decimal:
"""安全除法,分母为零或 None 时返回默认值"""
try:
d = Decimal(str(denominator))
if d == 0:
return default
return Decimal(str(numerator)) / d
except (ValueError, TypeError, ArithmeticError):
return default