|
@@ -0,0 +1,340 @@
|
|
|
+<?php
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+class BatteryAlarmKafkaAction extends Action {
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ public function kafka_index( ){
|
|
|
+ $broker_list = C('KAFKA_BROKER_LIST');
|
|
|
+ if (empty($broker_list)) {
|
|
|
+ exit("KAFKA_BROKER_LIST must be config!".PHP_EOL);
|
|
|
+ }
|
|
|
+ $group = C('WXT_DATA_KAFKA_GROUP');
|
|
|
+ if (empty($group)) {
|
|
|
+ exit("WXT_DATA_KAFKA_GROUP must be config!".PHP_EOL);
|
|
|
+ }
|
|
|
+ $topics = C('C61_BATTERY_ALARM_KAFKA_TOPIC');
|
|
|
+ if (empty($topics)) {
|
|
|
+ exit("C61_BATTERY_ALARM_KAFKA_TOPIC must be config!".PHP_EOL);
|
|
|
+ }
|
|
|
+ $topics = explode(',',$topics);
|
|
|
+ $conf = new RdKafka\Conf();
|
|
|
+ // Set a rebalance callback to log partition assignments (optional)(当有新的消费进程加入或者退出消费组时,kafka 会自动重新分配分区给消费者进程,这里注册了一个回调函数,当分区被重新分配时触发)
|
|
|
+ $conf->setRebalanceCb(function (RdKafka\KafkaConsumer $kafka, $err, array $partitions = null) {
|
|
|
+ switch ($err) {
|
|
|
+ case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
|
|
|
+ echo "Assign: ";
|
|
|
+ var_dump($partitions);
|
|
|
+ $kafka->assign($partitions);
|
|
|
+ break;
|
|
|
+
|
|
|
+ case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
|
|
|
+ echo "Revoke: ";
|
|
|
+ var_dump($partitions);
|
|
|
+ $kafka->assign(NULL);
|
|
|
+ break;
|
|
|
+
|
|
|
+ default:
|
|
|
+ throw new \Exception($err);
|
|
|
+ }
|
|
|
+ });
|
|
|
+ // Configure the group.id. All consumer with the same group.id will consume( 配置groud.id 具有相同 group.id 的consumer 将会处理不同分区的消息,所以同一个组内的消费者数量如果订阅了一个topic, 那么消费者进程的数量多于这个topic 分区的数量是没有意义的。)
|
|
|
+ // different partitions.
|
|
|
+ $conf->set('group.id', $group);
|
|
|
+ // Initial list of Kafka brokers(添加 kafka集群服务器地址)
|
|
|
+
|
|
|
+ $conf->set('metadata.broker.list', $broker_list);
|
|
|
+ $topicConf = new RdKafka\TopicConf();
|
|
|
+ // Set where to start consuming messages when there is no initial offset in
|
|
|
+ // offset store or the desired offset is out of range.
|
|
|
+ // 'smallest': start from the beginning
|
|
|
+ $topicConf->set('auto.offset.reset', 'earliest');
|
|
|
+ // Set the configuration to use for subscribed/assigned topics
|
|
|
+ $conf->setDefaultTopicConf($topicConf);
|
|
|
+ $consumer = new RdKafka\KafkaConsumer($conf);
|
|
|
+ // 订阅轨迹数据topic
|
|
|
+ $consumer->subscribe($topics);
|
|
|
+ while (true) {
|
|
|
+ $message = $consumer->consume(120*1000);
|
|
|
+ switch ($message->err) {
|
|
|
+ case RD_KAFKA_RESP_ERR_NO_ERROR:
|
|
|
+ $data = json_decode($message->payload,true);
|
|
|
+ if( $data ){
|
|
|
+ //var_dump($data);
|
|
|
+ $result = $this->handleAlarmData($data);
|
|
|
+ if (APP_DEBUG && isset($result['message'])) {
|
|
|
+ echo $result['message'] . PHP_EOL;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ break;
|
|
|
+ case RD_KAFKA_RESP_ERR__PARTITION_EOF:
|
|
|
+ echo "No more messages; will wait for more\n";
|
|
|
+ break;
|
|
|
+ case RD_KAFKA_RESP_ERR__TIMED_OUT:
|
|
|
+ echo "Timed out\n";
|
|
|
+ break;
|
|
|
+ default:
|
|
|
+ throw new \Exception($message->errstr(), $message->err);
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ private function handleAlarmData( $data ){
|
|
|
+
|
|
|
+ if(!is_array($data)){
|
|
|
+ return array('success'=>false,'message'=>'failed,data must be array!');
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ /*
|
|
|
+ $data = array(
|
|
|
+ 'tid' => $data['tid'],
|
|
|
+ 'alarm' => $data['alarm'],
|
|
|
+ 'is_low' => $data['is_low'],
|
|
|
+ 'is_charge' => $data['is_charge'],
|
|
|
+ 'is_charge_full' => $data['is_charge_full'],
|
|
|
+ 'battery_level' => $data['battery_level'],
|
|
|
+ 'timestamp' => $data['timestamp']
|
|
|
+ );*/
|
|
|
+
|
|
|
+ $cond=array('imei'=>$data['tid']);
|
|
|
+ $device_info=M('devices')->where($cond)->find();
|
|
|
+ if(!$device_info){
|
|
|
+ return array('success'=>false,'message'=>$data['tid'].'devices not existed');
|
|
|
+ }
|
|
|
+ $save_data_log=array(
|
|
|
+ 'imei'=>$data['tid'],
|
|
|
+ 'is_low' => $data['is_low'],
|
|
|
+ 'is_charge' => $data['is_charge'],
|
|
|
+ 'is_charge_full' => $data['is_charge_full'],
|
|
|
+ 'battery_level' => $data['battery_level'],
|
|
|
+ 'created_at'=>time(),
|
|
|
+ );
|
|
|
+ if($data['battery_level']==0){
|
|
|
+ return array('success'=>false,'message'=>$data['tid'].' battery_level is 0');
|
|
|
+ }
|
|
|
+ $table_name =getBatteryTableName();
|
|
|
+ M($table_name)->createAdd($save_data_log);
|
|
|
+ if(!$device_info['user_id']){
|
|
|
+ $res=M('devices')->createSave(array('imei'=>$data['tid']),array('battery_level'=> $data['battery_level']));
|
|
|
+ return array('success'=>false,'message'=>$data['tid'].' userid not existed');
|
|
|
+ }
|
|
|
+ $this->wxMsg = new \Jiaruan\WxTmp();
|
|
|
+ $send_user=M('users')->where(array('id'=>$device_info['user_id']))->find();
|
|
|
+ $alarm_state=0;
|
|
|
+ $save_data=array('battery_level'=> $data['battery_level']);
|
|
|
+ $alarm_time=$data['timestamp'];
|
|
|
+ if($alarm_time<(time()-3600)){
|
|
|
+ $alarm_time=time();
|
|
|
+ }
|
|
|
+ $alarm_data=array(
|
|
|
+ 'device_number'=>$device_info['imei'],
|
|
|
+ //'alarm_type'=>'low_recover',
|
|
|
+ 'created_at'=>time(),
|
|
|
+ 'creator_id'=>$device_info['creator_id'],
|
|
|
+ 'alarm_time'=>date('Y-m-d H:i:s',$alarm_time),
|
|
|
+ );
|
|
|
+ $send_interval = C('WX_BAT_ALARM_SEND_INTERVAL');
|
|
|
+ if (!$send_interval) {
|
|
|
+ $send_interval=600;
|
|
|
+ }
|
|
|
+ $check_cond=array(
|
|
|
+ 'device_number'=>$device_info['imei'],
|
|
|
+ 'created_at'=>array('gt',time()-$send_interval),
|
|
|
+ );
|
|
|
+ if($data['is_charge_full']){
|
|
|
+ //新版本充满只报一次
|
|
|
+ //充电充满
|
|
|
+ $alarm_data['alarm_type']='charge_full';
|
|
|
+ $check_cond['alarm_type']='charge_full';
|
|
|
+ if(!M('alarm_report')->where($check_cond)->count()){
|
|
|
+ $insertId=M('alarm_report')->createAdd($alarm_data);
|
|
|
+ // 电量充满模板ID
|
|
|
+ $full_template_id = C('WX_CHARGE_FULL_ALARM_TEMPLATE_ID');
|
|
|
+ if (!$full_template_id) {
|
|
|
+ create_log('WX_CHARGE_FULL_ALARM_TEMPLATE_ID required', 'add_alarm');
|
|
|
+ return array('success'=>false,'message'=>'WX_CHARGE_FULL_ALARM_TEMPLATE_ID required');
|
|
|
+ }
|
|
|
+
|
|
|
+ $sendmsg = array(
|
|
|
+ 'touser'=>$send_user['wx_open_id'],
|
|
|
+ 'template_id' => $full_template_id, //"NeSrB4ViIkRMOuCtoA6BIPPi_N5L-B5zaWcIXJZwkD0",
|
|
|
+ 'data' => [
|
|
|
+ 'first' => [ 'value' =>$send_user['realname'].'的电子卡已充电完成', 'color' => '#173177' ],
|
|
|
+ 'keyword1' => [ 'value' => $data['tid'], 'color' => '#173177' ],
|
|
|
+ 'keyword2' => [ 'value' => date('Y-m-d H:i:s'), 'color' => '#173177' ],
|
|
|
+ 'remark' => [ 'value' => '', 'color' => '#173177' ],
|
|
|
+ ],
|
|
|
+ );
|
|
|
+ $res=$this->wxMsg->sendMessage($sendmsg);
|
|
|
+ create_log($device_info['imei'].' charge_full user:'.$send_user['id'].' openid:'.$send_user['wx_open_id'].' res:'.json_encode($res), 'wx_push_result_log');
|
|
|
+ //推送记录
|
|
|
+ $push_res_data = array(
|
|
|
+ 'username'=>$send_user['realname'],
|
|
|
+ 'device_number'=>$device_info['imei'],
|
|
|
+ 'result'=> json_encode($res),
|
|
|
+ 'created_at'=>time(),
|
|
|
+ 'alarm_id'=>$insertId,
|
|
|
+ 'alarm_type'=>$alarm_data['alarm_type'],
|
|
|
+ 'creator_id'=>$device_info['creator_id']
|
|
|
+ );
|
|
|
+ M('wx_push_result_log')->createAdd($push_res_data);
|
|
|
+ }
|
|
|
+
|
|
|
+ }
|
|
|
+ if($data['is_charge'] || $data['is_charge_full']){
|
|
|
+ $recover_cond=array(
|
|
|
+ 'device_number'=>$device_info['imei'],
|
|
|
+ 'alarm_reason'=>'low_bat',
|
|
|
+ 'state'=>'start',
|
|
|
+ );
|
|
|
+ if(M("alarm_records")->where($recover_cond)->count()){
|
|
|
+ $recover_data=array(
|
|
|
+ 'end_time'=>time(),
|
|
|
+ 'state'=>'end',
|
|
|
+ 'result'=>'1',
|
|
|
+ );
|
|
|
+ M('alarm_records')->createSave($recover_cond,$recover_data);
|
|
|
+ if(M('alarm_records')->where(array('device_number'=>$device_info['imei'],'state'=>'start'))->count()){
|
|
|
+ $save_data['alarm_state']=1;
|
|
|
+ }else{
|
|
|
+ $save_data['alarm_state']=0;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /* //充电中 低电量告警恢复
|
|
|
+ $alarm_data['alarm_type']='is_charge';
|
|
|
+ $check_cond['alarm_type']='charge_full';
|
|
|
+ if(!M('alarm_report')->where($check_cond)->count()){
|
|
|
+ $insertId=M('alarm_report')->createAdd($alarm_data);
|
|
|
+
|
|
|
+ $recover_cond=array(
|
|
|
+ 'device_number'=>$device_info['imei'],
|
|
|
+ 'alarm_reason'=>'low_bat',
|
|
|
+ 'state'=>'start',
|
|
|
+ );
|
|
|
+ $recover_data=array(
|
|
|
+ 'end_time'=>time(),
|
|
|
+ 'state'=>'end',
|
|
|
+ 'result'=>'1',
|
|
|
+ );
|
|
|
+ M('alarm_records')->createSave($recover_cond,$recover_data);
|
|
|
+ if(M('alarm_records')->where(array('device_number'=>$device_info['imei'],'state'=>'start'))->count()){
|
|
|
+ $save_data['alarm_state']=1;
|
|
|
+ }else{
|
|
|
+ $save_data['alarm_state']=0;
|
|
|
+ }
|
|
|
+ //充电中 推送低电量恢复模板ID
|
|
|
+ $recover_template_id = C('WX_LOW_RECOVER_ALARM_TEMPLATE_ID');
|
|
|
+ if (!$recover_template_id) {
|
|
|
+ create_log('WX_LOW_RECOVER_ALARM_TEMPLATE_ID required', 'add_alarm');
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+
|
|
|
+ $sendmsg = array(
|
|
|
+ 'touser'=>$send_user['wx_open_id'],
|
|
|
+ 'template_id' => $recover_template_id, //"NeSrB4ViIkRMOuCtoA6BIPPi_N5L-B5zaWcIXJZwkD0",
|
|
|
+ 'data' => [
|
|
|
+ 'first' => [ 'value' =>$send_user['realname'].'的电子卡正在充电', 'color' => '#173177' ],
|
|
|
+ 'keyword1' => [ 'value' => $send_user['realname'], 'color' => '#173177' ],
|
|
|
+ 'keyword2' => [ 'value' => $data['tid'], 'color' => '#173177' ],
|
|
|
+ 'keyword3' => [ 'value' => $data['battery_level'], 'color' => '#173177' ],
|
|
|
+ 'remark' => [ 'value' => '', 'color' => '#173177' ],
|
|
|
+ ],
|
|
|
+ );
|
|
|
+ $res=$this->wxMsg->sendMessage($sendmsg);
|
|
|
+ //推送记录
|
|
|
+ $push_res_data = array(
|
|
|
+ 'username'=>$send_user['realname'],
|
|
|
+ 'device_number'=>$device_info['imei'],
|
|
|
+ 'result'=> json_encode($res),
|
|
|
+ 'created_at'=>time(),
|
|
|
+ 'alarm_id'=>$insertId,
|
|
|
+ 'alarm_type'=>$alarm_data['alarm_type'],
|
|
|
+ 'creator_id'=>$device_info['creator_id']
|
|
|
+ );
|
|
|
+ M('wx_push_result_log')->createAdd($push_res_data);
|
|
|
+ }*/
|
|
|
+ }
|
|
|
+ if($data['is_low'] && !$data['is_charge']){
|
|
|
+ //低电量告警
|
|
|
+ $save_data['alarm_state']=1;
|
|
|
+ //低电量告警
|
|
|
+ $alarm_data['alarm_type']='low_bat';
|
|
|
+ $insertId=M('alarm_report')->createAdd($alarm_data);
|
|
|
+ $cond=array(
|
|
|
+ 'device_number'=>$device_info['imei'],
|
|
|
+ 'alarm_reason'=>'low_bat',
|
|
|
+ 'state'=>'start',
|
|
|
+ );
|
|
|
+ //如果不存在告警 则添加
|
|
|
+ if(!M("alarm_records")->where($cond)->count()){
|
|
|
+ $records_save_data=array(
|
|
|
+ 'device_number'=>$device_info['imei'],
|
|
|
+ 'alarm_reason'=>'low_bat',
|
|
|
+ 'state'=>'start',
|
|
|
+ 'start_time'=>time(),
|
|
|
+ 'creator_id'=>$device_info['creator_id']
|
|
|
+ );
|
|
|
+ M('alarm_records')->createAdd($records_save_data);
|
|
|
+
|
|
|
+ // 低电量模板ID
|
|
|
+ $bat_template_id = C('WX_LOW_BAT_ALARM_TEMPLATE_ID');
|
|
|
+ if (!$bat_template_id) {
|
|
|
+ create_log('WX_LOW_BAT_ALARM_TEMPLATE_ID required', 'add_alarm');
|
|
|
+ return array('success'=>false,'message'=>'WX_CHARGE_FULL_ALARM_TEMPLATE_ID required');
|
|
|
+ }
|
|
|
+
|
|
|
+ $sendmsg = array(
|
|
|
+ 'touser'=>$send_user['wx_open_id'],
|
|
|
+ 'template_id' => $bat_template_id, //"NeSrB4ViIkRMOuCtoA6BIPPi_N5L-B5zaWcIXJZwkD0",
|
|
|
+ 'data' => [
|
|
|
+ 'first' => [ 'value' =>$send_user['realname'].'的电子卡电量过低,请及时充电', 'color' => '#173177' ],
|
|
|
+ 'keyword1' => [ 'value' => $data['tid'], 'color' => '#173177' ],
|
|
|
+ 'keyword2' => [ 'value' => '低电量告警', 'color' => '#173177' ],
|
|
|
+ 'keyword3' => [ 'value' => date('Y-m-d H:i:s'), 'color' => '#173177' ],
|
|
|
+ 'remark' => [ 'value' => '', 'color' => '#173177' ],
|
|
|
+ ],
|
|
|
+ );
|
|
|
+ $res=$this->wxMsg->sendMessage($sendmsg);
|
|
|
+ create_log($device_info['imei'].'is_low user:'.$send_user['id'].' openid:'.$send_user['wx_open_id'].' res:'.json_encode($res), 'wx_push_result_log');
|
|
|
+ //推送记录
|
|
|
+ $push_res_data = array(
|
|
|
+ 'username'=>$send_user['realname'],
|
|
|
+ 'device_number'=>$device_info['imei'],
|
|
|
+ 'result'=> json_encode($res),
|
|
|
+ 'created_at'=>time(),
|
|
|
+ 'alarm_id'=>$insertId,
|
|
|
+ 'alarm_type'=>$alarm_data['alarm_type'],
|
|
|
+ 'creator_id'=>$device_info['creator_id']
|
|
|
+ );
|
|
|
+ M('wx_push_result_log')->createAdd($push_res_data);
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ }
|
|
|
+ $cond=array('imei'=>$data['tid']);
|
|
|
+ //M('users')->createSave($cond,array('battery_level'=> $data['battery_level']));
|
|
|
+ $res=M('devices')->createSave($cond,$save_data);
|
|
|
+ //var_dump(M('devices')->getLastSql());
|
|
|
+ if($res){
|
|
|
+ return array('success'=>true,'message'=>$data['tid'].' update success,battery: '.$data['battery_level']);
|
|
|
+ }else{
|
|
|
+ return array('success'=>false,'message'=>$data['tid'].' update failed');
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+}
|