xybpkwebapi/src/common/common_functions.py

# coding:utf-8
import calendar
import hashlib
import datetime,time
import re
import M2Crypto
from PIL import Image,ImageDraw
import requests
from pypinyin import pinyin,lazy_pinyin
import common.error_info as ce

def get_month_dates(month="202008"):
    """
    """
    dates = []
    now = datetime.datetime.now().date()
    now_date_str = now.strftime("%Y%m")
    day_end = calendar.monthrange(int(now_date_str[0:4]),int(now_date_str[4:6]))[1]
    for i in range(1,day_end):
        dates.append(now_date_str+"%02d" % i)
    return dates

def check_password(new,old):
    """
    """
    np = hashlib.md5(new).hexdigest().upper()
    return np==old

def make_password(pwd,isdefault=None):
    """
    """
    return hashlib.md5(pwd).hexdigest().upper()

def addText(orgpath,string,path):
    img = Image.open(orgpath)
    size = img.size
    width = size[0] - 20
    high = size[1] - 20
    lenth = len(string)*3
    draw = ImageDraw.Draw(img)
    draw.text((width-lenth,high),string,fill='black')
    img.save(path)

def list_split(items, n):
    return [items[i:i+n] for i in range(0, len(items), n)]

def str_to_datetime(tm,format="%Y-%m-%d %H:%M:%S"):
    """
    """
    datetimestr = datetime.datetime.strptime(tm,format)
    return datetimestr

def datetime_to_str(tm,format="%Y-%m-%d %H:%M:%S"):
    """
    """
    datetimestr = datetime.datetime.strftime(tm,format)
    return datetimestr

def get_now_str(format="%Y-%m-%d %H:%M:%S"):
    """获取当前时间并转化成制定格式字符串
    """
    now = datetime.datetime.now()
    return datetime.datetime.strftime(now,format)

def check_pub_key(pub_key):
    """检查证书有效性
    """
    try:
        pub_key = M2Crypto.X509.load_cert_string(str(pub_key)) 
        return 1
    except:
        return 0

def check_priv_key(priv_key):
    """检查私钥有效性
    """
    try:
        M2Crypto.RSA.load_key_string(str(priv_key)) 
        return 1
    except:
        return 0


def check_pub_priv_key(pub_key, priv_key):
    if len(pub_key) == 0 and len(priv_key) == 0:
        return 0
    msg = "hello"
    try:
        cert = M2Crypto.X509.load_cert_string(str(pub_key))
        key = M2Crypto.RSA.load_key_string(str(priv_key))
        encrypted = cert.get_pubkey().get_rsa().public_encrypt(msg, M2Crypto.RSA.pkcs1_padding)
        decrypted = key.private_decrypt(encrypted, M2Crypto.RSA.pkcs1_padding)
        if msg != decrypted:
            return 0
            return errno.INVALID_CERT
    except:
        return 0
        return errno.INVALID_CERT
    return 1


def get_day_range(yesterday):
    """
    @attention: 获取昨天数据
    """
    sd = ed = yesterday.strftime("%Y%m%d")
    return sd,ed

def get_week_range(yesterday):
    """
    @attention: 获取最近一周数据
    """
    ed = yesterday.strftime("%Y%m%d")
    sd = yesterday - datetime.timedelta(days=6)
    sd = sd.strftime("%Y%m%d")
    return sd,ed

def get_month_range(yesterday,today_month,days):
    """
    @attention: 获取最近一个月数据
    """
    ed = yesterday.strftime("%Y%m%d")
    temp = datetime.datetime.strptime(today_month,"%Y%m")-datetime.timedelta(days=1)
    last_month = temp.strftime("%Y%m")
    sd = "%s%s"%(last_month,str(days).rjust(2,"0"))
    return sd,ed


