#include "mqtt_dp.h"
#include "string.h"
#include "../common.h"
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <curl/curl.h>
#include <string.h>
#include <sys/file.h>
#include <unistd.h>
#include "../cc_lc_com/collector-api.h"
#include "../common.h"
#include "../mqtt_emms2/mqtt_emms2.h"

static mqtt_dp_upgrade_t* upgrade = NULL;

int mqtt_dp_publish_message(mqtt_dp_var_t* var, char* pub_topic, char* data, int data_len);

int mqtt_process_login_message(mqtt_dp_var_t* var, char* type, cJSON* root)
{
    if (strcmp(type, FUNCTION_TYPE_POSTRESP) == 0)
    {
        var->status = MQTT_DP_DEVREPORTING;
    }
    return 0;
}

int mqtt_process_heartbeat_message(mqtt_dp_var_t* var, char* type, cJSON* root)
{
    if (strcmp(type, FUNCTION_TYPE_POSTRESP) == 0)
    {
        var->heartbeat.offline_count = 0;
    }
    return 0;
}

int upgrade_mqtt_config_operation(mqtt_dp_var_t* var)
{
    char* file = read_file_data(MQTT_OPERATION_FILE);
    int ret = -1;
    cJSON* root = NULL;
    cJSON* obj  = NULL;
    if(file != NULL)
    {
        ems_syslog(LOG_NOTICE, "read file %s failed", MQTT_OPERATION_FILE);
        root = cJSON_Parse(file);
    }
    if(root == NULL)
    {
        ems_syslog(LOG_NOTICE, "parse file %s failed", MQTT_OPERATION_FILE);
        root = cJSON_CreateObject();
    }

    obj = cJSON_GetObjectItem(root, var->station_id);
    if (obj != NULL)
    {
        cJSON_ReplaceItemInObject(root, var->station_id, cJSON_CreateNumber(1));
    }
    else
    {
        cJSON_AddNumberToObject(root, var->station_id, 1);
    }

    char* payload = cJSON_Print(root);
    if (payload != NULL)
    {
        write_file_data_safe(MQTT_OPERATION_FILE, payload, strlen(payload));
        free(payload);
        ret = 0;
    }

    if (file != NULL) free(file);
    if (root != NULL) cJSON_Delete(root);
    return ret;
}

int notify_set_config(mqtt_dp_set_config_t* config)
{
    if (config == NULL) return -1;

    cJSON* root = cJSON_CreateObject();
    if (root == NULL) return -1;

    cJSON_AddStringToObject(root, "sn", strdup(config->sn));
    cJSON_AddStringToObject(root, "file", strdup(config->file));
    cJSON_AddStringToObject(root, "md5", strdup(config->md5));
    if (strcmp(config->sn, config->var->topic_sn) == 0)
    {
        cJSON_AddStringToObject(root, "ip", strdup("127.0.0.1"));
    }
    else
    {
        char*           ip     = NULL;
        pcs_ctrl_var_t* var    = get_pcs_ctrl_var();
        lc_it*          list   = var->lc_list;
        int             lc_cnt = var->lc_list_len;
        for (int j = 0; j < lc_cnt; j++)
        {
            if (strcmp(config->sn, list[j].sn) == 0)
            {
                ip = strdup(list[j].ipaddr);
                break;
            }
        }
        if (ip)
        {
            cJSON_AddStringToObject(root, "ip", ip);
        }
        else
        {
            cJSON_Delete(root);
            return -1;
        }
    }
    char* payload = cJSON_Print(root);
    if (payload != NULL)
    {
        write_file_data_safe(MQTT_DP_CONFIG_SET_INFO, payload, strlen(payload));
        free(payload);
    }
    if (root) cJSON_Delete(root);

    return 0;
}

