如何在.NET Core项目中复用基于Sklearn的Python文本短语提取代码？

原Python代码 #

import pyodbc
from sklearn.feature_extraction.text import CountVectorizer

def connect_to_database(server, database, username, password):
    conn = pyodbc.connect(
        f'DRIVER=ODBC Driver 17 for SQL Server;SERVER={server};DATABASE={database};UID={username};PWD={password}'
    )
    return conn

def fetch_text_data(conn, sql_query):
    cursor = conn.cursor()
    cursor.execute(sql_query)
    text_data = [row.Message for row in cursor.fetchall()]
    return text_data

def extract_ngrams(text_data, n_gram_range=(2, 2), stop_words="english"):
    count = CountVectorizer(ngram_range=n_gram_range, stop_words=stop_words).fit(text_data)
    candidates = count.get_feature_names_out()
    return candidates

def main():
    server = 'server'
    database = 'database '
    username = 'username '
    password = 'password '
    sql_query = "my_sp"
    
    conn = connect_to_database(server, database, username, password)
    text_data = fetch_text_data(conn, sql_query)
    conn.close()
    
    candidates = extract_ngrams(text_data)
    
    for candidate in candidates:
        print(candidate)

if __name__ == "__main__":
    main()

需求说明 #

我希望将这段Python代码的功能集成到C# .NET Core MVC或控制台程序中。之前尝试用ML.NET等C# NuGet包实现相同的文本短语提取，但输出精度不如这段Python代码。请问能否通过制作DLL或其他方式实现该需求？

可行实现方案 #

1. 用Python.NET直接调用Python代码 #

Python.NET允许C#直接托管Python运行时，调用脚本中的方法，步骤如下：

• 安装NuGet包：Install-Package Python.Runtime
• 确保Python环境已安装pyodbc和scikit-learn依赖（执行pip install pyodbc scikit-learn）
• C#调用示例：

using Python.Runtime;

class Program
{
    static void Main(string[] args)
    {
        // 初始化Python运行时
        PythonEngine.Initialize();

        using (Py.GIL())
        {
            // 加载目标Python脚本（需将脚本放在项目输出目录）
            var pyModule = Py.Import("text_extractor");
            
            // 调用数据库连接方法
            var dbConn = pyModule.connect_to_database("your_server", "your_db", "your_user", "your_pwd");
            
            // 获取文本数据
            var textData = pyModule.fetch_text_data(dbConn, "my_sp");
            
            // 提取二元短语
            var ngramCandidates = pyModule.extract_ngrams(textData);
            
            // 遍历输出结果
            foreach (var candidate in ngramCandidates)
            {
                Console.WriteLine(candidate.ToString());
            }
            
            // 关闭数据库连接
            dbConn.InvokeMethod("close");
        }

        // 关闭Python运行时
        PythonEngine.Shutdown();
    }
}

注意：需将Python脚本保存为text_extractor.py，并设置为复制到输出目录。

2. 封装Python代码为REST API，C#通过HTTP调用 #

如果不想在C#项目中嵌入Python运行时，可以将Python代码封装为轻量API服务，C#通过HTTP请求调用：

• Python端Flask服务示例：

from flask import Flask, request, jsonify
import pyodbc
from sklearn.feature_extraction.text import CountVectorizer

app = Flask(__name__)

def connect_to_database(server, database, username, password):
    return pyodbc.connect(
        f'DRIVER=ODBC Driver 17 for SQL Server;SERVER={server};DATABASE={database};UID={username};PWD={password}'
    )

def fetch_text_data(conn, sql_query):
    cursor = conn.cursor()
    cursor.execute(sql_query)
    return [row.Message for row in cursor.fetchall()]

def extract_ngrams(text_data, n_gram_range=(2,2), stop_words="english"):
    vectorizer = CountVectorizer(ngram_range=n_gram_range, stop_words=stop_words).fit(text_data)
    return vectorizer.get_feature_names_out().tolist()

@app.post("/extract-ngrams")
def process_request():
    req_data = request.get_json()
    conn = connect_to_database(
        req_data["server"],
        req_data["database"],
        req_data["username"],
        req_data["password"]
    )
    text_data = fetch_text_data(conn, req_data["sql_query"])
    conn.close()
    return jsonify(extract_ngrams(text_data))

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=False)

• C#端调用示例：

using System.Net.Http.Json;

class Program
{
    static async Task Main(string[] args)
    {
        using var httpClient = new HttpClient();
        var requestBody = new
        {
            server = "your_server",
            database = "your_db",
            username = "your_user",
            password = "your_pwd",
            sql_query = "my_sp"
        };
        
        var response = await httpClient.PostAsJsonAsync("http://localhost:5000/extract-ngrams", requestBody);
        var ngramCandidates = await response.Content.ReadFromJsonAsync<List<string>>();
        
        if (ngramCandidates != null)
        {
            foreach (var candidate in ngramCandidates)
            {
                Console.WriteLine(candidate);
            }
        }
    }
}

需确保Flask已安装：pip install flask。

3. 打包Python代码为可执行文件，C#调用进程 #

使用PyInstaller将Python代码打包为独立可执行文件，C#通过启动进程捕获输出：

• 打包命令：pyinstaller --onefile text_extractor.py
• 修改Python脚本支持命令行参数（用于动态传递数据库信息）：

import sys

def main():
    if len(sys.argv) != 6:
        print("Usage: text_extractor.exe <server> <database> <username> <password> <sql_query>")
        sys.exit(1)
    
    server = sys.argv[1]
    database = sys.argv[2]
    username = sys.argv[3]
    password = sys.argv[4]
    sql_query = sys.argv[5]
    
    # 原逻辑不变...

• C#调用示例：

using System.Diagnostics;

class Program
{
    static void Main(string[] args)
    {
        var processStartInfo = new ProcessStartInfo
        {
            FileName = "text_extractor.exe",
            Arguments = "\"your_server\" \"your_db\" \"your_user\" \"your_pwd\" \"my_sp\"",
            RedirectStandardOutput = true,
            UseShellExecute = false,
            CreateNoWindow = true
        };
        
        using var process = Process.Start(processStartInfo);
        if (process != null)
        {
            while (!process.StandardOutput.EndOfStream)
            {
                var line = process.StandardOutput.ReadLine();
                if (!string.IsNullOrEmpty(line))
                {
                    Console.WriteLine(line);
                }
            }
            process.WaitForExit();
        }
    }
}

内容的提问来源于stack exchange，提问作者Sanaullah Ansari-KI