Merge branch 'master' of http://gitlab.ctune.cn/weijunh/DMapManager

@@ -12,15 +12,12 @@ from app.models import db
 from app.modules.auth.oauth2 import config_oauth, myCodeIDToken
 from app.modules.auth.oauth2 import config_oauth, myCodeIDToken
 from flasgger import Swagger
 from flasgger import Swagger
 import logging
 import logging
-from app.util.component.EntryData import EntryData
-from app.util.component.EntryDataVacuate import EntryDataVacuate
-import threading
 from app.util.component.StructuredPrint import StructurePrint
 from app.util.component.StructuredPrint import StructurePrint
 from app.util.component.PGUtil import PGUtil
 from app.util.component.PGUtil import PGUtil
 import os
 import os
-from app.modules.data.io.data_entry_center import data_entry_center
 from app.modules.monitor.schedule import start_schedule
 from app.modules.monitor.schedule import start_schedule
 
 
+
 class JSONEncoder(_JSONEncoder):
 class JSONEncoder(_JSONEncoder):
     """
     """
     因为decimal不能序列化，增加Flask对decimal类的解析
     因为decimal不能序列化，增加Flask对decimal类的解析
@@ -94,13 +91,6 @@ def create_app():
         for api in find_class(scan, BlueprintApi):
         for api in find_class(scan, BlueprintApi):
             app.register_blueprint(api.bp)
             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()
-
     # 不检测https
     # 不检测https
     os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'
     os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'
 
 

@@ -20,9 +20,12 @@ from app.util.component.StructuredPrint import StructurePrint
 import multiprocessing
 import multiprocessing
 import datetime
 import datetime
 from app.util.component.TaskController import TaskController
 from app.util.component.TaskController import TaskController
+from app.util.component.TaskWriter import TaskWriter
 
 
 class Api(ApiTemplate):
 class Api(ApiTemplate):
 
 
+    api_name = "数据下载"
+
     def process(self):
     def process(self):
 
 
         res = {}
         res = {}
@@ -47,7 +50,6 @@ class Api(ApiTemplate):
             db.session.add(task)
             db.session.add(task)
             db.session.commit()
             db.session.commit()
 
 
-
             res["data"] = "下载任务已提交!"
             res["data"] = "下载任务已提交!"
             res["result"] = True
             res["result"] = True
 
 
@@ -58,8 +60,8 @@ class Api(ApiTemplate):
 
 
     def download(self,task_guid,para):
     def download(self,task_guid,para):
 
 
-        sys_session = None
         ds: DataSource = None
         ds: DataSource = None
+        task_writer = None
 
 
         # 设置编码
         # 设置编码
         encoding = para.get("encoding")
         encoding = para.get("encoding")
@@ -72,12 +74,13 @@ class Api(ApiTemplate):
 
 
             #任务控制,等待执行
             #任务控制,等待执行
             TaskController.wait(task_guid)
             TaskController.wait(task_guid)
-
-            sys_session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
+            task_writer = TaskWriter(task_guid)
+            task_writer.update_task({"state":2,"update_time":datetime.datetime.now(),"process" : "下载中"})
+            task_writer.update_process("开始下载...")
 
 
             table_names = para.get("table_name").split(",")
             table_names = para.get("table_name").split(",")
             database_guid = para.get("database_guid")
             database_guid = para.get("database_guid")
-            database = sys_session.query(Database).filter_by(guid=database_guid).one_or_none()
+            database = task_writer.session.query(Database).filter_by(guid=database_guid).one_or_none()
             if not database:
             if not database:
                 raise Exception("数据库不存在！")
                 raise Exception("数据库不存在！")
 
 
@@ -89,36 +92,25 @@ class Api(ApiTemplate):
             if download_type.__eq__("shp"):
             if download_type.__eq__("shp"):
                 data = self.download_shp(table_names, ds)
                 data = self.download_shp(table_names, ds)
             if download_type.__eq__("gdb"):
             if download_type.__eq__("gdb"):
-                data = self.download_gdb(sys_session,table_names, ds, database_guid)
-
-            sys_session.query(Task).filter_by(guid=task_guid).update({"state":1,"update_time":datetime.datetime.now(),
-                                                                      "process" : "下载完成",
-                                                                      "parameter":data[0]["download_url"]})
-            sys_session.commit()
+                data = self.download_gdb(task_writer.session,table_names, ds, database_guid)
 
 
+            task_writer.update_task({"state":1,"update_time":datetime.datetime.now(),"process" : "下载完成","parameter":data[0]["download_url"]})
+            task_writer.update_process("下载完成！")
 
 
         except Exception as e:
         except Exception as e:
             try:
             try:
-                sys_session.query(Task).filter_by(guid=task_guid).update({"state": -1,"update_time":datetime.datetime.now(),
-                                                                          "process": "下载失败"})
-
-                message = "{} {}".format(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), e.__str__())
-                task_process_guid = uuid.uuid1().__str__()
-                task_process = Process(guid=task_process_guid, message=message, time=datetime.datetime.now(),
-                                       task_guid=task_guid)
-                sys_session.add(task_process)
-                sys_session.commit()
+                task_writer.update_task({"state": -1,"update_time":datetime.datetime.now(),"process": "下载失败"})
+                task_writer.update_process( e.__str__())
             except Exception as ee:
             except Exception as ee:
-                print(traceback.format_exc())
+                StructurePrint().print(ee.__str__())
             raise e
             raise e
         finally:
         finally:
             try:
             try:
                 if ds:
                 if ds:
                     ds.Destroy()
                     ds.Destroy()
-                if sys_session:
-                    sys_session.close()
-            except:
-                print(traceback.format_exc())
+                task_writer.close()
+            except Exception as e:
+                StructurePrint().print(e.__str__())
     
     
     
     
     def download_shp(self,table_names,ds):
     def download_shp(self,table_names,ds):
@@ -154,10 +146,7 @@ class Api(ApiTemplate):
                 StructurePrint().print("{}图层已下载{}个对象".format(table_name, count))
                 StructurePrint().print("{}图层已下载{}个对象".format(table_name, count))
 
 
         data_source.Destroy()
         data_source.Destroy()
-
-
         ZipUtil.create_zip(os.path.join(parent, "file_tmp", table_name+"_"+uuid_) + ".zip", [dirpath])
         ZipUtil.create_zip(os.path.join(parent, "file_tmp", table_name+"_"+uuid_) + ".zip", [dirpath])
-
         return "http://" + configure.deploy_ip_host + "/API/IO/Download/{}".format(table_name+"_"+uuid_ + ".zip")
         return "http://" + configure.deploy_ip_host + "/API/IO/Download/{}".format(table_name+"_"+uuid_ + ".zip")
     
     
     
     
@@ -175,8 +164,7 @@ class Api(ApiTemplate):
         gdb_path = os.path.join(parent, "file_tmp", uuid_+".gdb")
         gdb_path = os.path.join(parent, "file_tmp", uuid_+".gdb")
 
 
         gdb_ds: DataSource = gdb_driver.CreateDataSource(gdb_path)
         gdb_ds: DataSource = gdb_driver.CreateDataSource(gdb_path)
-    
-    
+
         for table_name in table_names:
         for table_name in table_names:
     
     
             layer: Layer = ds.GetLayerByName(table_name)
             layer: Layer = ds.GetLayerByName(table_name)
@@ -196,7 +184,6 @@ class Api(ApiTemplate):
             # schema = layer.schema
             # schema = layer.schema
             pg_layer.CreateFields(schema)
             pg_layer.CreateFields(schema)
 
 
-
             # gdb 不支持fid=0的要素，所以识别到后要+1
             # gdb 不支持fid=0的要素，所以识别到后要+1
             offset = 0
             offset = 0
             f1:Feature =  layer.GetNextFeature()
             f1:Feature =  layer.GetNextFeature()
@@ -218,8 +205,6 @@ class Api(ApiTemplate):
 
 
     
     
         return data
         return data
-    
-    
 
 
 
 
     api_doc={
     api_doc={
@@ -259,4 +244,7 @@ class Api(ApiTemplate):
             }
             }
             }
             }
         }
         }
-}
+
+}
+
+

@@ -18,9 +18,9 @@ from sqlalchemy.orm import Session
 import configure
 import configure
 import datetime
 import datetime
 import multiprocessing
 import multiprocessing
-from app.util.component.EntryDataVacuate import EntryDataVacuate
+from ..util.EntryDataVacuate import EntryDataVacuate
 from app.util.component.TaskController import TaskController
 from app.util.component.TaskController import TaskController
-
+from app.util.component.TaskWriter import TaskWriter
 class Api(ApiTemplate):
 class Api(ApiTemplate):
 
 
     api_name = "通过meta入库"
     api_name = "通过meta入库"
@@ -105,29 +105,23 @@ class Api(ApiTemplate):
 
 
     def entry(self,task_guid):
     def entry(self,task_guid):
 
 
-        sys_session: Session = None
+        task_writer = None
         this_task_layer = []
         this_task_layer = []
         try:
         try:
             #任务控制,等待执行
             #任务控制,等待执行
             TaskController.wait(task_guid)
             TaskController.wait(task_guid)
 
 
