#include "lc.h"
#include "../tag.h"
#include "../common.h"
#include "../ems/main.h"
#include "../opcua/ua_client.h"
#include "../ems/slave_ems.h"
#include "define.h"


char **get_dev_id_and_num(slave_ems *p, int *num) {
    if (!p || !p->node_sort) return NULL;

    cJSON *root = cJSON_Parse(p->node_sort);
    if (!root || !cJSON_IsArray(root)) {
        cJSON_Delete(root);
        *num = 0;
        return NULL;
    }
    int dev_count = cJSON_GetArraySize(root);
    *num = dev_count;
    p->devs = (device_t *)calloc(dev_count, sizeof(device_t));

    char **dev_ids = (char **)calloc(dev_count, sizeof(char*));
    for (int i = 0; i < dev_count; ++i) {
        cJSON *item = cJSON_GetArrayItem(root, i);
        if (!item) continue;
        cJSON *type = cJSON_GetObjectItem(item, "TYPE");
        cJSON *dev_no = cJSON_GetObjectItem(item, "DEV_NO");
        cJSON *disp = cJSON_GetObjectItem(item, "DISP");
        if (type && cJSON_IsString(type) && type->valuestring) {
            p->devs[i].dev_type = strdup(type->valuestring);
        } else {
            p->devs[i].dev_type = NULL;
        }
        if (dev_no && cJSON_IsString(dev_no) && dev_no->valuestring) {
          size_t len = snprintf(p->devs[i].no, DEV_NO_LEN, "%s.", p->no);
          if (dev_no && cJSON_IsString(dev_no) && dev_no->valuestring) {
            snprintf(p->devs[i].no + len, DEV_NO_LEN - len, "%s",
                     dev_no->valuestring);
            dev_ids[i] = strdup(dev_no->valuestring);
          } else {
            p->devs[i].no[0] = 0;
          }
        } else {
            p->devs[i].no[0] = 0;
        }
        if (disp && cJSON_IsString(disp) && disp->valuestring) {
            snprintf(p->devs[i].disp_name, DEV_NAME_LEN, "%s.%s",p->name, disp->valuestring);
        } else {
            p->devs[i].disp_name[0] = 0;
        }
    }
    cJSON_Delete(root);
    p->dev_sum = dev_count;
    return dev_ids;
}

char *get_dev_child_type_by_dev_info(slave_ems *p, int *num) {
    proto_syslog(LOG_ERR, "[get_dev_child_type_by_dev_info] Start - url: %s", p ? p->url : "NULL");

    if (!p || !p->dev_tree || !p->devs) {
        proto_syslog(LOG_ERR, "[get_dev_child_type_by_dev_info] Invalid parameters");
        return NULL;
    }

    cJSON *root = cJSON_Parse(p->dev_tree);
    if (!root) {
        proto_syslog(LOG_ERR, "[get_dev_child_type_by_dev_info] Parse dev_tree failed");
        return NULL;
    }

    cJSON *dev_tree = cJSON_GetObjectItem(root, "dev_tree");
    if (!dev_tree || !cJSON_IsArray(dev_tree)) {
        proto_syslog(LOG_ERR, "[get_dev_child_type_by_dev_info] Get dev_tree array failed");
        cJSON_Delete(root);
        return NULL;
    }

    int dev_count = p->dev_sum;
    proto_syslog(LOG_ERR, "[get_dev_child_type_by_dev_info] Processing %d devices", dev_count);

    for (int i = 0; i < dev_count; i++) {
        device_t *dev = &p->devs[i];
        dev->dev_child_type = NULL;
        int found = 0;

        for (int j = 0; j < cJSON_GetArraySize(dev_tree); j++) {
            cJSON *lc_item = cJSON_GetArrayItem(dev_tree, j);
            if (!lc_item) continue;

            cJSON *inner_dev_tree = cJSON_GetObjectItem(lc_item, "dev_tree");
            cJSON *devices_array = NULL;

            if (inner_dev_tree && cJSON_IsArray(inner_dev_tree)) {
                devices_array = inner_dev_tree;
            } else {
                devices_array = dev_tree;
                j = cJSON_GetArraySize(dev_tree);
            }

            for (int k = 0; k < cJSON_GetArraySize(devices_array); k++) {
                cJSON *one_dev = cJSON_GetArrayItem(devices_array, k);
                if (!one_dev) continue;

                cJSON *no_item = cJSON_GetObjectItem(one_dev, "no");
                if (no_item && cJSON_IsString(no_item) && strstr(dev->no, no_item->valuestring)) {
                    cJSON *subtype_item = cJSON_GetObjectItem(one_dev, "subtype");
                    if (subtype_item && cJSON_IsString(subtype_item)) {
                        dev->dev_child_type = strdup(subtype_item->valuestring);
                        cJSON *type_item = cJSON_GetObjectItem(one_dev, "type");
                        cJSON *name_item = cJSON_GetObjectItem(one_dev, "name");
                        proto_syslog(LOG_ERR, "[get_dev_child_type_by_dev_info] Matched: %s -> subtype=%s, type=%s, name=%s",
                            dev->no,
                            dev->dev_child_type,
                            type_item && cJSON_IsString(type_item) ? type_item->valuestring : "NULL",
                            name_item && cJSON_IsString(name_item) ? name_item->valuestring : "NULL");
                    } else {
                        proto_syslog(LOG_ERR, "[get_dev_child_type_by_dev_info] Matched %s but no subtype found", dev->no);
                    }
                    found = 1;
                    break;
                }
            }
            if (found) break;
        }

        if (!found) {
            proto_syslog(LOG_ERR, "[get_dev_child_type_by_dev_info] Not found: %s (%s)", dev->no, dev->disp_name);
        }
    }

    cJSON_Delete(root);
    if (num) *num = dev_count;
    proto_syslog(LOG_ERR, "[get_dev_child_type_by_dev_info] End - processed %d devices", dev_count);
    return NULL;
}

