import sys
import os
import json
etl_base = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(etl_base)
from etl.utils import getLogger
from etl.producer.kafka_client import kafka_client
from etl.db.db_client import json_to_sql
from etl.original_database.model_map import MODELS_MAP

logger = getLogger(__name__)
customer = kafka_client.create_customer("im_real")
logger.error(customer.topics())


def run_customer():
    try:
        for msg in customer:
            value = json.loads(msg.value)
            logger.debug(value)
            table = value.get("table")
            if table == 'im_real':
                tp = value.get("type")
                ts = value.get("ts")
                data = value.get("data")
                json_to_sql(MODELS_MAP[table], tp, data, 'id', ts)
            customer.commit()
    except Exception as e:
        logger.error(e)
    finally:
        customer.close()


run_customer()