-            sys_session: Session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
-
+            task_writer = TaskWriter(task_guid)
 
 
-            task:Task = sys_session.query(Task).filter_by(guid=task_guid).one_or_none()
+            task:Task = task_writer.session.query(Task).filter_by(guid=task_guid).one_or_none()
             parameter = json.loads(task.parameter)
             parameter = json.loads(task.parameter)
-
-            task.state = 2
-            task.process = "入库中"
-            sys_session.commit()
-
+            task_writer.update_task({"state": 2, "prcess": "入库中"})
 
 
             #处理修改入库信息
             #处理修改入库信息
-
             metas: list = json.loads(parameter.get("meta").__str__())
             metas: list = json.loads(parameter.get("meta").__str__())
             parameter["meta"] = metas
             parameter["meta"] = metas
 
 
-            database = sys_session.query(Database).filter_by(guid=task.database_guid).one_or_none()
+            database = task_writer.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))
             pg_ds: DataSource = PGUtil.open_pg_data_source(1, DES.decode(database.sqlalchemy_uri))
 
 
 
 
@@ -148,8 +142,8 @@ class Api(ApiTemplate):
                                              task_guid=task.guid,
                                              task_guid=task.guid,
                                              name=layer_name)
                                              name=layer_name)
 
 
-                    sys_session.add(iln)
-                    sys_session.commit()
+                    task_writer.session.add(iln)
+                    task_writer.session.commit()
                     this_task_layer.append(layer_name)
                     this_task_layer.append(layer_name)
                     # 修改表名
                     # 修改表名
                     meta["layer"][layer_name_origin] = layer_name
                     meta["layer"][layer_name_origin] = layer_name
@@ -160,23 +154,18 @@ class Api(ApiTemplate):
 
 
             #完成后
             #完成后
             for ln in this_task_layer:
             for ln in this_task_layer:
-                iln = sys_session.query(InsertingLayerName).filter_by(name=ln).one_or_none()
-                sys_session.delete(iln)
+                iln = task_writer.session.query(InsertingLayerName).filter_by(name=ln).one_or_none()
+                task_writer.session.delete(iln)
 
 
         except Exception as e:
         except Exception as e:
-            sys_session.query(Task).filter_by(guid=task_guid).update(
-                {"state": -1, "process": "入库失败"})
-
+            task_writer.update_task({"state": -1, "process": "入库失败"})
             for ln in this_task_layer:
             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()
+                iln = task_writer.session.query(InsertingLayerName).filter_by(name=ln).one_or_none()
+                task_writer.session.delete(iln)
             StructurePrint().print(e.__str__(), "error")
             StructurePrint().print(e.__str__(), "error")
         finally:
         finally:
-            sys_session.commit()
-            if sys_session:
-                sys_session.close()
+            task_writer.session.commit()
+            task_writer.close()
 
 
 
 
     api_doc={
     api_doc={

-# coding=utf-8
-#author:        4N
-#createtime:    2021/9/14
-#email:         nheweijun@sina.com
-
-import configure
-from ..models import InsertingLayerName, Database, Task,DES
-
-from sqlalchemy.orm import Session
-import multiprocessing
-from app.util.component.EntryDataVacuate import EntryDataVacuate
-import json
-from sqlalchemy import distinct
-import uuid
-from osgeo.ogr import DataSource
-from app.util.component.StructuredPrint import StructurePrint
-from app.util.component.PGUtil import PGUtil
-import time
-
-def data_entry_center():
-    running_dict = {}
-    sys_session: Session = PGUtil.get_db_session(
-        configure.SQLALCHEMY_DATABASE_URI)
-
-    while True:
-
-        try:
-            time.sleep(3)
-
-            # 已经结束的进程 从监测中删除
-            remove_process = []
-            for process, layer_names in running_dict.items():
-                if not process.is_alive():
-                    for l in layer_names:
-                        inserted = sys_session.query(
-                            InsertingLayerName).filter_by(name=l).one_or_none()
-                        if inserted:
-                            sys_session.delete(inserted)
-                    sys_session.commit()
-                    remove_process.append(process)
-            for process in remove_process:
-                running_dict.pop(process)
-
-
-            # 入库进程少于阈值，开启入库进程
-            inter_size = sys_session.query(
-                distinct(InsertingLayerName.task_guid)).count()
-
-            if inter_size < configure.entry_data_thread:
-                # 锁表
-                ready_task: Task = sys_session.query(Task).filter_by(state=0, task_type=1).order_by(
-                    Task.create_time).with_lockmode("update").limit(1).one_or_none()
-                if ready_task:
-
-                    try:
-                        parameter = json.loads(ready_task.parameter)
-                        StructurePrint().print("检测到入库任务")
-                        ready_task.state = 2
-                        ready_task.process = "入库中"
-                        sys_session.commit()
-
-                        metas: list = json.loads(
-                            parameter.get("meta").__str__())
-                        parameter["meta"] = metas
-
-                        database = sys_session.query(Database).filter_by(
-                            guid=ready_task.database_guid).one_or_none()
-                        pg_ds: DataSource = PGUtil.open_pg_data_source(
-                            1, DES.decode(database.sqlalchemy_uri))
-
-                        this_task_layer = []
-                        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=ready_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()
-                        entry_data_process = multiprocessing.Process(
-                            target=EntryDataVacuate().entry, args=(parameter,))
-                        entry_data_process.start()
-
-                        pid = entry_data_process.pid
-                        sys_session.query(Task).filter_by(guid=ready_task.guid).update({"task_pid":pid})
-                        sys_session.commit()
-
-                        running_dict[entry_data_process] = this_task_layer
-
-                    except Exception as e:
-                        sys_session.query(Task).filter_by(guid=ready_task.guid).update(
-                            {"state": -1, "process": "入库失败"})
-                        sys_session.commit()
-                        StructurePrint().print(e.__str__(), "error")
-                else:
-                    # 解表啊
-                    sys_session.commit()
-        except Exception as e:
-            sys_session.commit()
-            StructurePrint().print(e.__str__(), "error")

@@ -13,7 +13,7 @@ import json
 from app.util.component.ApiTemplate import ApiTemplate
 from app.util.component.ApiTemplate import ApiTemplate
 from .get_meta import Api as MetaApi
 from .get_meta import Api as MetaApi
 from threading import Thread
 from threading import Thread
-from app.util.component.EntryDataVacuate import EntryDataVacuate
+from ..util.EntryDataVacuate import EntryDataVacuate
 
 
 class Api(ApiTemplate):
 class Api(ApiTemplate):
 
 

@@ -5,14 +5,11 @@ from osgeo import gdal
 import os
 import os
 import uuid
 import uuid
 import shutil
 import shutil
-import time
 from app.modules.data.models import *
 from app.modules.data.models import *
 from app.util.component.PGUtil import PGUtil
 from app.util.component.PGUtil import PGUtil
 from app.util.component.StructuredPrint import StructurePrint
 from app.util.component.StructuredPrint import StructurePrint
 from sqlalchemy.orm import Session
 from sqlalchemy.orm import Session
 import configure
 import configure
-import math
-from functools import lru_cache
 import traceback
 import traceback
 import copy
 import copy
 from app.util.component.GeometryAdapter import GeometryAdapter
 from app.util.component.GeometryAdapter import GeometryAdapter

@@ -19,6 +19,7 @@ from app.util.component.StructuredPrint import StructurePrint
 from app.util.component.ApiTemplate import ApiTemplate
 from app.util.component.ApiTemplate import ApiTemplate
 from app.util.component.GeometryAdapter import GeometryAdapter
 from app.util.component.GeometryAdapter import GeometryAdapter
 from app.util.component.TaskController import TaskController
 from app.util.component.TaskController import TaskController
+from app.util.component.TaskWriter import TaskWriter
 import multiprocessing
 import multiprocessing
 import configure
 import configure
 
 
@@ -69,12 +70,17 @@ class Api(ApiTemplate):
         data_session=None
         data_session=None
         result = {}
         result = {}
         sys_session = None
         sys_session = None
+        task_writer = None
+
+
         db_tuple = PGUtil.get_info_from_sqlachemy_uri(DES.decode(database.sqlalchemy_uri))
         db_tuple = PGUtil.get_info_from_sqlachemy_uri(DES.decode(database.sqlalchemy_uri))
 
 
         try:
         try:
 
 
             #任务控制,等待执行
             #任务控制,等待执行
             TaskController.wait(task_guid)
             TaskController.wait(task_guid)
+            task_writer = TaskWriter(task_guid)
+
 
 
             sys_session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
             sys_session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
             sys_ds = PGUtil.open_pg_data_source(0,configure.SQLALCHEMY_DATABASE_URI)
             sys_ds = PGUtil.open_pg_data_source(0,configure.SQLALCHEMY_DATABASE_URI)
@@ -145,24 +151,19 @@ class Api(ApiTemplate):
             sys_session.commit()
             sys_session.commit()
             result["data"] = "刷新数据成功！"
             result["data"] = "刷新数据成功！"
             result["state"] = 1
             result["state"] = 1
-            sys_session.query(Task).filter_by(guid=task_guid).update(
-                {"state": 1, "update_time": datetime.datetime.now(),"process":"更新成功"})
-            sys_session.commit()
+
+            task_writer.update_task({"state": 1, "update_time": datetime.datetime.now(),"process":"更新成功"})
+
 
 
         except Exception as e:
         except Exception as e:
             try:
             try:
-                print(traceback.format_exc())
-                sys_session.query(Task).filter_by(guid=task_guid).update(
-                    {"state": -1, "update_time": datetime.datetime.now(),"process":"更新失败"})
-                message = "{} {}".format(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), e.__str__())
-                task_process_guid = uuid.uuid1().__str__()
-                task_process = Process(guid=task_process_guid, message=message, time=datetime.datetime.now(),
-                                       task_guid=task_guid)
-                sys_session.add(task_process)
-                sys_session.commit()
+                StructurePrint().print(traceback.format_exc())
+                task_writer.update_task({"state": -1, "update_time": datetime.datetime.now(),"process":"更新失败"})
+                task_writer.update_process(e.__str__())
             except Exception as ee:
             except Exception as ee:
-                print(traceback.format_exc())
+                StructurePrint().print(traceback.format_exc())
         finally:
         finally:
+            task_writer.close()
             if pg_ds:
             if pg_ds:
                 pg_ds.Destroy()
                 pg_ds.Destroy()
             if data_session:
             if data_session:

@@ -18,6 +18,7 @@ from app.util.component.VacuateConf import VacuateConf
 from app.util.component.GeometryAdapter import GeometryAdapter
 from app.util.component.GeometryAdapter import GeometryAdapter
 
 
 from app.util.component.TaskController import TaskController
 from app.util.component.TaskController import TaskController
+from app.util.component.TaskWriter import TaskWriter
 from osgeo.ogr import DataSource,Layer,Geometry
 from osgeo.ogr import DataSource,Layer,Geometry
 from osgeo import ogr
 from osgeo import ogr
 
 
@@ -100,7 +101,7 @@ class Api(ApiTemplate):
 
 
     def task(self,table,task_guid):
     def task(self,table,task_guid):
 
 
-        sys_session = None
+        task_writer = None
         pg_session = None
         pg_session = None
         pg_ds = None
         pg_ds = None
         vacuate_process = None
         vacuate_process = None
@@ -109,25 +110,22 @@ class Api(ApiTemplate):
             #任务控制,等待执行
             #任务控制,等待执行
             TaskController.wait(task_guid)
             TaskController.wait(task_guid)
 
 
+            task_writer = TaskWriter(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":"精华中"})
+            task_writer.update_table(table.guid,{"is_vacuate": 2, "update_time": datetime.datetime.now()})
+            task_writer.update_task({"state":2,"process":"精华中"})
 
 
-            database = sys_session.query(Database).filter_by(guid=table.database_guid).one_or_none()
+            database = task_writer.session.query(Database).filter_by(guid=table.database_guid).one_or_none()
             database_sqlalchemy_uri = str(database.sqlalchemy_uri)
             database_sqlalchemy_uri = str(database.sqlalchemy_uri)
+
             pg_session = PGUtil.get_db_session(DES.decode(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))
             pg_ds :DataSource= PGUtil.open_pg_data_source(0,DES.decode(database.sqlalchemy_uri))
 
 
-
             #删除原有数据
             #删除原有数据
-            tvs = sys_session.query(TableVacuate).filter_by(table_guid=table.guid).all()
+            tvs = task_writer.session.query(TableVacuate).filter_by(table_guid=table.guid).all()
             for tv in tvs :
             for tv in tvs :
-                sys_session.delete(tv)
-
-            sys_session.commit()
-            #长时间的连接，导致后续的session超时，现在先断开
-            sys_session.close()
+                task_writer.session.delete(tv)
+            task_writer.session.commit()
 
 
 
 
             # 创建抽稀过程
             # 创建抽稀过程
@@ -150,9 +148,6 @@ class Api(ApiTemplate):
 
 
             vacuate_process.set_vacuate_count()
             vacuate_process.set_vacuate_count()
 
 
-            # 重连
-            sys_session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
-
             #新增
             #新增
             if configure.VACUATE_DB_URI:
             if configure.VACUATE_DB_URI:
                 user, passwd, host, port, datab = PGUtil.get_info_from_sqlachemy_uri(configure.VACUATE_DB_URI)
                 user, passwd, host, port, datab = PGUtil.get_info_from_sqlachemy_uri(configure.VACUATE_DB_URI)
@@ -169,41 +164,30 @@ class Api(ApiTemplate):
                                              name=vacuate_process.vacuate_layers[l].GetName(),
                                              name=vacuate_process.vacuate_layers[l].GetName(),
                                              pixel_distance=vacuate_process.this_gridsize[l],
                                              pixel_distance=vacuate_process.this_gridsize[l],
                                              connectstr=DES.encode(connectstr))
                                              connectstr=DES.encode(connectstr))
