DataProcessService/DataProcessService/ytWarterMkProcess/databasethread.cpp

#include "databasethread.h"
//#include "waterMk01.h"
#include "AlarmReport.h"

#define HostName  "47.98.201.187"
#define HostPort 3306
#define UserName "uskyIOT"
#define PassWord "uskyIOT"
#define DatabaseName "jdxf"


#define sql_debug

//extern MksensorList *mkList;

DatabaseThread::DatabaseThread(QObject *parent) :
    QThread(parent)
{
    c = NULL;
    keep = false;
    hour = 255;
    sqlList.clear();
    alarmRepList.clear();
    db = QSqlDatabase::addDatabase("QMYSQL","mksensor_db");
    db.setHostName(QString(HostName));
    db.setPort(HostPort);
    db.setUserName(QString(UserName));
    db.setPassword(QString(PassWord));
    db.setDatabaseName(QString(DatabaseName));
    db.open();
    mcdb = QSqlDatabase::addDatabase("QMYSQL","mksensor_sdb");
    mcdb.setHostName(QString("47.111.81.118"));
    mcdb.setPort(3306);
    mcdb.setUserName(QString("root"));
    mcdb.setPassword(QString("Yt2021"));
    mcdb.setDatabaseName(QString("jx_cover"));
    redisconnected=-1;
//    mcdb.open();
}

void DatabaseThread::appendSql(QString sql,int type)
{
    if(type==1){
        sqlList.append(sql);
    }else if (type==2) {
        mcsqlList.append(sql);
    }
}

void DatabaseThread::appendAlarm(AlarmRep rep,int type)
{
    if(type==1){
        alarmRepList.append(rep);
    }else if (type==2) {
        mcalarmRepList.append(rep);
    }
}

bool DatabaseThread::chktable(QString tbname)
{
    QStringList tblist = db.tables();
    for(int i=0;i<tblist.size();i++){
        if(QString::compare(tblist.at(i),tbname)==0)
            return true;
    }
    return false;
}

void DatabaseThread::check_devicelist()
{
    if(redisconnected==-1){
        redisconnected=redis_init();
        if(redisconnected==-1){
            printf("redis init failed\n");
        }
        redis_save(AUTH);
        redis_save(DBID);
    }

    printf("database thread check_devicelist()\n");
    uint nowTime = QDateTime::currentDateTime().toTime_t();
    hour=QTime::currentTime().hour();
    QString sql = QString("select owner_code, owner_name, unitinfo, owner_xh, company, dwtype from sp_owner where dwtype in (2,4,128) and owner_xh !=''");
    QSqlQuery qry = db.exec(sql);
    int nrow= 0;
    while(qry.next()){
        if(nrow>10239)
            break;
//        QString tbname = QString("sp_d%1").arg(qry.value(0).toString());
//        if(!chktable(tbname)){
//            db.exec(QString::fromUtf8("create table `%1` (`id` bigint(20) unsigned not null auto_increment primary key, `device_code` varchar(30) not null, `port` varchar(55) default null, `time` datetime default null, `status` varchar(50) default null, `address` varchar(50) default null, `ncmd` varchar(50) default null, `data1` varchar(50) default null, `data2` varchar(255) default null, `data3` varchar(255) default null, `data4` varchar(255) default null, `data5` varchar(255) default null )").arg(tbname));
//            db.exec(QString::fromUtf8("insert into sp_owner_status (id, device_id, dwtype, point_name, point_code, point_data, data_time, content) values (null, '%1', %2, '状态', 1, '', '1970-01-01 08:00:00','')").arg(qry.value(0).toString()).arg(qry.value(5).toInt()));
//            db.exec(QString::fromUtf8("insert into sp_owner_status (id, device_id, dwtype, point_name, point_code, point_data, data_time, content) values (null, '%1', %2, '剩余电量', 2, '', '1970-01-01 08:00:00','')").arg(qry.value(0).toString()).arg(qry.value(5).toInt()));
//            db.exec(QString::fromUtf8("insert into sp_owner_status (id, device_id, dwtype, point_name, point_code, point_data, data_time, content) values (null, '%1', %2, '无线信号', 3, '', '1970-01-01 08:00:00','')").arg(qry.value(0).toString()).arg(qry.value(5).toInt()));
//            db.exec(QString::fromUtf8("insert into sp_owner_status (id, device_id, dwtype, point_name, point_code, point_data, data_time, content) values (null, '%1', %2, '水压/水位', 4, '', '1970-01-01 08:00:00','')").arg(qry.value(0).toString()).arg(qry.value(5).toInt()));
//        }
        sql = QString("select true_status from sp_devices_status where deviceid='%1'")
                .arg(qry.value(0).toString());
        QSqlQuery qry2 = db.exec(sql);

        uint redistime = 0;

        //mkList->sensors[nrow].Enabled=0x01;
        if(qry2.next()){
            if(qry2.value(0).toUInt()==0){
                //mkList->sensors[nrow].LastCommtime = nowTime;
                redistime = nowTime;
            }

        }
        //sprintf(mkList->sensors[nrow].Device_Code,"%s",qry.value(0).toString().toUtf8().data());
        //sprintf(mkList->sensors[nrow].Device_Name,"%s",qry.value(1).toString().toUtf8().data());
        //sprintf(mkList->sensors[nrow].Device_Info,"%s",qry.value(2).toString().toUtf8().data());
        //sprintf(mkList->sensors[nrow].Device_Type,"%s",qry.value(5).toString().toUtf8().data());
        //sprintf(mkList->sensors[nrow].Company_Code,"%s",qry.value(4).toString().toUtf8().data());
        //printf("Device_Code:%s\n",QString(mkList->sensors[nrow].Device_Code).toUtf8().data());
        //printf("Device_Code:%s\n",QString(mkList->sensors[nrow].Company_Code).toUtf8().data());

        QString data = QString("hmset %1 row %2 devicecode %3 devicename %4 deviceinfo %5 devicetype %6 companycode %7 enabled %8 onlineflag %9 lastcommtime %10").arg(QString("%1%2").arg("sp_owner8-").arg(qry.value(0).toString().toUtf8().data())).arg(nrow).arg(qry.value(0).toString().toUtf8().data()).arg(qry.value(1).toString().toUtf8().data()).arg(qry.value(2).toString().toUtf8().data()).arg(qry.value(5).toString().toUtf8().data()).arg(qry.value(4).toString().toUtf8().data()).arg(QString("0x01")).arg(QString("0x00")).arg(redistime);
        redis_save(data.toUtf8());

        nrow++;
    }

    redis_save((QString("hmset tablelen sp_owner8 %1").arg(nrow)).toUtf8());
    redis_free();
    redisconnected = -1;

//    for(int i=nrow;i<10240;i++){
//        mkList->sensors[i].Enabled=0x00;
//        mkList->sensors[i].LastCommtime = 0;
//    }

}