static void handler_data_changed(UA_Client *client, UA_UInt32 subId, void *subContext, UA_UInt32 monId, void *monContext, UA_DataValue *value)
{
    tag_t *tag = (tag_t *)monContext;
    UA_Variant val = value->value;

    if (val.type == &UA_TYPES[UA_TYPES_INT32])
    {
        tag->data_type = 0;
        tag->read_cache.to_int = *(UA_Int32 *)val.data;
        proto_syslog(LOG_NOTICE, "[%s]%s changed, value:%ld", ((slave_ems *)subContext)->url, tag->dev_tag_name, tag->read_cache.to_int);
    }
    else if (val.type == &UA_TYPES[UA_TYPES_DOUBLE])
    {
        tag->data_type = 1;
        tag->read_cache.to_float = *(UA_Double *)val.data;
        proto_syslog(LOG_NOTICE, "[%s]%s changed, value:%f", ((slave_ems *)subContext)->url, tag->dev_tag_name, tag->read_cache.to_float);
    }
    else
    {
        proto_syslog(LOG_ERR, "[%s] data type error", ((slave_ems *)subContext)->url);
    }
}


static void handler_online_changed(UA_Client *client, UA_UInt32 subId, void *subContext, UA_UInt32 monId, void *monContext, UA_DataValue *value)
{
    if (!value || !value->hasValue)
    {
        proto_syslog(LOG_ERR, "[%s] Online status change: Invalid data value", ((slave_ems *)subContext)->url);
        return;
    }

    device_t *dev = (device_t *)monContext;
    UA_Variant val = value->value;
    if (val.type == &UA_TYPES[UA_TYPES_INT32])
    {
        dev->online = *(UA_Int32 *)val.data;
        proto_syslog(LOG_NOTICE, "[%s] %s.Online=%d", ((slave_ems *)subContext)->url, dev->disp_name, dev->online);

    }
    else
    {
        proto_syslog(LOG_ERR, "[%s] Online status change: Unexpected data type", ((slave_ems *)subContext)->url);
    }
}



slave_ems *new_slave_ems(const char *url,const char *no,const char *name, int en_data_sub)
{
    slave_ems *p = (slave_ems *)malloc(sizeof(slave_ems));
    snprintf(p->url, sizeof(p->url), "%s", url);
    snprintf(p->no, sizeof(p->no), "%s", no);
    snprintf(p->name, sizeof(p->name), "%s", name);
    p->en_data_sub = en_data_sub;
    p->client = NULL;
    p->devs = NULL;
    p->dev_hash = NULL;
    p->dev_sum = 0;
    p->dev_info = NULL;
    p->dev_tree = NULL;
    p->dev_group_info = NULL;
    p->node_sort = NULL;
    p->dev_group_info = NULL; 
    //
    return p;
}

