klammertext/mac/eval_python.cpp

#include "eval_python.h"
#include "show.h"
#include "util.h"
#include "log.h"

std::regex Eval_python::statement_delimiter("\\s*;\\s*");

Eval_python::Eval_python(Machine& machine, const Locator& loc)
    : m_machine(machine)
    , m_loc(loc)
    , m_globals(nullptr)
    , m_locals(nullptr)
{
    (void)K::log(3);
    // Only initialize if Python is not already initialized
    if (!Py_IsInitialized()) {
#if PY_VERSION_HEX >= 0x030B0000
        // Python 3.11+ uses PyConfig API
        PyConfig config;
        PyConfig_InitPythonConfig(&config);
        Py_InitializeFromConfig(&config);
        PyConfig_Clear(&config);
#else
        Py_Initialize();
#endif
    }
    add_module_path("..");
    add_module_path(".");
    m_globals = PyDict_New();
    m_locals = PyDict_New();
    PyDict_SetItemString(m_globals, "__builtins__", PyEval_GetBuiltins());
    // The machine's result text, so :after_apply phase functions can take it
    // as an argument (the Python counterpart of a :cpp phase function reading
    // machine.m_result). Set directly rather than through the state's
    // python_code() because document text cannot be safely embedded in a
    // quoted Python source string.
    PyObject* result_text = PyUnicode_FromString(m_machine.m_result.c_str());
    if (result_text) {
        PyDict_SetItemString(m_globals, "K_result", result_text);
        Py_DECREF(result_text);
    }
    import_module("inspect", false);
    if (!m_machine.m_state.m_frames.empty()) {
        PyRun_String(m_machine.m_state.python_code().c_str(), Py_file_input, m_globals, m_locals);
    }
}

Eval_python::~Eval_python()
{
    (void)K::log(3, "destructor");
    // Clean up our objects BEFORE finalizing Python
    if (m_globals) {
        Py_DECREF(m_globals);
        m_globals = nullptr;
    }
    if (m_locals) {
        Py_DECREF(m_locals);
        m_locals = nullptr;
    }
    // Don't call Py_Finalize() here - it can cause double-free if other
    // Eval_python objects exist or if Python is used elsewhere.
    // Python will clean up automatically at program exit.
}

std::string remove_string_values(std::string s)
{
    return std::regex_replace(s, std::regex(R"(\".*?\")"), "\"\"");
}

void Eval_python::add_module_path(const std::string& path)
{
    PyObject* sys_path = PySys_GetObject("path");  // Borrowed reference
    if (sys_path) {
        PyObject* py_path = PyUnicode_FromString(path.c_str());
        if (py_path) {
            PyList_Insert(sys_path, 0, py_path);  // Insert at front for priority
            Py_DECREF(py_path);
        }
    }
}

strings_t Eval_python::parse_modules(std::string code)
{
    (void)K::log(3, code);
    code = remove_string_values(code); // Hack! Don't look for module patterns in strings.
    std::regex module_re(R"(([A-Za-z]\w*)\.[A-Za-z_]\w*)");
    auto code_begin = std::sregex_iterator(code.begin(), code.end(), module_re);
    auto code_end = std::sregex_iterator();
    std::vector<std::string> modules;
    for (std::sregex_iterator it = code_begin; it != code_end; ++it) {
        modules.push_back((*it).str(1));
    }
    return modules;
}    

