Bladeren bron

知识服务[连接应用状态由“直接写入数据库”改为“查询时校验后返回”]

liangxunge 1 maand geleden
bovenliggende
commit
9318c270fd

+ 0 - 6
api/controllers/console/external_application.py

@@ -57,9 +57,6 @@ class ExternalApplicationListApi(Resource):
         parser.add_argument(
             "method", type=str, location="json", nullable=False, help="Invalid external_application method."
         )
-        parser.add_argument(
-            "status", type=bool, location="json", nullable=False, help="Invalid external_application status."
-        )
         args = parser.parse_args()
         external_application = ExternalApplicationService.save_external_application(args)
         return external_application, 200
@@ -94,9 +91,6 @@ class ExternalApplicationApi(Resource):
         parser.add_argument(
             "method", type=str, location="json", nullable=False, help="Invalid external_application method."
         )
-        parser.add_argument(
-            "status", type=bool, location="json", nullable=False, help="Invalid external_application status."
-        )
         args = parser.parse_args()
         external_application = ExternalApplicationService.update_external_application(args, external_application_id)
         return external_application, 200

+ 40 - 1
api/models/external_application.py

@@ -1,3 +1,7 @@
+import re
+from urllib.parse import urlparse
+
+import requests
 from sqlalchemy import func
 
 from .engine import db
@@ -15,8 +19,43 @@ class ExternalApplication(db.Model):
     type = db.Column(db.String(255), nullable=False)
     url = db.Column(db.String(255), nullable=False)
     method = db.Column(db.String(255), nullable=False)
-    status = db.Column(db.Boolean, nullable=False, server_default=db.text("true"))
+    # status = db.Column(db.Boolean, nullable=False, server_default=db.text("true"))
     created_by = db.Column(StringUUID, nullable=False)
     created_at = db.Column(db.DateTime, nullable=False, server_default=func.current_timestamp())
     updated_by = db.Column(StringUUID, nullable=True)
     updated_at = db.Column(db.DateTime, nullable=False, server_default=func.current_timestamp())
+
+    @property
+    def status(self):
+        # 0). 使用正则表达式验证URL格式
+        regex = re.compile(
+            r'^(?:http|ftp)s?://'  # http:// or https://
+            r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+(?:[A-Z]{2,6}\.?|[A-Z0-9-]{2,}\.?)|'  # domain...
+            r'localhost|'  # localhost...
+            r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}|'  # ...or ipv4
+            r'\[?[A-F0-9]*:[A-F0-9:]+\]?)'  # ...or ipv6
+            r'(?::\d+)?'  # optional port
+            r'(?:/?|[/?]\S+)$', re.IGNORECASE)
+        is_valid_format = re.match(regex, self.url) is not None
+        if not is_valid_format:
+            return False
+        # 1). 使用urllib库进行URL解析
+        parsed_url = urlparse(self.url)
+        # 2). 检查URL的scheme部分
+        is_valid_scheme = parsed_url.scheme in ["http", "https"]
+        if not is_valid_scheme:
+            return False
+        # 3). 检查URL的netloc部分
+        is_valid_netloc = parsed_url.netloc
+        if not is_valid_netloc:
+            return False
+        # 4). 使用requests库发送HTTP请求
+        try:
+            response = requests.get(self.url)
+            if response.status_code == 200:
+                return True
+            else:
+                return False
+        except requests.exceptions.RequestException:
+            return False
+

+ 0 - 2
api/services/external_application_service.py

@@ -54,7 +54,6 @@ class ExternalApplicationService:
             type=args["type"],
             url=args["url"],
             method=args["method"],
-            status=args["status"],
             created_by=current_user.id,
             updated_by=current_user.id,
         )
@@ -71,7 +70,6 @@ class ExternalApplicationService:
         external_application.type = args["type"]
         external_application.url = args["url"]
         external_application.method = args["method"]
-        external_application.status = args["status"]
         external_application.updated_by = current_user.id
         external_application.updated_at = datetime.now()
         db.session.add(external_application)