def list_group_by(olist,key,sort=None):
    """
    """
    nlist = []
    tmp = {}
    for ol in olist:
        kkey = ol[key]
        if not tmp.has_key(kkey):
            tmp[kkey] = [0]
        else:
            tmp[kkey].append(ol)
    for k,v in tmp.items():
        dct = {key:k,"data":v,"count":len(v)}
        nlist.append(dct)
    if sort:
        nlist = sorted(nlist,key=lambda x:x["count"])
    return nlist

def get_need_params(*need_parms,**kwargs):
    """
    """
    newdct = {}
    need_parms = set(need_parms).intersection(set(kwargs.keys()))
    for k in need_parms:
        newdct[k] = kwargs.get(k)
    return newdct

def check_params(*need_parms,**kwargs):
    if not set(need_parms).issubset(set(kwargs.keys())):
        miss = list(set(need_parms)-set(kwargs.keys()))
        miss = ",".join(miss)
        return "缺少参数:{}".format(miss)
    for nk in need_parms:
        if not kwargs.get(nk):
            return "缺少参数值:{}!".format(nk)
    return None

def get_page_list(list,page,page_size=20):
    """
    """
    page = int(page)
    page_size = int(page_size)
    if page and page_size:
        start = (page - 1)*page_size
        end = page * page_size
        count = len(list)
        list = list[start:end]
    else:
        count = len(list)
    return count,list

def get_ip(request):
    if request.META.has_key('HTTP_X_REAL_IP'):
        ip = request.META['HTTP_X_REAL_IP']
    elif request.META.has_key('HTTP_X_FORWARDED_FOR'):
        ip = request.META['HTTP_X_FORWARDED_FOR']
    else:
        ip = request.META['REMOTE_ADDR']
    return ip

def get_city_from_ip(ip):
    url = "https://sp1.baidu.com/8aQDcjqpAAV3otqbppnN2DJv/api.php?query=%s&co=&resource_id=5809&t=%s" \
        % (ip,str(int(time.time()*1000)))
    try:
        result = requests.get(url).json()
        location = result.get("data")[0].get("location")
        location = location.split("省")[-1].split("市")[0]
    except Exception as e:
        print(e)
        location = None
    return location


def get_name_pinyin(name):
    pinyin = lazy_pinyin(name)
    if len(pinyin)>1:
        return "".join([x[0] for x in pinyin]).upper()
    else:
        return pinyin[0].upper()

def calc_age(birthday):
    """
    """
    now = datetime.datetime.now()
    birdate = str_to_datetime(birthday,"%Y-%m-%d")
    years = (now - birdate).days/365
    return years


def check_sign(func):
    apikey = "9edf5d26-6907-4534-8907-e4c3f8ed53c8"
    def __wrapper(*args,**kwargs):
        print(kwargs)
        timestamp = kwargs.get("timestamp")
        #apikey = kwargs.get("apikey")
        sign = kwargs.get("sign")

        doctor_id = kwargs.get("doctor_id")
        hospital_id = kwargs.get("hospital_id")
        patient_id = kwargs.get("patient_id")

        if abs(time.time() - int(timestamp)) > 60*10:
            raise ce.TipException(u"请求超时!")

        sign_str = "apikey=%s;timestamp=%s" % (apikey,timestamp)
        if doctor_id:
            sign_str = "apikey=%s;doctor_id=%s;timestamp=%s" % (apikey,doctor_id,timestamp)
        if doctor_id and hospital_id:
            sign_str = "apikey=%s;doctor_id=%s;hospital_id=%s;timestamp=%s" % (apikey,doctor_id,hospital_id,timestamp)
        if doctor_id and hospital_id and patient_id:
            sign_str = "apikey=%s;doctor_id=%s;hospital_id=%s;patient_id=%s;timestamp=%s" % (apikey,doctor_id,hospital_id,patient_id,timestamp)


        req_sign = make_password(sign_str)
        print(sign_str)
        print(req_sign)
        if not (sign == req_sign):
            raise ce.TipException(u"签名错误!")

        res = func(*args,**kwargs)
        return res
    return __wrapper


if __name__ == "__main__":
    pass
    print make_password("hnwz@2021")