-                sys_session.add(table_vacuate)
+                task_writer.session.add(table_vacuate)
 
 
-            sys_session.query(Task).filter_by(guid=task_guid).update({"state":1,"update_time":datetime.datetime.now(),
-                                                                      "process": "精化完成"})
-            sys_session.query(Table).filter_by(guid=table.guid).update(
-                {"is_vacuate": 1, "update_time": datetime.datetime.now()})
-            sys_session.commit()
+            task_writer.update_task({"state":1,"update_time":datetime.datetime.now(),"process": "精化完成"})
+            task_writer.update_table(table.guid, {"is_vacuate": 1, "update_time": datetime.datetime.now()})
 
 
         except Exception as e:
         except Exception as e:
             try:
             try:
-                if not sys_session.is_active:
-                    sys_session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
-                sys_session.query(Task).filter_by(guid=task_guid).update({"state": -1,"update_time":datetime.datetime.now(),
-                                                                          "process": "精化失败"})
-                sys_session.query(Table).filter_by(guid=table.guid).update(
-                    {"is_vacuate": 0, "update_time": datetime.datetime.now()})
-
-                message = "{} {}".format(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), e.__str__())
-                task_process_guid = uuid.uuid1().__str__()
-                task_process = Process(guid=task_process_guid, message=message, time=datetime.datetime.now(),
-                                       task_guid=task_guid)
-                sys_session.add(task_process)
-                sys_session.commit()
+                task_writer.update_task({"state": -1,"update_time":datetime.datetime.now(),"process": "精化失败"})
+                task_writer.update_table(table.guid, {"is_vacuate": 0, "update_time": datetime.datetime.now()})
+                task_writer.update_process( e.__str__())
+
+                task_writer.session.commit()
+
                 if vacuate_process:
                 if vacuate_process:
                     vacuate_process.rollback()
                     vacuate_process.rollback()
 
 
-                print(traceback.format_exc())
+                StructurePrint().print(traceback.format_exc())
             except Exception as ee:
             except Exception as ee:
-                print(traceback.format_exc())
+                StructurePrint().print(traceback.format_exc())
         finally:
         finally:
             try:
             try:
+                task_writer.close()
                 if vacuate_process:
                 if vacuate_process:
                     vacuate_process.end()
                     vacuate_process.end()
-                if sys_session:
-                    sys_session.close()
                 if pg_session:
                 if pg_session:
                     pg_session.close()
                     pg_session.close()
                 if pg_ds:
                 if pg_ds:
@@ -233,15 +217,6 @@ class Api(ApiTemplate):
         }
         }
     }
     }
 
 
-class SysSession:
-    def __init__(self):
-        self.session : Session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
-
-    def close(self):
-        try:
-            self.session.close()
-        except:
-            pass
 
 
 class VacuateProcess:
 class VacuateProcess:
 
 

@@ -5,10 +5,9 @@
 
 
 import datetime
 import datetime
 import traceback
 import traceback
-from ..models import Table, Database, DES,Task,db,TableVacuate
+from ..models import Table, Database, DES,Task,db,TableVacuate,Process
 from app.util.component.ApiTemplate import ApiTemplate
 from app.util.component.ApiTemplate import ApiTemplate
 from app.util.component.PGUtil import PGUtil
 from app.util.component.PGUtil import PGUtil
-from app.util.component.EntryDataVacuate import Process
 from app.util.component.StructuredPrint import StructurePrint
 from app.util.component.StructuredPrint import StructurePrint
 import multiprocessing
 import multiprocessing
 import uuid
 import uuid
@@ -18,6 +17,7 @@ from osgeo import ogr
 
 
 from app.util.component.VacuateConf import VacuateConf
 from app.util.component.VacuateConf import VacuateConf
 from app.util.component.TaskController import TaskController
 from app.util.component.TaskController import TaskController
+from app.util.component.TaskWriter import TaskWriter
 
 
 class Api(ApiTemplate):
 class Api(ApiTemplate):
     api_name = "单独抽稀"
     api_name = "单独抽稀"
@@ -100,7 +100,7 @@ class Api(ApiTemplate):
 
 
     def task(self,table,task_guid,grids):
     def task(self,table,task_guid,grids):
 
 
-        sys_session = None
+        task_write = None
         pg_session = None
         pg_session = None
         pg_ds = None
         pg_ds = None
         vacuate_process = None
         vacuate_process = None
@@ -110,27 +110,23 @@ class Api(ApiTemplate):
 
 
             #任务控制,等待执行
             #任务控制,等待执行
             TaskController.wait(task_guid)
             TaskController.wait(task_guid)
