python-fastapi-mm-zhcs-yj-api/utils/__init__.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
from fastapi import Request, Header
import time
from datetime import datetime
# from dateutil import relativedelta as rdelta
from pprint import pprint
import hashlib
import uuid
from sqlalchemy.orm import Session
from models import *
from exceptions import ParamException
import re


def rnd():
    return datetime.now().strftime("%Y%m%d%H%M")


def new_guid() -> str:
    return str(uuid.uuid1())


def md5(val: str) -> str:
    m = hashlib.md5()
    m.update(val.encode('utf-8'))
    val_md5 = m.hexdigest()
    return val_md5


def unixtimestamp():
    return int(time.time())


def unixstamp():
    return int(time.time())


def get_model_dict(model):
    return dict((column.name, getattr(model, column.name))
                for column in model.__table__.columns)


def from_timestamp(timestamp: int):
    if timestamp > 0:
        time_local = time.localtime(timestamp)
        dt = time.strftime("%Y-%m-%d %H:%M:%S", time_local)
        return dt


def from_timestamp2(timestamp: int):
    if timestamp > 0:
        time_local = time.localtime(timestamp)
        dt = time.strftime("%Y-%m-%d %H:%M", time_local)
        return dt

def get_datetime_str(val: datetime) -> str:
    return val.strftime("%Y-%m-%d %H:%M:%S")
    

def null2int(val):
    if val is None:
        return 0
    else:
        return val


def get_req_param(params: dict, name: str) -> any:
    if name not in params:
        raise ParamException(name)
    return params[name]


def get_req_param_optional(params: dict, name: str) -> any:
    if name not in params:
        return ''
    return params[name]


# 下发时间超过一个月的数据，删除字段“核处情况”、“身份证件号码”、“手机号码”
def need_tuomin(xfsj: datetime):
    if_tuomin = False
    diff = datetime.now() - xfsj
    if diff.days > 14:
        if_tuomin = True
    return if_tuomin


# 身份证脱敏
def sfz_tuomin(sfzh: str) -> str:
    if sfzh is None or len(sfzh) < 10:
        return sfzh

    return_sfzh = sfzh[0:6]
    for n in range(6, len(sfzh) - 4):
        return_sfzh = return_sfzh + '*'

    return return_sfzh + sfzh[-4:]


# 手机号码脱敏
def sj_tuomin(lxdh: str) -> str:
    if lxdh is None or len(lxdh) <= 6:
        return lxdh

    return_lxdh = lxdh[0:4]
    for n in range(4, len(lxdh) - 4):
        return_lxdh = return_lxdh + '*'

    return return_lxdh + lxdh[-4:]


# 姓名脱敏
def xm_tuomin(xm: str) -> str:
    if len(xm) == 1:
        return xm

    if len(xm) == 2:
        return xm[0:1] + '*'

    else:
        return xm[0:1] + '**'


def get_admin_area_code(areaCode: str = Header(None)) -> str:
    if areaCode is None or areaCode == "":
        return "4409"
    else:
        return re.sub(r"0+$", "", areaCode)