int mqtt_publish_simple_result(mqtt_dp_var_t* var, char* type, char* func_id, int seq, int result, char* info_key, cJSON* info)
{
    int   ret     = -1;
    char* topic   = malloc(256);
    char* payload = malloc(2048);
    if (topic && payload)
    {
        snprintf(topic, 256, "emms2/%s/%s/%s", type, var->topic_sn, func_id);
        int len = snprintf(payload,
                           1024,
                           "{\"funcId\":\"%s\",\"lcSN\": \"%s\",\"seq\": %d,\"time\":%lu,\"result\": %d",
                           func_id,
                           var->topic_sn,
                           seq,
                           time(NULL),
                           result);
        if (info_key && info)
        {
            char* i = cJSON_Print(info);
            if (i)
            {
                len += snprintf(payload + len, 1024 - len, ",\"%s\":%s", info_key, i);
                free(i);
            }
        }
        len += snprintf(payload + len, 1024 - len, "}");

        ret = mqtt_dp_publish_message(var, topic, payload, len);
    }
    if (topic) free(topic);
    if (payload) free(payload);
    return ret;
}

void* mqtt_dp_set_config_task(void* arg)
{
    mqtt_dp_set_config_t* config = (mqtt_dp_set_config_t*)arg;
    if (config == NULL) return NULL;
    int ret = -1;
    if (http_post_download_file(config->url, MQTT_DP_CONFIG_SET_DIR, config->file, NULL) == 0)
    {
        if (notify_set_config(config) != 0)
            ems_syslog(LOG_ERR, "notify_set_config failed");
        else
            ret = 0;
    }
    else
    {
        ems_syslog(LOG_ERR, "http_post_download_file config failed");
    }

    if (mqtt_publish_simple_result(config->var, FUNCTION_TYPE_SETRESP, FUNCTION_ID_CONFIG, config->seq, ret, NULL, NULL) < 0)
    {
        ems_syslog(LOG_ERR, "mqtt_publish_message failed, Set Config answer failed");
    }

    return NULL;
}

int mqtt_dp_set_config_task_start(mqtt_dp_set_config_t* arg)
{
    pthread_create(&arg->tid, NULL, mqtt_dp_set_config_task, arg);
    pthread_detach(arg->tid);
    return 0;
}

int mqtt_dp_get_ems_config(mqtt_dp_var_t* var, char* sn)
{
    if (var == NULL || sn == NULL || sn[0] == '\0')
        return -1;

    char md5path[256]  = {0};
    char md5[33]       = {0};
    char filepath[256] = {0};
    char filename[128] = {0};
    snprintf(filename, sizeof(filename), "%s.tar.gz", sn);
    snprintf(filepath, sizeof(filepath), "%s/%s", MQTT_DP_CONFIG_TAR_DIR, filename);
    if (access(filepath, F_OK) != 0)
    {
        ems_syslog(LOG_ERR, "file %s not exist", filepath);
        return -1;
    }
    get_file_md5(filepath, md5);

    snprintf(md5path, sizeof(md5path), "%s%s", MQTT_DP_CONFIG_MD5_FILE, var->station_id);

    char*  his      = read_file_data(md5path);
    cJSON* md5_root = NULL;

    if (his != NULL)
    {
        md5_root = cJSON_Parse(his);
        if (md5_root == NULL)
        {
            md5_root = cJSON_CreateObject();
        }
    }
    else
    {
        md5_root = cJSON_CreateObject();
    }

    cJSON* item = cJSON_GetObjectItem(md5_root, filename);

    if (http_post_upload_file(var->base_param.up_url, filename, filepath, NULL, NULL) == 0)
    {
        if (item == NULL)
            cJSON_AddItemToObject(md5_root, filename, cJSON_CreateString(md5));
        else
            cJSON_ReplaceItemInObject(md5_root, filename, cJSON_CreateString(md5));
    }
    else
    {
        ems_syslog(LOG_ERR, "upload config file %s failed, md5: %s", filepath, md5);
    }
    char* md5str = cJSON_Print(md5_root);
    write_file_data(md5path, md5str, strlen(md5str));
    free(md5str);

    if (var->base_param.operation == 0)
    {
        // 开始周期检测
        var->config.operation     = 1;
        var->base_param.operation = 1;
        upgrade_mqtt_config_operation(var);
    }

    return 0;
}