+            task_write =  TaskWriter(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":"精华中"})
+            task_write.update_table(table.guid,{"is_vacuate": 2, "update_time": datetime.datetime.now()})
+            task_write.update_task({"state":2,"process":"精华中"})
 
 
-
-            database = sys_session.query(Database).filter_by(guid=table.database_guid).one_or_none()
+            database = task_write.session.query(Database).filter_by(guid=table.database_guid).one_or_none()
             database_sqlalchemy_uri = str(database.sqlalchemy_uri)
             database_sqlalchemy_uri = str(database.sqlalchemy_uri)
             pg_session = PGUtil.get_db_session(DES.decode(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))
             pg_ds :DataSource= PGUtil.open_pg_data_source(1,DES.decode(database_sqlalchemy_uri))
 
 
             #删除原有数据
             #删除原有数据
             for grid in grids:
             for grid in grids:
-                tvs = sys_session.query(TableVacuate).filter_by(pixel_distance=grid,table_guid=table.guid).all()
+                tvs = task_write.session.query(TableVacuate).filter_by(pixel_distance=grid,table_guid=table.guid).all()
                 for tv in tvs :
                 for tv in tvs :
-                    sys_session.delete(tv)
+                    task_write.session.delete(tv)
+            task_write.session.commit()
 
 
-            sys_session.commit()
-            #长时间的连接，导致后续的session超时，现在先断开
-            sys_session.close()
 
 
             # 创建抽稀过程
             # 创建抽稀过程
             options = ["OVERWRITE=yes", "GEOMETRY_NAME={}".format(PGUtil.get_geo_column(table.name,pg_session)),
             options = ["OVERWRITE=yes", "GEOMETRY_NAME={}".format(PGUtil.get_geo_column(table.name,pg_session)),
@@ -152,8 +148,6 @@ class Api(ApiTemplate):
 
 
             vacuate_process.set_vacuate_count()
             vacuate_process.set_vacuate_count()
 
 
-            #重新连接
-            sys_session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
 
 
             #新增
             #新增
             if configure.VACUATE_DB_URI:
             if configure.VACUATE_DB_URI:
@@ -172,41 +166,26 @@ class Api(ApiTemplate):
                                              name=layer_name,
                                              name=layer_name,
                                              pixel_distance=vacuate_process.this_gridsize[l],
                                              pixel_distance=vacuate_process.this_gridsize[l],
                                              connectstr=DES.encode(connectstr))
                                              connectstr=DES.encode(connectstr))
-                sys_session.add(table_vacuate)
+                task_write.session.add(table_vacuate)
+            task_write.update_task({"state":1,"update_time":datetime.datetime.now(),"process" : "精化完成"})
+            task_write.update_table(table.guid,{"is_vacuate": 1, "update_time": datetime.datetime.now()})
 
 
-            sys_session.query(Task).filter_by(guid=task_guid).update({"state":1,"update_time":datetime.datetime.now(),
-                                                                      "process" : "精化完成"})
-            sys_session.query(Table).filter_by(guid=table.guid).update(
-                {"is_vacuate": 1, "update_time": datetime.datetime.now()})
-            sys_session.commit()
 
 
         except Exception as e:
         except Exception as e:
             try:
             try:
-                if not sys_session.is_active:
-                    sys_session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
-                sys_session.query(Task).filter_by(guid=task_guid).update({"state": -1,"update_time":datetime.datetime.now(),
-                                                                          "process": "精化失败"})
-                sys_session.query(Table).filter_by(guid=table.guid).update(
-                    {"is_vacuate": origin_vacuate, "update_time": datetime.datetime.now()})
-
-                message = "{} {}".format(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), e.__str__())
-                task_process_guid = uuid.uuid1().__str__()
-                task_process = Process(guid=task_process_guid, message=message, time=datetime.datetime.now(),
-                                       task_guid=task_guid)
-                sys_session.add(task_process)
-                sys_session.commit()
+                task_write.update_task({"state": -1,"update_time":datetime.datetime.now(),"process": "精化失败"})
+                task_write.update_table(table.guid, {"is_vacuate": origin_vacuate, "update_time": datetime.datetime.now()})
+                task_write.update_process( e.__str__())
                 if vacuate_process:
                 if vacuate_process:
                     vacuate_process.rollback()
                     vacuate_process.rollback()
-
                 print(traceback.format_exc())
                 print(traceback.format_exc())
             except Exception as ee:
             except Exception as ee:
                 print(traceback.format_exc())
                 print(traceback.format_exc())
         finally:
         finally:
             try:
             try:
+                task_write.close()
                 if vacuate_process:
                 if vacuate_process:
                     vacuate_process.end()
                     vacuate_process.end()
-                if sys_session:
-                    sys_session.close()
                 if pg_session:
                 if pg_session:
                     pg_session.close()
                     pg_session.close()
                 if pg_ds:
                 if pg_ds:

@@ -4,7 +4,8 @@
 #email:         nheweijun@sina.com
 #email:         nheweijun@sina.com
 
 
 from app.models import db
 from app.models import db
-from sqlalchemy import Column, Integer, String, ForeignKey, Text, DateTime, Time,Float,Binary
+from sqlalchemy import Column, Integer, String, ForeignKey, Text, DateTime, Time,Float,Binary,Sequence
+from sqlalchemy import Sequence
 from sqlalchemy.orm import relationship
 from sqlalchemy.orm import relationship
 import base64
 import base64
 from pyDes import *
 from pyDes import *

@@ -29,6 +29,7 @@ class Api(ApiTemplate):
                     exec_result = os.popen('taskkill.exe /pid:' + str(pid))
                     exec_result = os.popen('taskkill.exe /pid:' + str(pid))
 
 
                 else:
                 else:
+                    #分布式下，很难吧，还需要记录机器，分布式是不可行的
                     os.kill(pid,signal.SIGILL)
                     os.kill(pid,signal.SIGILL)
             except Exception as e:
             except Exception as e:
                 StructurePrint.print("Kill task 失败")
                 StructurePrint.print("Kill task 失败")

@@ -144,7 +144,10 @@ class EntryDataVacuate:
 
 
 
 
         for i in range(ds.GetLayerCount()):
         for i in range(ds.GetLayerCount()):
+
             layer: Layer = ds.GetLayer(i)
             layer: Layer = ds.GetLayer(i)
+            if layer.GetName().lower() not in meta.get("layer").keys():
+                continue
             is_success, new_layer_name = self.entry_one_layer(layer,this_task,meta)
             is_success, new_layer_name = self.entry_one_layer(layer,this_task,meta)
             new_layer_names.append(new_layer_name)
             new_layer_names.append(new_layer_name)
             is_successes.append(is_success)
             is_successes.append(is_success)
@@ -330,7 +333,12 @@ class ThisTask:
                       is_vacuate=is_vacuate
                       is_vacuate=is_vacuate
                       )
                       )
         # 删除遗留业务数据
         # 删除遗留业务数据
-        history_table = self.sys_session.query(Table).filter_by(name=new_layer_name,database_guid=self.database.guid).all()
+        try:
+            history_table = self.sys_session.query(Table).filter_by(name=new_layer_name,database_guid=self.database.guid).all()
+        except:
+            self.sys_session: Session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
+            history_table = self.sys_session.query(Table).filter_by(name=new_layer_name,database_guid=self.database.guid).all()
+
         if history_table:
         if history_table:
             for ht in history_table:
             for ht in history_table:
                 self.sys_session.delete(ht)
                 self.sys_session.delete(ht)

@@ -21,6 +21,7 @@ from osgeo.gdal import *
 import traceback
 import traceback
 import os
 import os
 from app.util.component.TaskController import TaskController
 from app.util.component.TaskController import TaskController
+from app.util.component.TaskWriter import TaskWriter
 
 
 class Api(ApiTemplate):
 class Api(ApiTemplate):
 
 
@@ -76,21 +77,17 @@ class Api(ApiTemplate):
         return res
         return res
 
 
     def build_pyramid_task(self,image_guid,task_guid,data_servers,path):
     def build_pyramid_task(self,image_guid,task_guid,data_servers,path):
-        sys_session = None
-        try:
 
 
+        task_writer = None
 
 
+        try:
             #任务控制,等待执行
             #任务控制,等待执行
             TaskController.wait(task_guid)
             TaskController.wait(task_guid)
+            task_writer = TaskWriter(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})
-            sys_session.query(Task).filter_by(guid=task_guid).update({"state": 2})
-
-            sys_session.commit()
-            sys_session.close()
-
+            task_guid.sys_session.query(Image).filter_by(guid=image_guid).update({"has_pyramid": -1})
+            task_writer.update_task({"state": 2})
 
 
             #所有数据节点的影像都要建金字塔
             #所有数据节点的影像都要建金字塔
             update_size = None
             update_size = None
@@ -109,11 +106,9 @@ class Api(ApiTemplate):
                     update_size = json.loads(thrift_connect.client.getInfo(path)).get("size")
                     update_size = json.loads(thrift_connect.client.getInfo(path)).get("size")
                     thrift_connect.close()
                     thrift_connect.close()
 
 
-            #重新连接，防止session超时
-            sys_session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
+            task_writer.update_task({"state":1,"update_time":datetime.datetime.now()})
 
 
-            sys_session.query(Task).filter_by(guid=task_guid).update({"state":1,"update_time":datetime.datetime.now()})
-            image:Image = sys_session.query(Image).filter_by(guid=image_guid).one_or_none()
+            image:Image = task_writer.session.query(Image).filter_by(guid=image_guid).one_or_none()
 
 
             if not overview_count:
             if not overview_count:
                 overview_count = 0
                 overview_count = 0
