异步导入配方

调用方法

#include "RecipeImportAsync.h"

bool started = RecipeImportAsync::start(
    "产品配方",             // T_RCP 配方变量名
    "recipe/product.csv",  // U盘内相对路径
    "配方导入状态"          // T_VCHAR 结果变量名
);

if (!started) {
    // 任务没有启动，具体原因已写入“配方导入状态”
}

将下面的代码保存为 RecipeImportAsync.h

#pragma once

#include <atomic>
#include <cctype>
#include <exception>
#include <fstream>
#include <list>
#include <memory>
#include <string>
#include <thread>
#include <vector>

#include "base/Constants.h"
#include "base/Interface.h"
#include "db/IDataBase.h"
#include "type/Trcp.h"
#include "util/StringUtil.h"
#include "util/Util.h"

/**
 * 客户工程侧的异步配方 CSV 导入器。
 *
 * recipeName 必须是已存在的 T_RCP 组态变量；relativeCsvPath 是 U 盘内的
 * 相对路径（模拟器下相对于 ./data）；resultVar 必须是已存在的 T_VCHAR 变量。
 *
 * 返回值只表示异步任务是否成功启动，最终结果通过 resultVar 返回：
 *   正在导入...
 *   导入成功: N条
 *   导入失败: reason
 *
 * 导入过程中不要编辑或保存同一个配方。
 */
