zkyuan/src/common/common_functions.py

# coding:utf-8
import calendar
import hashlib
import datetime
import re
import M2Crypto
from PIL import Image,ImageDraw
import requests

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):
    """
    """
    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 isWechat(ua):
    """
    """
    if "micromessenger" in ua.lower():
        return True
    else:
        return False

def get_city_by_ip(ip):
    """
    """
    url = "http://api.map.baidu.com/location/ip?ip={}&ak=dYrYC6F0YePk51zYG2H5kSyAbU0nX4NR".format(ip)
    res = requests.get(url)
    if res.status_code == 200:
        if res.json().get("status") == 0:
            city = res.json()["content"]["address_detail"]["city"]
            return city
        return None
    return None

def get_weathear_by_city(district_id):
    """
    """
    url = "http://api.map.baidu.com/weather/v1/?district_id={}&data_type=all&ak=dYrYC6F0YePk51zYG2H5kSyAbU0nX4NR".format(district_id)
    res = requests.get(url)
    if res.status_code == 200:
        if res.json().get("status") == 0:
            city = res.json()["result"]["location"]["city"]
            temp = res.json()["result"]["now"]["temp"]
            wind_class = res.json()["result"]["now"]["wind_class"]
            wind_dir = res.json()["result"]["now"]["wind_dir"]
            text = res.json()["result"]["now"]["text"]
            return city,temp,wind_class,wind_dir,text
        return None
    return None

if __name__ == "__main__":
    print make_password("123456")
    pass