int mqtt_process_config_message(mqtt_dp_var_t* var, char* type, cJSON* root)
{
    if (strcmp(type, FUNCTION_TYPE_SET) == 0)
    {
        //  todo 下载配置包
        cJSON* info = cJSON_GetObjectItem(root, "info");
        if (info == NULL)
        {
            return -1;
        }
        cJSON* url  = cJSON_GetObjectItem(info, "url");
        cJSON* md5  = cJSON_GetObjectItem(info, "md5");
        cJSON* file = cJSON_GetObjectItem(info, "file");
        cJSON* sn   = cJSON_GetObjectItem(info, "sn");
        cJSON* seq  = cJSON_GetObjectItem(root, "seq");

        if (url == NULL || url->valuestring == NULL || url->valuestring[0] == '\0' ||
            md5 == NULL || md5->valuestring == NULL || md5->valuestring[0] == '\0' ||
            file == NULL || file->valuestring == NULL || file->valuestring[0] == '\0' ||
            sn == NULL || sn->valuestring == NULL || sn->valuestring[0] == '\0' ||
            seq == NULL || seq->valueint == 0)
        {
            return -1;
        }
        mqtt_dp_set_config_t* config = malloc(sizeof(mqtt_dp_set_config_t));
        if (config == NULL)
        {
            return -1;
        }
        memset(config, 0, sizeof(mqtt_dp_set_config_t));
        config->var = var;
        config->seq = seq->valueint;
        strcpy(config->url, url->valuestring);
        strcpy(config->md5, md5->valuestring);
        strcpy(config->file, file->valuestring);
        strcpy(config->sn, sn->valuestring);

        mqtt_dp_set_config_task_start(config);
    }
    if (strcmp(type, FUNCTION_TYPE_GET) == 0)
    {
        // 上报配置文件包
        int    ret  = -1;
        cJSON* info = cJSON_GetObjectItem(root, "info");
        cJSON* seq  = cJSON_GetObjectItem(root, "seq");
        if (info != NULL && seq != NULL && seq->valueint != 0)
        {
            cJSON* sn = cJSON_GetObjectItem(info, "sn");
            if (sn != NULL && sn->valuestring != NULL && sn->valuestring[0] != '\0')
            {
                ret = mqtt_dp_get_ems_config(var, sn->valuestring);
            }
            mqtt_publish_simple_result(var, FUNCTION_TYPE_GETRESP, FUNCTION_ID_CONFIG, seq->valueint, ret, NULL, NULL);
        }

    }
    return 0;
}

int mqtt_process_monitor_message(mqtt_dp_var_t* var, char* type, cJSON* root)
{
    return 0;
}
int mqtt_process_lcinfo_message(mqtt_dp_var_t* var, char* type, cJSON* root)
{
    return 0;
}

static int mqtt_dp_download_lc_callback(void* clientp, int dltotal, int dlnow, double ultotal, double ulnow)
{
    static time_t lc_last_time = 0;
    time_t        now          = time(NULL);

    if (now - lc_last_time >= 3 || (dlnow == dltotal && dltotal != 0))
    {
        char* payload = malloc(1024);
        char* topic   = malloc(256);
        if (payload == NULL || topic == NULL)
        {
            ems_syslog(LOG_ERR, "malloc failed");
            return -1;
        }
        mqtt_dp_lc_t* lc = upgrade->lc_info[upgrade->do_index];
        if (lc == NULL)
        {
            ems_syslog(LOG_ERR, "lc is null");
            return -1;
        }
        snprintf(topic, 256, "emms2/" FUNCTION_TYPE_SETRESP "/%s/%s", upgrade->var->topic_sn, FUNCTION_ID_PROGRESS);
        for (int i = 0; i < lc->num; i++)
        {
            int p = (ulnow / ultotal * 100.0);
            int len = snprintf(payload,
                               1024,
                               "{\"funcId\":\"Progress\",\"lcSN\": \"%s\",\"seq\": %d,\"time\":%lu, \"info\":{\"type\":\"upgrade\" ,\"sn\": \"%s\",\"code\":1,\"progress\":%d}}",
                               upgrade->var->topic_sn,
                               upgrade->seq,
                               time(NULL),
                               lc->sn[i],
                               p);
            if (mqtt_dp_publish_message(upgrade->var, topic, payload, len) < 0)
            {
                ems_syslog(LOG_ERR, "mqtt publish message failed");
            }
        }
        free(topic);
        free(payload);
        lc_last_time = now;
    }
    return 0;
}

