def acctounting(self):
        if not self.account:
            return logger.error(
                "[Acct] Received an accounting update request but user[%s] not exists"% self.request.account_number)  

        ticket = Storage(**self.request)
        _datetime = datetime.datetime.now() 
        online = self.get_online(ticket.nas_addr,ticket.acct_session_id)    
        if not online:
            session_time = ticket.acct_session_time 
            stop_time = _datetime.strftime( "%Y-%m-%d %H:%M:%S")
            start_time = (_datetime - datetime.timedelta(seconds=int(session_time))).strftime( "%Y-%m-%d %H:%M:%S")
            ticket.acct_start_time = start_time
            ticket.acct_stop_time = stop_time
            ticket.start_source= STATUS_TYPE_STOP
            ticket.stop_source = STATUS_TYPE_STOP
            self.add_ticket(ticket)
        else:
            self.del_online(ticket.nas_addr,ticket.acct_session_id)
            ticket.acct_start_time = online.acct_start_time
            ticket.acct_stop_time= _datetime.strftime( "%Y-%m-%d %H:%M:%S")
            ticket.start_source = online.start_source
            ticket.stop_source = STATUS_TYPE_STOP
            self.add_ticket(ticket)

            self.billing(online)
            logger.info('%s Accounting stop request, remove online'% self.account.account_number)

    def process(self, *args, **kwargs):
        self.logtimes()
        with make_db(self.db) as db:
            try:
                nodes = db.query(models.TrNode)
                for node in nodes:
                    r = db.query(
                        func.sum(models.TrOnline.input_total).label("input_total"),
                        func.sum(models.TrOnline.output_total).label("output_total")
                    ).filter(
                        models.TrOnline.account_number == models.TrAccount.account_number,
                        models.TrAccount.customer_id == models.TrCustomer.customer_id,
                        models.TrCustomer.node_id == node.id
                    ).first()
                    if r and all([r.input_total,r.output_total]):
                        stat = models.TrFlowStat()
                        stat.node_id = node.id
                        stat.stat_time = int(time.time())
                        stat.input_total = r.input_total
                        stat.output_total = r.output_total
                        db.add(stat)

                # clean expire data
                _time = int(time.time()) - (86400 * 2)
                db.query(models.TrFlowStat).filter(models.TrFlowStat.stat_time < _time).delete()

                db.commit()
                logger.info("flow stat task done")
            except Exception as err:
                db.rollback()
                logger.error('flow_stat_job err,%s'%(str(err)))
        
        return self.get_notify_interval()

 def process(self, message):
     datagram, host, port =  umsgpack.unpackb(message[0])
     reply = self.processAuth(datagram, host, port)
     logger.info("[Radiusd] :: Send radius response: %s" % repr(reply))
     if self.config.system.debug:
         logger.debug(reply.format_str())
     self.pusher.push(umsgpack.packb([reply.ReplyPacket(),host,port]))

    def send_mail(self, mailto, topic, content, tls=False,**kwargs):
        message = email.MIMEText.MIMEText(content,'html', 'utf-8')
        message["Subject"] = email.Header.Header(topic,'utf-8')
        message["From"] = self.from_addr
        message["To"] = mailto
        message["Accept-Language"]="zh-CN"
        message["Accept-Charset"]="ISO-8859-1,utf-8"
        if not tls:
            logger.info('send mail:%s:%s:%s'%(self.smtp_server,self.smtp_port,mailto))
            return sendmail(self.smtp_server, self.from_addr, mailto, message,
                        port=self.smtp_port, username=self.smtp_user, password=self.smtp_pwd)
        else:
            logger.info('send tls mail:%s:%s:%s'%(self.smtp_server,self.smtp_port,mailto))
            contextFactory = ContextFactory()
            resultDeferred = Deferred()
            senderFactory = ESMTPSenderFactory(
                self.smtp_user,
                self.smtp_pwd,
                self.from_addr,
                mailto,
                StringIO(message.as_string()),
                resultDeferred,
                contextFactory=contextFactory,
                requireAuthentication=(self.smtp_user and self.smtp_pwd),
                requireTransportSecurity=tls)

            reactor.connectTCP(self.smtp_server, self.smtp_port, senderFactory)
            return resultDeferred

    def process(self, *args, **kwargs):
        self.logtimes()
        with make_db(self.db) as db:
            try:
                nodes = db.query(models.TrNode)
                for node in nodes:
                    online_count = db.query(models.TrOnline.id).filter(
                        models.TrOnline.account_number == models.TrAccount.account_number,
                        models.TrAccount.customer_id == models.TrCustomer.customer_id,
                        models.TrCustomer.node_id == node.id
                    ).count()
                    stat = models.TrOnlineStat()
                    stat.node_id = node.id
                    stat.stat_time = int(time.time())
                    stat.total = online_count
                    db.add(stat)

                # clean expire data
                _time = int(time.time()) - (86400 * 2)
                db.query(models.TrOnlineStat).filter(models.TrOnlineStat.stat_time < _time).delete()

                db.commit()
                logger.info("online stat task done")
            except Exception as err:
                db.rollback()
                logger.exception(err)
        
        return self.get_notify_interval()

 def start(self):
     self.start_expire_notify()
     self.start_ddns_update()
     self.start_radius_stat_update()
     self.start_online_stat_task()
     self.start_flow_stat_task()
     logger.info('init task done')

    def event_toughcloud_mail_account_open(self, userinfo):
        """ toughCloud mail api open notify without password event """
        if not userinfo:
            return

        if not userinfo.get('email'):
            logger.error('user email is None exit')
            return

        try:
            api_secret = self.get_param_value("toughcloud_license")
            service_mail = self.get_param_value("toughcloud_service_mail")
            if not service_mail:
                return
            api_token = yield tools.get_sys_token()
            params = dict(
                token=api_token.strip(),
                action='email',
                mailto=userinfo.get('email'),
                tplname=self.MAIL_TPLNAME,
                customer=utils.safestr(userinfo.get('realname')),
                username=userinfo.get('account_number'),
                product=utils.safestr(userinfo.get('product_name')),
                expire=userinfo.get('expire_date'),
                service_call=self.get_param_value("toughcloud_service_call", ''),
                service_mail=service_mail,
                nonce=str(int(time.time()))
            )
            params['sign'] = apiutils.make_sign(api_secret.strip(), params.values())
            resp = yield httpclient.fetch(self.MAIL_APIURL, postdata=urlencode(params))
            logger.info(resp.body)
            logger.info('open account send email without password success')
        except Exception as err:
            logger.exception(err)

    def event_toughcloud_mail_account_expire(self, userinfo):
        """ toughcloud mail api notify event """
        if not userinfo:
            return

        api_secret = self.get_param_value("toughcloud_license")
        service_mail=self.get_param_value("toughcloud_service_mail")
        if not service_mail:
            return
        api_token = yield tools.get_sys_token()
        params = dict(
            token=api_token.strip(),
            mailto=userinfo.email,
            tplname=self.MAIL_TPLNAME,
            customer=utils.safestr(userinfo.realname),
            username=userinfo.account_number,
            product=utils.safestr(userinfo.product_name),
            expire=userinfo.expire_date,
            service_call=self.get_param_value("toughcloud_service_call",''),
            service_mail=service_mail,
            nonce = str(int(time.time()))
        )
        params['sign'] = apiutils.make_sign(api_secret.strip(), params.values())
        try:
            resp = yield httpclient.fetch(self.MAIL_APIURL, postdata=urlencode(params))
            logger.info(resp.body)
        except Exception as err:
            logger.exception(err)

    def processAcct(self, datagram, host, port):
        try:
            bas = self.find_nas(host)
            if not bas:
                raise PacketError('[Radiusd] :: Dropping packet from unknown host %s' % host)

            secret, vendor_id = bas['bas_secret'], bas['vendor_id']
            req = self.createAcctPacket(packet=datagram, 
                dict=self.dict, secret=six.b(str(secret)),vendor_id=vendor_id)

            logger.info("[Radiusd] :: Received radius request: %s" % (repr(req)))
            if self.config.system.debug:
                logger.debug(req.format_str())

            if req.code != packet.AccountingRequest:
                raise PacketError('non-AccountingRequest packet on authentication socket')

            if not req.VerifyAcctRequest():
                raise PacketError('VerifyAcctRequest error')

            reply = req.CreateReply()
            status_type = req.get_acct_status_type()
            if status_type in self.acct_class:
                acct_func = self.acct_class[status_type](self,req.get_ticket()).acctounting
                reactor.callLater(0.1,acct_func)
            return reply
        except Exception as err:
            errstr = 'RadiusError:Dropping invalid acct packet from {0} {1},{2}'.format(
                host, port, utils.safeunicode(err))
            logger.error(errstr)
            import traceback
            traceback.print_exc()

 def process(self, *args, **kwargs):
     next_interval = self.get_notify_interval()
     user_total = 0
     online_total = 0
     with make_db(self.db) as db:
         try:
             user_total = db.query(models.TrAccount).count()
             online_total = db.query(models.TrOnline).count()
         except Exception as err:
             pass
     try:
         api_url = "https://www.toughcloud.net/api/v1/ping"
         api_token = yield tools.get_sys_token()
         params = dict(
             token=api_token,
             app="toughradius",
             ver=__version__,
             release=self.config.system.get('release',"standard"),
             unum=user_total,
             onum=online_total,
             dist=' '.join(pf.linux_distribution()),
         )
         param_str = urlencode(params)
         resp = yield httpclient.fetch(api_url+"?"+param_str,followRedirect=True)
         logger.info("toughcloud ping resp code: %s"%resp.code)
     except Exception as err:
         logger.error(err)
     defer.returnValue(next_interval)

    def acctounting(self):
        if not self.account:
            dispatch.pub(UNLOCK_ONLINE_EVENT,
                self.request.account_number,self.request.nas_addr, self.request.acct_session_id,async=True)
            return logger.error(
                "[Acct] Received an accounting update request but user[%s] not exists"% self.request.account_number)     

        ticket = Storage(**self.request)
        online = self.get_online(ticket.nas_addr,ticket.acct_session_id)     
        if not online:         
            sessiontime = ticket.acct_session_time
            updatetime = datetime.datetime.now()
            _starttime = updatetime - datetime.timedelta(seconds=sessiontime)       
            online = Storage(
                account_number = self.account.account_number,
                nas_addr = ticket.nas_addr,
                acct_session_id = ticket.acct_session_id,
                acct_start_time = _starttime.strftime( "%Y-%m-%d %H:%M:%S"),
                framed_ipaddr = ticket.framed_ipaddr,
                mac_addr = ticket.mac_addr or '',
                nas_port_id = ticket.nas_port_id or '',
                billing_times = ticket.acct_session_time,
                input_total = self.get_input_total(),
                output_total = self.get_output_total(),
                start_source = STATUS_TYPE_UPDATE
            )
            self.add_online(online)

        self.billing(online)
        logger.info('%s Accounting update request, update online'% self.account.account_number)


        
        

    def event_unlock_online(self, account_number, nas_addr, acct_session_id):
        logger.info("event unlock online [username:{0}] {1} {2}".format(account_number, nas_addr, acct_session_id))
        nas = self.db.query(models.TrBas).filter_by(ip_addr=nas_addr).first()
        if nas_addr and  not nas:
            self.db.query(models.TrOnline).filter_by(
                nas_addr=nas_addr,acct_session_id=acct_session_id).delete()
            self.db.commit()
            return

        online = self.db.query(models.TrOnline).filter_by(
                nas_addr=nas_addr, acct_session_id=acct_session_id).first()

        authorize.disconnect(
            int(nas.vendor_id or 0),
            self.dictionary,
            nas.bas_secret,
            nas.ip_addr,
            coa_port=int(nas.coa_port or 3799),
            debug=True,
            User_Name=account_number,
            NAS_IP_Address=nas.ip_addr,
            Acct_Session_Id=acct_session_id
        ).addCallback(
            self.onSendResp, self.get_request(online)).addErrback(
            self.onSendError, self.get_request(online))

    def bill_botimes(self,online, product):
        #买断时长
        logger.info('%s > Buyout long time billing ' % self.account.account_number)
        time_length = self.get_user_time_length()
        sessiontime = self.request.acct_session_time
        billing_times = online.billing_times
        acct_times = sessiontime - billing_times
        if acct_times < 0:
            acct_times = 0
        user_time_length = time_length - acct_times
        if user_time_length < 0 :
            user_time_length = 0

        self.update_billing(Storage(
            account_number = online.account_number,
            nas_addr = online.nas_addr,
            acct_session_id = online.acct_session_id,
            acct_start_time = online.acct_start_time,
            acct_session_time = self.request.acct_session_time,
            input_total = self.get_input_total(),
            output_total = self.get_output_total(),
            acct_times = acct_times,
            acct_flows = 0,
            acct_fee = 0,
            actual_fee = 0,
            balance = 0,
            time_length = user_time_length,
            flow_length = 0,
            is_deduct = 1,
            create_time = datetime.datetime.now().strftime( "%Y-%m-%d %H:%M:%S")
        ))
    
        if user_time_length == 0 :
            dispatch.pub(UNLOCK_ONLINE_EVENT,
                online.account_number,online.nas_addr, online.acct_session_id,async=True)

    def process(self, *args, **kwargs):
        with make_db(self.db) as db:
            try:
                nodes = db.query(models.TrNode)
                for node in nodes:
                    online_count = (
                        db.query(models.TrOnline.id)
                        .filter(
                            models.TrOnline.account_number == models.TrAccount.account_number,
                            models.TrAccount.customer_id == models.TrCustomer.customer_id,
                            models.TrCustomer.node_id == node.id,
                        )
                        .count()
                    )
                    stat = models.TrOnlineStat()
                    stat.node_id = node.id
                    stat.stat_time = int(time.time())
                    stat.total = online_count
                    db.add(stat)
                db.commit()
                logger.info("online stat task done")
            except Exception as err:
                db.rollback()
                logger.error("online_stat_job err,%s" % (str(err)))

        return 120.0

 def bill_boflows(self, online, product):
     #买断流量
     logger.info('%s > Buyout flow billing ' % self.account.account_number)
     flow_length = self.get_user_flow_length()
     output_total = self.get_output_total()
     billing_output_total = online.output_total
     acct_flows = output_total - billing_output_total
     user_flow_length = flow_length - acct_flows
     if user_flow_length < 0 :
         user_flow_length = 0
         
     self.update_billing(Storage(
         account_number = online.account_number,
         nas_addr = online.nas_addr,
         acct_session_id = online.acct_session_id,
         acct_start_time = online.acct_start_time,
         acct_session_time = self.request.acct_session_time,
         input_total = self.get_input_total(),
         output_total = self.get_output_total(),
         acct_times = 0,
         acct_flows = acct_flows,
         acct_fee = 0,
         actual_fee = 0,
         balance = 0,
         time_length = 0,
         flow_length = user_flow_length,
         is_deduct = 1,
         create_time = datetime.datetime.now().strftime( "%Y-%m-%d %H:%M:%S")
     ))
     
     if user_flow_length == 0 :
         self.disconnect(online)

    def process(self):
        conn = self.db()
        try:
            nas_list = conn.query(models.TrwBas)
            for nas in nas_list:
                if not nas.dns_name:
                    continue
                results, _, _ = yield client.lookupAddress(nas.dns_name)
                if not results:
                    logger.error("domain {0} resolver empty".format(nas.dns_name))

                if results[0].type == dns.A:
                    ipaddr = ".".join(str(i) for i in struct.unpack("BBBB", results[0].payload.address))
                    if ipaddr:
                        nas.ip_addr = ipaddr
                        conn.commit()
                        logger.info("domain {0} resolver {1}  success".format(nas.dns_name,ipaddr))
                else:
                    logger.info("domain {0} no ip address,{1}".format(nas.dns_name, repr(results)))

        except Exception as err:
            logger.error('ddns process error %s' % utils.safeunicode(err.message))
        finally:
            conn.close()

        reactor.callLater(60, self.process, )

    def event_toughcloud_sms_account_open(self, userinfo):
        """ toughCloud sms api open notify event """
        if not userinfo:
            return

        if not userinfo.get('phone'):
            logger.error('user phone is None exit')
            return

        api_secret = self.get_param_value("toughcloud_license")
        api_token = yield tools.get_sys_token()
        params = dict(
            token=api_token.strip(),
            action='sms',
            tplname=self.SMS_TPLNAME,
            phone=userinfo.get('phone'),
            customer=utils.safestr(userinfo.get('realname')),
            username=userinfo.get('account_number'),
            product=utils.safestr(userinfo.get('product_name')),
            password=userinfo.get('password'),
            expire=userinfo.get('expire_date'),
            nonce=str(int(time.time()))
        )
        params['sign'] = apiutils.make_sign(api_secret.strip(), params.values())
        try:
            resp = yield httpclient.fetch(self.SMS_APIURL, postdata=urlencode(params))
            logger.info(resp.body)
            logger.info('open account send short message success')
        except Exception as err:
            logger.exception(err)

 def onSendResp(self, resp, disconnect_req):
     if disconnect_req and self.db.query(models.TrOnline).filter_by(
             nas_addr=disconnect_req.nas_addr,
             acct_session_id=disconnect_req.acct_session_id).count() > 0:
         radius_acct_stop.RadiusAcctStop(
             self.dbengine,self.mcache,self.aes,disconnect_req).acctounting()
     logger.info(u"send disconnect ok! coa resp : %s" % resp)

 def update(self):
     logger.info('Alive session  %s' % self.session_id)
     self.session_data['Acct-Status-Type'] = 3
     self.session_data["Acct-Output-Octets"]  +=  random.randint(1048576, 8192000)
     self.session_data["Acct-Input-Octets"]  +=  random.randint(1048576, 4096000)
     self.session_data['Acct-Session-Time'] = (int(time.time()) - self.session_start)
     acct_resp = yield self.radius.send_acct(**self.session_data)
     defer.returnValue(acct_resp)

 def __init__(self, config, service='auth'):
     self.config = config
     self.service = service
     self.pusher = ZmqPushConnection(ZmqFactory(), ZmqEndpoint('bind', 'ipc:///tmp/radiusd-%s-message' % service))
     self.puller = ZmqPullConnection(ZmqFactory(), ZmqEndpoint('bind', 'ipc:///tmp/radiusd-%s-result' % service))
     self.puller.onPull = self.reply
     logger.info("init %s master pusher : %s " % (self.service, self.pusher))
     logger.info("init %s master puller : %s " % (self.service, self.puller))