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("power_oa")
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':
                if table in MODELS_MAP.keys():
                    tp = value.get("type")
                    ts = value.get("ts")
                    data = value.get("data")
                    old = value.get("old")
                    json_to_sql(MODELS_MAP[table], tp, data, old, 'id', ts)

            customer.commit()
    except Exception as e:
        logger.error(e)
    finally:
        customer.close()


run_customer()