int clean_slave_ems(slave_ems *ems)
{
    int dev_sum = ems->dev_sum;
    ems->dev_sum = 0;
    if (ems->devs != NULL)
    {
        for (int i = 0; i < dev_sum; ++i)
        {
            device_t *dev = &ems->devs[i];
            if (dev->dev_type) {
                free(dev->dev_type);
                dev->dev_type = NULL;
            }
            if (dev->dev_child_type) {
                free(dev->dev_child_type);
                dev->dev_child_type = NULL;
            }

            for (int j = 0; j < dev->tags.ro_tags_size; j++) {
                tag_t *tag = dev->tags.ro_tags[j];
                if (tag) {
                    if (tag->desc) free(tag->desc);
                    free(tag);
                }
                dev->tags.ro_tags[j] = NULL;
            }
            for (int j = 0; j < dev->tags.rw_tags_size; j++) {
                tag_t *tag = dev->tags.rw_tags[j];
                if (tag) {
                    if (tag->desc) free(tag->desc);
                    free(tag);
                }
                dev->tags.rw_tags[j] = NULL;
            }
            for (int j = 0; j < dev->tags.sys_info_tags_size; j++) {
                tag_t *tag = dev->tags.sys_info_tags[j];
                if (tag) {
                    if (tag->desc) free(tag->desc);
                    free(tag);
                }
                dev->tags.sys_info_tags[j] = NULL;
            }
        }
        free(ems->devs);
        ems->devs = NULL;
    }

    if (ems->client != NULL)
    {
        UA_Client_disconnect(ems->client);
        UA_Client_delete(ems->client);
        ems->client = NULL;
    }

    if (ems->dev_hash != NULL)
    {
        hash_intptr_removeall(&ems->dev_hash);
        ems->dev_hash = NULL;
    }

    if (ems->dev_info != NULL)
    {
        free(ems->dev_info);
        ems->dev_info = NULL;
    }

    if (ems->dev_tree != NULL)
    {
        free(ems->dev_tree);
        ems->dev_tree = NULL;
    }
    if (ems->dev_group_info != NULL)
    {
        free(ems->dev_group_info);
        ems->dev_group_info = NULL;
    }
    if (ems->node_sort != NULL)
    {
        free(ems->node_sort);
        ems->node_sort = NULL;
    }
    return 0;
}

static char **ems_opcua_get_all_dev(slave_ems *p, int *dev_num)
{
    *dev_num = 0;
    char **client_node_id_var = get_dev_id_and_num(p, dev_num);//获取设备名，数量

    if (!client_node_id_var) {
        return NULL;
    }
    // 删除ems
    int del_num = *dev_num + 1;
    for (int i = 0; i < *dev_num; i++) {
        if (strcmp(client_node_id_var[i], "EMS") == 0) {
            del_num = i;
            free(client_node_id_var[i]);
            while ((del_num + 1) < *dev_num) {
                client_node_id_var[del_num] = client_node_id_var[del_num + 1];
                del_num++;
            }
            (*dev_num)--;
            break;
        }
    }
    client_node_id_var[*dev_num] = NULL;
    p->dev_sum = *dev_num;
    return client_node_id_var;
}

int free_char_p_array(char **p, int len)
{
    if (!p)
        return 0;
    for (int i = 0; i < len; i++)
    {
        free(p[i]);
    }
    free(p);
    return 0;
}

static int slave_ems_connect(slave_ems *this)
{
    char url[128];
    snprintf(url, sizeof(url), "opc.tcp://%s:4840", this->url);
    this->client = ua_connect_opcua(url);
    if (this->client == NULL)
    {
        return -1;
    }
    return 0;
}

