data-verification/Monitor/balance_monitoring.py

# -*- encoding=utf8 -*-

import time
import logging
import pandas as pd
import pymysql
from DBUtils.PooledDB import PooledDB

#配置输出日志格式
LOG_FORMAT = '%(asctime)s %(filename)s[line:%(lineno)d] %(name)s %(levelname)s %(pathname)s  %(message)s'

#配置输出时间格式
DATE_FORMAT = '%Y-%m-%d %H-%M-%S %a'

logging.basicConfig(level = logging.INFO,
                    format = LOG_FORMAT,
                    datefmt = DATE_FORMAT,
                    filename = r"./balance_monitoring.log")



#数据库连接
def conMysql(mysql_host,mysql_port,mysql_user,mysql_password,mysql_db):
    db = pymysql.connect(host=mysql_host,port=mysql_port,user=mysql_user, password=mysql_password,db=mysql_db,charset = 'utf8',
                        autocommit = 1)
    print("数据库连接成功")
    return db

#创建数据库连接池
def createPool(db_config):
   spool = PooledDB(pymysql, 5, **db_config) 
   return spool


#查询余额
def getData(sql,db):
  df = pd.read_sql(sql,con=db)
  db.close
  #print(df)
  return df

def saveData():
  ys_df = chanel_df[(chanel_df['supplier_name'] == '亚杉') & (chanel_df['balance'] < 0)]
  mf_df = chanel_df[(chanel_df['supplier_name'] == '满帆起航') & (chanel_df['balance'] <= -45000)]
  zr_df = chanel_df[(chanel_df['supplier_name'] == '兆蓉') & (chanel_df['balance'] <= 0)]
  zrwt_df = chanel_df[(chanel_df['supplier_name'] == '兆蓉WT') & (chanel_df['balance'] <= 5000)]
  yxj_df = chanel_df[(chanel_df['supplier_name'] == '易迅捷') & (chanel_df['balance'] <= 10000)]
  sht_df = cus_df[(cus_df['customer_id'] == 54) & (cus_df['available_balance'] < 50000)] #十荟团
  ylb_df = cus_df[(cus_df['customer_id'] == 47) & (cus_df['available_balance'] < 40000)] #云喇叭
  fql_df = cus_df[(cus_df['customer_id'] == 66) & (cus_df['available_balance'] < 20000)] #分期乐
  #print(fql_df)
  timestamp = time.time()
  str_time = time.strftime('%H:%M',time.localtime(timestamp))
  str_time_int = int(str(str_time).split(':')[-1])
  print(str(str_time).split(':')[-1])
  ins_sql = """ INSERT  INTO balance_monitoring VALUES(0,%s,%s,1,%s)  """
  msg = str_time + ': {} 余额为 {}，请及时充值'
  cus_msg = str_time + ': {} 可用额度为 {}；额度不足，请及时处理'
  ord_msg = str_time + ': 超过1小时订单总数量为：{},手机号为：\n {}'
  sup_msg = str_time + '：超过30分钟无成功的通道：\n{}'
  
  #print(ord_df) 
  #print(sup_df['channel_id'])
  #mon_cursor = mon_db.cursor()
  try:
    #print(3333)
    mon_cursor = mon_db.cursor()
    #print(mon_cursor)
    if(mf_df.empty is False):
        #print(444)
        sup_name = mf_df['supplier_name'].values[0]
        balance = mf_df['balance'].values[0]
        mf_msg = msg.format(sup_name,balance)
        mon_cursor.execute(ins_sql, (mf_msg,group_name1,int(timestamp))) 
        #print(444)
    if(zr_df.empty is False):
        sup_name = zr_df['supplier_name'].values[0]
        balance = zr_df['balance'].values[0]
        zr_msg = msg.format(sup_name,balance)
        mon_cursor.execute(ins_sql, (zr_msg,group_name1,int(timestamp)))

    if(zrwt_df.empty is False):
        sup_name = zrwt_df['supplier_name'].values[0]
        balance = zrwt_df['balance'].values[0]
        zrwt_msg = msg.format(sup_name,balance)
        mon_cursor.execute(ins_sql, (zrwt_msg,group_name1,int(timestamp)))
        #print(sup_name)
    if(yxj_df.empty is False):
        sup_name = yxj_df['supplier_name'].values[0]
        balance = yxj_df['balance'].values[0]
        yxj_msg = msg.format(sup_name,balance)
        mon_cursor.execute(ins_sql, (yxj_msg,group_name1,int(timestamp)))
        #print(sup_name)

    if((sht_df.empty is False)  and (str_time_int % 30 == 0)):
        cus_name = sht_df['customer_name'].values[0]
        balance = sht_df['available_balance'].values[0]
        sht_msg = cus_msg.format(cus_name,balance)
        mon_cursor.execute(ins_sql,(sht_msg,group_name3,int(timestamp)))
        #print(cus_name) 
        
    if((ylb_df.empty is False)  and (str_time_int % 30 == 0)):
        cus_name = ylb_df['customer_name'].values[0]
        balance = ylb_df['available_balance'].values[0]
        ylb_msg = cus_msg.format(cus_name,balance)
        mon_cursor.execute(ins_sql,(ylb_msg,group_name3,int(timestamp)))
        #print(222)
        
    if((fql_df.empty is False)  and (str_time_int % 30 == 0)):
        cus_name = fql_df['customer_name'].values[0]
        balance = fql_df['available_balance'].values[0]
        fql_msg = cus_msg.format(cus_name,balance)
        mon_cursor.execute(ins_sql,(fql_msg,group_name3,int(timestamp)))
        #print('aaa')
    
    #print(str_time_int % 10)
    if((ord_df.empty is False) and (str_time_int % 10 == 0)):
        print(len(ord_df))
        total_num = ord_df.shape[0]
        phone_list = []
        if total_num >10:
           phone_list =  ord_df['used_mobile'].head(10).tolist()
        else:
           phone_list =  ord_df['used_mobile'].values.tolist()
        #print(total_num)
        #print(phone_list)
        ord_msg1 = ord_msg.format(total_num,phone_list)
        #print(ord_msg1)
        mon_cursor.execute(ins_sql,(ord_msg1,group_name4,int(timestamp)))
        logging.info(ord_msg1)
    #sup_total_list = ['83','84','85','89','90','91','95','104','105','116','117','118','119','120','121','122','123']
    sup_total_list = ['83','84','85','89','90','91','95','104','105','116','117','121','122','123']
    sup_list = list(sup_df['channel_id'].values)
    #print(sup_list)
    dif_list = [i for i in sup_total_list if i not in sup_list]
    #print(dif_list)
    if (len(dif_list) > 0) and  (str_time_int % 10 == 0):
        for i in range(len(dif_list)):
            if dif_list[i] == '83':
               dif_list[i] = '满帆移动网厅'
            elif dif_list[i] == '84':
               dif_list[i] = '满帆联通网厅'
            elif dif_list[i] == '85':
               dif_list[i] = '满帆电信网厅'
            elif dif_list[i] == '89':
               dif_list[i] = '兆蓉移动'
            elif dif_list[i] == '90':
               dif_list[i] = '兆蓉联通'
            elif dif_list[i] == '91':
               dif_list[i] = '兆蓉电信'
            elif dif_list[i] == '95':
               dif_list[i] = '兆蓉移动网厅'
            elif dif_list[i] == '104':
               dif_list[i] = '亚杉移动网厅'
            elif dif_list[i] == '105':
               dif_list[i] = '兆蓉电信网厅'
            elif dif_list[i] == '116':
               dif_list[i] = '兆蓉联通网厅'
            elif dif_list[i] == '117':
               dif_list[i] = '亚杉电信网厅'
            elif dif_list[i] == '118':
               dif_list[i] = '智信全国移动'
            elif dif_list[i] == '119':
               dif_list[i] = '智信全联通'
            elif dif_list[i] == '120':
               dif_list[i] = '智信全国电信'
            elif dif_list[i] == '121':
               dif_list[i] = '易迅捷移动网厅'
            elif dif_list[i] == '122':
               dif_list[i] = '易迅捷电信网厅'
            elif dif_list[i] == '123':
               dif_list[i] = '易迅捷联通网厅'
        dif_list = sorted(dif_list)
        #print(dif_list)
        sup_msg1 = sup_msg.format(dif_list)       
        mon_cursor.execute(ins_sql,(sup_msg1,group_name2,int(timestamp)))
        logging.info(sup_msg1)
    #print(dif_list)
  
  except:
      #mon_db.rollback()
      print('数据回滚')
   
  finally:
      mon_cursor.close
      mon_db.close  


