#!/usr/bin/env python
import os
import json
import signal
import asyncio
import argparse
import aiomysql
from datetime import datetime
from nats.aio.client import Client as NATS
from stan.aio.client import Client as STAN
from kazoo.client import KazooClient
from config import *
from ticker import Ticker


""" Consume log channel from nats message queue. """


FLAG_QUIT = False
ONE_DAY = 60 * 60 * 24
TIME_FORMAT = '%Y-%m-%d %H:%M:%S'


async def on_disconnected():
    print('connection lost')


async def on_closed():
    print('connection closed')


async def on_streaming_conn_lost():
    print('lost connection')


async def on_reconnected():
    print('re connected')


def on_signal(signum, frame):
    global FLAG_QUIT
    print('received signal, num=', signum)
    FLAG_QUIT = True


def retrieve_streaming_conf():
    with open(js_conf) as h:
        data = h.read()
        c = json.loads(data)

    zk = KazooClient(hosts=','.join(c['zoo']['hosts']))
    zk.start()
    zk.add_auth('digest', '{}:{}'.format(c['zoo']['username'], c['zoo']['password']))
    data, _ = zk.get('/kg/conf/servers/mq')
    sc = json.loads(data)
    zk.stop()
    zk.close()
    return sc


async def main(lp):
    """
    entry of process

    :param lp: loop
    :return: None
    """
    parser = argparse.ArgumentParser(description='timer task for checking if user active')
    parser.add_argument('-c', '--clientId', action='store', dest='clientId', default=client_id, type=str, metavar='clientId', help='set  client id of consumer')
    parser.add_argument('-d', '--db', action='store', dest='db', default=log_db_name, type=str, metavar='db', help='db name')
    parser.add_argument('-t', '--table', action='store', dest='table', default=log_table_name, type=str, metavar='table', help='set table where to store log')
    args = parser.parse_args()

    # retrieve streaming config
    stream_conf = retrieve_streaming_conf()

    nc = NATS()
    await nc.connect(
        name=args.clientId,
        verbose=True,
        servers=stream_conf['addrs'],
        io_loop=lp,
        user=stream_conf['username'],
        password=stream_conf['password'],
        disconnected_cb=on_disconnected,
        reconnected_cb=on_reconnected,
        closed_cb=on_closed,
        no_echo=False,
    )

    sc = STAN()
    await sc.connect(
        cluster_id=stream_conf['cluster_id'],
        client_id=args.clientId,
        nats=nc,
        conn_lost_cb=on_streaming_conn_lost,
    )

    mysql = await aiomysql.create_pool(host=my_host, port=my_port, user=my_user, password=my_password, db=args.db, loop=lp)
    def_guid = 168000
    def_level = 'debug'

    async def exec_sql(sql):
        conn = await mysql.acquire()
        cur = await conn.cursor()
        await cur.execute(sql)
        await conn.commit()
        await cur.close()
        mysql.release(conn)

    async def on_tick(n):
        now = datetime.now().timestamp()
        _30_days_before = now - 30 * ONE_DAY
        sql = "DELETE FROM {} WHERE `when`<'{}'".format(args.table, datetime.fromtimestamp(_30_days_before).strftime(TIME_FORMAT))
        print(sql)
        await exec_sql(sql)

    async def on_message(msg):
        try:
            when = datetime.fromtimestamp(int(msg.timestamp / 1000000000)).strftime(TIME_FORMAT)
            # parse
            data = json.loads(msg.data)
            guid = data['guid'] if 'guid' in data and type(data['guid']) is int else def_guid
            level = data['level'] if 'level' in data and type(data['level']) is str else def_level
            sql = "INSERT INTO {}(`guid`,`from`,`when`,`seq`,`level`,`content`)VALUES({},'{}','{}',{},'{}','{}')".format(args.table, guid, msg.sub.subject, when, msg.seq, level, msg.data.decode('utf-8'))
            print(msg.seq, msg.data.decode('utf-8'))
            await exec_sql(sql)
        except Exception as e:
            print('exception:', e)

    subs = [await sc.subscribe(subject='{}.{}'.format(subject, lobby), cb=on_message, queue=queue_name, durable_name=args.clientId, start_at='last_received') for lobby in lobbies]

    signal.signal(signal.SIGINT, on_signal)
    print('press CTRL + C to quit. pid=', os.getpid())

    tk = Ticker(ONE_DAY, on_tick)
    print('setup ticker, period=', ONE_DAY, ' seconds')

    async def wait_infinite():
        global FLAG_QUIT
        while not FLAG_QUIT:
            await asyncio.sleep(5)

    await wait_infinite()
    tk.stop()
    [await sub.unsubscribe() for sub in subs]
    await sc.close()
    await nc.close()
    await mysql.wait_closed()
    print('working stopped.')


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop))
    loop.close()