@@ -128,20 +123,12 @@ class Api(ApiTemplate):
             image.overview_count = overview_count
             image.overview_count = overview_count
             if update_size:
             if update_size:
                 image.size = update_size
                 image.size = update_size
-
         except:
         except:
-            sys_session.query(Image).filter_by(guid=image_guid).update({"has_pyramid": 0})
-            sys_session.query(Task).filter_by(guid=task_guid).update({"state": -1,"update_time":datetime.datetime.now()})
+            task_writer.session.query(Image).filter_by(guid=image_guid).update({"has_pyramid": 0})
+            task_writer.update_task({"state": -1,"update_time":datetime.datetime.now()})
         finally:
         finally:
-            sys_session.commit()
-            if sys_session:
-                try:
-                    sys_session.close()
-                except:
-                    pass
-
-
-
+            task_writer.session.commit()
+            task_writer.close()
 
 
     def buildOverview(self,path):
     def buildOverview(self,path):
         image: Dataset = gdal.Open(path, 1)
         image: Dataset = gdal.Open(path, 1)

@@ -10,8 +10,9 @@ import time
 from app.modules.service.models import ImageService
 from app.modules.service.models import ImageService
 from app.modules.service.models import TileScheme,Service
 from app.modules.service.models import TileScheme,Service
 from app.util.component.ModelVisitor import ModelVisitor
 from app.util.component.ModelVisitor import ModelVisitor
-
+from app.util.component.StructuredPrint import StructurePrint
 import json
 import json
+import traceback
 
 
 class Cache:
 class Cache:
 
 
@@ -29,17 +30,23 @@ class Cache:
             GLOBAL_DIC["zookeeper_updatetime"] = time.time()
             GLOBAL_DIC["zookeeper_updatetime"] = time.time()
         else:
         else:
             if not zoo.connected:
             if not zoo.connected:
-                zoo.start()
+                try:
+                    zoo.start()
+                except:
+                    pass
 
 
         # 更新zoo
         # 更新zoo
         if not  GLOBAL_DIC.get("zookeeper_updatetime"):
         if not  GLOBAL_DIC.get("zookeeper_updatetime"):
             GLOBAL_DIC["zookeeper_updatetime"] = time.time()
             GLOBAL_DIC["zookeeper_updatetime"] = time.time()
         if time.time() - GLOBAL_DIC["zookeeper_updatetime"] > 15:
         if time.time() - GLOBAL_DIC["zookeeper_updatetime"] > 15:
-            #释放
+            #释放,高并发下可行吗，线程安全问题
             try:
             try:
                 zoo.stop()
                 zoo.stop()
-            except:
-                pass
+                zoo.close()
+                del zoo
+            except Exception as e:
+                StructurePrint().print("关闭zoo失败")
+                StructurePrint().print(e.__str__())
             zoo: KazooClient = KazooClient(hosts=configure.zookeeper, timeout=1)
             zoo: KazooClient = KazooClient(hosts=configure.zookeeper, timeout=1)
             zoo.start()
             zoo.start()
             GLOBAL_DIC["zookeeper"] = zoo
             GLOBAL_DIC["zookeeper"] = zoo

@@ -8,6 +8,8 @@ from osgeo import gdal,ogr
 import uuid
 import uuid
 import time
 import time
 import os
 import os
+import json
+
 
 
 def get_info_from_sqlachemy_uri(uri):
 def get_info_from_sqlachemy_uri(uri):
     parts = uri.split(":")
     parts = uri.split(":")
@@ -39,121 +41,158 @@ def open_pg_data_source(iswrite, uri):
         raise Exception("打开数据源失败！")
         raise Exception("打开数据源失败！")
     return ds
     return ds
 
 
-def move(geo:Geometry,offx,offy):
 
 
-    g = geo.GetGeometryRef(0)
-    num = g.GetPointCount()
+def move_digui(coords,offx,offy):
+    if isinstance(coords[0],list):
+        for coord in coords:
+            move_digui(coord, offx, offy)
+    else :
+        coords[0] += offx
+        coords[1] += offy
 
 
-    coor = []
-    try:
-        for j in range(num):
-            point = g.GetPoint(j)
-            x = point[0]
-            y = point[1]
-            x += offx
-            y += offy
-            coor.append([x, y])
-        if num==1:
-            point = ogr.Geometry(ogr.wkbPoint)
-            point.AddPoint(coor[0][0], coor[0][1])
-            return point
-        elif coor[0].__eq__(coor[-1]):
-            ring = ogr.Geometry(ogr.wkbLinearRing)
-            for co in coor:
-                ring.AddPoint(co[0], co[1])
-            poly = ogr.Geometry(ogr.wkbPolygon)
-            poly.AddGeometry(ring)
-            return poly
-        else :
-            line = ogr.Geometry(ogr.wkbLineString)
-            for co in coor:
-                line.AddPoint(co[0], co[1])
-            return line
-    except:
-        return None
+def move(geo,offx,offy):
 
 
 
 
-def copydata():
+    geojson_str = geo.ExportToJson()
+    geojson = json.loads(geojson_str)
 
 
+    coords = geojson["coordinates"]
+    move_digui(coords, offx, offy)
+    og = ogr.CreateGeometryFromJson(json.dumps(geojson))
 
 
-    bound = ""
+    return og
 
 
-    work_dir =r"E:\D2\桌面\FSSD_RES_PY_FWM"
 
 
-    data_path=r"E:\D2\桌面\FSSD_RES_PY_FWM\fs4326.shp"
+def copydata(bound,input,output):
+    work_dir = os.path.dirname(os.path.abspath(__file__))
 
 
-    base_name="fs900w"
+    shp_driver: Driver = ogr.GetDriverByName("ESRI Shapefile")
+    bound_ds : DataSource = shp_driver.Open(bound, 0)
+    bound_layer :Layer = bound_ds.GetLayer(0)
 
 
-    driver: Driver = ogr.GetDriverByName("ESRI Shapefile")
-    ds: DataSource = driver.Open(data_path, 1)
+    bf = bound_layer.GetNextFeature()
+    bound_geom = bf.GetGeometryRef()
 
 
-    bound_ds : DataSource = driver.Open(bound, 1)
-    bound_layer :Layer = bound_ds.GetLayer(0)
-    bound_geom = bound_layer.GetNextFeature().GetGeometryRef()
 
 
 
 
-    if not ds:
+    gdb_driver: Driver = ogr.GetDriverByName("FileGDB")
+    # gdb_ds: DataSource = gdb_driver.Open(gdb, 0)
+    # layer: Layer = gdb_ds.GetLayerByName(layer_name)
+
+    input_ds: DataSource = shp_driver.Open(input, 0)
+    layer: Layer = input_ds.GetLayer(0)
+
+    if not layer:
         raise Exception("打开数据失败！")
         raise Exception("打开数据失败！")
 
 
-    layer: Layer = ds.GetLayer(0)
+
+    output_ds: DataSource = gdb_driver.CreateDataSource(output)
 
 
 
 
+    gdb_layer:Layer = output_ds.CreateLayer(layer.GetName(),layer.GetSpatialRef(),layer.GetGeomType())
 
 
-    schema = layer.schema
-    schema = [s for s in schema if not s.GetName().lower().__eq__("objectid")]
-    for s in schema :
-        if s.GetName().lower().__eq__("objectid") and not s.GetTypeName().__eq__("Interger"):
-            s.SetType(ogr.OFTInteger)
+    gdb_layer.CreateFields(layer.schema)
 
 
 
 
 
 
-    gdb_driver:Driver = ogr.GetDriverByName("FileGDB")
-    gdb_ds: DataSource = gdb_driver.CreateDataSource(os.path.join(work_dir,"{}.gdb".format(base_name)))
-    gdb_layer:Layer = gdb_ds.CreateLayer(base_name,  layer.GetSpatialRef(),layer.GetGeomType())
 
 
-    # gdb_layer.CreateFields(schema)
+    #
+    bound_exent = bound_layer.GetExtent()
 
 
-    print(gdb_layer.GetFIDColumn())
+    layer_extent = layer.GetExtent()
+
+    xxrange=layer_extent[1]-layer_extent[0]
+    yrange = layer_extent[3]-layer_extent[2]
+
+
+    left = (int((layer_extent[0] - bound_exent[0]) / xxrange) + 1) * (-1)
+    right = int((bound_exent[1] - layer_extent[1]) / xxrange) + 1
+    up = int((bound_exent[3] - layer_extent[3]) / yrange) + 1
+    down =  (int((layer_extent[2] - bound_exent[2]) / yrange) + 1) * (-1)
 
 
-    extent = layer.GetExtent()
-    xxrange=extent[1]-extent[0]
-    yrange = extent[3]-extent[2]
-    feature_defn: FeatureDefn = layer.GetLayerDefn()
 
 
     begin = time.time()
     begin = time.time()
-    count=0
-    work_dir =os.path.dirname(os.path.abspath(__file__))
+    count=1
+
 
 
     for f in layer:
     for f in layer:
 
 