static int mqtt_dp_download_ems_callback(void* clientp, int dltotal, int dlnow, double ultotal, double ulnow)
{
    static time_t ems_last_time = 0;
    time_t        now           = time(NULL);

    if (now - ems_last_time >= 3 || (dlnow == dltotal && dlnow != 0))
    {
        int p = (ulnow / ultotal * 100.0);
        cJSON* info = cJSON_CreateObject();
        cJSON_AddItemToObject(info, "type", cJSON_CreateString("upgrade"));
        cJSON_AddItemToObject(info, "sn", cJSON_CreateString(upgrade->var->topic_sn));
        cJSON_AddNumberToObject(info, "code", 1);
        cJSON_AddNumberToObject(info, "progress", p);
        if (mqtt_publish_simple_result(upgrade->var, FUNCTION_TYPE_SETRESP, FUNCTION_ID_PROGRESS, upgrade->seq, 0, "info", info) < 0)
        {
            ems_syslog(LOG_ERR, "mqtt publish message failed");
        }
        ems_last_time = now;
        cJSON_Delete(info);
    }
    return 0;
}

static int mqtt_db_publish_upgress_start(mqtt_dp_upgrade_t* upgrade, int index)
{
    char* payload = malloc(2048);
    char* topic   = malloc(256);
    if (payload != NULL && topic != NULL)
    {
        snprintf(topic, 256, "emms2/" FUNCTION_TYPE_SETRESP "/%s/%s", upgrade->var->topic_sn, FUNCTION_ID_PROGRESS);
        if (index == 0 && upgrade->ems == 1)
        {
            int len = snprintf(payload,
                               1024,
                               "{\"funcId\":\"%s\",\"lcSN\": \"%s\", \"seq\": %d,\"time\":%lu, \"info\":{\"type\":\"upgrade\" ,\"sn\": \"%s\",\"code\":2,\"progress\":0}}",
                               FUNCTION_ID_PROGRESS,
                               upgrade->var->topic_sn,
                               upgrade->seq,
                               time(NULL),
                               upgrade->var->topic_sn);

            if (mqtt_dp_publish_message(upgrade->var, topic, payload, len) < 0)
            {
                ems_syslog(LOG_ERR, "mqtt publish message failed");
            }
        }
        else if (index > 0 && index - 1 == upgrade->do_index)
        {
            mqtt_dp_lc_t* lc = upgrade->lc_info[index - 1];
            for (int i = 0; i < lc->num; i++)
            {
                if (payload != NULL && topic != NULL)
                {
                    int len = snprintf(payload,
                                       1024,
                                       "{\"funcId\":\"Progress\",\"lcSN\": \"%s\",\"seq\": %d,\"time\":%lu, \"info\":{\"type\":\"upgrade\" ,\"sn\": \"%s\",\"code\":2,\"progress\":0}}",
                                       upgrade->var->topic_sn,
                                       upgrade->seq,
                                       time(NULL),
                                       lc->sn[i]);

                    if (mqtt_dp_publish_message(upgrade->var, topic, payload, len) < 0)
                    {
                        ems_syslog(LOG_ERR, "mqtt publish message failed");
                    }
                }
            }
        }
    }
    free(topic);
    free(payload);
    return 0;
}

