自定义本地 OCR

app.py

@@ -1,3 +1,6 @@
+import subprocess
+
+import requests
 from PyQt5.QtWidgets import *
 from PyQt5.QtCore import *
 from PyQt5.QtGui import *
@@ -17,6 +20,7 @@
 import utils.port
 import utils.update
 import utils.hwnd
+import utils.offline_ocr
 
 import ui.login
 import ui.register
@@ -184,19 +188,48 @@ def clickSettin(self) :
         self.range_ui.close()
         self.settin_ui.show()
 
+    # 本地 OCR 是否本机运行
+    def is_local_offline_ocr(self):
+        return not not self.yaml["offline_ocr_cmd"]
 
-    # 自动打开本地OCR
-    def autoOpenOfflineOCR(self) :
+    # 本地 OCR 是否运行
+    def is_offline_ocr_running(self):
+        try:
+            return requests.head(self.yaml["offline_ocr_url"], timeout=5).headers.get("Dango-OCR") == "OK"
+        except Exception:
+            return False
 
-        if not self.config["offlineOCR"] :
+    # 启动本地 OCR
+    def start_offline_ocr(self):
+        if not self.is_local_offline_ocr():
             return
-        if not utils.port.detectPort(self.yaml["port"]) :
-            try :
-                # 启动本地OCR
-                os.startfile(self.yaml["ocr_cmd_path"])
-            except Exception :
-                self.logger.error(format_exc())
-
+        args = ["START", "离线 OCR"]
+        cmd = self.yaml["offline_ocr_cmd"]
+
+        is_list = isinstance(cmd, list)
+        if is_list:
+            file = str(cmd[0])
+        else:
+            file = str(cmd)
+        _, ext = os.path.splitext(file)
+        if ext.lower() == ".cmd":
+            args.append("ConHost")
+        args.append(file)
+
+        if is_list:
+            for arg in cmd[1:]:
+                args.append(str(arg))
+        subprocess.Popen(args, shell=True)
+
+    # 自动打开本地 OCR
+    def autoOpenOfflineOCR(self):
+        if not self.config["offlineOCR"] or self.is_offline_ocr_running():
+            return
+        try:
+            # 启动本地 OCR
+            self.start_offline_ocr()
+        except Exception:
+            self.logger.error(format_exc())
 
     # 初始化资源
     def InitLoadFile(self) :

translator/ocr/dango.py

@@ -315,7 +315,7 @@ def dangoOCR(object, test=False) :
 
 
 # 本地OCR
-def offlineOCR(object) :
+def offlineOCR(object):
 
     image_path = os.path.join(os.getcwd(), "config", "image.jpg")
     # try :
@@ -324,14 +324,21 @@ def offlineOCR(object) :
     # except Exception :
     #     object.logger.error(format_exc())
 
-    url = "http://127.0.0.1:6666/ocr/api"
+    url = object.yaml["offline_ocr_url"]
     language = object.config["language"]
     body = {
-        "ImagePath": image_path,
         "Language": language
     }
 
-    res = utils.http.post(url, body, object.logger)
+    if object.is_local_offline_ocr():
+        body["ImagePath"] = image_path
+        res = utils.http.post(url, body, object.logger)
+    else:
+        with open(image_path, 'rb') as file:
+            res = utils.http.post(url, body, object.logger, files={
+                'Image': file
+            })
+
     if not res :
         return False, "本地OCR错误: 本地OCR所使用的端口可能被占用, 请重启电脑以释放端口后重试\n如果频繁出现, 建议切换其他OCR使用"
 

ui/settin.py

@@ -2203,35 +2203,35 @@ def resetSwitch(self, switch_type) :
             self.online_ocr_probation_switch.mousePressEvent(1)
             self.online_ocr_probation_switch.updateValue()
 
-
     # 运行本地OCR
-    def runOfflineOCR(self) :
+    def runOfflineOCR(self):
+        if not self.object.is_local_offline_ocr():
+            return
 
