2021.12.02 任务控制使用新方式

nheweijun
app/__init__.py
app/modules/data/io/data_download_task.py
app/modules/data/io/data_entry_by_meta.py
app/modules/data/manager/table_refresh.py
app/modules/data/manager/table_vacuate.py
app/modules/data/manager/table_vacuate_one.py
app/modules/data/models/__init__.py
app/modules/service/image/image_build_pyramid.py
app/util/component/TaskController.py
--- a/app/__init__.py
查看文件 @ec3d2c0
+++ b/app/__init__.py
查看文件 @ec3d2c0
@@ -79,12 +79,9 @@ def create_app():
     # 日志
     logging.basicConfig(level=configure.log_level)
-    log_file = os.path.join(os.path.dirname(os.path.dirname(
-        os.path.realpath(__file__))), "logs", "log.txt")
-    handler = logging.FileHandler(
-        log_file, encoding='UTF-8')   # 设置日志字符集和存储路径名字
-    logging_format = logging.Formatter(
-        '[%(levelname)s] %(asctime)s %(message)s')
+    log_file = os.path.join(os.path.dirname(os.path.dirname(os.path.realpath(__file__))), "logs", "log.txt")
+    handler = logging.FileHandler(log_file, encoding='UTF-8')   # 设置日志字符集和存储路径名字
+    logging_format = logging.Formatter('[%(levelname)s] %(asctime)s %(message)s')
     handler.setFormatter(logging_format)
     app.logger.addHandler(handler)
@@ -97,16 +94,13 @@ def create_app():
             app.register_blueprint(api.bp)
     # 入库监测线程
-    @app.before_first_request
-    def data_entry_process():
-        StructurePrint().print("start listen")
-        process = threading.Thread(target=data_entry_center)
-        process.start()
+    # @app.before_first_request
+    # def data_entry_process():
+    #     StructurePrint().print("start listen")
+    #     process = threading.Thread(target=data_entry_center)
+    #     process.start()
     # 不检测https
     os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'
-
-
-
     return app
--- a/app/modules/data/io/data_download_task.py
查看文件 @ec3d2c0
+++ b/app/modules/data/io/data_download_task.py
查看文件 @ec3d2c0
@@ -19,6 +19,7 @@ from app.util.component.ZipUtil import ZipUtil
 from app.util.component.StructuredPrint import StructurePrint
 import multiprocessing
 import datetime
+from app.util.component.TaskController import TaskController
 class Api(ApiTemplate):
@@ -69,6 +70,9 @@ class Api(ApiTemplate):
         try:
+            #任务控制,等待执行
+            TaskController.wait(task_guid)
+
             sys_session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
             table_names = para.get("table_name").split(",")
--- a/app/modules/data/io/data_entry_by_meta.py
查看文件 @ec3d2c0
+++ b/app/modules/data/io/data_entry_by_meta.py
查看文件 @ec3d2c0
@@ -13,8 +13,13 @@ import json
 import re
 from app.util.component.ApiTemplate import ApiTemplate
 from app.util.component.PGUtil import PGUtil
+from app.util.component.StructuredPrint import StructurePrint
+from sqlalchemy.orm import Session
+import configure
 import datetime
-
+import multiprocessing
+from app.util.component.EntryDataVacuate import EntryDataVacuate
+from app.util.component.TaskController import TaskController
 class Api(ApiTemplate):
@@ -68,7 +73,11 @@ class Api(ApiTemplate):
                     except:
                        print("关闭数据库失败！")
                 return res
-    
+
+            entry_thread = multiprocessing.Process(target=self.entry,args=(self.para.get("task_guid"),))
+            entry_thread.start()
+
+
             # 录入数据后台进程,录入主函数为entry
             # 初始化task
             task = Task(guid=self.para.get("task_guid"),
@@ -81,10 +90,12 @@ class Api(ApiTemplate):
                         database_guid=self.para.get("database_guid"),
                         catalog_guid=self.para.get("catalog_guid"),
                         process="等待入库",
-                        parameter=json.dumps(self.para))
+                        parameter=json.dumps(self.para),
+                        task_pid=entry_thread.pid)
             db.session.add(task)
             db.session.commit()
-    
+
+
             res["result"] = True
             res["msg"] = "数据录入提交成功！"
             res["data"] = self.para["task_guid"]
@@ -92,6 +103,81 @@ class Api(ApiTemplate):
             raise e
         return res