static int mqtt_db_publish_upgress_failure(mqtt_dp_upgrade_t* upgrade, int index, int code)
{
    char* payload = malloc(2048);
    char* topic   = malloc(256);
    if (payload != NULL && topic != NULL)
    {
        snprintf(topic, 256, "emms2/" FUNCTION_TYPE_SETRESP "/%s/%s", upgrade->var->topic_sn, FUNCTION_ID_PROGRESS);
        if (index == 0 && upgrade->ems == 1)
        {
            int len = snprintf(payload,
                               1024,
                               "{\"funcId\":\"%s\",\"lcSN\": \"%s\", \"seq\": %d,\"time\":%lu, \"info\":{\"type\":\"upgrade\" ,\"sn\": \"%s\",\"code\":3,\"progress\":%d}}",
                               FUNCTION_ID_PROGRESS,
                               upgrade->var->topic_sn,
                               upgrade->seq,
                               time(NULL),
                               upgrade->var->topic_sn,
                               code);

            if (mqtt_dp_publish_message(upgrade->var, topic, payload, len) < 0)
            {
                ems_syslog(LOG_ERR, "mqtt publish message failed");
            }
        }
        else if (index > 0 && index - 1 == upgrade->do_index)
        {
            mqtt_dp_lc_t* lc = upgrade->lc_info[index - 1];
            for (int i = 0; i < lc->num; i++)
            {
                if (payload != NULL && topic != NULL)
                {
                    int len = snprintf(payload,
                                       1024,
                                       "{\"funcId\":\"Progress\",\"lcSN\": \"%s\",\"seq\": %d,\"time\":%lu, \"info\":{\"type\":\"upgrade\" ,\"sn\": \"%s\",\"code\":3,\"progress\":%d}}",
                                       upgrade->var->topic_sn,
                                       upgrade->seq,
                                       time(NULL),
                                       lc->sn[i],
                                       code);

                    if (mqtt_dp_publish_message(upgrade->var, topic, payload, len) < 0)
                    {
                        ems_syslog(LOG_ERR, "mqtt publish message failed");
                    }
                }
            }
        }
    }
    free(topic);
    free(payload);
    return 0;
}



static int notify_ems_upgrade_monitor(mqtt_dp_upgrade_t* upgrade, int index)
{
    cJSON* root = NULL;
    int    ret  = -1;

    if (index == 0 && upgrade->ems == 1)
    {
        root = cJSON_CreateObject();
        cJSON_AddStringToObject(root, "file", upgrade->file);
        cJSON_AddStringToObject(root, "md5", upgrade->md5);
        cJSON* sn = cJSON_AddArrayToObject(root, "sn");
        cJSON* ip = cJSON_AddArrayToObject(root, "ip");
        cJSON_AddItemToArray(sn, cJSON_CreateString(upgrade->var->topic_sn));
        cJSON_AddItemToArray(ip, cJSON_CreateString("127.0.0.1"));
    }
    else if (index > 0 && index - 1 == upgrade->do_index)
    {
        // 1 从机
        int i = index - 1;
        if (upgrade->lc_info[i] == NULL) goto END;

        root             = cJSON_CreateObject();
        mqtt_dp_lc_t* lc = upgrade->lc_info[i];
        cJSON_AddStringToObject(root, "file", lc->file);
        cJSON_AddStringToObject(root, "md5", lc->md5);
        cJSON* sn = cJSON_AddArrayToObject(root, "sn");
        cJSON* ip = cJSON_AddArrayToObject(root, "ip");
        for (int i = 0; i < lc->num; i++)
        {
            cJSON_AddItemToArray(sn, cJSON_CreateString(lc->sn[i]));
            cJSON_AddItemToArray(ip, cJSON_CreateString(lc->ip[i]));
        }
    }
    sleep(1);
    mqtt_db_publish_upgress_start(upgrade, index);
    sleep(1);
    char* str = cJSON_PrintUnformatted(root);
    if (str)
    {
        int fd  = open(UPGRADE_INFO_FILE, O_WRONLY | O_APPEND | O_CREAT, 0666);
        if (fd < 0)
        {
            char tmp[256];
            snprintf(tmp, sizeof(tmp), "%s", UPGRADE_INFO_FILE);
            for (char* p = strchr(tmp + 1, '/'); p; p = strchr(p + 1, '/'))
            {
                *p = '\0';
                mkdir(tmp, 0777);
                *p = '/';
            }
            fd = open(UPGRADE_INFO_FILE, O_WRONLY | O_CREAT, 0666);
        }
        if (fd > 0)
        {
            flock(fd, LOCK_EX); // 获取排他锁
            ret = write(fd, str, strlen(str)) > 0 ? 0 : -1;
            if(ret == 0)
            {
                write(fd, "\n", 1);
            }
            flock(fd, LOCK_UN); // 释放锁
            close(fd);
        }
        
        free(str);
    }

END:
    if (root) cJSON_Delete(root);
    return ret;
}