class RecipeImportAsync {
public:
    static bool start(const std::string& recipeName,
                      const std::string& relativeCsvPath,
                      const std::string& resultVar) {
        using namespace stt;
        using namespace stt::constant;
        using namespace stt::data;
        using namespace stt::db;
        using namespace stt::util;

        IRuntime* runtime = getRuntime();
        IVarMgr vm = runtime ? runtime->getIVarMgr() : NULL;
        if(!runtime || !vm || recipeName.empty() || relativeCsvPath.empty()
                || resultVar.empty()) {
            setResult(vm, resultVar, "导入失败: 参数无效");
            return false;
        }

        Ivariable statusVar = vm->get(resultVar);
        if(!statusVar || statusVar->type() != DataType::T_VCHAR) {
            DEBUG::setLastErrorString("Recipe import result variable must be T_VCHAR: " + resultVar);
            return false;
        }

        Ivariable recipeVar = vm->get(recipeName);
        if(!recipeVar || recipeVar->type() != DataType::T_RCP) {
            setResult(vm, resultVar, "导入失败: 配方变量不存在");
            DEBUG::setLastErrorString("Recipe variable does not exist or is not T_RCP: " + recipeName);
            return false;
        }

        HmiAppBase* app = runtime->getAppInstance(std::this_thread::get_id());
        if(!app) {
            setResult(vm, resultVar, "导入失败: 必须从应用线程启动");
            return false;
        }

        bool expected = false;
        if(!running().compare_exchange_strong(expected, true)) {
            setResult(vm, resultVar, "导入失败: 另一导入任务正在运行");
            return false;
        }

        std::string fullPath;
        std::string pathError;
        if(!resolvePath(relativeCsvPath, fullPath, pathError)) {
            running().store(false);
            setResult(vm, resultVar, "导入失败: " + pathError);
            return false;
        }

        TRecipe* recipe = static_cast<TRecipe*>(recipeVar);
        IDatabase* recipeDb = recipe->m_db;
        const std::string tableName = recipe->m_tbl;
        const std::string keyName = recipe->m_key;
        const std::vector<TTable::Column> columns = recipe->m_columns;
        if(!recipeDb || recipeDb->getDBType() != DBTYPE::SQLITE3
                || recipeDb->m_url.empty() || tableName.empty() || columns.empty()) {
            running().store(false);
            setResult(vm, resultVar, "导入失败: 仅支持有效的SQLite配方");
            return false;
        }

        // 使用同一文件的独立 SQLite 连接，避免后台批处理与运行时默认连接
        // 共用 m_transing 状态。连接由 DBMgr 管理，可供下次导入复用。
        const std::string importDbUrl = makePrivateDbUrl(recipeDb->m_url);
        IDatabase* importDb = DBMgr::getdb(importDbUrl, DBTYPE::SQLITE3,
                                            importDbUrl);
        if(!importDb || importDb == recipeDb) {
            running().store(false);
            setResult(vm, resultVar, "导入失败: 无法建立独立数据库连接");
            return false;
        }

        std::shared_ptr<TaskResult> result;
        std::shared_ptr<TTable> imported;
        AsynWorker* worker = NULL;
        try {
            result.reset(new TaskResult());
            imported.reset(new TTable("__recipe_import"));
            worker = new AsynWorker(app);
        } catch(...) {
            delete worker;
            running().store(false);
            setResult(vm, resultVar, "导入失败: 内存不足，无法启动任务");
            return false;
        }

        // 先释放旧的内存行，避免导入时同时保留两份大配方。数据库原表会一直
        // 保留到临时表完整写入后才在短事务中替换。
        std::vector<TTable::Row>().swap(recipe->m_rows);
        recipe->notify();

        setResult(vm, resultVar, "正在导入...");

        worker->invoke([importDb, tableName, keyName, columns, fullPath,
                        imported, result]() {
            runImport(importDb, tableName, keyName, columns, fullPath,
                      *imported, *result);
        })->onFinished([vm, recipeName, resultVar, imported, result]() {
            running().store(false);
            Ivariable current = vm ? vm->get(recipeName) : NULL;
            TRecipe* currentRecipe = current && current->type() == stt::constant::DataType::T_RCP
                    ? static_cast<TRecipe*>(current) : NULL;

            if(currentRecipe && (result->ok || result->originalRowsLoaded)) {
                currentRecipe->m_rows.swap(imported->m_rows);
                currentRecipe->notify();
            }

            if(result->ok) {
                setResult(vm, resultVar,
                        "导入成功: " + std::to_string(result->rowCount) + "条");
            } else {
                std::string message = "导入失败: " + result->error;
                if(!result->originalRowsLoaded) {
                    message += "；原配方仍在数据库，重启后可恢复";
                }
                setResult(vm, resultVar, message);
            }
        });

        if(!worker->start()) {
            running().store(false);
            recipe->drawData(true);
            setResult(vm, resultVar, "导入失败: 无法启动后台任务");
            return false;
        }
        return true;
    }

private:
    struct TaskResult {
        bool ok;
        bool originalRowsLoaded;
        std::size_t rowCount;
        std::string error;

        TaskResult() : ok(false), originalRowsLoaded(false), rowCount(0) {}
    };

    struct ImportAbort {};

    static std::atomic<bool>& running() {
        static std::atomic<bool> value(false);
        return value;
    }

    static void setResult(stt::IVarMgr vm, const std::string& name,
                          const std::string& value) {
        if(vm && !name.empty()) vm->setChar(name, value);
    }

    static bool resolvePath(const std::string& relativePath,
                            std::string& fullPath, std::string& error) {
        using namespace stt::util;
        if(relativePath.empty() || relativePath[0] == '/'
                || relativePath[0] == '\\' || relativePath.find(':') != std::string::npos
                || relativePath.find("..") != std::string::npos) {
            error = "文件路径必须是U盘内的相对路径";
            return false;
        }
#if __arm__
        std::vector<std::string> disks = SysInfo::getHotPlugin()
                ? SysInfo::getUdisksHot() : SysInfo::getUdisks();
        if(disks.empty()) {
            error = "未发现U盘";
            return false;
        }
        fullPath = disks[0] + "/" + relativePath;
#else
        fullPath = "./data/" + relativePath;
#endif
        return true;
    }