+    def entry(self,task_guid):
+
+        sys_session: Session = None
+        this_task_layer = []
+        try:
+            #任务控制,等待执行
+            TaskController.wait(task_guid)
+
+            sys_session: Session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
+
+
+            task:Task = sys_session.query(Task).filter_by(guid=task_guid).one_or_none()
+            parameter = json.loads(task.parameter)
+
+            task.state = 2
+            task.process = "入库中"
+            sys_session.commit()
+
+
+            #处理修改入库信息
+
+            metas: list = json.loads(parameter.get("meta").__str__())
+            parameter["meta"] = metas
+
+            database = sys_session.query(Database).filter_by(guid=task.database_guid).one_or_none()
+            pg_ds: DataSource = PGUtil.open_pg_data_source(1, DES.decode(database.sqlalchemy_uri))
+
+
+            for meta in metas:
+                overwrite = parameter.get("overwrite", "no")
+
+                for layer_name_origin, layer_name in meta.get("layer").items():
+                    origin_name = layer_name
+                    no = 1
+
+                    while (overwrite.__eq__("no") and pg_ds.GetLayerByName(layer_name)) or sys_session.query(
+                            InsertingLayerName).filter_by(name=layer_name).one_or_none():
+                        layer_name = origin_name + "_{}".format(no)
+                        no += 1
+
+                    # 添加到正在入库的列表中
+                    iln = InsertingLayerName(guid=uuid.uuid1().__str__(),
+                                             task_guid=task.guid,
+                                             name=layer_name)
+
+                    sys_session.add(iln)
+                    sys_session.commit()
+                    this_task_layer.append(layer_name)
+                    # 修改表名
+                    meta["layer"][layer_name_origin] = layer_name
+            pg_ds.Destroy()
+
+            #入库
+            EntryDataVacuate().entry(parameter)
+
+            #完成后
+            for ln in this_task_layer:
+                iln = sys_session.query(InsertingLayerName).filter_by(name=ln).one_or_none()
+                sys_session.delete(iln)
+
+        except Exception as e:
+            sys_session.query(Task).filter_by(guid=task_guid).update(
+                {"state": -1, "process": "入库失败"})
+
+            for ln in this_task_layer:
+                iln = sys_session.query(InsertingLayerName).filter_by(name=ln).one_or_none()
+                sys_session.delete(iln)
+
+            sys_session.commit()
+            StructurePrint().print(e.__str__(), "error")
+        finally:
+            sys_session.commit()
+            if sys_session:
+                sys_session.close()
+
     api_doc={
     "tags":["IO接口"],
--- a/app/modules/data/manager/table_refresh.py
查看文件 @ec3d2c0
+++ b/app/modules/data/manager/table_refresh.py
查看文件 @ec3d2c0
@@ -18,7 +18,7 @@ from app.util.component.PGUtil import PGUtil
 from app.util.component.StructuredPrint import StructurePrint
 from app.util.component.ApiTemplate import ApiTemplate
 from app.util.component.GeometryAdapter import GeometryAdapter
-
+from app.util.component.TaskController import TaskController
 import multiprocessing
 import configure
@@ -72,6 +72,10 @@ class Api(ApiTemplate):
         db_tuple = PGUtil.get_info_from_sqlachemy_uri(DES.decode(database.sqlalchemy_uri))
         try:
+
+            #任务控制,等待执行
+            TaskController.wait(task_guid)
+
             sys_session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
             sys_ds = PGUtil.open_pg_data_source(0,configure.SQLALCHEMY_DATABASE_URI)
--- a/app/modules/data/manager/table_vacuate.py
查看文件 @ec3d2c0
+++ b/app/modules/data/manager/table_vacuate.py
查看文件 @ec3d2c0
@@ -17,7 +17,7 @@ from app.util.component.PGUtil import PGUtil
 from app.util.component.VacuateConf import VacuateConf
 from app.util.component.GeometryAdapter import GeometryAdapter
-
+from app.util.component.TaskController import TaskController
 from osgeo.ogr import DataSource,Layer,Geometry
 from osgeo import ogr
@@ -105,15 +105,18 @@ class Api(ApiTemplate):
         pg_ds = None
         vacuate_process = None
         try:
-            sys_session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
-            sys_session.query(Table).filter_by(guid=table.guid).update({"is_vacuate": 2, "update_time": datetime.datetime.now()})
+            #任务控制,等待执行
+            TaskController.wait(task_guid)
+
+            sys_session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
+            sys_session.query(Table).filter_by(guid=table.guid).update({"is_vacuate": 2, "update_time": datetime.datetime.now()})
+            sys_session.query(Task).filter_by(guid=task_guid).update({"state":2,"process":"精华中"})
             database = sys_session.query(Database).filter_by(guid=table.database_guid).one_or_none()
             database_sqlalchemy_uri = str(database.sqlalchemy_uri)
             pg_session = PGUtil.get_db_session(DES.decode(database.sqlalchemy_uri))
-
             pg_ds :DataSource= PGUtil.open_pg_data_source(0,DES.decode(database.sqlalchemy_uri))
@@ -126,6 +129,7 @@ class Api(ApiTemplate):
             #长时间的连接，导致后续的session超时，现在先断开
             sys_session.close()
+
             # 创建抽稀过程
             options = ["OVERWRITE=yes", "GEOMETRY_NAME={}".format(PGUtil.get_geo_column(table.name,pg_session)),
                        "PRECISION=NO"]
--- a/app/modules/data/manager/table_vacuate_one.py
查看文件 @ec3d2c0
+++ b/app/modules/data/manager/table_vacuate_one.py
查看文件 @ec3d2c0
@@ -17,6 +17,7 @@ from osgeo.ogr import DataSource,Layer,Geometry
 from osgeo import ogr
 from app.util.component.VacuateConf import VacuateConf
+from app.util.component.TaskController import TaskController
 class Api(ApiTemplate):
     api_name = "单独抽稀"
@@ -106,16 +107,19 @@ class Api(ApiTemplate):
         origin_vacuate = table.is_vacuate
         try:
+
+            #任务控制,等待执行
+            TaskController.wait(task_guid)
+
             sys_session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
             sys_session.query(Table).filter_by(guid=table.guid).update(
                 {"is_vacuate": 2, "update_time": datetime.datetime.now()})
+            sys_session.query(Task).filter_by(guid=task_guid).update({"state":2,"process":"精华中"})
-            database = sys_session.query(Database).filter_by(guid=table.database_guid).one_or_none()
+            database = sys_session.query(Database).filter_by(guid=table.database_guid).one_or_none()
             database_sqlalchemy_uri = str(database.sqlalchemy_uri)
-
             pg_session = PGUtil.get_db_session(DES.decode(database_sqlalchemy_uri))
-
             pg_ds :DataSource= PGUtil.open_pg_data_source(1,DES.decode(database_sqlalchemy_uri))
             #删除原有数据
--- a/app/modules/data/models/__init__.py
查看文件 @ec3d2c0
+++ b/app/modules/data/models/__init__.py
查看文件 @ec3d2c0
@@ -103,6 +103,10 @@ class Task(db.Model):
     process = Column(Text)
     create_time = Column(DateTime)
     update_time = Column(DateTime)
+    #0等待
+    #1完成
+    #-1失败
+    #2正在进行中
     state = Column(Integer)
     #数据源外键
     database_guid = Column(String(256), ForeignKey('dmap_database.guid'))
--- a/app/modules/service/image/image_build_pyramid.py
查看文件 @ec3d2c0
+++ b/app/modules/service/image/image_build_pyramid.py
查看文件 @ec3d2c0
@@ -16,10 +16,11 @@ import configure
 from app.util.component.PGUtil import PGUtil
 from .util.Cache import Cache
 import json
-from osgeo import gdal,osr
+from osgeo import gdal
 from osgeo.gdal import *
 import traceback
 import os
+from app.util.component.TaskController import TaskController
 class Api(ApiTemplate):
@@ -77,6 +78,11 @@ class Api(ApiTemplate):
     def build_pyramid_task(self,image_guid,task_guid,data_servers,path):
         sys_session = None
         try:
+
+
+            #任务控制,等待执行
+            TaskController.wait(task_guid)
+
             sys_session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
             #进入创建金字塔的状态
             sys_session.query(Image).filter_by(guid=image_guid).update({"has_pyramid": -1})
--- a/app/util/component/TaskController.py 0 → 100644
查看文件 @ec3d2c0
+++ b/app/util/component/TaskController.py 0 → 100644
查看文件 @ec3d2c0
+# coding=utf-8
+#author:        4N
+#createtime:    2021/12/2
+#email:         nheweijun@sina.com
+
+from app.modules.data.models import Task
+from app.util.component.StructuredPrint import StructurePrint
+import configure
+import time
+from app.util.component.PGUtil import PGUtil
+
+class TaskController:
+    '''
+    此类是面向切面的类，控制各种任务是否可以进入执行状态
+    '''
+
+    @classmethod
+    def pass_check(cls,sys_session,task_guid):
+
+        # 任务类型
+        # 1：入库任务
+        # 2：抽稀任务
+        # 3：数据库刷新任务
+        # 4：数据下载任务
+        # 5：影像金字塔任务
+
+        task = sys_session.query(Task).filter_by(guid=task_guid).one_or_none()
+
+        check = True
+        if task.task_type==1:
+            #控制正在运行的任务数量
+            running_count = sys_session.query(Task).filter(Task.state == 2).filter(Task.task_type == 1).count()
+            first_task = sys_session.query(Task).filter(Task.state == 0).filter(Task.task_type == 1).order_by(Task.create_time).first()
+            if (running_count > configure.entry_data_thread) or (not first_task.guid.__eq__(task_guid)):
+                StructurePrint().print("等待入库")
+                check = False
+        if task.task_type==2:
+            running_count = sys_session.query(Task).filter(Task.state == 2).filter(Task.task_type == 2).count()
+            first_task = sys_session.query(Task).filter(Task.state == 0).filter(Task.task_type == 2).order_by(Task.create_time).first()
+            if (running_count > configure.entry_data_thread) or (not first_task.guid.__eq__(task_guid)):
+                StructurePrint().print("等待精化")
+                check = False
+        if task.task_type==3:
+            pass
+        if task.task_type==4:
+            pass
+        if task.task_type==5:
+            pass
+
+        return check
+
+    @classmethod
+    def wait(cls,task_guid):
+        sys_session= PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
+        while not cls.pass_check(sys_session,task_guid):
+            time.sleep(3)
+        sys_session.close()