void DatabaseThread::stop()
{
    keep = false;
}

void DatabaseThread::run()
{
    db.open();
    QString ip,port,name,passwd;
    QString mqttsql = QString("select ip, port, username, passwd from yt_dataprocessservice_mqtt where type = 'ytDP0008';");
    QSqlQuery mqttqry = db.exec(mqttsql);
    while(mqttqry.next())
    {
        ip = mqttqry.value(0).toString();
        port = mqttqry.value(1).toString();
        name = mqttqry.value(2).toString();
        passwd = mqttqry.value(3).toString();
    }
    emit mqttInfo(ip,port,name,passwd);
    mqttqry.clear();
    db.close();

    keep = true;
    printf("database thread start\n");
    while(keep){
        if(!db.isOpen()){
            db.open();
        }else{
            if(QTime::currentTime().hour()!=hour){
                check_devicelist();
            }
            if(alarmRepList.length()>0){
                while(alarmRepList.length()>0){
                    AlarmRep rep = alarmRepList.first();
                    QSqlQuery qry = db.exec(rep.Sql);
#ifdef sql_debug
                    emit SqlLog(QString("[ %1 sql] ").arg(QDateTime::currentDateTime().toString("yyyy-MM-dd HH:mm:ss.zzz"))+rep.Sql);
#endif
                    quint64 lastid = qry.lastInsertId().toULongLong()&0xffffffffffffffff;
                    if(lastid>0){
                        emit AlarmReport(rep.DeviceId,lastid,rep.AlarmType,rep.AlarmTime);
                    }
                    alarmRepList.removeFirst();
                    usleep(1000);
                }
            }
            if(sqlList.length()>0){
                db.exec(sqlList.first());
#ifdef sql_debug
                emit SqlLog(QString("[ %1 sql ] ").arg(QDateTime::currentDateTime().toString("yyyy-MM-dd HH:mm:ss.zzz"))+sqlList.first());
#endif
                sqlList.removeFirst();
            }
            db.close();
        }
        usleep(10000);
    }
    if(db.isOpen())
        db.close();
}

int DatabaseThread::redis_init()
{
    c = redisConnect(REDIS_HOST, REDIS_PORT);
    if (NULL == c || c->err) {
        if(c) {
            printf("Redis [%s:%d], Error:[%s]\n", REDIS_HOST, REDIS_PORT, c->errstr);
            redisFree(c);
        } else {
            printf("Redis [%s:%d] failure\n", REDIS_HOST, REDIS_PORT);
        }
        return -1;
    }

    return 0;
}

void DatabaseThread::redis_free()
{
    if (c) {
        redisFree(c);
    }
    c = NULL;
}

