PCA_Test.py

@@ -7,7 +7,8 @@ PCA source code
 
 import numpy as np
 import json
-import pymysql
+import dmPython
+from urllib.parse import urlparse, parse_qs
 import requests
 import config
 
@@ -145,32 +146,45 @@ class MSSQL:
         self.conn.close()
 
 
-class MySQL:
+def _parse_dm_jdbc_url(jdbc_url):
-    def __init__(self, host, user, pwd, database):
+    if not jdbc_url:
-        self.host = host
+        raise NameError("没有设置数据库URL")
+    url = jdbc_url
+    if url.startswith("jdbc:"):
+        url = url[len("jdbc:"):]
+    parsed = urlparse(url)
+    host = parsed.hostname
+    port = parsed.port or 5236
+    schema = None
+    if parsed.query:
+        params = parse_qs(parsed.query)
+        schema_list = params.get("schema") or params.get("SCHEMA")
+        if schema_list:
+            schema = schema_list[0]
+    return host, port, schema
+
+
+class DaMeng:
+    def __init__(self, jdbc_url, user, pwd):
+        self.jdbc_url = jdbc_url
         self.user = user
         self.pwd = pwd
-        self.db = database
+        self.conn = None
 
     def __GetConnect(self):
-        """建立 MySQL 连接并返回 cursor"""
+        """建立达梦连接并返回 cursor"""
-        if not self.db:
+        host, port, schema = _parse_dm_jdbc_url(self.jdbc_url)
-            raise NameError("没有设置数据库信息")
+        self.conn = dmPython.connect(
-
-        self.conn = pymysql.connect(
-            host=self.host,
-            port=config._PORT,            # MySQL 端口，例：3306
             user=self.user,
             password=self.pwd,
-            database=self.db,
+            server=host,
-            charset="utf8mb4",
+            port=port
-            cursorclass=pymysql.cursors.Cursor
         )
-
         cur = self.conn.cursor()
         if not cur:
             raise NameError("连接数据库失败")
-
+        if schema:
+            cur.execute(f"SET SCHEMA {schema}")
         return cur
 
     def ExecQuery(self, sql):
@@ -188,13 +202,13 @@ class MySQL:
         self.conn.commit()
         self.conn.close()
 def get_model_by_ID(model_id, version="v-test"):
-    ms = MySQL(host=config._SQL_IP, user="root", pwd="powerSIS#123", database="alert")
+    ms = DaMeng(jdbc_url=config._DM_URL, user=config._DM_USER, pwd=config._DM_PASSWORD)
     resList = ms.ExecQuery(f"SELECT Model_info FROM model_cfg where model_id={model_id}")
     return json.loads(resList[0][0])
 
 
 def get_model_by_id_and_version(model_id, version):
-    ms = MySQL(host=config._SQL_IP, user="root", pwd="powerSIS#123", database="alert")
+    ms = DaMeng(jdbc_url=config._DM_URL, user=config._DM_USER, pwd=config._DM_PASSWORD)
     resList = ms.ExecQuery(f"SELECT Model_info FROM model_version where model_id={model_id} and version='{version}'")
     return json.loads(resList[0][0])
 

config.py

@@ -3,4 +3,7 @@
 _SQL_IP = "120.26.116.243"  # 数据库IP
 _EXA_IP = "120.26.116.243"  # EXA IP
 _CLEAN_IP = "172.28.137.231:8098"  # EXA清洗接口IP
-_PORT = 3306  # port
+_PORT = 3306  # port
+_DM_URL = "jdbc:dm://120.26.116.243:5236?schema=ALERT"
+_DM_USER = "SYSDBA"
+_DM_PASSWORD = "powerSIS123"