void destory_upgrade_struct(mqtt_dp_upgrade_t* upgrade)
{
    if (upgrade == NULL) return;
    for (int i = 0; i < upgrade->lc_num; i++)
    {
        if (upgrade->lc_info[i] != NULL)
        {
            for (int j = 0; j < upgrade->lc_info[i]->num; j++)
            {
                free(upgrade->lc_info[i]->sn[j]);
                free(upgrade->lc_info[i]->ip[j]);
            }
            free(upgrade->lc_info[i]);
        }
    }
    free(upgrade);
}

void* mqtt_db_upgrade_ems_task(void* arg)
{
    if (arg == NULL || upgrade == NULL || arg != upgrade) return NULL;
    for (int i = 0; i < upgrade->lc_num; i++)
    {
        if (upgrade->lc_info[i] == NULL || upgrade->lc_info[i]->sn == NULL || upgrade->lc_info[i]->ip == NULL) continue;
        upgrade->do_index = i;
        int ret           = http_post_download_file(upgrade->lc_info[i]->url, UPGRADE_DOWNLOAD_DIR, upgrade->lc_info[i]->file, mqtt_dp_download_lc_callback);
        if (ret == 0)
        {
            if (notify_ems_upgrade_monitor(upgrade, i + 1) < 0)
            {
                ems_syslog(LOG_ERR, "notify_ems_upgrade_monitor failed");
            }
        }
        else
        {
            mqtt_db_publish_upgress_failure(upgrade, i + 1, ret);
            ems_syslog(LOG_ERR, "http_post_download_file failed, %s", upgrade->lc_info[i]->url);
        }
    }
    if (upgrade->ems == 1)
    {
        int ret = http_post_download_file(upgrade->url, UPGRADE_DOWNLOAD_DIR, upgrade->file, mqtt_dp_download_ems_callback);
        sleep(1);
        if (ret == 0)
        {
            if (notify_ems_upgrade_monitor(upgrade, 0) < 0)
            {
                ems_syslog(LOG_ERR, "notify_ems_upgrade_monitor failed");
            }
        }
        else
        {
            mqtt_db_publish_upgress_failure(upgrade, 0, ret);
            ems_syslog(LOG_ERR, "http_post_download_file failed, %s", upgrade->url);
        }
    }
    destory_upgrade_struct(upgrade);
    upgrade = NULL;
    return NULL;
}

int mqtt_process_progress_message(mqtt_dp_var_t* var, char* type, cJSON* root)
{
    return 0;
}

// static int get_lc_list_ip(mqtt_dp_lc_t* lc)
// {
//     pcs_ctrl_var_t* var    = get_pcs_ctrl_var();
//     lc_it*          list   = var->lc_list;
//     int             lc_cnt = var->lc_list_len;
//     for (int i = 0; i < lc->num; i++)
//     {
//         memset(lc->ip[i], 0, sizeof(lc->ip[i]));
//         for (int j = 0; j < lc_cnt; j++)
//         {
//             if (strcmp(lc->sn[i], list[j].sn) == 0)
//             {
//                 lc->ip[i] = strdup(list[j].ipaddr);
//                 break;
//             }
//         }
//     }
//     return 0;
// }