static int slave_ems_set_lc_cabname(slave_ems *this)
{
    pcs_ctrl_var_t *var = get_pcs_ctrl_var();
    int cur_sum = var->lc_list_len;
    lc_it *lc_cfg = var->lc_list;
    char set_json[128];
	char *uc_ret;
    int find = 0;
    int ret = 0;

	if (this == NULL)
		return -1;

    for (int i = 0; i < cur_sum; i++)
    {
        if (strcmp(this->no, lc_cfg[i].no) == 0) 
        {
            snprintf(set_json, sizeof(set_json),  "{\"name\":\"%s\"}", lc_cfg[i].name);
            find = 1;
            proto_syslog(LOG_NOTICE, "lc no[%s] set cabname:[%s]", this->no, set_json);
            break;
        }
    }

    if(find == 1)
    {
		uc_ret = ua_client_call_write_cfg_by_hd(this->client, LC_SET_CABNAME, set_json, strlen(set_json));
		if (uc_ret == NULL)
			ret = -1;
		else
			free(uc_ret);
    }
    return ret;
}

static int slave_ems_get_cfg(slave_ems *this)
{
    this->dev_tree = ua_client_call_read_cfg_by_hd(this->client, DEV_TREE_NAME);
    this->dev_group_info = ua_client_call_read_cfg_by_hd(this->client, DEV_GROUP_INFO_NAME);
    this->node_sort = ua_client_call_read_cfg_by_hd(this->client, NODE_SORT);
    return 0;
}

static int slave_ems_set_cfg(slave_ems *this)
{
    slave_ems_set_lc_cabname(this);
    return 0;
}

static int slave_ems_creat_dev(slave_ems *p)
{
    int ret = 0;
    int dev_sum = 0;

    char **dev = ems_opcua_get_all_dev(p, &dev_sum);
    if (dev == NULL)
    {
        ret = -2;
        goto slave_ems_creat_dev_err;
    }
    proto_syslog(LOG_NOTICE, "dev_sum: %d", dev_sum);

    device_t *p_dev = NULL;
    for (int i = 0; i < dev_sum; i++)
    {
        p_dev = &p->devs[i];
        proto_syslog(LOG_NOTICE, "dev[%d]:%s", i, p_dev->no);

        int ret_hash = hash_intptr_addptr(&p->dev_hash, p_dev->no, strlen(p_dev->no), p_dev->no);
        if (ret_hash < 0)
        {
            proto_syslog(LOG_ERR, "[%s] hash_intptr_addptr failed, key: %s", p->url, p_dev->no);
            ret = -4;
            break;
        }
    }
    //
    char *old_dev = ua_client_call_read_cfg_by_hd(p->client, DEV_INFO_CONFIG_NAME);
    char *new_dev = add_group_to_devices(old_dev, p->no);

    p->dev_info = new_dev;
    get_dev_child_type_by_dev_info(p, &dev_sum);//获取设备子类型
    if (new_dev)
    {
        free(old_dev);
    }
    //
slave_ems_creat_dev_err:
    free_char_p_array(dev, dev_sum);
    return ret;
}

