|
|
@@ -1,675 +0,0 @@
|
|
|
-import json
|
|
|
-import os
|
|
|
-import time as stime
|
|
|
-import redis
|
|
|
-import h5py
|
|
|
-from os import path
|
|
|
-import re
|
|
|
-from datetime import timedelta
|
|
|
-from datetime import datetime
|
|
|
-import numpy as np
|
|
|
-from matplotlib.figure import Figure
|
|
|
-from matplotlib import ticker
|
|
|
-from io import BytesIO
|
|
|
-import logging
|
|
|
-
|
|
|
-from refill import ProfitHandler
|
|
|
-
|
|
|
-class QueueListener(object):
|
|
|
- queue_name = 'REFILL_MONITOR_QUEUE'
|
|
|
- latest_delta = 10
|
|
|
- pos_map = {
|
|
|
- 'commit': 0, 'success': 1, 'fail': 2
|
|
|
- }
|
|
|
-
|
|
|
- def __init__(self):
|
|
|
- self._mquit = False
|
|
|
- self._mRHost = ''
|
|
|
- self._mRPort = 6379
|
|
|
- self._file_names = {
|
|
|
- 'profit': {
|
|
|
- 'name': '/var/www/html/data/stdata/profit.hdf5',
|
|
|
- 'handler': None
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- def set_redis(self, rhost, rport):
|
|
|
- self._mRHost = rhost
|
|
|
- self._mRPort = rport
|
|
|
-
|
|
|
- def stop(self):
|
|
|
- self._mquit = True
|
|
|
- pass
|
|
|
-
|
|
|
- def prepare_data(self):
|
|
|
- def open(file):
|
|
|
- if path.exists(file):
|
|
|
- hfive = h5py.File(file, 'a')
|
|
|
- else:
|
|
|
- hfive = h5py.File(file, 'w')
|
|
|
- return hfive
|
|
|
-
|
|
|
- while self._mquit == False:
|
|
|
- try:
|
|
|
- pool = redis.ConnectionPool(host=self._mRHost, port=self._mRPort, db=0)
|
|
|
- r = redis.Redis(connection_pool=pool)
|
|
|
-
|
|
|
- for _key,val in self._file_names.items():
|
|
|
- hfive = open(val['name'])
|
|
|
- if _key == 'profit':
|
|
|
- val['handler'] = ProfitHandler(hfive)
|
|
|
-
|
|
|
- self.read(self.queue_name,r)
|
|
|
- except Exception as ex:
|
|
|
- print(ex)
|
|
|
- finally:
|
|
|
- stime.sleep(1)
|
|
|
-
|
|
|
- def read(self,queue,redis):
|
|
|
- while self._mquit == False:
|
|
|
- item = redis.brpop(queue, 1)
|
|
|
- if item is None:
|
|
|
- continue
|
|
|
- else:
|
|
|
- val = json.loads(item[1])
|
|
|
- self._parse(val)
|
|
|
-
|
|
|
- def _parse(self,val):
|
|
|
- method = val['method']
|
|
|
- params = val['params']
|
|
|
-
|
|
|
-
|
|
|
- def parase(self, hfive, text, val, prefix):
|
|
|
- items = re.split(r'-', text)
|
|
|
- if len(items) != 5:
|
|
|
- return False
|
|
|
-
|
|
|
- (mchid, quality, card_type, amount, time) = items
|
|
|
- pos = self.pos_map[f'{prefix}']
|
|
|
-
|
|
|
- time = int(time)
|
|
|
- today = self.day_stamp(time)
|
|
|
- path = f'/{today}/{mchid}/{quality}/{card_type}/{amount}'
|
|
|
- if path not in hfive:
|
|
|
- dim = len(self.pos_map)
|
|
|
- hfive[path] = np.zeros((dim, 86400))
|
|
|
-
|
|
|
- diff = time - today
|
|
|
- if diff < 0:
|
|
|
- print(diff)
|
|
|
- hfive[path][pos, diff] = val
|
|
|
- print(path, pos, prefix, diff, time, val, hfive[path][pos, diff])
|
|
|
- pass
|
|
|
-
|
|
|
- def day_stamp(self, stamp):
|
|
|
- stamp = int(stamp)
|
|
|
- x = stime.gmtime(stamp + 8 * 3600)
|
|
|
- diff = timedelta(hours=x.tm_hour, minutes=x.tm_min, seconds=x.tm_sec)
|
|
|
- today = stamp - diff.total_seconds()
|
|
|
- return int(today)
|
|
|
-
|
|
|
- def _days(self, root):
|
|
|
- result = []
|
|
|
- try:
|
|
|
- for name, sub in root.items():
|
|
|
- if isinstance(sub, h5py.Group):
|
|
|
- result.append(name)
|
|
|
- except Exception as ex:
|
|
|
- print(ex)
|
|
|
- finally:
|
|
|
- return result
|
|
|
-
|
|
|
- def days(self):
|
|
|
- try:
|
|
|
- hfive = h5py.File(self._file_name, 'r')
|
|
|
- root = hfive.require_group('/')
|
|
|
- days = self._days(root)
|
|
|
- hfive.close()
|
|
|
- return days
|
|
|
- except Exception as ex:
|
|
|
- print(ex)
|
|
|
- return []
|
|
|
-
|
|
|
- def paths(self, time_stamp):
|
|
|
- try:
|
|
|
- day_stamp = self.day_stamp(time_stamp)
|
|
|
- hfive = h5py.File(self._file_name, 'r')
|
|
|
- group = hfive.require_group(f'/{day_stamp}')
|
|
|
- paths = self.dir(group)
|
|
|
- hfive.close()
|
|
|
- return paths
|
|
|
- except Exception as ex:
|
|
|
- print(ex)
|
|
|
- return []
|
|
|
-
|
|
|
- def dir(self, group):
|
|
|
- result = []
|
|
|
- for name, sub in group.items():
|
|
|
- if isinstance(sub, h5py.Group):
|
|
|
- result.extend(self.dir(sub))
|
|
|
- else:
|
|
|
- result.append(sub.name)
|
|
|
- return result
|
|
|
-
|
|
|
- def _all_none(self, **kwargs):
|
|
|
- for key, val in kwargs.items():
|
|
|
- if val is not None:
|
|
|
- return False
|
|
|
- return True
|
|
|
-
|
|
|
- def _merge_path(self,paths):
|
|
|
- result = {}
|
|
|
- for path in paths:
|
|
|
- items = re.split(r'/', path)
|
|
|
- if len(items) != 6:
|
|
|
- continue
|
|
|
- (_, _sday, _mchid, _quality, _card_type, _amount) = items
|
|
|
-
|
|
|
- _mchid = int(_mchid)
|
|
|
- if _mchid not in result:
|
|
|
- result[_mchid] = []
|
|
|
- result[_mchid].append(path)
|
|
|
-
|
|
|
- return result
|
|
|
-
|
|
|
- def draw_plot(self, start_time, interval=300, **kwargs):
|
|
|
- logger = logging.getLogger('app')
|
|
|
- hfive = h5py.File(self._file_name, 'r')
|
|
|
- try:
|
|
|
- day_stamp = self.day_stamp(start_time)
|
|
|
- start_pos = start_time - day_stamp
|
|
|
-
|
|
|
- cur_day = self.day_stamp(stime.time())
|
|
|
- if day_stamp == cur_day:
|
|
|
- end_pos = int(stime.time()) - day_stamp
|
|
|
- else:
|
|
|
- end_pos = -1
|
|
|
-
|
|
|
- fig = Figure(figsize=(16, 8))
|
|
|
- ax = fig.subplots()
|
|
|
-
|
|
|
- dim = len(self.pos_map)
|
|
|
- predata = np.zeros((dim, 86400))
|
|
|
- x = np.arange(0, 86400, interval)
|
|
|
-
|
|
|
- sub_count = 0
|
|
|
- filer_text, paths = self.datasets(hfive, start_time, **kwargs)
|
|
|
- if self._all_none(**kwargs):
|
|
|
- paths = self._merge_path(paths)
|
|
|
- for mchid, data in self._read_dict_data(hfive, paths):
|
|
|
- predata = predata + data
|
|
|
- path = f'{mchid}'
|
|
|
- ret = self._draw_plot(ax, x, day_stamp, start_pos, end_pos, data, interval, path)
|
|
|
- if ret:
|
|
|
- sub_count += 1
|
|
|
- pass
|
|
|
- else:
|
|
|
- for path, data in self.read_data(hfive, paths):
|
|
|
- data = np.array(data)
|
|
|
- predata = predata + data
|
|
|
- ret = self._draw_plot(ax, x, day_stamp, start_pos, end_pos, data, interval, path)
|
|
|
- if ret:
|
|
|
- sub_count += 1
|
|
|
- if sub_count > 1:
|
|
|
- self._draw_plot(ax, x, day_stamp, start_pos, end_pos, predata, interval, filer_text)
|
|
|
-
|
|
|
- ax.legend()
|
|
|
- ax.grid()
|
|
|
- ax.set_title('success ratio')
|
|
|
- ax.set(xlabel='time', ylabel='ratio')
|
|
|
- fig.autofmt_xdate()
|
|
|
- fig.subplots_adjust(left=0.05, right=0.999, top=0.95, bottom=0.1)
|
|
|
-
|
|
|
- buf = BytesIO()
|
|
|
- fig.savefig(buf, format="png")
|
|
|
- return buf
|
|
|
- except Exception as ex:
|
|
|
- print(ex)
|
|
|
- finally:
|
|
|
- hfive.close()
|
|
|
-
|
|
|
- def read_data(self, hfive, paths):
|
|
|
- for path in paths:
|
|
|
- yield path, hfive[path]
|
|
|
-
|
|
|
- def _read_dict_data(self, hfive, mchPaths):
|
|
|
- for mchid, paths in mchPaths.items():
|
|
|
- dim = len(self.pos_map)
|
|
|
- predata = np.zeros((dim, 86400))
|
|
|
- for path in paths:
|
|
|
- predata += hfive[path]
|
|
|
- yield mchid, predata
|
|
|
-
|
|
|
- def datasets(self, hfive, start_time, **kwargs):
|
|
|
- logger = logging.getLogger('app')
|
|
|
-
|
|
|
- day_stamp = self.day_stamp(start_time)
|
|
|
- sday = f'{day_stamp}'
|
|
|
- root = hfive.require_group('/')
|
|
|
- days = self._days(root)
|
|
|
- if sday not in days:
|
|
|
- return False
|
|
|
-
|
|
|
- group = hfive.require_group(sday)
|
|
|
- dsets = self.dir(group)
|
|
|
-
|
|
|
- mchid = quality = card_type = amount = None
|
|
|
- for key, val in kwargs.items():
|
|
|
- if val is None:
|
|
|
- continue
|
|
|
- if key == 'mchid':
|
|
|
- mchid = val
|
|
|
- elif key == 'quality':
|
|
|
- quality = f'{val}'
|
|
|
- elif key == 'card_type':
|
|
|
- card_type = f'{val}'
|
|
|
- elif key == 'amount':
|
|
|
- amount = f'{val}'
|
|
|
- else:
|
|
|
- continue
|
|
|
- return self._filter(dsets, mchid=mchid, quality=quality, card_type=card_type, amount=amount)
|
|
|
-
|
|
|
- def _filter(self, dsets, mchid=None, quality=None, card_type=None, amount=None):
|
|
|
- filer_text = ''
|
|
|
- if mchid is not None:
|
|
|
- filer_text = mchid
|
|
|
- if quality is not None:
|
|
|
- filer_text = filer_text + f"-qua:{quality}"
|
|
|
- if card_type is not None:
|
|
|
- filer_text = filer_text + f"-type:{card_type}"
|
|
|
- if amount is not None:
|
|
|
- filer_text = filer_text + f"-amount:{amount}"
|
|
|
-
|
|
|
- paths = []
|
|
|
- for text in dsets:
|
|
|
- items = re.split(r'/', text)
|
|
|
- if len(items) != 6:
|
|
|
- return False
|
|
|
- (_, _sday, _mchid, _quality, _card_type, _amount) = items
|
|
|
- if (mchid is not None) and (_mchid != mchid):
|
|
|
- continue
|
|
|
- if (quality is not None) and (_quality != quality):
|
|
|
- continue
|
|
|
- if (card_type is not None) and (_card_type != card_type):
|
|
|
- continue
|
|
|
- if (amount is not None) and (_amount != amount):
|
|
|
- continue
|
|
|
- paths.append(text)
|
|
|
-
|
|
|
- return filer_text, paths
|
|
|
-
|
|
|
- def _draw_plot(self, ax, x, day_stamp, start_pos, end_pos, data, interval=300, path=''):
|
|
|
- # 'commit': 0, 'succ': 1, 'fail': 2
|
|
|
- logging.getLogger('app').debug("path=%s", path)
|
|
|
-
|
|
|
- all = data[1] + data[2]
|
|
|
- all = all.reshape((-1, interval))
|
|
|
- all = np.sum(all, axis=1)
|
|
|
-
|
|
|
- commit = data[0]
|
|
|
- commit = commit.reshape((-1, interval))
|
|
|
- commit = np.sum(commit, axis=1)
|
|
|
-
|
|
|
- ySucc = data[1]
|
|
|
- ySucc = ySucc.reshape((-1, interval))
|
|
|
- ySucc = np.sum(ySucc, axis=1)
|
|
|
-
|
|
|
- yFail = data[2]
|
|
|
- yFail = yFail.reshape((-1, interval))
|
|
|
- yFail = np.sum(yFail, axis=1)
|
|
|
-
|
|
|
- if end_pos == -1:
|
|
|
- pos = np.where(x >= start_pos)
|
|
|
- x = x[pos]
|
|
|
- ySucc = ySucc[pos]
|
|
|
- all = all[pos]
|
|
|
- commit = commit[pos]
|
|
|
- yFail = yFail[pos]
|
|
|
- else:
|
|
|
- pos = np.where(start_pos <= x)
|
|
|
- x = x[pos]
|
|
|
- ySucc = ySucc[pos]
|
|
|
- all = all[pos]
|
|
|
- commit = commit[pos]
|
|
|
- yFail = yFail[pos]
|
|
|
-
|
|
|
- pos = np.where(x < end_pos)
|
|
|
- x = x[pos]
|
|
|
- ySucc = ySucc[pos]
|
|
|
- all = all[pos]
|
|
|
- commit = commit[pos]
|
|
|
- yFail = yFail[pos]
|
|
|
-
|
|
|
- succ_count = int(np.sum(ySucc))
|
|
|
- all_count = int(np.sum(all))
|
|
|
- commit_count = int(np.sum(commit))
|
|
|
- fail_count = int(np.sum(yFail))
|
|
|
-
|
|
|
- if all_count < 1:
|
|
|
- return False
|
|
|
-
|
|
|
- pos = np.where(ySucc > all)
|
|
|
- ySucc[pos] = all[pos]
|
|
|
-
|
|
|
- ySucc = ySucc / (all + 0.00000001)
|
|
|
- xs = np.array([stime.strftime('%H:%M', stime.localtime(d + day_stamp)) for d in x])
|
|
|
- ax.yaxis.set_major_formatter(ticker.PercentFormatter(xmax=1, decimals=0))
|
|
|
- ax.plot(xs, ySucc, ls='--', marker='o', label=self._label(path, succ_count, all_count,commit_count,fail_count))
|
|
|
- return True
|
|
|
-
|
|
|
- def _label(self, path, succ_count, all, commit_count, fail_count):
|
|
|
- ratio = 0.00
|
|
|
- if all > 0:
|
|
|
- ratio = round(succ_count * 100 / all, 2)
|
|
|
-
|
|
|
- items = re.split(r'/', path)
|
|
|
- if len(items) == 6:
|
|
|
- (_, _sday, _chname, _quality, _card_type, _amount) = items
|
|
|
- card_type = ''
|
|
|
- if _card_type == '1':
|
|
|
- card_type = 'SY'
|
|
|
- elif _card_type == '2':
|
|
|
- card_type = 'SH'
|
|
|
- elif _card_type == '4':
|
|
|
- card_type = 'YD'
|
|
|
- elif _card_type == '5':
|
|
|
- card_type = 'LT'
|
|
|
- elif _card_type == '6':
|
|
|
- card_type = 'DX'
|
|
|
- elif _card_type == '7':
|
|
|
- card_type = 'TH'
|
|
|
- return f"{_chname}-{_quality}-{card_type}-{_amount}:{succ_count}/{all} = {ratio}% {commit_count}:{fail_count}"
|
|
|
- else:
|
|
|
- if path == '' or path is None:
|
|
|
- path = 'average'
|
|
|
- return f"{path}:{succ_count}/{all} = {ratio}% {commit_count}:{fail_count}"
|
|
|
- pass
|
|
|
-
|
|
|
- #统计机构当前时间之前序列时间成功率
|
|
|
- def _merge_mobile_path(self,paths):
|
|
|
- result = {}
|
|
|
- for path in paths:
|
|
|
- items = re.split(r'/', path)
|
|
|
- if len(items) != 6:
|
|
|
- continue
|
|
|
- (_, _sday, _mchid, _quality, _card_type, _amount) = items
|
|
|
-
|
|
|
- _card_type = int(_card_type)
|
|
|
- if _card_type not in [4, 5, 6]:
|
|
|
- continue
|
|
|
-
|
|
|
- _mchid = int(_mchid)
|
|
|
- if _mchid not in result:
|
|
|
- result[_mchid] = []
|
|
|
- result[_mchid].append(path)
|
|
|
- return result
|
|
|
-
|
|
|
- def _merge_data(self, hfive, paths):
|
|
|
- dim = len(self.pos_map)
|
|
|
- predata = np.zeros((dim, 86400))
|
|
|
- for path in paths:
|
|
|
- predata += hfive[path]
|
|
|
- return predata
|
|
|
-
|
|
|
- def _calc_mratio(self,data,startes,end):
|
|
|
- succ = data[1]
|
|
|
- fail = data[2]
|
|
|
- x = np.arange(0, 86400, 1)
|
|
|
-
|
|
|
- result = {}
|
|
|
- for start in startes:
|
|
|
- if end - start < 0:
|
|
|
- start_pos = 0
|
|
|
- else:
|
|
|
- start_pos = end - start
|
|
|
-
|
|
|
- pos = np.where(x >= start_pos)
|
|
|
- t = x[pos]
|
|
|
- _fail = fail[pos]
|
|
|
- _succ = succ[pos]
|
|
|
-
|
|
|
- pos = np.where(t < end)
|
|
|
- _fail = _fail[pos]
|
|
|
- _succ = _succ[pos]
|
|
|
-
|
|
|
- succs = int(np.sum(_succ))
|
|
|
- fails = int(np.sum(_fail))
|
|
|
- ratio = round((succs + 0.00001) / (succs + fails + 0.00001), 4)
|
|
|
- result[start] = [succs, fails, ratio]
|
|
|
- return result
|
|
|
-
|
|
|
- def mratios(self, time_stamp,presecs):
|
|
|
- paths = self.paths(time_stamp)
|
|
|
- mchid_paths = self._merge_mobile_path(paths)
|
|
|
- day_stamp = self.day_stamp(time_stamp)
|
|
|
-
|
|
|
- mratios = {}
|
|
|
- hfive = h5py.File(self._file_name, 'r')
|
|
|
- for mchid, paths in mchid_paths.items():
|
|
|
- mdata = self._merge_data(hfive,paths)
|
|
|
- result = self._calc_mratio(mdata,presecs,time_stamp - day_stamp)
|
|
|
- mratios[mchid] = result
|
|
|
- hfive.close()
|
|
|
- return mratios
|
|
|
-
|
|
|
- def calc_ratio(self):
|
|
|
- import json
|
|
|
-
|
|
|
- r = None
|
|
|
- try:
|
|
|
- pool = redis.ConnectionPool(host=self._mRHost, port=self._mRPort, db=0)
|
|
|
- r = redis.Redis(connection_pool=pool)
|
|
|
- except Exception as ex:
|
|
|
- print(ex)
|
|
|
-
|
|
|
- while True:
|
|
|
- try:
|
|
|
- time_sec = int(stime.time())
|
|
|
- presecs = [900, 1800, 3600, 7200, 86400]
|
|
|
- mratios = self.mratios(time_sec, presecs)
|
|
|
-
|
|
|
- if len(mratios) != 0:
|
|
|
- r.set(f"nc_merchant_ratios", json.dumps(mratios))
|
|
|
- r.publish('refill',json.dumps({'type':'mch_ratio','value':0}))
|
|
|
- print('push msg=',mratios)
|
|
|
- except Exception as ex:
|
|
|
- print(ex)
|
|
|
- finally:
|
|
|
- stime.sleep(2)
|
|
|
-
|
|
|
- #以下为按照卡类型计算成功率代码
|
|
|
- def _merge_mobile_type_path(self,paths,card_type=None):
|
|
|
- result = {}
|
|
|
- for path in paths:
|
|
|
- items = re.split(r'/', path)
|
|
|
- if len(items) != 6:
|
|
|
- continue
|
|
|
- (_, _sday, _mchid, _quality, _card_type, _amount) = items
|
|
|
-
|
|
|
- _card_type = int(_card_type)
|
|
|
- if _card_type not in [4, 5, 6]:
|
|
|
- continue
|
|
|
- if card_type is not None and _card_type != card_type:
|
|
|
- continue
|
|
|
-
|
|
|
- _mchid = int(_mchid)
|
|
|
- if _mchid not in result:
|
|
|
- result[_mchid] = []
|
|
|
- result[_mchid].append(path)
|
|
|
- return result
|
|
|
-
|
|
|
- def mratio_types(self, time_stamp,presecs):
|
|
|
- paths = self.paths(time_stamp)
|
|
|
-
|
|
|
- mchid_paths = self._merge_mobile_type_path(paths)
|
|
|
- day_stamp = self.day_stamp(time_stamp)
|
|
|
-
|
|
|
- card_types = {None: 'ALL', 4: 'YD', 5: 'LT', 6: 'DX'}
|
|
|
-
|
|
|
- mratios = {}
|
|
|
- hfive = h5py.File(self._file_name, 'r')
|
|
|
- for mchid, paths in mchid_paths.items():
|
|
|
- mch_ratios = {}
|
|
|
- for card_type, name in card_types.items():
|
|
|
- print('card_type=', card_type, 'name=', name)
|
|
|
- if card_type is None:
|
|
|
- cur_paths = paths
|
|
|
- else:
|
|
|
- cur_paths = self._merge_mobile_type_path(paths, card_type)
|
|
|
- if len(cur_paths) == 0:
|
|
|
- continue
|
|
|
- cur_paths = cur_paths[mchid]
|
|
|
- mdata = self._merge_data(hfive,cur_paths)
|
|
|
- result = self._calc_mratio(mdata,presecs,time_stamp - day_stamp)
|
|
|
- mch_ratios[name] = result
|
|
|
- mratios[mchid] = mch_ratios
|
|
|
- hfive.close()
|
|
|
- return mratios
|
|
|
-
|
|
|
- def calc_ratios(self):
|
|
|
- import json
|
|
|
-
|
|
|
- r = None
|
|
|
- try:
|
|
|
- pool = redis.ConnectionPool(host=self._mRHost, port=self._mRPort, db=0)
|
|
|
- r = redis.Redis(connection_pool=pool)
|
|
|
- except Exception as ex:
|
|
|
- print(ex)
|
|
|
-
|
|
|
- while True:
|
|
|
- try:
|
|
|
- time_sec = int(stime.time())
|
|
|
- presecs = [900, 1800, 3600, 7200, 86400]
|
|
|
- mratios = self.mratio_types(time_sec, presecs)
|
|
|
-
|
|
|
- if len(mratios) != 0:
|
|
|
- r.set(f"nc_merchant_card_type_ratios", json.dumps(mratios))
|
|
|
- # r.publish('refill',json.dumps({'type':'mch_ratio','value':0}))
|
|
|
- print('push msg=', mratios)
|
|
|
- except Exception as ex:
|
|
|
- print(ex)
|
|
|
- finally:
|
|
|
- stime.sleep(2)
|
|
|
-
|
|
|
- ####################################################################################################################
|
|
|
- ####
|
|
|
- def _calc_mcount(self,data,startes,end):
|
|
|
- succ = data[1]
|
|
|
- fail = data[2]
|
|
|
- x = np.arange(0, 86400, 1)
|
|
|
-
|
|
|
- result = {}
|
|
|
- for start in startes:
|
|
|
- if end - start < 0:
|
|
|
- start_pos = 0
|
|
|
- else:
|
|
|
- start_pos = end - start
|
|
|
-
|
|
|
- pos = np.where(x >= start_pos)
|
|
|
- t = x[pos]
|
|
|
- _succ = succ[pos]
|
|
|
- _fail = fail[pos]
|
|
|
-
|
|
|
- pos = np.where(t < end)
|
|
|
- _succ = _succ[pos]
|
|
|
- _fail = _fail[pos]
|
|
|
-
|
|
|
- succs = int(np.sum(_succ))
|
|
|
- fails = int(np.sum(_fail))
|
|
|
- ratio = round((succs + 0.00001) / (succs + fails + 0.00001), 4)
|
|
|
- result[start] = [succs, fails, ratio]
|
|
|
- return result
|
|
|
-
|
|
|
- def merchant_rmobile_path(self, paths):
|
|
|
- result = {}
|
|
|
- for path in paths:
|
|
|
- items = re.split(r'/', path)
|
|
|
- if len(items) != 6:
|
|
|
- continue
|
|
|
- (_, _sday, _mchid, _quality, _card_type, _amount) = items
|
|
|
-
|
|
|
- _card_type = int(_card_type)
|
|
|
- if _card_type not in [4, 5, 6]:
|
|
|
- continue
|
|
|
-
|
|
|
- _mchid = int(_mchid)
|
|
|
- if _mchid not in result:
|
|
|
- result[_mchid] = []
|
|
|
- result[_mchid].append(path)
|
|
|
- return result
|
|
|
-
|
|
|
- def rmobile_path(self, paths):
|
|
|
- result = []
|
|
|
- for path in paths:
|
|
|
- items = re.split(r'/', path)
|
|
|
- if len(items) != 6:
|
|
|
- continue
|
|
|
- (_, _sday, _mchid, _quality, _card_type, _amount) = items
|
|
|
-
|
|
|
- _card_type = int(_card_type)
|
|
|
- if _card_type not in [4, 5, 6]:
|
|
|
- continue
|
|
|
- result.append(path)
|
|
|
- return result
|
|
|
-
|
|
|
- def mch_count(self, paths, presecs,time_stamp):
|
|
|
- hfive = h5py.File(self._file_name, 'r')
|
|
|
- mdata = self._merge_data(hfive,paths)
|
|
|
- day_stamp = self.day_stamp(time_stamp)
|
|
|
- result = self._calc_mcount(mdata,presecs,time_stamp - day_stamp)
|
|
|
- hfive.close()
|
|
|
-
|
|
|
- return result
|
|
|
-
|
|
|
- def mch_detail_count(self, paths, presecs,time_stamp):
|
|
|
- hfive = h5py.File(self._file_name, 'r')
|
|
|
-
|
|
|
- result = {}
|
|
|
- for path in paths:
|
|
|
- items = re.split(r'/', path)
|
|
|
- if len(items) != 6:
|
|
|
- continue
|
|
|
- (_, _sday, _mchid, _quality, _card_type, _amount) = items
|
|
|
- key = f"{_mchid}-{_quality}-{_card_type}-{_amount}"
|
|
|
- mdata = hfive[path]
|
|
|
- day_stamp = self.day_stamp(time_stamp)
|
|
|
- result[key] = self._calc_mcount(mdata,presecs,time_stamp - day_stamp)
|
|
|
- hfive.close()
|
|
|
- return result
|
|
|
-
|
|
|
- def mch_counts(self):
|
|
|
- import json
|
|
|
-
|
|
|
- r = None
|
|
|
- try:
|
|
|
- pool = redis.ConnectionPool(host=self._mRHost, port=self._mRPort, db=0)
|
|
|
- r = redis.Redis(connection_pool=pool)
|
|
|
- except Exception as ex:
|
|
|
- print(ex)
|
|
|
-
|
|
|
- while True:
|
|
|
- try:
|
|
|
- time_stamp = int(stime.time())
|
|
|
- presecs = [900, 1800, 3600, 7200, 86400]
|
|
|
- all_paths = self.paths(time_stamp)
|
|
|
- mchid_paths = self.merchant_rmobile_path(all_paths)
|
|
|
-
|
|
|
- gross = {}
|
|
|
- for mchid, paths in mchid_paths.items():
|
|
|
- counts = self.mch_count(paths, presecs, time_stamp)
|
|
|
- gross[mchid] = counts
|
|
|
-
|
|
|
- paths = self.rmobile_path(all_paths)
|
|
|
- detail = self.mch_detail_count(paths, presecs, time_stamp)
|
|
|
-
|
|
|
- result = {'gross': gross, 'detail': detail}
|
|
|
- if len(result) != 0:
|
|
|
- r.set(f"nc_merchant_refill_counts", json.dumps(result))
|
|
|
- r.publish('refill',json.dumps({'type':'mch_counts','value':0}))
|
|
|
- print('push msg=', result)
|
|
|
- except Exception as ex:
|
|
|
- print(ex)
|
|
|
- finally:
|
|
|
- stime.sleep(2)
|
|
|
-
|
|
|
-queueListener = QueueListener()
|
stanley-king