-        # 杀死本地可能在运行ocr进程
-        utils.offline_ocr.killOfflineOCR(self.object.yaml["port"])
+        # 杀死可能在运行的本地 OCR
+        utils.offline_ocr.killOfflineOCR(self.object)
 
-        # 检查端口是否被占用
-        if utils.port.detectPort(self.object.yaml["port"]) :
+        # 本地 OCR 是否运行
+        if self.object.is_offline_ocr_running():
             utils.message.MessageBox("本地OCR运行失败",
                                      "本地OCR已启动, 请不要重复运行!     ")
-        else :
-            try :
-                # 启动本地OCR
-                os.startfile(self.object.yaml["ocr_cmd_path"])
-            except FileNotFoundError :
+        else:
+            try:
+                # 启动本地 OCR
+                self.object.start_offline_ocr()
+            except FileNotFoundError:
                 utils.message.MessageBox("本地OCR运行失败",
                                          "本地OCR还未安装, 请先安装!     ")
-            except Exception :
+            except Exception:
                 self.logger.error(format_exc())
                 utils.message.MessageBox("本地OCR运行失败",
                                          "原因: %s"%format_exc())
 
-
     # 测试本地OCR
-    def testOfflineOCR(self) :
+    def testOfflineOCR(self):
 
         # 检查端口是否被占用
-        if not utils.port.detectPort(self.object.yaml["port"]) :
+        if not self.object.is_offline_ocr_running():
             utils.message.MessageBox("测试失败",
                                      "本地OCR还没运行成功，不可以进行测试     \n"
                                      "请先启动本地OCR, 并保证其运行正常")

ui/switch.py

@@ -227,7 +227,7 @@ def updateValue(self):
     def mousePressEvent(self, event) :
 
         # 通过检查端口占用校验本地OCR是否运行成功
-        sign = detectPort(self.object.yaml["port"])
+        sign = self.object.is_offline_ocr_running()
         if not self.checked and not sign :
             MessageBox("本地OCR使用失败", "请先运行本地OCR, 待运行成功后再打开此开关\n"
                                           "若运行时间较长或者运行失败, 可通过交流群联系客服协助      \n"

utils/config.py

@@ -24,8 +24,10 @@ def openConfig(logger) :
             "user": "",
             "password": "",
             "dict_info_url": "https://dango.c4a15wh.cn/DangoTranslate/ShowDict",
-            "ocr_cmd_path": ".\ocr\startOCR.cmd",
-            "port": 6666,
+            # 本地 OCR 接口地址
+            "offline_ocr_url": "http://127.0.0.1:6666/ocr/api",
+            # 本地 OCR 启动命令，可为：字符串、数组、False。当 False 时，表示不启动
+            "offline_ocr_cmd": ".\\ocr\\startOCR.cmd",
         }
 
     # 2022.02.19 添加新参数

utils/http.py

@@ -6,42 +6,41 @@
 
 
 # 发送http请求
-def post(url, body, logger, headers=None, timeout=5) :
+def post(url, body, logger, headers=None, timeout=5, files=None):
 
     proxies = {
         "http": None,
         "https": None
     }
     result = {}
 
-    try :
+    try:
         # 消除https警告
         requests.packages.urllib3.disable_warnings()
-    except Exception :
+    except Exception:
         pass
 
     response = None
-    try :
-        if headers :
-            response = requests.post(url, headers=headers, data=json.dumps(body), proxies=proxies, verify=False, timeout=timeout)
-        else :
-            response = requests.post(url, data=json.dumps(body), proxies=proxies, verify=False, timeout=timeout)
-        try :
+    try:
+        if not files:
+            body = json.dumps(body)
+        response = requests.post(url, headers=headers, data=body, files=files, proxies=proxies,
+                                 verify=False, timeout=timeout)
+        try:
             response.encoding = "utf-8"
             result = json.loads(response.text)
             response.close()
-        except Exception :
+        except Exception:
             response.encoding = "gb18030"
             result = json.loads(response.text)