static int slave_ems_creat_dev_tag(slave_ems *p)
{
    int dev_sum = p->dev_sum;
    char* node_sort = p->node_sort;
    proto_forward_t *var = get_pcs_ctrl_var()->device_layer_ptr;
    if (!node_sort) return -1;

    cJSON *root = cJSON_Parse(node_sort);
    if (!root || !cJSON_IsArray(root)) {
        cJSON_Delete(root);
        return -2;
    }

    for (int i = 0; i < dev_sum; i++) {
        device_t *dev = &p->devs[i];
        cJSON *dev_item = cJSON_GetArrayItem(root, i);
        if (!dev_item) continue;

        dev->tags.rw_tags_size = 0;
        dev->tags.ro_tags_size = 0;
        dev->tags.sys_info_tags_size = 0;
        struct {const char *key; tag_t **arr; int *size; int max;} fields[] = {
            {"SETTING", dev->tags.rw_tags, &dev->tags.rw_tags_size, WR_TAGS_MAX},
            {"ALARM", dev->tags.ro_tags, &dev->tags.ro_tags_size, RO_TAGS_MAX},
            {"DATA", dev->tags.ro_tags, &dev->tags.ro_tags_size, RO_TAGS_MAX},
            {"STATUS", dev->tags.sys_info_tags, &dev->tags.sys_info_tags_size, SYS_INFO_TAGS_MAX},
        };

        for (int f = 0; f < 4; f++) {
            cJSON *arr = cJSON_GetObjectItem(dev_item, fields[f].key);
            if (!arr || !cJSON_IsArray(arr)) continue;
            int tag_num = cJSON_GetArraySize(arr);

            for (int j = 0; j < tag_num && *(fields[f].size) < fields[f].max; j++) {
                cJSON *tag_obj = cJSON_GetArrayItem(arr, j);
                if (!tag_obj) continue;

                if (strcmp(fields[f].key, "STATUS") == 0) {
                    cJSON *jname = cJSON_GetObjectItem(tag_obj, "NAME");
                    if (!(jname && cJSON_IsString(jname) && strstr(jname->valuestring, "Online"))) {
                        continue;
                    }
                }
                tag_t *tag = (tag_t *)calloc(1, sizeof(tag_t));
                cJSON *jname = cJSON_GetObjectItem(tag_obj, "NAME");
                if (jname && cJSON_IsString(jname)) {
                    snprintf(tag->name, sizeof(tag->name), "%s", jname->valuestring);
                }
                cJSON *jdisp = cJSON_GetObjectItem(tag_obj, "DISP");
                if (jdisp && cJSON_IsString(jdisp)) {
                    snprintf(tag->disp_name, sizeof(tag->disp_name), "%s", jdisp->valuestring);
                }

                cJSON *jtype = cJSON_GetObjectItem(tag_obj, "TYPE");
                if (jtype && cJSON_IsNumber(jtype)) {
                    tag->data_type = jtype->valueint;
                }

                cJSON *jscale = cJSON_GetObjectItem(tag_obj, "SCALE");
                if (jscale && cJSON_IsNumber(jscale)) {
                    tag->scale = jscale->valuedouble;
                }
                cJSON *jprop = cJSON_GetObjectItem(tag_obj, "PROPERTY");
                if (jprop && cJSON_IsNumber(jprop)) {
                    tag->property = jprop->valueint;
                }

                cJSON *jdesc = cJSON_GetObjectItem(tag_obj, "DESC");
                if (jdesc && cJSON_IsString(jdesc)) {
                    tag->desc = strdup(jdesc->valuestring);
                }

                snprintf(tag->dev_tag_name, sizeof(tag->dev_tag_name), "%s.%s", dev->no, tag->name);
                tag->active = 0;
                tag->dev_ptr = dev;

                int ret = hash_intptr_addptr(&var->tag_map_hash, tag->dev_tag_name,strlen(tag->dev_tag_name), tag);
                if(ret != 0){
                    proto_syslog(LOG_ERR, "hash_intptr_addptr failed, key: %s", tag->dev_tag_name);
                }
                fields[f].arr[*(fields[f].size)] = tag;
                (*(fields[f].size))++;
            }
        }
    }

    cJSON_Delete(root);
    return 0;
}
//设备数据回调
static int slave_ems_tag_data_change_sub(slave_ems *slave, UA_CreateSubscriptionResponse *response, char *sub_id_str, void *context)
{
    UA_MonitoredItemCreateRequest monRequest = UA_MonitoredItemCreateRequest_default(UA_NODEID_STRING(1, sub_id_str));
    UA_MonitoredItemCreateResult monResponse = UA_Client_MonitoredItems_createDataChange(slave->client, response->subscriptionId,
                                                                                         UA_TIMESTAMPSTORETURN_BOTH,
                                                                                         monRequest, context, handler_data_changed, NULL);
    if (monResponse.statusCode == UA_STATUSCODE_GOOD)
    {
        proto_syslog(LOG_NOTICE, "[%s] Monitoring %s DataChange succeeded", slave->url, sub_id_str);
        return 0;
    }
    else
    {
        proto_syslog(LOG_ERR, "[%s] Monitoring %s DataChange error", slave->url, sub_id_str);
        return -1;
    }
}
//设备在线状态回调
static int slave_ems_dev_online_change_sub(slave_ems *slave, UA_CreateSubscriptionResponse *response, char *sub_id_str, void *context)
{
    char new_name[64] = {0};
    strcpy(new_name,sub_id_str);
    strcat(new_name,".Online");
    UA_MonitoredItemCreateRequest monRequest = UA_MonitoredItemCreateRequest_default(UA_NODEID_STRING(1, new_name));
    UA_MonitoredItemCreateResult monResponse = UA_Client_MonitoredItems_createDataChange(slave->client, response->subscriptionId,
                                                                                         UA_TIMESTAMPSTORETURN_BOTH,
                                                                                         monRequest, context, handler_online_changed, NULL);
    if (monResponse.statusCode == UA_STATUSCODE_GOOD)
    {
        proto_syslog(LOG_NOTICE, "[%s] Monitoring %s DataChange succeeded", slave->url, sub_id_str);
        return 0;
    }
    else
    {
        proto_syslog(LOG_ERR, "[%s] Monitoring %s DataChange error", slave->url, sub_id_str);
        return -1;
    }
}