void Eval_python::import_module(const std::string& module_name, bool verify)
{
    (void)K::log(3, module_name);

    std::string module_check =
        "\"" + module_name + "\" in locals() and inspect.isclass(" + module_name + ")"; 
    if (verify && eval_expression(module_check, false) == "True") {
        return;
    }
    PyObject* module = PyImport_ImportModule(module_name.c_str());
    if (module == nullptr) {
        // Extract the Python traceback before clearing the error.
        // This reveals the actual source of the failure (e.g., a syntax
        // error in a transitively imported module), not just the top-level
        // module name that failed to load.
        std::string detail;
        PyObject* ptype;
        PyObject* pvalue;
        PyObject* ptraceback;
        PyErr_Fetch(&ptype, &pvalue, &ptraceback);
        if (pvalue) {
            PyErr_NormalizeException(&ptype, &pvalue, &ptraceback);
            PyObject* str = PyObject_Str(pvalue);
            if (str) {
                detail = PyUnicode_AsUTF8(str);
                Py_DECREF(str);
            }
            // Format the traceback if available
            if (ptraceback) {
                PyObject* tb_module = PyImport_ImportModule("traceback");
                if (tb_module) {
                    PyObject* format_tb = PyObject_GetAttrString(tb_module, "format_exception");
                    if (format_tb) {
                        PyObject* args = PyTuple_Pack(3, ptype, pvalue, ptraceback);
                        PyObject* tb_list = PyObject_CallObject(format_tb, args);
                        if (tb_list) {
                            PyObject* separator = PyUnicode_FromString("");
                            PyObject* joined = PyUnicode_Join(separator, tb_list);
                            if (joined) {
                                detail = PyUnicode_AsUTF8(joined);
                                Py_DECREF(joined);
                            }
                            Py_DECREF(separator);
                            Py_DECREF(tb_list);
                        }
                        Py_XDECREF(args);
                        Py_DECREF(format_tb);
                    }
                    Py_DECREF(tb_module);
                }
            }
        }
        Py_XDECREF(ptype);
        Py_XDECREF(pvalue);
        Py_XDECREF(ptraceback);
        PyErr_Clear();

        std::string message = "Cannot import module \"" + module_name + "\"";
        if (!detail.empty()) {
            message += ":\n\n" + detail;
        }
        throw Argument_error(message, m_loc, false);
    }
    // PyDict_SetItemString steals a reference, so we don't need to DECREF module
    // The dictionary will own the reference
    PyDict_SetItemString(m_globals, module_name.c_str(), module);
}

std::string Eval_python::get_result(PyObject* result_object)
{
    std::string result {};
    if (result_object) {
       const char* value = PyUnicode_AsUTF8(result_object);
       result = std::string(value);
       Py_DECREF(result_object);
    } else {
        std::cout << red;
        PyErr_Print();
        throw Parsing_error("Python code error in @eval", m_loc);
    }
    return result;
}

std::string Eval_python::eval_expression(const std::string& expression, bool import_modules)
{
    (void)K::log(3, expression);
    // msg() << "expression: " << expression << "\n";
    if (import_modules && expression.find('.') != std::string::npos) {
        for (const auto& m : parse_modules(expression)) {
            import_module(m);
        }
    }
    return get_result(
        PyRun_String(
            std::string("str(" + expression +")").c_str(),
            Py_eval_input, m_globals, m_locals));
}

std::string Eval_python::eval_statements(const std::string& script)
{
    (void)K::log(3);
    strings_t statements = regex_split(script, statement_delimiter);
    for (auto iter = statements.begin(); iter < statements.end() - 1; iter++) {
        (void)K::log(3, "  Run: " + (*iter));
        PyRun_String(iter->c_str(), Py_file_input, m_globals, m_locals);
    }
    (void)K::log(3, "  Result from: " + statements.back());
    return get_result(
        PyRun_String(std::string("str("+statements.back()+")").c_str(),
            Py_eval_input, m_globals, m_locals));
}

std::string Eval_python::eval(std::string code)
{
    (void)K::log(3, code);
    code = m_machine.m_state.subst(code, true);
    // Kept as a record of a value worth watching; line comments rather than a
    // block, so the msg() guard can see it is inert (doc/check_output_policy.sh
    // is line-based and cannot tell it is inside a /* */).
    // msg() << "\n"
    //  << std::string(80, '-') << "\n"
    //  << code << "\n"
    //  << std::string(80, '-') << "\n";
    if (std::regex_search(code, statement_delimiter)) {
        return eval_statements(code);
    } else {
        return eval_expression(code);
    }
}

std::string Eval_python::eval_katom_list(
    katom_list& katoms, const katom_iter& begin, const katom_iter& end)
{
    (void)K::log(3, katoms);
    katom_iter code_begin = begin + 1;
    katom_iter code_end = end - 1;
    std::string code_result = eval(as_string(code_begin, code_end, true));
    (void)K::log(3, "code_result:", code_result);
    katom_list code_katoms = m_machine.process(code_result, command_name);
    for (auto kiter = code_begin; kiter < code_end; kiter++) {
        kiter->m_type = katom_t::replaced;
    }
    katoms.insert(end, code_katoms.begin(), code_katoms.end());
    return code_result;
}