balance_monitoring.py.20201127.1237 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433
  1. # -*- encoding=utf8 -*-
  2. import re
  3. import time
  4. import json
  5. import logging
  6. import pandas as pd
  7. import pymysql
  8. from DBUtils.PooledDB import PooledDB
  9. #配置输出日志格式
  10. LOG_FORMAT = '%(asctime)s %(filename)s[line:%(lineno)d] %(name)s %(levelname)s %(pathname)s %(message)s'
  11. #配置输出时间格式
  12. DATE_FORMAT = '%Y-%m-%d %H-%M-%S %a'
  13. logging.basicConfig(level = logging.INFO,
  14. format = LOG_FORMAT,
  15. datefmt = DATE_FORMAT,
  16. filename = r"./balance_monitoring.log")
  17. comment_re = re.compile(
  18. '(^)?[^\S\n]*/(?:\*(.*?)\*/[^\S\n]*|/[^\n]*)($)?',
  19. re.DOTALL | re.MULTILINE
  20. )
  21. #数据库连接
  22. def conMysql(mysql_host,mysql_port,mysql_user,mysql_password,mysql_db):
  23. db = pymysql.connect(host=mysql_host,port=mysql_port,user=mysql_user, password=mysql_password,db=mysql_db,charset = 'utf8',
  24. autocommit = 1)
  25. print("数据库连接成功")
  26. return db
  27. #创建数据库连接池
  28. def createPool(db_config):
  29. spool = PooledDB(pymysql, 5, **db_config)
  30. return spool
  31. #查询余额
  32. def getData(sql,db):
  33. df = pd.read_sql(sql,con=db)
  34. db.close
  35. #print(df)
  36. return df
  37. #解析带注释的json
  38. def parse_json(filename):
  39. with open(filename,encoding='utf8') as f:
  40. content = ''.join(f.readlines())
  41. ## Looking for comments
  42. match = comment_re.search(content)
  43. while match:
  44. # single line comment
  45. content = content[:match.start()] + content[match.end():]
  46. match = comment_re.search(content)
  47. #print(content)
  48. # Return json file
  49. return json.loads(content)
  50. def saveData():
  51. data_json = parse_json(r'./config.json')
  52. #print(json_data)
  53. '''
  54. ys_df = chanel_df[(chanel_df['supplier_name'] == '亚杉') & (chanel_df['balance'] < 0)]
  55. mf_df = chanel_df[(chanel_df['supplier_name'] == '满帆起航') & (chanel_df['balance'] <= -45000)]
  56. zr_df = chanel_df[(chanel_df['supplier_name'] == '兆蓉') & (chanel_df['balance'] <= 0)]
  57. zrwt_df = chanel_df[(chanel_df['supplier_name'] == '兆蓉WT') & (chanel_df['balance'] <= 5000)]
  58. yxj_df = chanel_df[(chanel_df['supplier_name'] == '易迅捷') & (chanel_df['balance'] <= 10000)]
  59. fy_df = chanel_df[(chanel_df['supplier_name'] == '枫叶') & (chanel_df['balance'] <= 10000)]
  60. zx_df = chanel_df[(chanel_df['supplier_name'] == '智信') & (chanel_df['balance'] <= 2000)]
  61. sht_df = cus_df[(cus_df['customer_id'] == 54) & (cus_df['available_balance'] < 50000)] #十荟团
  62. ylb_df = cus_df[(cus_df['customer_id'] == 47) & (cus_df['available_balance'] < 40000)] #云喇叭
  63. fql_df = cus_df[(cus_df['customer_id'] == 66) & (cus_df['available_balance'] < 20000)] #分期乐
  64. '''
  65. #print(fql_df)
  66. timestamp = time.time()
  67. str_time = time.strftime('%H:%M',time.localtime(timestamp))
  68. str_time_int = int(str(str_time).split(':')[-1])
  69. print(str(str_time).split(':')[-1])
  70. ins_sql = """ INSERT INTO balance_monitoring VALUES(0,%s,%s,1,%s) """
  71. msg = str_time + ': {} 余额为 {},请及时充值'
  72. cus_msg = str_time + ': {} 可用额度为 {};额度不足,请及时处理'
  73. ord_msg = str_time + ': 超过1小时订单总数量为:{},手机号为:\n {}'
  74. sup_msg = str_time + ':超过30分钟无成功的通道:\n{}'
  75. ord_count_msg = ' {} 面额 {} :{} ; '
  76. #art_msg = str_time + ':自动补单2中充值订单 :{}'
  77. #print(ord_df)
  78. #print(sup_df['channel_id'])
  79. #mon_cursor = mon_db.cursor()
  80. try:
  81. #print(3333)
  82. mon_cursor = mon_db.cursor()
  83. #print(mon_cursor)
  84. for item in data_json:
  85. #print(item)
  86. channels = item['channels']
  87. customers = item['customers']
  88. #print(channels)
  89. #print(customes)
  90. for channel in channels:
  91. supplier_name = channel['supplier_name']
  92. balance = channel['balance']
  93. #print(supplier_name)
  94. #print(balance)
  95. ch_df = chanel_df[(chanel_df['supplier_name'] == supplier_name) & (chanel_df['balance'] < balance)]
  96. #print(ch_df)
  97. if ch_df.empty is False:
  98. sup_name = ch_df['supplier_name'].values[0]
  99. balance = ch_df['balance'].values[0]
  100. channel_msg = msg.format(sup_name,balance)
  101. mon_cursor.execute(ins_sql, (channel_msg,group_name1,int(timestamp)))
  102. #print(88888)
  103. for customer in customers:
  104. customer_id = customer['customer_id']
  105. available_balance = customer['available_balance']
  106. customer_df = cus_df[(cus_df['customer_id'] == customer_id) & (cus_df['available_balance'] < available_balance)]
  107. #if customer_df.empty is False:
  108. if((customer_df.empty is False) and (str_time_int % 30 == 0)):
  109. cus_name = customer_df['customer_name'].values[0]
  110. balance = customer_df['available_balance'].values[0]
  111. customer_msg = cus_msg.format(cus_name,balance)
  112. mon_cursor.execute(ins_sql,(customer_msg,group_name3,int(timestamp)))
  113. #print(customer_msg)
  114. '''
  115. if(mf_df.empty is False):
  116. #print(444)
  117. sup_name = mf_df['supplier_name'].values[0]
  118. balance = mf_df['balance'].values[0]
  119. mf_msg = msg.format(sup_name,balance)
  120. mon_cursor.execute(ins_sql, (mf_msg,group_name1,int(timestamp)))
  121. #print(444)
  122. if(zr_df.empty is False):
  123. sup_name = zr_df['supplier_name'].values[0]
  124. balance = zr_df['balance'].values[0]
  125. zr_msg = msg.format(sup_name,balance)
  126. mon_cursor.execute(ins_sql, (zr_msg,group_name1,int(timestamp)))
  127. if(zrwt_df.empty is False):
  128. sup_name = zrwt_df['supplier_name'].values[0]
  129. balance = zrwt_df['balance'].values[0]
  130. zrwt_msg = msg.format(sup_name,balance)
  131. mon_cursor.execute(ins_sql, (zrwt_msg,group_name1,int(timestamp)))
  132. #print(sup_name)
  133. if(yxj_df.empty is False):
  134. sup_name = yxj_df['supplier_name'].values[0]
  135. balance = yxj_df['balance'].values[0]
  136. yxj_msg = msg.format(sup_name,balance)
  137. mon_cursor.execute(ins_sql, (yxj_msg,group_name1,int(timestamp)))
  138. #print(sup_name)
  139. if(fy_df.empty is False):
  140. sup_name = fy_df['supplier_name'].values[0]
  141. balance = fy_df['balance'].values[0]
  142. fy_msg = msg.format(sup_name,balance)
  143. mon_cursor.execute(ins_sql, (fy_msg,group_name1,int(timestamp)))
  144. #print(sup_name)
  145. if(zx_df.empty is False):
  146. sup_name = zx_df['supplier_name'].values[0]
  147. balance = zx_df['balance'].values[0]
  148. zx_msg = msg.format(sup_name,balance)
  149. mon_cursor.execute(ins_sql, (zx_msg,group_name1,int(timestamp)))
  150. #print(sup_name)
  151. if((sht_df.empty is False) and (str_time_int % 30 == 0)):
  152. cus_name = sht_df['customer_name'].values[0]
  153. balance = sht_df['available_balance'].values[0]
  154. sht_msg = cus_msg.format(cus_name,balance)
  155. mon_cursor.execute(ins_sql,(sht_msg,group_name3,int(timestamp)))
  156. #print(cus_name)
  157. if((ylb_df.empty is False) and (str_time_int % 30 == 0)):
  158. cus_name = ylb_df['customer_name'].values[0]
  159. balance = ylb_df['available_balance'].values[0]
  160. ylb_msg = cus_msg.format(cus_name,balance)
  161. mon_cursor.execute(ins_sql,(ylb_msg,group_name3,int(timestamp)))
  162. #print(222)
  163. if((fql_df.empty is False) and (str_time_int % 30 == 0)):
  164. cus_name = fql_df['customer_name'].values[0]
  165. balance = fql_df['available_balance'].values[0]
  166. fql_msg = cus_msg.format(cus_name,balance)
  167. mon_cursor.execute(ins_sql,(fql_msg,group_name3,int(timestamp)))
  168. #print('aaa')
  169. '''
  170. print(str_time_int % 10)
  171. if((ord_df.empty is False) and (str_time_int % 10 == 0)):
  172. #print(len(ord_df))
  173. total_num = ord_df.shape[0]
  174. phone_list = []
  175. if total_num >10:
  176. phone_list = ord_df['used_mobile'].head(10).tolist()
  177. else:
  178. phone_list = ord_df['used_mobile'].values.tolist()
  179. #print(total_num)
  180. #print(phone_list)
  181. ord_msg1 = ord_msg.format(total_num,phone_list)
  182. #print(ord_msg1)
  183. mon_cursor.execute(ins_sql,(ord_msg1,group_name4,int(timestamp)))
  184. logging.info(ord_msg1)
  185. #sup_total_list = ['83','84','85','89','90','91','95','104','105','116','117','118','119','120','121','122','123']
  186. #补单中的订单:
  187. if(ord_count_df.empty is False) and (str_time_int % 10 == 0):
  188. #if(ord_count_df.empty is False) :
  189. ord_count_list = ord_count_df.values.tolist()
  190. #print(ord_count_list)
  191. temp_msg = ''
  192. total_count = 0
  193. for i in range(len(ord_count_list)):
  194. total_count += ord_count_list[i][2]
  195. if i == 0:
  196. temp_msg = str_time + ' 补单中的订单数量为:\n {} 面额 {} :{} ; '.format(ord_count_list[i][0],
  197. int(ord_count_list[i][1]),ord_count_list[i][2])
  198. else:
  199. temp_msg += ord_count_msg.format(ord_count_list[i][0],int(ord_count_list[i][1]),ord_count_list[i][2])
  200. temp_msg = temp_msg + '补单总数量为:【{}】。'.format(total_count)
  201. #print(temp_msg)
  202. mon_cursor.execute(ins_sql,(temp_msg,group_name5,int(timestamp)))
  203. #自动补单2中的订单
  204. if(art_df.empty is False):
  205. #print(art_df)
  206. art_count_list = art_df.values.tolist()
  207. #print(art_count_list)
  208. temp_msg = ''
  209. total_count = 0
  210. for i in range(len(art_count_list)):
  211. total_count += art_count_list[i][2]
  212. #print(total_count)
  213. if i == 0:
  214. temp_msg = str_time + ' 补单2中的订单数量为:\n {} 面额 {} :{} ; '.format(art_count_list[i][0],
  215. int(art_count_list[i][1]),art_count_list[i][2])
  216. #iprint(temp_msg)
  217. else:
  218. temp_msg += ord_count_msg.format(art_count_list[i][0],int(art_count_list[i][1]),art_count_list[i][2])
  219. temp_msg = temp_msg + '补单2总数量为:【{}】。'.format(total_count)
  220. #print(temp_msg)
  221. mon_cursor.execute(ins_sql,(temp_msg,group_name6,int(timestamp)))
  222. sup_total_list = ['83','84','85','89','90','91','95','105','116','118','119','120','121','122',
  223. '123','125','126','127','142','143','144']
  224. sup_list = list(sup_df['channel_id'].values)
  225. #print(sup_list)
  226. dif_list = [i for i in sup_total_list if i not in sup_list]
  227. #print(dif_list)
  228. if (len(dif_list) > 0) and (str_time_int % 10 == 0):
  229. for i in range(len(dif_list)):
  230. if dif_list[i] == '83':
  231. dif_list[i] = '满帆移动WT'
  232. elif dif_list[i] == '84':
  233. dif_list[i] = '满帆联通WT'
  234. elif dif_list[i] == '85':
  235. dif_list[i] = '满帆电信WT'
  236. elif dif_list[i] == '89':
  237. dif_list[i] = '兆蓉移动'
  238. elif dif_list[i] == '90':
  239. dif_list[i] = '兆蓉联通'
  240. elif dif_list[i] == '91':
  241. dif_list[i] = '兆蓉电信'
  242. elif dif_list[i] == '95':
  243. dif_list[i] = '兆蓉移动WT'
  244. elif dif_list[i] == '142':
  245. dif_list[i] = '亚杉移动WT2'
  246. elif dif_list[i] == '105':
  247. dif_list[i] = '兆蓉电信WT'
  248. elif dif_list[i] == '116':
  249. dif_list[i] = '兆蓉联通WT'
  250. elif dif_list[i] == '144':
  251. dif_list[i] = '亚杉电信WT2'
  252. elif dif_list[i] == '118':
  253. dif_list[i] = '智信全国移动'
  254. elif dif_list[i] == '119':
  255. dif_list[i] = '智信全国联通'
  256. elif dif_list[i] == '120':
  257. dif_list[i] = '智信全国电信'
  258. elif dif_list[i] == '121':
  259. dif_list[i] = '易迅捷移动WT'
  260. elif dif_list[i] == '122':
  261. dif_list[i] = '易迅捷电信WT'
  262. elif dif_list[i] == '123':
  263. dif_list[i] = '易迅捷联通WT'
  264. elif dif_list[i] == '143':
  265. dif_list[i] = '亚杉联通WT2'
  266. elif dif_list[i] == '145':
  267. dif_list[i] = '幻星移动WT'
  268. elif dif_list[i] == '146':
  269. dif_list[i] = '幻星联通WT'
  270. elif dif_list[i] == '147':
  271. dif_list[i] = '幻星电信WT'
  272. elif dif_list[i] == '148':
  273. dif_list[i] = '立纵电信WT'
  274. elif dif_list[i] == '149':
  275. dif_list[i] = '立纵联通WT'
  276. elif dif_list[i] == '150':
  277. dif_list[i] = '立纵移动WT'
  278. elif dif_list[i] == '151':
  279. dif_list[i] = 'A11移动WT'
  280. elif dif_list[i] == '152':
  281. dif_list[i] = 'A11联通WT'
  282. elif dif_list[i] == '153':
  283. dif_list[i] = 'A11电信WT'
  284. elif dif_list[i] == '125':
  285. dif_list[i] = '枫叶移动WT'
  286. elif dif_list[i] == '126':
  287. dif_list[i] = '枫叶联通WT'
  288. elif dif_list[i] == '127':
  289. dif_list[i] = '枫叶电信WT'
  290. dif_list = sorted(dif_list)
  291. #print(dif_list)
  292. sup_msg1 = sup_msg.format(dif_list)
  293. mon_cursor.execute(ins_sql,(sup_msg1,group_name2,int(timestamp)))
  294. logging.info(sup_msg1)
  295. #print(dif_list)
  296. except:
  297. #mon_db.rollback()
  298. print('数据回滚')
  299. finally:
  300. mon_cursor.close
  301. mon_db.close
  302. if __name__ == '__main__':
  303. #数据库连接信息
  304. chanel_db_config = {
  305. 'host' : '47.95.217.180',
  306. 'port' : 3306,
  307. 'user' : 'root',
  308. 'password' : '93DkChZMgZRyCbWh',
  309. 'db' : 'fmp',
  310. 'charset' : 'utf8',
  311. 'autocommit' : 1
  312. }
  313. #监控消息数据库连接信息
  314. mon_db_config = {
  315. 'host' : '127.0.0.1',
  316. 'port' : 9001,
  317. 'user' : 'root',
  318. 'password' : 'nibuzhidaowozhidao',
  319. 'db' : 'monitoring',
  320. 'charset' : 'utf8',
  321. 'autocommit' : 1
  322. }
  323. #数据库连接
  324. #db = conMysql(mysql_host,mysql_port,mysql_user,mysql_password,mysql_db)
  325. chanel_db = createPool(chanel_db_config).connection()
  326. mon_db = createPool(mon_db_config).connection()
  327. #通道余额
  328. chanel_sql = '''SELECT supplier_name,balance FROM channel_supplier
  329. WHERE
  330. supplier_name like '%兆蓉%' OR supplier_name like '%满帆%' OR supplier_name like '%亚杉%'
  331. OR supplier_name like '%易迅捷%' OR supplier_name like '%枫叶%' OR supplier_name like '%智信%'
  332. '''
  333. #客户余额
  334. cus_sql = '''
  335. SELECT customer_id,customer_name,(balance + credit_amount - current_amount) 'available_balance'
  336. FROM customer_info
  337. WHERE customer_id IN(47,54)
  338. '''
  339. #超过1小时未处理订单
  340. ord_sql = ''' SELECT used_mobile,(UNIX_TIMESTAMP(now()) - UNIX_TIMESTAMP(apply_date)) 'used_time'
  341. FROM flow_order_info
  342. WHERE
  343. (UNIX_TIMESTAMP(now()) - UNIX_TIMESTAMP(apply_date)) >=3600 AND status NOT IN(4,6) AND
  344. channel_id NOT IN ( 101,102,103)
  345. ORDER BY used_time DESC;
  346. '''
  347. #半小时之内成功的通道
  348. sup_sql = '''
  349. SELECT
  350. channel_id
  351. FROM
  352. flow_order_info
  353. WHERE
  354. (UNIX_TIMESTAMP(now()) - UNIX_TIMESTAMP(check_time)) <=1800 AND
  355. status = 6 AND
  356. channel_id NOT IN (92,93,94,96,97,98,101,102,103,107,108,109,110,111,112,113,114,115)
  357. GROUP BY channel_id;
  358. '''
  359. #自动补单充值中的订单数量:
  360. ord_count_sql = """ SELECT
  361. substring_index(mobile_home,'-',-1) provider , flow_amount,count(*) count_num
  362. FROM
  363. flow_order_info
  364. WHERE
  365. status NOT IN(4,6) AND channel_id IN ( 96,97,98)
  366. GROUP BY provider,flow_amount ORDER BY provider,flow_amount"""
  367. #自动补单2充值中的订单
  368. art_sql = """ SELECT
  369. substring_index(mobile_home,'-',-1) provider , flow_amount,count(*) count_num
  370. FROM
  371. flow_order_info
  372. WHERE
  373. status NOT IN(4,6) AND channel_id IN ( 113,114,115)
  374. GROUP BY provider,flow_amount ORDER BY provider,flow_amount """
  375. chanel_df = getData(chanel_sql,chanel_db)
  376. cus_df = getData(cus_sql,chanel_db)
  377. ord_df = getData(ord_sql,chanel_db)
  378. sup_df = getData(sup_sql,chanel_db)
  379. ord_count_df = getData(ord_count_sql,chanel_db)
  380. art_df = getData(art_sql,chanel_db)
  381. chanel_db.close
  382. #群名:
  383. group_name1 = '通道余额监控群'
  384. group_name2 = '30分钟无成功通道监控群'
  385. group_name3 = '客户授信监控群'
  386. group_name4 = '超时订单监控群'
  387. group_name5 = '补单订单数量监控群'
  388. group_name6 = '自动补单2监控群'
  389. saveData()