char* remove_slave_ems_no(const char* input) {
    if (!input) return NULL;

    const char* dot_pos = strchr(input, '.');
    if (!dot_pos) return strdup(input);

    return strdup(dot_pos + 1);
}

static int slave_ems_data_change_sub(slave_ems *slave)
{
    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    // 调整订阅参数，增加容错时间
    request.requestedPublishingInterval = 1000.0;
    request.requestedLifetimeCount = 300;
    request.requestedMaxKeepAliveCount = 30;
    
    UA_CreateSubscriptionResponse response = UA_Client_Subscriptions_create(slave->client, request, slave, NULL, NULL);

    if (response.responseHeader.serviceResult == UA_STATUSCODE_GOOD)
    {
        proto_syslog(LOG_NOTICE, "Create %s subscription succeeded", slave->url);
    }
    else
    {
        proto_syslog(LOG_ERR, "Create %s subscription error", slave->url);
        return -1;
    }

    // 添加计数器，用于定期发送活跃信号
    int tag_count = 0;
    const int ITERATE_INTERVAL = 50;  // 每订阅 50 个点位调用一次 run_iterate

    for (int i = 0; i < slave->dev_sum; i++)
    {
        device_t *dev = &slave->devs[i];

        // 设备上线订阅
        char *modified_dev_no = remove_slave_ems_no(dev->no);
        if (modified_dev_no) {
            slave_ems_dev_online_change_sub(slave, &response, modified_dev_no, dev);
            free(modified_dev_no);
            tag_count++;
        } else {
            slave_ems_dev_online_change_sub(slave, &response, dev->no, dev);
            tag_count++;
        }

        // 读点数据订阅
        for (int j = 0; j < dev->tags.ro_tags_size; j++)
        {
            tag_t *tag = dev->tags.ro_tags[j];
            char *modified_tag_name = remove_slave_ems_no(tag->dev_tag_name);
            if (modified_tag_name) {
                slave_ems_tag_data_change_sub(slave, &response, modified_tag_name, tag);
                free(modified_tag_name);
            } else {
                slave_ems_tag_data_change_sub(slave, &response, tag->dev_tag_name, tag);
            }
            tag_count++;
            
            if (tag_count % ITERATE_INTERVAL == 0) {
                UA_Client_run_iterate(slave->client, 100);
            }
        }

        // 写点数据订阅
        for (int j = 0; j < dev->tags.rw_tags_size; j++)
        {
            tag_t *tag = dev->tags.rw_tags[j];
            char *modified_tag_name = remove_slave_ems_no(tag->dev_tag_name);
            if (modified_tag_name) {
                slave_ems_tag_data_change_sub(slave, &response, modified_tag_name, tag);
                free(modified_tag_name);
            } else {
                slave_ems_tag_data_change_sub(slave, &response, tag->dev_tag_name, tag);
            }
            tag_count++;
            
            if (tag_count % ITERATE_INTERVAL == 0) {
                UA_Client_run_iterate(slave->client, 100);
            }
        }
    }

    UA_Client_run_iterate(slave->client, 1000);

    return 0;
}

int slave_ems_print(slave_ems *slave)
{
    proto_syslog(LOG_NOTICE, "url:%s", slave->url);
    proto_syslog(LOG_NOTICE, "slave_ems[%s] dev_sum:%d", slave->url, slave->dev_sum);
    for (int i = 0; i < slave->dev_sum; i++)
    {
        device_t *dev = &slave->devs[i];
        proto_syslog(LOG_NOTICE, "slave_ems[%s]; dev[%d]:%s", slave->url, i, dev->disp_name);
        //
        for (int j = 0; j < dev->tags.ro_tags_size; j++)
        {
            tag_t *data = dev->tags.ro_tags[j];
            proto_syslog(LOG_NOTICE, "slave_ems[%s]; dev[%d]:%s; ro_data[%d]:%s", slave->url, i, dev->disp_name, j, data->dev_tag_name);
        }
        //
        for (int j = 0; j < dev->tags.rw_tags_size; j++)
        {
            tag_t *data = dev->tags.rw_tags[j];
            proto_syslog(LOG_NOTICE, "slave_ems[%s]; dev[%d]:%s; rw_data[%d]:%s", slave->url, i, dev->disp_name, j, data->dev_tag_name);
        }
    }
    return 0;
}