-        if count%10000==0:
-            print(count)
-            with open(os.path.join(work_dir, "copy.txt"), "w") as fi:
-                fi.write("已完成{}".format(count))
-        g:Geometry=f.GetGeometryRef()
-        new_f:Feature = copy.copy(f)
-        for xt in range(3):
-            for yt in range(3):
-                out_g = move(g,xxrange*xt,yrange*yt)
-                new_f.SetGeometry(out_g)
-                new_f.SetFID(count)
-                dd:FieldDefn=new_f.GetField("OBJECTID")
-                # new_f.UnsetField("OBJECTID")
-                gdb_layer.CreateFeature(new_f)
-                count += 1
+        new_f = copy.copy(f)
+        for xt in range(left,right+1,1):
+            for yt in range(down,up+1,1):
+                out_g = move(f.GetGeometryRef(),xxrange*xt,yrange*yt)
+
+                if out_g.Intersect(bound_geom):
+                    out_g = out_g.Intersection(bound_geom)
+                    new_f.SetGeometry(out_g)
+                    new_f.SetFID(count)
+                    gdb_layer.CreateFeature(new_f)
+                    count += 1
+                    if count % 10000 == 0:
+                        # print(count)
+                        with open(os.path.join(work_dir, "copy.txt"), "w") as fi:
+                            fi.write("已完成{}".format(count))
 
 
     print(time.time()-begin)
     print(time.time()-begin)
 
 
-    gdb_ds.Destroy()
+    input_ds.Destroy()
 
 
+def envelop_2_polygon(env):
+    ring = ogr.Geometry(ogr.wkbLinearRing)
+    ring.AddPoint(env[0], env[2])
+    ring.AddPoint(env[0], env[3])
+    ring.AddPoint(env[1], env[3])
+    ring.AddPoint(env[1], env[2])
+    ring.AddPoint(env[0], env[2])
+    # Create polygon
+    poly = ogr.Geometry(ogr.wkbPolygon)
+    poly.AddGeometry(ring)
+    return poly
 
 
+def clip():
 
 
-if __name__ == '__main__':
-    bound_shp = ""
-    gdb = ""
-    layer_name = ""
+    input = r"E:\Data\copy\x.shp"
+    output = r"E:\Data\copy\xianorigin.shp"
+    gdal.SetConfigOption("SHAPE_ENCODING", "UTF-8")
+    shp_driver: Driver = ogr.GetDriverByName("ESRI Shapefile")
 
 
-    copydata()
+    input_ds: DataSource = shp_driver.Open(input, 0)
+    layer: Layer = input_ds.GetLayer(0)
+
+    bound_geo = envelop_2_polygon([112.84,112.952,22.914,22.985])
+
+    if not layer:
+        raise Exception("打开数据失败！")
+
+    output_ds: DataSource = shp_driver.CreateDataSource(output)
+    #
+    # out_layer: Layer = output_ds.CreateLayer(layer.GetName(), layer.GetSpatialRef(), layer.GetGeomType())
+    # out_layer.CreateFields(layer.schema)
+    for f in layer:
+        new_f:Feature = copy.copy(f)
+        out_g = f.GetGeometryRef()
+        if out_g.Intersect(bound_geo):
+            out_g = out_g.Intersection(bound_geo)
+            new_f.SetGeometry(out_g)
+            new_f.SetField("ttt",1)
+            # new_f.UnsetField("OBJECTID")
+            # out_layer.CreateFeature(new_f)
+
+    # output_ds.Destroy()
+
+
+
+if __name__ == '__main__':
+    # bound_shp = "E:\Data\广东边界\gdsample.shp"
+    # input = r"E:\Data\copy\origin.shp"
+    # output = r"E:\Data\copy\re.shp"
+    # copydata(bound_shp,input,output)
+    #
+    # bound_shp = "/root/CopyData/gdsample.shp"
+    # input = r"/root/CopyData/origin.shp"
+    # output = r"/root/CopyData/re.gdb"
+    # copydata(bound_shp,input,output)
+    clip()
 
 
 
 
 
 

-# coding=utf-8
-#author:        4N
-#createtime:    2021/6/11
-#email:         nheweijun@sina.com
-import copy
-from osgeo.ogr import *
-from osgeo import gdal,ogr
-import uuid
-import time
-import os
-import json
-
-
-def get_info_from_sqlachemy_uri(uri):
-    parts = uri.split(":")
-    user = parts[1][2:]
-
-    password_list = parts[2].split("@")
-    if password_list.__len__() > 2:
-        password = "@".join(password_list[:-1])
-    else:
-        password = parts[2].split("@")[0]
-    host = parts[2].split("@")[-1]
-    port = parts[3].split("/")[0]
-    database = parts[3].split("/")[1]
-
-    return user, password, host, port, database
-
-def open_pg_data_source(iswrite, uri):
-    """
-    # 获取PostGIS数据源
-    :return:
-    """
-    db_conn_tuple = get_info_from_sqlachemy_uri(uri)
-    fn = "PG: user=%s password=%s host=%s port=%s dbname=%s " % db_conn_tuple
-    driver = ogr.GetDriverByName("PostgreSQL")
-    if driver is None:
-        raise Exception("打开PostgreSQL驱动失败，可能是当前GDAL未支持PostgreSQL驱动！")
-    ds = driver.Open(fn, iswrite)
-    if ds is None:
-        raise Exception("打开数据源失败！")
-    return ds
-
-
-def move_digui(coords,offx,offy):
-    if isinstance(coords[0],list):
-        for coord in coords:
-            move_digui(coord, offx, offy)
-    else :
-        coords[0] += offx
-        coords[1] += offy
-
-def move(geo,offx,offy):
-
-
-    geojson_str = geo.ExportToJson()
-    geojson = json.loads(geojson_str)
-
-    coords = geojson["coordinates"]
-    move_digui(coords, offx, offy)
-    og = ogr.CreateGeometryFromJson(json.dumps(geojson))
-
-    return og
-
-
-def copydata(bound,input,output):
-    work_dir = os.path.dirname(os.path.abspath(__file__))
-
-    shp_driver: Driver = ogr.GetDriverByName("ESRI Shapefile")
-    bound_ds : DataSource = shp_driver.Open(bound, 0)
-    bound_layer :Layer = bound_ds.GetLayer(0)
-
-    bf = bound_layer.GetNextFeature()
-    bound_geom = bf.GetGeometryRef()
-
-
-
-    gdb_driver: Driver = ogr.GetDriverByName("FileGDB")
-    # gdb_ds: DataSource = gdb_driver.Open(gdb, 0)
-    # layer: Layer = gdb_ds.GetLayerByName(layer_name)
-
-    input_ds: DataSource = shp_driver.Open(input, 0)
-    layer: Layer = input_ds.GetLayer(0)
-
-    if not layer:
-        raise Exception("打开数据失败！")
-
-
-    output_ds: DataSource = gdb_driver.CreateDataSource(output)
-
-
-    gdb_layer:Layer = output_ds.CreateLayer(layer.GetName(),layer.GetSpatialRef(),layer.GetGeomType())
-
-    gdb_layer.CreateFields(layer.schema)
-
-
-
-
-    #
-    bound_exent = bound_layer.GetExtent()
-
-    layer_extent = layer.GetExtent()
-
-    xxrange=layer_extent[1]-layer_extent[0]
-    yrange = layer_extent[3]-layer_extent[2]
-
-
-    left = (int((layer_extent[0] - bound_exent[0]) / xxrange) + 1) * (-1)
-    right = int((bound_exent[1] - layer_extent[1]) / xxrange) + 1
-    up = int((bound_exent[3] - layer_extent[3]) / yrange) + 1
-    down =  (int((layer_extent[2] - bound_exent[2]) / yrange) + 1) * (-1)
-
-
-    begin = time.time()
-    count=1
-
-
-    for f in layer:
-
-        new_f = copy.copy(f)
-        for xt in range(left,right+1,1):
-            for yt in range(down,up+1,1):
-                out_g = move(f.GetGeometryRef(),xxrange*xt,yrange*yt)
-
-                if out_g.Intersect(bound_geom):
-                    out_g = out_g.Intersection(bound_geom)
-                    new_f.SetGeometry(out_g)
-                    new_f.SetFID(count)
-                    gdb_layer.CreateFeature(new_f)
-                    count += 1
-                    if count % 10000 == 0:
-                        # print(count)
-                        with open(os.path.join(work_dir, "copy.txt"), "w") as fi:
-                            fi.write("已完成{}".format(count))
-
-    print(time.time()-begin)
-
-    input_ds.Destroy()
-
-def envelop_2_polygon(env):
-    ring = ogr.Geometry(ogr.wkbLinearRing)
-    ring.AddPoint(env[0], env[2])
-    ring.AddPoint(env[0], env[3])
-    ring.AddPoint(env[1], env[3])
-    ring.AddPoint(env[1], env[2])
-    ring.AddPoint(env[0], env[2])
-    # Create polygon
-    poly = ogr.Geometry(ogr.wkbPolygon)
-    poly.AddGeometry(ring)
-    return poly
-
-def clip():
-
-    input = r"E:\Data\copy\x.shp"
-    output = r"E:\Data\copy\xianorigin.shp"
-    gdal.SetConfigOption("SHAPE_ENCODING", "UTF-8")
-    shp_driver: Driver = ogr.GetDriverByName("ESRI Shapefile")
-
-    input_ds: DataSource = shp_driver.Open(input, 0)
-    layer: Layer = input_ds.GetLayer(0)
-
-    bound_geo = envelop_2_polygon([112.84,112.952,22.914,22.985])
-
-    if not layer:
-        raise Exception("打开数据失败！")
-
-    output_ds: DataSource = shp_driver.CreateDataSource(output)
-    #
-    # out_layer: Layer = output_ds.CreateLayer(layer.GetName(), layer.GetSpatialRef(), layer.GetGeomType())
-    # out_layer.CreateFields(layer.schema)
-    for f in layer:
-        new_f:Feature = copy.copy(f)
-        out_g = f.GetGeometryRef()
-        if out_g.Intersect(bound_geo):
-            out_g = out_g.Intersection(bound_geo)
-            new_f.SetGeometry(out_g)
-            new_f.SetField("ttt",1)
-            # new_f.UnsetField("OBJECTID")
-            # out_layer.CreateFeature(new_f)
-
-    # output_ds.Destroy()
-
-
-
-if __name__ == '__main__':
-    # bound_shp = "E:\Data\广东边界\gdsample.shp"
-    # input = r"E:\Data\copy\origin.shp"
-    # output = r"E:\Data\copy\re.shp"
-    # copydata(bound_shp,input,output)
-    #
-    # bound_shp = "/root/CopyData/gdsample.shp"
-    # input = r"/root/CopyData/origin.shp"
-    # output = r"/root/CopyData/re.gdb"
-    # copydata(bound_shp,input,output)
-    clip()
-
-
-
-
-
-
-
-
-
-
-