    static std::string makePrivateDbUrl(const std::string& source) {
        std::string url = source.compare(0, 5, "file:") == 0
                ? source : "file:" + source;
        url += url.find('?') == std::string::npos ? '?' : '&';
        url += "cache=private&stt_recipe_import=1";
        return url;
    }

    static std::string quoteIdentifier(const std::string& value) {
        std::string out = "\"";
        for(std::size_t i = 0; i < value.size(); ++i) {
            if(value[i] == '\"') out += "\"\"";
            else out += value[i];
        }
        out += "\"";
        return out;
    }

    static bool parseCsvLine(const std::string& source,
                             std::vector<std::string>& fields) {
        fields.clear();
        std::string field;
        bool quoted = false;
        for(std::size_t i = 0; i < source.size(); ++i) {
            char ch = source[i];
            if(ch == '\"') {
                if(quoted && i + 1 < source.size() && source[i + 1] == '\"') {
                    field += '\"';
                    ++i;
                } else {
                    quoted = !quoted;
                }
            } else if(ch == ',' && !quoted) {
                fields.push_back(field);
                field.clear();
            } else if(ch != '\r') {
                field += ch;
            }
        }
        if(quoted) return false;
        fields.push_back(field);
        return true;
    }

    static std::string makeInsertSql(const std::string& table,
                                     const std::string& columnList,
                                     const std::vector<std::string>& fields) {
        std::string sql = "INSERT INTO " + quoteIdentifier(table)
                + " (" + columnList + ") VALUES (";
        for(std::size_t i = 0; i < fields.size(); ++i) {
            if(i) sql += ',';
            sql += "'" + stt::util::StringUtil::sqlEsc(fields[i]) + "'";
        }
        sql += ')';
        return sql;
    }

    static bool executeBatch(stt::db::IDatabase* db,
                             std::list<std::string>& sqls,
                             TaskResult& result) {
        if(sqls.empty()) return true;
        if(!db->execute(sqls)) {
            result.error = db->infoContent.empty() ? "数据库批量写入失败" : db->infoContent;
            return false;
        }
        sqls.clear();
        std::this_thread::yield();
        return true;
    }