void *UAClient_thread(void *ua_client_parm)
{
    slave_ems *p = (slave_ems *)ua_client_parm;
    int ret = 0;
    while (1)
    {
        ret = slave_ems_connect(p);
        if (ret != 0)
        {
            proto_syslog(LOG_ERR, "slave_ems_connect[%s] failed, ret: %d", p->url, ret);
            sleep(5);
            continue;
        }
        slave_ems_get_cfg(p);
        slave_ems_set_cfg(p);
        if (p->en_data_sub != 0)
        {
            ret = slave_ems_creat_dev(p);
            if (ret != 0)
            {
                proto_syslog(LOG_ERR, "slave_ems_creat_dev[%s] failed, ret: %d", p->url, ret);
            }
            else
            {
                ret = slave_ems_creat_dev_tag(p);
                if (ret == 0)
                {
                    slave_ems_print(p);
                    ret = slave_ems_data_change_sub(p);
                }
                else
                {
                    proto_syslog(LOG_ERR, "slave_ems_creat_dev_tag failed, ret: %d", ret);
                }
            }
        }
        while (ret == 0)
        {
            if (UA_STATUSCODE_GOOD != UA_Client_run_iterate(p->client, 1000)) // 每1秒处理一次事件
            {
                ret++;
            }
        }
        clean_slave_ems(p);
        proto_syslog(LOG_WARNING, "slave_ems[%s] restart", p->url);
        sleep(5);
    }
    return NULL;
}

int slave_ems_start(slave_ems *slave)
{
    pthread_t UAClient_thread_id = PTHREAD_NULL;
    if (0 == pthread_create(&UAClient_thread_id, NULL, (void *)UAClient_thread, slave))
    {
        pthread_setname_np(UAClient_thread_id, slave->no);
    }
    else
    {
        return -1;
    }
    return 0;
}

tag_t **slave_ems_get_ro_tags(slave_ems *slave, const char *dev_name, int *num)
{
    device_t *dev = NULL;
    if (hash_intptr_findptr(slave->dev_hash, dev_name, strlen(dev_name), (void **)&dev) < 0)
    {
        if (num) *num = 0;
        return NULL;
    }
    if (num) *num = dev->tags.ro_tags_size;
    return dev->tags.ro_tags;
}

tag_t **slave_ems_get_rw_tags(slave_ems *slave, const char *dev_name, int *num)
{
    device_t *dev = NULL;
    if (hash_intptr_findptr(slave->dev_hash, dev_name, strlen(dev_name), (void **)&dev) < 0)
    {
        if (num) *num = 0;
        return NULL;
    }
    if (num) *num = dev->tags.rw_tags_size;
    return dev->tags.rw_tags;
}


char *slave_ems_get_dev_info(slave_ems *slave)
{
    if (!slave) return NULL;
    return slave->dev_info;
}

char *slave_ems_get_dev_tree(slave_ems *slave)
{
    if (!slave) return NULL;
    return slave->dev_tree;
}

char *slave_ems_get_dev_group_info(slave_ems *slave)
{
    if (!slave) return NULL;
    return slave->dev_group_info;
}

char *slave_ems_get_dev_node_sort(slave_ems *slave)
{
    return slave->node_sort;
}

//添加dev_name前缀，设备上云使用
static void process_node(cJSON* node, const char * target, const char* prefix) {
    if (node == NULL || prefix == NULL) return;

    if (node->type == cJSON_Object) {
        cJSON* child = node->child;
        while (child != NULL) {
            if (child->string && (strcmp(child->string, target) == 0)) {
                if (child->type == cJSON_String && child->valuestring != NULL) {
                    size_t prefix_len = strlen(prefix);
                    size_t value_len = strlen(child->valuestring);
                    
                    char* new_value = (char*)malloc(prefix_len + 1 + value_len + 1);
                    
                    if (new_value) {
                        strcpy(new_value, prefix);
                        strcat(new_value, ".");
                        strcat(new_value, child->valuestring);
                        
                        cJSON_SetValuestring(child, new_value);
                        free(new_value);
                    }
                }
            }
            process_node(child, target, prefix);
            child = child->next;
        }
    } else if (node->type == cJSON_Array) {
        cJSON* element = node->child;
        while (element != NULL) {
            process_node(element, target, prefix);
            element = element->next;
        }
    }
}