-
-from osgeo.ogr import *
-from osgeo import ogr
-from osgeo import gdal
-import os
-import uuid
-import shutil
-import time
-from app.modules.data.models import *
-from app.util.component.PGUtil import PGUtil
-from app.util.component.StructuredPrint import StructurePrint
-from sqlalchemy.orm import Session
-import configure
-import copy
-import datetime
-class EntryData:
-
-    def entry(self,parameter):
-        meta:dict = parameter.get("meta")
-
-        #设置编码
-        encoding = parameter.get("encoding")
-        if encoding:
-            gdal.SetConfigOption("SHAPE_ENCODING",encoding)
-        else:
-            gdal.SetConfigOption("SHAPE_ENCODING", "GBK")
-
-        #如果包含cpg文件，优先使用cpg文件中声明的编码
-        encoding_cpg = meta.get("encoding")
-        if encoding_cpg:
-            gdal.SetConfigOption("SHAPE_ENCODING", encoding_cpg)
-
-
-        # 初始化任务
-        this_task = ThisTask(parameter)
-        data_path = meta.get("data_path")
-
-        try:
-            this_task.write_process("入库任务初始化...")
-
-            this_task.update({"process": "入库中"})
-            is_success = None
-            if not data_path:
-                raise Exception("数据错误！")
-            # 分为shp和gdb 2种录入形式
-
-            # 开始事务
-            this_task.pg_ds.StartTransaction()
-            if data_path.endswith("shp"):
-                is_success,new_layer_name = self.entry_shp(data_path,this_task)
-
-            if data_path.endswith("gdb"):
-                is_success,new_layer_names = self.entry_gdb(data_path,this_task)
-
-            this_task.write_process("数据入库结束。")
-
-            if is_success:
-                # 更新任务为成功任务
-                this_task.pg_ds.CommitTransaction()
-                this_task.update({"state": 1,"process":"入库完成","update_time": datetime.datetime.now()})
-            else:
-                # 更新任务为失败任务
-                this_task.update({"state": -1, "process": "入库失败", "update_time": datetime.datetime.now()})
-                # rollback
-                this_task.pg_ds.RollbackTransaction()
-        except Exception as e:
-            this_task.write_process("{} 任务结束！".format(e.__str__()))
-            this_task.update({"state": -1, "process": "入库失败", "update_time": datetime.datetime.now()})
-            StructurePrint().print(e.__str__(),"ERROR")
-            # rollback
-            this_task.pg_ds.RollbackTransaction()
-        finally:
-            this_task.end()
-            try:
-                file_tmp_path = os.path.join(data_path.split("file_tmp")[0],"file_tmp")
-                dir_path = os.path.dirname(data_path)
-                i=0
-                while not os.path.dirname(dir_path).__eq__(file_tmp_path) and i<30:
-                    dir_path = os.path.dirname(dir_path)
-                    i+=1
-                if i<30:
-                    shutil.rmtree(dir_path,True)
-                    StructurePrint().print("删除文件成功！")
-                else:
-                    raise Exception("找不到文件！")
-
-            except Exception as e:
-                StructurePrint().print(e.__str__(), "ERROR")
-                StructurePrint().print("删除文件失败！","ERROR")
-
-    def entry_shp(self,data_path,this_task):
-        '''
-        录入shp
-        :param data_path:
-        :return:
-        '''
-
-        driver: Driver = ogr.GetDriverByName("ESRI Shapefile")
-        ds: DataSource = driver.Open(data_path, 1)
-        if not ds:
-            raise Exception("打开数据失败！")
-        layer: Layer = ds.GetLayer(0)
-
-        return self.entry_one_layer(layer, this_task)
-
-    def entry_gdb(self,data_path,this_task):
-        '''
-        录入gdb
-        :param data_path:
-        :return:
-        '''
-
-        is_successes = []
-        new_layer_names=[]
-        driver: Driver = ogr.GetDriverByName("OpenFileGDB")
-        ds: DataSource = driver.Open(data_path, 0)
-        if not ds:
-            raise Exception("打开数据失败！")
-
-
-        for i in range(ds.GetLayerCount()):
-            layer: Layer = ds.GetLayer(i)
-            is_success, new_layer_name = self.entry_one_layer(layer,this_task)
-            new_layer_names.append(new_layer_name)
-            is_successes.append(is_success)
-
-        if is_successes.__contains__(False):
-            return False,new_layer_names
-        else:
-            return True,new_layer_names
-
-    def entry_one_layer(self,layer: Layer,this_task):
-
-        # this_task.pg_ds.StartTransaction()
-        new_layer_name = None
-
-
-        try:
-            # 图层设置
-            parameter = this_task.parameter
-
-            meta: dict = parameter.get("meta")
-            overwrite = parameter.get("overwrite") if parameter.get("overwrite") is not None and parameter.get("overwrite")=="yes" else "no"
-            geom_name = parameter.get("geom_name") if parameter.get("geom_name") is not None else "geom"
-            fid =  parameter.get("fid") if parameter.get("fid") is not None else "fid"
-            options = ["OVERWRITE={}".format(overwrite), "FID={}".format(fid), "GEOMETRY_NAME={}".format(geom_name),"PRECISION=NO"]
-
-
-
-            # # 中文名处理
-            # chinese = is_chinese(new_layer_name)
-            # if chinese:
-            #     new_layer_name = new_layer_name.__hash__()
-
-            # 将线/面转多线多面，dmap只支持多线面
-            geom_type = self.change_geom_type(layer.GetGeomType())
-
-            # 更改图层名
-            change_name = False
-            origin_name = layer.GetName().lower()
-
-            # 新图层名
-            new_layer_name: str = meta.get("layer").get(origin_name)
-            origin_name = new_layer_name
-            no = 1
-            while overwrite.__eq__("no") and this_task.pg_ds.GetLayerByName(new_layer_name) :
-                change_name=True
-                new_layer_name = origin_name+"_{}".format(no)
-                no+=1
-
-            if change_name:
-                this_task.write_process("{}图层已存在，更名为{}入库".format(origin_name, new_layer_name))
-
-
-            this_task.write_process("{}图层正在入库...".format(new_layer_name))
-
-            pg_layer: Layer = this_task.pg_ds.CreateLayer(new_layer_name, layer.GetSpatialRef(), geom_type, options)
-
-            # 复制原图层的属性
-            # 去掉fid的属性
-            schema = [sche for sche in layer.schema if not sche.name.__eq__(fid)]
-
-            pg_layer.CreateFields(schema)
-
-            count =0
-            this_time = time.time()
-            for feature in layer:
-                count+=1
-                if count%10000==0:
-                    StructurePrint().print("{}图层已入库{}个对象".format(new_layer_name,count))
-                    # print(time.time()-this_time)
-                    this_time=time.time()
-                geo :Geometry = feature.GetGeometryRef()
-                # 如果是空对象不录入
-                if geo is not None:
-                    if geo.IsEmpty():
-                        this_task.write_process("FID:{}要素的空间字段为空，跳过该要素！".format(feature.GetFID()))
-                        StructurePrint().print("FID:{}要素的空间字段为空，跳过该要素！".format(feature.GetFID()),"WARN")
-                        continue
-                out_feature: Feature = copy.copy(feature)
-
-                if geo is not None:
-                    out_geom:Geometry = self.change_geom(geo, geom_type)
-                    out_feature.SetGeometry(out_geom)
-
-                pg_layer.CreateFeature(out_feature)
-
-            # 注册图层信息
-            this_task.register_table(pg_layer,new_layer_name,overwrite,parameter.get("creator"))
-            this_task.write_process("{}图层入库成功。".format(new_layer_name))
-        except Exception as e:
-            # this_task.pg_ds.RollbackTransaction()
-            this_task.write_process("{}入库失败，数据回滚！原因：{}".format(new_layer_name,e.__str__()))
-            StructurePrint().print("{}入库失败，数据回滚！原因：{}".format(new_layer_name,e.__str__()), "error")
-            return False,new_layer_name
-        # finally:
-
-        return True,new_layer_name
-
-    def entry_feature(self):
-        pass
-
-
-    def change_geom_type(self,raw):
-        if raw.__eq__(-2147483646):
-            return 5
-        if raw.__eq__(-2147483645):
-            return 6
-        if raw==2 or raw ==3:
-            return raw+3
-        if raw==4:
-            return 1
-        return raw
-
-    def get_table_type(self,raw):
-        if raw==4 or raw ==5 or raw ==6:
-            return raw-3
-        return raw
-
-    def change_geom(self,geo:Geometry,geom_type):
-        '''
-        转换空间对象的类型，以适应dmap只支持Multi类型
-        :param geo:
-        :param geom_type:
-        :return: 转换后的空间对象
-        '''
-
-
-        # Point = 1,
-        # LineString = 2,
-        # Polygon = 3,
-        # MultiPoint = 4,
-        # MultiLineString = 5,
-        # MultiPolygon = 6,
-        # GeometryCollection = 7,
-        # CircularString = 8,
-        # CompoundCurve = 9,
-        # CurvePolygon = 10,
-        # MultiCurve = 11,
-        # MultiSurface = 12,
-        # PolyhedralSurface = 15,
-        # LinearRing = 101,
-
-        # MultiPointZ = -2147483644
-        # MultiPolygonZ = -2147483643
-        # MultiLineStringZ = -2147483642
-
-
-        # PointZ = -2147483647
-        # LINESTRINGZ=-2147483646
-        # POLYGONZ=-2147483645
-
-        if geom_type==5 or geom_type.__eq__(-2147483646):
-            return ogr.ForceToMultiLineString(geo)
-        if geom_type==6 or geom_type.__eq__(-2147483645):
-            return ogr.ForceToMultiPolygon(geo)
-        if geom_type==1:
-            # 多点转单点，会有问题，只拿了第一个点
-            xy = geo.GetPoint()
-            point = ogr.Geometry(ogr.wkbPoint)
-            point.AddPoint(xy[0], xy[1])
-            return point
-        return geo
-
-    def write_task_process(self,session, task_guid, message):
-        '''
-        写详细过程
-        :param session:
-        :param task_guid:
-        :param message:
-        :return:
-        '''
-
-        task_process_guid = uuid.uuid1().__str__()
-        task_process = Process(guid=task_process_guid,
-                               message=message,
-                               time=datetime.datetime.now(),
-                               task_guid=task_guid)
-        session.add(task_process)
-
-
-    def data_check(self):
-        pass
-
-class ThisTask:
-
-    def __init__(self, parameter):
-        try:
-            self.sys_session: Session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
-        except Exception as e:
-            raise Exception("打开数据库失败！")
-        self.parameter = parameter
-
-        # # 初始化task
-        # task = Task(guid=parameter.get("task_guid"), name=parameter.get("task_name"),create_time=datetime.datetime.now(),state=0)
-        # self.sys_session.add(task)
-        # self.sys_session.commit()
-
-        self.task = self.sys_session.query(Task).filter_by(guid=parameter.get("task_guid"))
-
-        self.database = self.sys_session.query(Database).filter_by(
-            guid=parameter.get("database_guid")).one_or_none()
-
-        self.catalog_guid = parameter.get("catalog_guid")
-
-        self.pg_ds: DataSource = PGUtil.open_pg_data_source(1, DES.decode(self.database.sqlalchemy_uri))
-
-        if self.pg_ds is None:
-            raise Exception("打开系统数据库失败！")
-
-    def update(self, update_dict):
-        self.task.update(update_dict)
-        self.sys_session.commit()
-
-    def write_process(self, message):
-        message = "{} {}".format(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), message)
-        task_process_guid = uuid.uuid1().__str__()
-        task_process = Process(guid=task_process_guid, message=message, time=datetime.datetime.now(),
-                               task_guid=self.parameter.get("task_guid"))
-        self.sys_session.add(task_process)
-        self.sys_session.commit()
-
-    def register_table(self, layer: Layer, new_layer_name, overwrite, creator):
-        '''
-        注册表
-        :param layer: 图层
-        :param new_layer_name: 图层名
-        :return: 表名
-        '''
-
-        # 在覆盖模式下，删除原有记录
-        if overwrite.__eq__("yes"):
-            old_table = self.sys_session.query(Table).filter_by(name=new_layer_name).one_or_none()
-            if old_table:
-                self.sys_session.delete(old_table)
-                self.sys_session.commit()
-
-        this_time = datetime.datetime.now()
-
-        ext = layer.GetExtent()
-        if ext[0] < 360:
-            ext = [round(e, 6) for e in ext]
-        else:
-            ext = [round(e, 2) for e in ext]
-
-        # extent = "西：{}，东：{}，南：{}，北：{}".format(ext[0],ext[1],ext[2],ext[3])
-        extent = "{},{},{},{}".format(ext[0], ext[1], ext[2], ext[3])
-
-        # 其他相同库也更新
-        connectstr = self.database.connectstr
-        databs = self.sys_session.query(Database).filter_by(connectstr=connectstr).all()
-        databs_guid = [d.guid for d in databs]
-        for d_guid in databs_guid:
-            table_guid = uuid.uuid1().__str__()
-            table = Table(guid=table_guid,
-                          database_guid=d_guid,
-                          # alias=new_layer_name,
-                          creator=creator,
-                          name=new_layer_name, create_time=this_time, update_time=this_time,
-                          catalog_guid=self.catalog_guid, table_type=self.get_table_type(layer.GetGeomType()),
-                          extent=extent,
-                          feature_count=layer.GetFeatureCount()
-                          )
-            self.sys_session.add(table)
-
-            feature_defn: FeatureDefn = layer.GetLayerDefn()
-
-            for i in range(feature_defn.GetFieldCount()):
-                field_defn: FieldDefn = feature_defn.GetFieldDefn(i)
-                field_name = field_defn.GetName().lower()
-                field_alias = field_name if field_defn.GetAlternativeName() is None or field_defn.GetAlternativeName().__eq__(
-                    "") else field_defn.GetAlternativeName()
-                column = Columns(guid=uuid.uuid1().__str__(), table_guid=table_guid,
-                                 name=field_name, alias=field_alias, create_time=this_time, update_time=this_time)
-                self.sys_session.add(column)
-
-        self.sys_session.commit()
-
-    def end(self):
-        if self.sys_session:
-            self.sys_session.close()
-        if self.pg_ds:
-            self.pg_ds.Destroy()
-
-    def get_table_type(self,raw):
-        if raw==4 or raw ==5 or raw ==6:
-            return raw-3
-        return raw