-    except json.decoder.JSONDecodeError :
-        try :
-            logger.error("post %s error, httpcode is %s, response is %s"%(url, response.status_code, response.text))
-        except Exception :
+    except json.decoder.JSONDecodeError:
+        try:
+            logger.error("post %s error, httpcode is %s, response is %s" % (url, response.status_code, response.text))
+        except Exception:
             pass
-    except Exception :
+    except Exception:
         logger.error(format_exc())
 
-
     return result
 
 

utils/offline_ocr.py

@@ -1,3 +1,5 @@
+import subprocess
+
 from PyQt5.QtCore import *
 from traceback import format_exc
 import shutil
@@ -15,20 +17,27 @@
 
 
 # 安装本地ocr
-def install_offline_ocr(object) :
+def install_offline_ocr(object):
+
+    cmd = object.yaml["offline_ocr_cmd"]
+    if cmd is False:
+        return
+
+    if isinstance(cmd, list) and len(cmd):
+        cmd = cmd[0]
 
     # 判断本地ocr是否已安装
-    if os.path.exists(object.yaml["ocr_cmd_path"]) :
+    if os.path.exists(cmd):
         utils.message.MessageBox("安装失败",
                                  "本地OCR已安装, 不需要重复安装!     ")
         return
 
-    try :
+    try:
         # 杀死本地ocr进程
-        killOfflineOCR(object.yaml["port"])
+        object.kill_offline_ocr_is_running()
         # 删除旧文件
         shutil.rmtree(OCR_INSTALL_PATH)
-    except Exception :
+    except Exception:
         object.logger.error(format_exc())
 
     object.settin_ui.progress_bar.finish_sign = False
@@ -44,12 +53,37 @@ def install_offline_ocr(object) :
 
 
 # 杀死本地ocr进程
-def killOfflineOCR(port) :
-
-    popen_content = os.popen("netstat -ano |findstr %s"%port).read()
-    if popen_content :
-        pid = popen_content.split(" ")[-1]
-        os.popen("taskkill /f /t /im %s"%pid)
+def killOfflineOCR(object):
+    # path = ".\\ocr\\startOCR.cmd"
+    # path = os.path.realpath(path)
+    # path = path.replace("\\", "\\\\")
+    # subprocess.Popen(["WMIC", "PROCESS", "WHERE", "ExecutablePath='%s'" % path, "CALL", "Terminate"], shell=True)\
+    #     .wait()
+
+    if not object.is_local_offline_ocr():
+        return
+    cmd = object.yaml["offline_ocr_cmd"]
+    if not isinstance(cmd, list):
+        cmd = [cmd]
+    _, ext = os.path.splitext(str(cmd[0]))
+    is_con_host = ext.lower() == ".cmd"
+
+    arr = []
+    for v in cmd:
+        v = str(v)
+        if " " in v:
+            v = "\"" + v + "\""
+        arr.append(v.replace("\\", "\\\\"))
+
+    cl = arr[0]
+    if len(arr) > 1:
+        if is_con_host:
+            space = ' '
+        else:
+            space = '  '
+        cl += space + ' '.join(arr[1:])
+    subprocess.Popen(["WMIC", "PROCESS", "WHERE", "CommandLine='%s'" % cl, "CALL", "Terminate"], shell=True)\
+        .wait()
 
 
 # 是否卸载本地ocr
@@ -67,10 +101,10 @@ def whether_uninstall_offline_ocr(object) :
 
 
 # 卸载本地ocr
-def uninstall_offline_ocr(object) :
+def uninstall_offline_ocr(object):
 
     # 杀死本地ocr进程解除占用
-    killOfflineOCR(object.yaml["port"])
+    utils.offline_ocr.killOfflineOCR(object)
 
     # 删除本地ocr
     try:

utils/port.py

@@ -1,15 +1,15 @@
 import socket
 
 
-def detectPort(port=6666) :
+def detectPort(port=6666):
 
-    s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
     try:
         s.connect(("127.0.0.1", int(port)))
         s.shutdown(2)
-        sign =  True
-    except Exception :
+        sign = True
+    except Exception:
         sign = False
     s.close()
 
-    return sign
+    return sign