cJSON* string_to_json_and_add_prefix(const char* json_str,const char* no, const char* name) {
    cJSON* json_obj = cJSON_Parse(json_str);
    process_node(json_obj, "no", no);
    process_node(json_obj, "name", name);

    if (json_obj == NULL) {
        proto_syslog(LOG_ERR, "cjSON parse faild");
        return NULL;
    }
    return json_obj;
}
//添加group键
char* add_group_to_devices(const char* json_str,const char *no) {
    cJSON* root = cJSON_Parse(json_str);
    if (root == NULL) {
        return NULL;
    }

    cJSON* dev_class = cJSON_GetObjectItemCaseSensitive(root, "dev_class");
    if (dev_class == NULL || !cJSON_IsArray(dev_class)) {
        cJSON_Delete(root);
        return NULL;
    }

    cJSON* dev_class_item;
    cJSON_ArrayForEach(dev_class_item, dev_class) {
        cJSON* type_item = cJSON_GetObjectItemCaseSensitive(dev_class_item, "type");
        if (type_item == NULL || !cJSON_IsString(type_item)) {
            continue;
        }
        const char* group_name = no;
        cJSON* devs = cJSON_GetObjectItemCaseSensitive(dev_class_item, "devs");
        if (devs == NULL || !cJSON_IsArray(devs)) {
            continue;
        }

        cJSON* dev;
        cJSON_ArrayForEach(dev, devs) {
            cJSON_AddStringToObject(dev, "group", group_name);
        }
    }

    char* new_json = cJSON_Print(root);
    cJSON_Delete(root);

    return new_json;
}

void print_cjson(const cJSON* json, int formatted) {
    if (json == NULL) {
        fprintf(stderr, "Error: NULL cJSON pointer\n");
        return;
    }

    char* json_str = formatted ? 
        cJSON_Print(json) :      // 格式化输出
        cJSON_PrintUnformatted(json); // 紧凑输出

    if (json_str == NULL) {
        fprintf(stderr, "Error: Failed to generate JSON string\n");
        return;
    }

    printf("%s\n", json_str);
    free(json_str);
}


char* remove_substring(const char* no, const char* tag_name) {
    if (!no || !tag_name) return NULL;

    if (strlen(tag_name) == 0) {
        return strdup(no);
    }

    char* target = (char*)malloc(strlen(tag_name) + 2);
    if (!target) return NULL;
    strcpy(target, tag_name);
    strcat(target, ".");

    char* found = strstr(no, target);
    if (found) {
        size_t prefix_len = found - no;
        const char* suffix = found + strlen(target);
        
        char* result = (char*)malloc(prefix_len + strlen(suffix) + 1);
        if (!result) {
            free(target);
            return NULL;
        }
        
        memcpy(result, no, prefix_len);
        strcpy(result + prefix_len, suffix);
        
        free(target);
        return result;
    }

    char* result = strdup(no);
    free(target);
    return result;
}

int slave_ems_get_dev_oline(slave_ems *slave, const char *dev_name)
{
    device_t *dev = NULL;
    if (hash_intptr_findptr(slave->dev_hash, dev_name, strlen(dev_name), (void **)&dev) < 0)
    {
        return 01;
    }
    return dev->online;
}


const device_t *slave_ems_get_dev(slave_ems *slave, const char *dev_name)
{
    device_t *dev = NULL;
    if (hash_intptr_findptr(slave->dev_hash, dev_name, strlen(dev_name), (void **)&dev) < 0)
    {
        return NULL;
    }
    return dev;
}

int slave_ems_set_dev_group_info(slave_ems *slave, const char *json_str)
{
    char cfg_name[128] = {0};
    snprintf(cfg_name, sizeof(cfg_name), "%s", DEV_GROUP_INFO_NAME);
    if (NULL == ua_client_call_write_cfg_by_hd(slave->client, cfg_name, json_str, strlen(json_str)))
    {
        return -1;
    }
    return 0;
}