@@ -52,6 +52,15 @@ class TaskController:
     @classmethod
     @classmethod
     def wait(cls,task_guid):
     def wait(cls,task_guid):
         sys_session= PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
         sys_session= PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
+        #三秒检测一次是否可以执行任务
+        wait_time = 0
         while not cls.pass_check(sys_session,task_guid):
         while not cls.pass_check(sys_session,task_guid):
             time.sleep(3)
             time.sleep(3)
+            wait_time += 3
+            if wait_time > 21600:
+                try:
+                    sys_session.close()
+                except Exception as e:
+                    pass
+                raise Exception("任务等待超时")
         sys_session.close()
         sys_session.close()

+# coding=utf-8
+#author:        4N
+#createtime:    2021/12/3
+#email:         nheweijun@sina.com
+
+import datetime
+import uuid
+from app.modules.data.models import Process,Task,Table
+from app.modules.service.models import Image
+import configure
+from app.util.component.PGUtil import PGUtil
+
+def check_session(fun):
+    '''
+    检查session的装饰器
+    :param fun:
+    :return:
+    '''
+
+    def wrap(self,*args, **kwargs):
+        if not self.sys_session.is_active:
+            self.sys_session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
+        fun(self,*args, **kwargs)
+
+    return wrap
+
+class TaskWriter:
+
+    def __init__(self,task_guid):
+        self.sys_session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
+        self.task_guid = task_guid
+
+    @check_session
+    def update_task(self,update_info,commit=True):
+        self.sys_session.query(Task).filter_by(guid=self.task_guid).update(update_info)
+        if commit:
+            self.sys_session.commit()
+
+    @check_session
+    def update_process(self,msg,commit=True):
+        message = "{} {}".format(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), msg)
+        task_process_guid = uuid.uuid1().__str__()
+        task_process = Process(guid=task_process_guid, message=message, time=datetime.datetime.now(),
+                               task_guid=self.task_guid)
+
+        self.sys_session.add(task_process)
+        if commit:
+            self.sys_session.commit()
+
+    @check_session
+    def update_table(self,table_guid,update_info,commit=True):
+        self.sys_session.query(Table).filter_by(guid=table_guid).update(update_info)
+        if commit:
+            self.sys_session.commit()
+
+    @check_session
+    def update_image(self,image_guid,update_info,commit=True):
+        self.sys_session.query(Image).filter_by(guid=image_guid).update(update_info)
+        if commit:
+            self.sys_session.commit()
+
+    def close(self):
+        try:
+            self.sys_session.commit()
+            self.sys_session.close()
+        except:
+            pass
+
+    @property
+    def session(self):
+        if not self.sys_session.is_active:
+            self.sys_session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
+        return self.sys_session