    static void runImport(stt::db::IDatabase* db,
                          const std::string& tableName,
                          const std::string& keyName,
                          const std::vector<stt::data::TTable::Column>& columns,
                          const std::string& fullPath,
                          stt::data::TTable& imported,
                          TaskResult& result) {
        static const std::size_t kBatchSize = 500;
        static const std::size_t kMaxLineSize = 1024 * 1024;
        static const std::size_t kMaxRows = 10000;
        static const std::size_t kMaxCells = 200000;
        static const std::size_t kMaxCsvBytes = 16 * 1024 * 1024;
        const std::string tempTable = "__RCP_IMPORT_TMP";
        std::string columnList;

        try {
            imported.setNotifier(false);
            for(std::size_t i = 0; i < columns.size(); ++i) {
                if(i) columnList += ',';
                columnList += quoteIdentifier(columns[i].id);
                imported.addColumn(columns[i].id, columns[i].def, columns[i].type);
            }

            std::ifstream input(fullPath.c_str(), std::ios::in | std::ios::binary);
            if(!input.is_open()) {
                result.error = "无法打开文件";
                throw ImportAbort();
            }
            input.seekg(0, std::ios::end);
            const std::streamoff fileSize = input.tellg();
            if(fileSize < 0 || static_cast<unsigned long long>(fileSize)
                    > static_cast<unsigned long long>(kMaxCsvBytes)) {
                result.error = "CSV超过16MB安全上限";
                throw ImportAbort();
            }
            input.seekg(0, std::ios::beg);

            std::string line;
            std::vector<std::string> fields;
            if(!std::getline(input, line)) {
                result.error = "文件内容为空";
                throw ImportAbort();
            }
            std::size_t csvBytes = line.size() + 1;
            if(line.size() > kMaxLineSize) {
                result.error = "CSV表头过长";
                throw ImportAbort();
            }
            if(line.size() >= 3 && static_cast<unsigned char>(line[0]) == 0xEF
                    && static_cast<unsigned char>(line[1]) == 0xBB
                    && static_cast<unsigned char>(line[2]) == 0xBF) {
                line.erase(0, 3);
            }
            if(!parseCsvLine(line, fields) || fields.size() != columns.size()) {
                result.error = "CSV表头列数与配方不一致";
                throw ImportAbort();
            }

            db->execute("DROP TABLE IF EXISTS " + quoteIdentifier(tempTable));
            const std::string createSql = "CREATE TABLE " + quoteIdentifier(tempTable)
                    + " AS SELECT " + columnList + " FROM " + quoteIdentifier(tableName)
                    + " WHERE 0";
            if(!db->execute(createSql)) {
                result.error = db->infoContent.empty() ? "无法创建导入临时表" : db->infoContent;
                throw ImportAbort();
            }

            std::list<std::string> batch;
            std::size_t lineNumber = 1;
            while(std::getline(input, line)) {
                ++lineNumber;
                csvBytes += line.size() + 1;
                if(csvBytes > kMaxCsvBytes) {
                    result.error = "CSV超过16MB安全上限";
                    break;
                }
                if(line.empty() || line == "\r") continue;
                if(line.size() > kMaxLineSize) {
                    result.error = "第" + std::to_string(lineNumber) + "行过长";
                    break;
                }
                if(!parseCsvLine(line, fields)) {
                    result.error = "第" + std::to_string(lineNumber) + "行CSV格式错误";
                    break;
                }
                if(fields.size() != columns.size()) {
                    result.error = "第" + std::to_string(lineNumber) + "行列数不一致";
                    break;
                }
                if(result.rowCount >= kMaxRows
                        || (result.rowCount + 1) > kMaxCells / columns.size()) {
                    result.error = "数据超过10000行或200000个单元格安全上限";
                    break;
                }

                imported.addRow(fields);
                batch.push_back(makeInsertSql(tempTable, columnList, fields));
                ++result.rowCount;
                if(batch.size() >= kBatchSize && !executeBatch(db, batch, result)) break;
            }

            if(result.error.empty() && !input.eof()) result.error = "读取U盘文件失败";
            if(result.error.empty() && !executeBatch(db, batch, result)) {
                // executeBatch supplies the error text.
            }

            if(result.error.empty()) {
                std::list<std::string> publish;
                publish.push_back("DELETE FROM " + quoteIdentifier(tableName));
                publish.push_back("INSERT INTO " + quoteIdentifier(tableName)
                        + " (" + columnList + ") SELECT " + columnList
                        + " FROM " + quoteIdentifier(tempTable));
                publish.push_back("DROP TABLE " + quoteIdentifier(tempTable));
                if(db->execute(publish)) {
                    result.ok = true;
                    return;
                }
                result.error = db->infoContent.empty() ? "替换配方表失败" : db->infoContent;
            }
        } catch(const ImportAbort&) {
            // result.error 已在失败位置设置。
        } catch(const std::exception& e) {
            result.error = std::string("导入异常: ") + e.what();
        } catch(...) {
            result.error = "未知导入异常";
        }

        db->execute("DROP TABLE IF EXISTS " + quoteIdentifier(tempTable));
        // 原表未被替换或最终事务已回滚。失败恢复也在后台完成，避免
        // onFinished 再同步扫描整张配方表。
        if(!columnList.empty()) {
            imported.m_rows.clear();
            imported.m_columns.clear();
            std::string restoreSql = "SELECT " + columnList + " FROM "
                    + quoteIdentifier(tableName);
            if(!keyName.empty()) restoreSql += " ORDER BY " + quoteIdentifier(keyName);
            db->query(restoreSql, &imported);
            result.originalRowsLoaded = db->infoContent == "OK";
        }
        if(result.error.empty()) result.error = "未知错误";
    }
};