int DatabaseThread::redis_save(const char *cmd)
{
    int i = 0;
    redisReply *r = NULL;
    if (NULL == cmd) {
        return -1;
    }

    printf("%s, c %p\n", cmd,c);

    r = (redisReply *)redisCommand(c, cmd);
    if (NULL == r) {
        printf("Error[%d:%s]", c->err, c->errstr);
        return -1;
    }

    switch(r->type) {
    case REDIS_REPLY_STATUS:
        printf("type:%s, reply->len:%ld reply->str:%s\n", "REDIS_REPLY_STATUS", r->len, r->str);
        break;
    case REDIS_REPLY_ERROR:
        printf("type:%s, reply->len:%ld reply->str:%s\n", "REDIS_REPLY_ERROR", r->len, r->str);
        break;
    case REDIS_REPLY_INTEGER:
        printf("type:%s, reply->integer:%lld\n", "REDIS_REPLY_INTEGER", r->integer);
        break;
    case REDIS_REPLY_NIL:
        printf("type:%s, no data\n", "REDIS_REPLY_NIL");
        break;
    case REDIS_REPLY_STRING:
        printf("type:%s, reply->len:%ld reply->str:%s\n", "REDIS_REPLY_STRING", r->len, r->str);
        break;
    case REDIS_REPLY_ARRAY:
        printf("type:%s, reply->elements:%ld\n", "REDIS_REPLY_ARRAY", r->elements);
        for (i = 0; i < r->elements; i++) {
            printf("%d: %s\n", i, r->element[i]->str);
        }
        break;
    default:
        printf("unkonwn type:%d\n", r->type);
        break;
    }

    /*release reply and context */
    freeReplyObject(r);
    return 0;
}

QString DatabaseThread::redis_qstring(const char *cmd)
{
    redisReply *r = NULL;
    if (NULL == cmd) {
        return "error";
    }

    printf("%s\n", cmd);

    r = (redisReply *)redisCommand(c, cmd);
    if (NULL == r) {
        printf("Error[%d:%s]", c->err, c->errstr);
        return "error";
    }

    switch(r->type) {
    case REDIS_REPLY_STATUS:
        printf("type:%s, reply->len:%ld reply->str:%s\n", "REDIS_REPLY_STATUS", r->len, r->str);
        break;
    case REDIS_REPLY_ERROR:
        printf("type:%s, reply->len:%ld reply->str:%s\n", "REDIS_REPLY_ERROR", r->len, r->str);
        break;
    case REDIS_REPLY_INTEGER:
        printf("type:%s, reply->integer:%lld\n", "REDIS_REPLY_INTEGER", r->integer);
        break;
    case REDIS_REPLY_NIL:
        printf("type:%s, no data\n", "REDIS_REPLY_NIL");
        break;
    case REDIS_REPLY_STRING:
        printf("type:%s, reply->len:%ld reply->str:%s\n", "REDIS_REPLY_STRING", r->len, r->str);
        break;
    case REDIS_REPLY_ARRAY:
        printf("type:%s, reply->elements:%ld\n", "REDIS_REPLY_ARRAY", r->elements);
        for (int i = 0; i < r->elements; i++) {
            printf("%d: %s\n", i, r->element[i]->str);
        }
        break;
    default:
        printf("unkonwn type:%d\n", r->type);
        break;
    }

    QString data = r->str;

    /*release reply and context */
    freeReplyObject(r);
    return data;
}

int DatabaseThread::redis_int(const char *cmd)
{
    redisReply *r = NULL;
    if (NULL == cmd) {
        return -1;
    }

    printf("%s\n", cmd);

    r = (redisReply *)redisCommand(c, cmd);
    if (NULL == r) {
        printf("Error[%d:%s]", c->err, c->errstr);
        return -1;
    }

    switch(r->type) {
    case REDIS_REPLY_STATUS:
        printf("type:%s, reply->len:%ld reply->str:%s\n", "REDIS_REPLY_STATUS", r->len, r->str);
        break;
    case REDIS_REPLY_ERROR:
        printf("type:%s, reply->len:%ld reply->str:%s\n", "REDIS_REPLY_ERROR", r->len, r->str);
        break;
    case REDIS_REPLY_INTEGER:
        printf("type:%s, reply->integer:%lld\n", "REDIS_REPLY_INTEGER", r->integer);
        break;
    case REDIS_REPLY_NIL:
        printf("type:%s, no data\n", "REDIS_REPLY_NIL");
        break;
    case REDIS_REPLY_STRING:
        printf("type:%s, reply->len:%ld reply->str:%s\n", "REDIS_REPLY_STRING", r->len, r->str);
        break;
    case REDIS_REPLY_ARRAY:
        printf("type:%s, reply->elements:%ld\n", "REDIS_REPLY_ARRAY", r->elements);
        for (int i = 0; i < r->elements; i++) {
            printf("%d: %s\n", i, r->element[i]->str);
        }
        break;
    default:
        printf("unkonwn type:%d\n", r->type);
        break;
    }

    int data = r->integer;

    /*release reply and context */
    freeReplyObject(r);
    return data;
}