if __name__ == '__main__':

  #数据库连接信息
  chanel_db_config = {
    'host' : '47.95.217.180',
    'port' : 3306,
    'user' : 'root',
    'password' : '93DkChZMgZRyCbWh',
    'db' : 'fmp',
    'charset' : 'utf8',
    'autocommit' : 1
  }
  #监控消息数据库连接信息
  mon_db_config = {
          'host' : '127.0.0.1',
          'port' : 9001,
          'user' : 'root',
          'password' : 'nibuzhidaowozhidao',
          'db' : 'monitoring',
          'charset' : 'utf8',
          'autocommit' : 1
  }
  #数据库连接 
  #db = conMysql(mysql_host,mysql_port,mysql_user,mysql_password,mysql_db)
  chanel_db = createPool(chanel_db_config).connection()
  mon_db =  createPool(mon_db_config).connection()
  #通道余额
  chanel_sql = '''SELECT supplier_name,balance  FROM channel_supplier
                  WHERE  
                  supplier_name like '%兆蓉%' OR supplier_name like '%满帆%' OR supplier_name like '%亚杉%'
		  OR supplier_name like '%易迅捷%'
               '''
  
  #客户余额
  cus_sql = ''' 
                SELECT customer_id,customer_name,(balance + credit_amount - current_amount) 'available_balance'
		FROM customer_info
		WHERE  customer_id IN(47,54) 
  		 '''
  
  #超过1小时未处理订单
  ord_sql = ''' SELECT  used_mobile,(UNIX_TIMESTAMP(now()) - UNIX_TIMESTAMP(apply_date)) 'used_time'
                FROM flow_order_info
                WHERE
                  (UNIX_TIMESTAMP(now()) - UNIX_TIMESTAMP(apply_date)) >=3600 AND status NOT IN(4,6)
                ORDER BY used_time DESC;
            '''
  
  #半小时之内成功的通道
  sup_sql = '''
             SELECT 
		channel_id
             FROM 
		flow_order_info 
             WHERE 
                (UNIX_TIMESTAMP(now()) - UNIX_TIMESTAMP(check_time)) <=1800 AND
                status = 6  AND 
                channel_id NOT IN (92,93,94,96,97,98,101,102,103,107,108,109,110,111,112,113,114,115)
             GROUP BY channel_id;  
            '''
  
  chanel_df = getData(chanel_sql,chanel_db)
  cus_df = getData(cus_sql,chanel_db)
  ord_df = getData(ord_sql,chanel_db)
  sup_df = getData(sup_sql,chanel_db)
  
  chanel_db.close
  
  #群名：
  group_name1 = '通道余额监控群'
  group_name2 = '30分钟无成功通道监控群'
  group_name3 = '客户授信监控群'
  group_name4 = '超时订单监控群'
  saveData()