# coding:utf-8
import time
from multiprocessing import Process

from kafka import TopicPartition

from settings import settings
from v2 import *
from v2.struct_cache import StructCacheFile, StructCacheRedis


class XProcess(Process):

    def __init__(self, partition, lock, ipsearch, log, rdb=None, event_attr=None):
        super(XProcess, self).__init__()
        # self.daemon = True
        self.partition = partition
        self.lock = lock
        self.ipsearch = ipsearch
        self.rdb = rdb
        self.log = log
        self.event_attr = event_attr

    def run(self):
        db_client = CK(**settings.CK_CONFIG)
        sketch = Sketch(db_client, StructCacheRedis(self.rdb))
        handler_event = HandlerEvent(db_client, settings.GAME, self.ipsearch)
        handler_user = HandlerUser(db_client, settings.GAME)
        transmitter = Transmitter(db_client, settings.GAME, sketch, self.log, self.lock, self.event_attr,
                                  self.partition)
        transmitter.start_ping()
        transmitter.add_source(handler_event, 5000, 60)
        transmitter.add_source(handler_user, 500, 60)

        last_ts = int(time.time())
        consumer, kafka_client = create_consumer(self.partition)

        for msg in consumer():
            data = msg.value
            type_ = data['#type']
            del data['#type']
            ts = int(time.time())
            try:
                data['properties']['unique_id'] = f'{msg.topic}-{msg.partition}-{msg.offset}'
            except:
                pass

            if msg.topic == 'debug':
                self.log.info(data)

            if 'user' in type_:
                # continue
                obj = getattr(handler_user, type_)
                handler_user.receive_data.append(UserAct(obj, data))
                if len(handler_user.receive_data) >= 1000 or last_ts + 60 < ts:
                    last_ts = ts
                    handler_user.execute()

            elif 'track' in type_:
                # continue
                if data['#event_name'] == 'pay':
                    self.log.info(f'topid->{msg.topic} | partition->{msg.partition} | offset->{msg.offset} | data-> {data}')

                obj = getattr(handler_event, type_)
                obj(data)
            elif type_ == settings.STOP_SIGNAL:
                # continue
                # 1 小时内有效
                self.log.info(type_)
                if data.get('#time', 0) + 3600 < int(time.time()):
                    continue
                # 停止消费kafka
                self.log.info(f'进程{self.partition} 等待90秒')
                time.sleep(90)
                self.log.info(f'进程{self.partition} 写入数据')
                transmitter.run(kafka_client)
                self.log.info(f'进程{self.partition} 结束')
                kafka_client.commit()
                kafka_client.close()

                break
            elif type_ == 'test':
                self.log.info(f'topid->{msg.topic} | partition->{msg.partition} | offset->{msg.offset} | data-> {data}')
            else:
                continue

            transmitter.run(kafka_client)

        while True:
            time.sleep(5)
            self.log.info(f'消费分区{self.partition} 已结束。。。')