# -*- 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()