int mqtt_process_upgrade_message(mqtt_dp_var_t* var, char* type, cJSON* root)
{
    int ret = -1;
    srand(time(0));
    if (strcmp(type, FUNCTION_TYPE_SET) == 0)
    {
        cJSON* seq_json = cJSON_GetObjectItem(root, "seq");
        cJSON* info     = cJSON_GetObjectItem(root, "info");
        int    size     = cJSON_GetArraySize(info);
        if (seq_json == NULL || info == NULL || size <= 0)
        {
            ems_syslog(LOG_ERR, "mqtt_process_upgrade_message error, key info is Lacked");
            return ret;
        }

        upgrade = malloc(sizeof(mqtt_dp_upgrade_t));
        if (upgrade == NULL)
        {
            return ret;
        }
        memset(upgrade, 0, sizeof(mqtt_dp_upgrade_t));
        upgrade->var     = var;
        upgrade->seq     = seq_json->valueint;
        upgrade->lc_info = malloc(sizeof(mqtt_dp_lc_t*) * size);

        for (int i = 0; i < size; i++)
        {
            cJSON* item = cJSON_GetArrayItem(info, i);
            if (item == NULL) continue;
            cJSON* lc      = cJSON_GetObjectItem(item, "lc");
            cJSON* sn_list = cJSON_GetObjectItem(item, "sn");
            cJSON* url     = cJSON_GetObjectItem(item, "url");
            cJSON* file    = cJSON_GetObjectItem(item, "file");
            cJSON* md5     = cJSON_GetObjectItem(item, "md5");
            if (lc == NULL || lc->valueint <= 0 ||
                sn_list == NULL || cJSON_GetArraySize(sn_list) <= 0 ||
                url == NULL || url->valuestring == NULL || url->valuestring[0] == '\0' ||
                md5 == NULL || md5->valuestring == NULL || md5->valuestring[0] == '\0' ||
                file == NULL || file->valuestring == NULL || file->valuestring[0] == '\0') continue;

            if (lc->valueint == 1)
            {
                cJSON* sn = cJSON_GetArrayItem(sn_list, 0);
                if (sn == NULL) continue;
                if (strcmp(sn->valuestring, var->topic_sn) != 0) continue;
                if (upgrade->ems == 0)
                    upgrade->ems++;
                else
                {
                    ems_syslog(LOG_ERR, "ems upgrade error, ems is already upgrade");
                    continue;
                }
                local_strlcpy(upgrade->url, url->valuestring, sizeof(upgrade->url));
                local_strlcpy(upgrade->md5, md5->valuestring, sizeof(upgrade->md5));
                char file_name[256] = {0};
                snprintf(file_name,sizeof(file_name), "%d_%s", rand(), file->valuestring);
                local_strlcpy(upgrade->file, file_name, sizeof(upgrade->file));
            }
            else if (lc->valueint == 2)
            {
                mqtt_dp_lc_t* lc = malloc(sizeof(mqtt_dp_lc_t));
                if (lc == NULL) continue;

                memset(lc, 0, sizeof(mqtt_dp_lc_t));
                local_strlcpy(lc->url, url->valuestring, sizeof(lc->url));
                local_strlcpy(lc->md5, md5->valuestring, sizeof(lc->md5));
                char file_name[256] = {0};
                snprintf(file_name,sizeof(file_name), "%d_%s", rand(), file->valuestring);
                local_strlcpy(lc->file,file_name, sizeof(lc->file));

                int sn_size = cJSON_GetArraySize(sn_list);
                lc->sn = malloc(sizeof(char*) * sn_size);
                lc->ip = malloc(sizeof(char*) * sn_size);
                memset(lc->sn, 0, sizeof(char*) * sn_size);
                memset(lc->ip, 0, sizeof(char*) * sn_size);

                pcs_ctrl_var_t* var    = get_pcs_ctrl_var();
                lc_it*          list   = var->lc_list;
                int             lc_cnt = var->lc_list_len;

                for (int j = 0; j < sn_size; j++)
                {
                    cJSON* sn = cJSON_GetArrayItem(sn_list, j);
                    if (sn == NULL || sn->valuestring == NULL || sn->valuestring[0] == '\0') continue;
                    int index = lc->num++;
                    lc->sn[index] = strdup(sn->valuestring);
                    for( int k = 0; k < lc_cnt; k++)
                    {
                        if (strcmp(lc->sn[index], list[k].sn) == 0)
                        {
                            lc->ip[index] = strdup(list[k].ipaddr);
                            break;
                        }
                    }
                }
                upgrade->lc_info[upgrade->lc_num++] = lc;
            }
        }

        if (upgrade->ems > 0 || upgrade->lc_num > 0)
        {
            ret = 0;
            pthread_create(&upgrade->tid, NULL, mqtt_db_upgrade_ems_task, upgrade);
            pthread_detach(upgrade->tid);
        }
    }

    return 0;
}