sanya-dbms/processing/tools/PostgreSQL/PostgreSQL.py

# -*- coding: utf-8 -*-

__author__ = 'wanger'
__date__ = '2024-08-20'
__copyright__ = '(C) 2024 by siwei'
__revision__ = '1.0'

import time
from typing import Optional
import os
import psycopg2
import uuid
from ..FTP.FtpUitl import FtpOper
from ..FTP.FtpConfig import *
from ..FileUtils import getInputFileName
from ..StringUtils import (getNow, base64str, enbase64str)
from datetime import datetime


class PostgreSQL:
    # 矢量数据元数据表
    Vector_Storage = "t_vector_storage"
    # 矢量数据服务发布表
    Vector_Server = "t_vector_server"
    # 附件表
    Vector_FJ = "t_vector_fj"
    # 字段表
    Vector_Field = "t_vector_field"
    # 平台资源目录表
    Portal_Zyml = "t_yzt_zyml"
    # 备份表后缀名
    BackTable = "_back"
    # 序列名称后缀
    Sequence = "_id_seq"
    # 版本控制表
    Vector_Version = "t_vector_version"
    # 版本控制详情表
    Vector_Version_Details = "t_vector_version_details"

    def __init__(
            self,
            host: Optional[str] = "127.0.0.1",  # default host during installation
            port: Optional[str] = "5432",  # default port during pg installation
            user: Optional[str] = "postgres",  # default user during pg installation
            password: Optional[str] = "postgres",  # default password during pg installation
            dbname: Optional[str] = "real3d",  # default dbname during pg installation
            schema: Optional[str] = None
    ):
        # 配置数据库连接参数并指定schema
        self.connparams = {
            "dbname": dbname,
            "user": user,
            "password": password,
            "host": host,
            "port": port,
            "options": "-c search_path=otherSchema," + schema if schema is not None else None
        }
        self.conn = psycopg2.connect(**self.connparams)
        # 创建一个游标对象
        self.cur = self.conn.cursor()
        # 执行一个查询
        # self.cur.execute("SELECT 省,类型 from \"XZQH3857\";")
        # 获取查询结果
        # rows = self.cur.fetchall()
        # 打印结果
        # for row in rows:
        #     for v in row:
        #         print(v)

    def execute(self, sql):
        # 执行一个查询
        self.cur.execute(sql)
        # 获取查询结果
        return self.cur.fetchall()

    def close(self):
        # 关闭游标和连接
        self.cur.close()
        self.conn.close()

    # 根据名称删除元数据信息
    def deleteVectorStorage(self, name):
        sql = "delete from {} where name = '{}'".format(self.Vector_Storage, name)
        self.cur.execute(sql)
        self.conn.commit()

    # 保存元数据信息
    def insertVectorStorage(self, id, name, year, xmlx, sjly, xzqh, xzqh_field, ywlx, sjlx, sjywz, glbm, table_alias):
        sql = "insert into {} (id, name, year, xmlx, sjly, xzqh, xzqh_field, ywlx, sjlx, sjywz, glbm, table_alias)  values ('{}','{}', '{}', (select bsm from t_vector_zyml where name = '{}'), '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}')".format(
            self.Vector_Storage, id, name, year.toString("yyyy"), xmlx, sjly, xzqh, xzqh_field, ywlx, sjlx, sjywz, glbm,
            table_alias)
        self.cur.execute(sql)
        self.conn.commit()

    # 元数据入库，参数parameters
    def metadataStorage(self, parameters, ssxzqh, fileliststr, ywlx, glbm, ogrLayer, zyml):
        record_id = ""
        id = uuid.uuid4().__str__()
        id = id.replace("-", "")
        # print(parameters)
        if (parameters.get("TABLE") is None or parameters.get("TABLE") == "") and ogrLayer is not None:
            record_id = parameters.get("SCHEMA") + "." + getInputFileName(ogrLayer).lower()
        elif ogrLayer is not None:
            record_id = parameters.get("SCHEMA") + "." + ogrLayer.lower()
        else:
            record_id = parameters.get("SCHEMA") + "." + parameters.get("TABLE")
        table_alias = parameters.get("TABLE_ALIAS")
        if table_alias is not None and table_alias != '':  # 配置表别名
            self.setTableAlias(table=self.getTableName(recordid=record_id),
                               alias=table_alias)
        else:
            table_alias = ''
        self.deleteVectorStorage(name=record_id)
        self.deleteVectorVersion(tablename=record_id)
        self.insertVectorStorage(id=id, name=record_id, year=parameters.get("VECTOR_YEAR"),
                                 xmlx=zyml, sjly=parameters.get("VECTOR_SJLY"),
                                 xzqh=ssxzqh, xzqh_field=parameters.get("XZQH_FIELD"),
                                 ywlx=ywlx, sjlx=parameters.get("SOURCE_TYPE"), sjywz=ogrLayer,
                                 glbm=glbm, table_alias=table_alias)
        self.insertVectorVersion(tablename=record_id)
        # 附件上传
        if fileliststr is not None and fileliststr != "":
            self.uploadAttachment(layername=record_id, layerid=id, fileliststr=fileliststr)

    # 删除版本控制表
    def deleteVectorVersion(self, tablename):
        sql = "delete from {} where tablename = '{}'".format(self.Vector_Version, tablename)
        self.cur.execute(sql)
        self.conn.commit()

    # 删除版本数据
    def deleteVectorRecords(self, tablename, rksj_sw):
        sql = "delete from {} t where t.rksj_sw > '{}'".format(tablename, rksj_sw)
        self.cur.execute(sql)
        self.conn.commit()
        sql = "update {} set version = '{}' where tablename = '{}'".format(self.Vector_Version, rksj_sw,
                                                                           tablename)
        self.cur.execute(sql)
        self.conn.commit()

    # 插入版本数据
    def insertVectorRecords(self, tablename, curversion, targetversion):
        sql = "select insertsql from {} t where t.tablename = '{}' and  t.version > '{}' and t.version <= '{}'".format(
            self.Vector_Version_Details, tablename, curversion, targetversion)
        self.cur.execute(sql)
        rows = self.cur.fetchall()
        for row in rows:
            sql = enbase64str(row[0])
            self.cur.execute(sql)
        sql = "update {} set version = '{}' where tablename = '{}'".format(self.Vector_Version, targetversion,
                                                                           tablename)
        self.cur.execute(sql)
        self.conn.commit()

    # 创建版本控制表
    def insertVectorVersion(self, tablename):
        formatted_time = getNow()
        # 输出结果
        print("当前时间：{}".format(formatted_time))
        sql = "insert into {} (tablename, version, type) values ('{}', '{}', '{}')".format(self.Vector_Version,
                                                                                           tablename, formatted_time,
                                                                                           "base")
        self.cur.execute(sql)
        self.conn.commit()
        sql = "alter table {} add column rksj_sw timestamp".format(tablename)
        self.cur.execute(sql)
        self.conn.commit()
        sql = "update {} SET rksj_sw = to_timestamp('{}', 'YYYY-MM-DD HH24:MI:SS')".format(tablename, formatted_time)
        self.cur.execute(sql)
        self.conn.commit()
        details = "insert into {} (tablename, version, insertsql) values ('{}', '{}', '{}')".format(
            self.Vector_Version_Details, tablename, formatted_time,
            "")
        print(details)
        self.cur.execute(details)
        self.conn.commit()

    # 获取文件名称
    def getInputFileName(self, file):
        filename = file.split("/")[len(file.split("/")) - 1]
        return filename[0: filename.find(".")]

    # 附件上传FTP
    def uploadAttachment(self, layername, layerid, fileliststr):
        print("开始上传附件啦！！")
        self.ftpOper = FtpOper()
        self.ftpOper.ftp.connect(ftpHost, ftpPort)
        self.ftpOper.ftp.login(ftpUsername, ftpPassword)
        ftp_path = '/三亚数管/' + layername  # 图层FTP目录地址
        if not self.ftpOper.is_exist('/三亚数管/'):
            self.ftpOper.makedir('三亚数管', '/', '/三亚数管/')
        if self.ftpOper.is_exist(ftp_path):  # 已有此图层目录，则删除
            self.ftpOper.deletedir(ftp_path)
        self.ftpOper.makedir(layername, '/三亚数管/', ftp_path)  # 重新创建图层目录
        file_paths_list = fileliststr.split(",")
        if len(file_paths_list):
            for file_path in file_paths_list:
                if os.path.exists(file_path):
                    filename = os.path.basename(file_path)
                    filenamenoexten, fileextension = os.path.splitext(filename)
                    self.ftpOper.uploadfile(file_path, ftp_path)
                    self.insertVectorFJ(uuid.uuid4().__str__(), layerid, layername, filename, fileextension,
                                        ftp_path + '/' + filename,
                                        file_path, self.ftpOper.ftp.host + ':' + str(self.ftpOper.ftp.port),
                                        str(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())))
        self.ftpOper.ftp.close()

    # 保存附件信息
    def insertVectorFJ(self, id, layerid, layername, name, fjlx, fjdz, fjwz, dldz, cjsj):
        sql = "insert into {} (id,layerid, layername, name, fjlx, fjdz, fjwz,dldz,cjsj)  values ('{}','{}','{}', '{}', '{}', '{}', '{}', '{}', '{}')".format(
            self.Vector_FJ, id, layerid, layername, name, fjlx, fjdz, fjwz, dldz, cjsj)
        self.cur.execute(sql)
        self.conn.commit()

    # 服务发布表插入
    def insertVectorServer(self, id, fwqlx, sjy, fwlx, fwdz, fwmc, fwgzkj, fwys, qpfa, zymlbsm, layergroup, zoommin,
                           zoommax, layername):
        # 数据库管理系统服务相关表插入
        sql = "insert into {} (id, fwqlx, sjy, fwlx, fwdz, fwmc, fwgzkj, fwys, qpfa, zymlbsm, layergroup, zoommin, zoommax, layername)  values ('{}','{}','{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}')".format(
            self.Vector_Server, id, fwqlx, sjy, fwlx, fwdz, fwmc, fwgzkj, fwys, qpfa, zymlbsm, layergroup, zoommin,
            zoommax, layername)
        self.cur.execute(sql)
        self.conn.commit()

    # 平台资源目录表插入
    def insertPortalZymlServer(self, id, fwqlx, sjy, fwlx, fwdz, fwmc, fwgzkj, fwys, qpfa, zymlbsm, layergroup, zoommin,
                               zoommax, layername):
        # 平台资源目录服务相关表插入
        sql = "insert into {} (bsm, name, type, pbsm, url, state, parent, server_type, fwmc, " \
              "fwgzkj, fwys, qpfa, layergroup, format, maximumlevel, minimumlevel)  values ('{}','{}','{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}')".format(
            self.Portal_Zyml, id, fwmc, fwlx, zymlbsm, fwdz, '1', '0', fwqlx, layername, fwgzkj, fwys, qpfa,
            layergroup, 'image/png', zoommax, zoommin)
        self.cur.execute(sql)
        self.conn.commit()

    # 获取资源目录
    def getZyml(self):
        self.cur.execute(
            "select t.bsm, t.name, t.pbsm from t_yzt_zyml t where t.parent = '1' order by t.lev,t.pbsm , t.sort")
        rows = self.cur.fetchall()
        return rows

    def getVectorZyml(self):
        self.cur.execute(
            "select bsm,name from t_vector_zyml t order by t.sort,t.name")
        rows = self.cur.fetchall()
        return rows

    # 获取行政区划
    def getXzqh(self):
        self.cur.execute(
            "select t.id, t.name, case when t.pid = '0' then '' else t.pid end from vector.xzqh t order by t.id")
        rows = self.cur.fetchall()
        return rows

    # 获取矢量数据字典类型
    def getVectorYwlx(self):
        self.cur.execute(
            "select distinct(t.ywlx) from t_vector_field t ")
        rows = self.cur.fetchall()
        return rows

    # 获取部门列表
    def getDeptList(self):
        self.cur.execute(
            "select dept_name from  (select distinct(dept_name),dept_id from sys_dept t where t.del_flag = '0' order by t.dept_id) a")
        rows = self.cur.fetchall()
        return rows

    # 根据用户名称查询有权限维护的数据库表，返回别名或者表名
    def getManagerTables(self, username):
        sql = f'select t.id "id",name "name", case when t.table_alias != \'\' then  t.table_alias else t.name end as "alias", t.ywlx "ywlx" ' \
              f'from {self.Vector_Storage} t where t.glbm = (select dept_name from sys_dept d where d.dept_id = (select dept_id from sys_user u where u.user_name = \'{username}\' )) and t.sjlx = \'vector\''
        self.cur.execute(sql)
        rows = self.cur.fetchall()
        return rows

    # 根据业务类型和表名查询必填的字段名
    def getMustRequireField(self, tablename, ywlx):
        arr = tablename.split(".")
        sql = f'select lower(f.name) from {self.Vector_Field} f where f.ywlx = \'{ywlx}\' and lower(f.name) in (' \
              f'select column_name from information_schema.columns where table_schema = \'{arr[0]}\' and table_name = \'{arr[1]}\') and f.nullable = \'1\''
        self.cur.execute(sql)
        rows = self.cur.fetchall()
        fields = []
        for row in rows:
            fields.append(row[0])
        return fields

    def getAllTableField(self, tablename):
        arr = tablename.split(".")
        sql = f'select column_name from information_schema.columns where table_schema = \'{arr[0]}\' and table_name = \'{arr[1]}\' and lower(column_name) not in (\'id\',\'geom\', \'shape_leng\', \'shape_area\')'
        self.cur.execute(sql)
        rows = self.cur.fetchall()
        fields = []
        for row in rows:
            fields.append(row[0])
        return fields

    # 保存矢量要素
    def insertVectorFeature(self, tablename, fields, values, wkt, wktsrid, geomtype, rksj):
        arr = tablename.split(".")
        insertvalues = ""
        # print(values)
        for value in values:
            if value == None:
                insertvalues += "null,"
            else:
                b = self.is_string(value)
                if b == True:
                    insertvalues += f'\'{value}\','
                else:
                    insertvalues += f'{value},'
        # print(insertvalues)
        # print(insertvalues[:-1])
        # 判断几何类型是否一致
        if wkt.startswith(geomtype) == False:
            if geomtype.startswith("MULTI"):  # 转换为Multi
                wkt = 'MULTI' + wkt
                wkt = wkt.replace("((", "(((").replace("))", ")))")
            else:
                wkt = wkt.replace("(((", "((").replace(")))", "))").replace("MULTI", "")
        sql = f'insert into {arr[0]}."{arr[1]}" ({",".join(fields)} , geom) values ({insertvalues[:-1]} , public.st_geomfromtext( \'{wkt}\', {wktsrid}))'
        sql = sql.replace("None", "null")
        # print(sql)
        self.cur.execute(sql)
        self.conn.commit()
        # 插入到更新详情表
        details = "insert into {} (tablename, version, insertsql) values ('{}', '{}', '{}')".format(
            self.Vector_Version_Details, tablename, rksj,
            base64str(sql))
        # print(details)
        self.cur.execute(details)
        self.conn.commit()

    def updateVectorVersion(self, tablename, version):
        details = "update {} set rksj_sw = to_timestamp('{}', 'YYYY-MM-DD HH24:MI:SS') where rksj_sw is null".format(
            tablename, version)
        self.cur.execute(details)
        self.conn.commit()
        details = "update {} set version = '{}', type = 'update' where tablename = '{}'".format(self.Vector_Version,
                                                                                                version, tablename)
        self.cur.execute(details)
        self.conn.commit()

    # 通过表名称获取版本集合
    def getTableVersions(self, tablename):
        sql = "select distinct(version) from {} t  where t.tablename = '{}' order by version desc".format(
            self.Vector_Version_Details, tablename)
        print(sql)
        self.cur.execute(sql)
        rows = self.cur.fetchall()
        return rows

    # 通过表名称获取当前版本
    def getTableCurVersion(self, tablename):
        sql = "select version from {} t where t.tablename = '{}'".format(
            self.Vector_Version, tablename)
        print(sql)
        self.cur.execute(sql)
        rows = self.cur.fetchall()
        return rows[0][0]

    # 获取矢量数据表的坐标系
    def getVectorTableSrid(self, tablename):
        arr = tablename.split(".")
        sql = f'select public.st_srid(geom) from {arr[0]}."{arr[1]}" limit 1'
        self.cur.execute(sql)
        rows = self.cur.fetchall()
        for row in rows:
            return row[0]

    # 获取矢量数据表的空间类型
    def getVectorTableGeomType(self, tablename):
        arr = tablename.split(".")
        sql = f'select replace(upper(public.ST_GeometryType(geom)),\'ST_\',\'\') from {arr[0]}."{arr[1]}" limit 1'
        self.cur.execute(sql)
        rows = self.cur.fetchall()
        for row in rows:
            return row[0]

    def getInstersectsCount(self, tablename, tablesrid, wkts, wktsrid):
        arr = tablename.split(".")
        wktsql = ""
        for i in range(len(wkts)):
            wkt = wkts[i]
            if i > 0:
                wktsql += ' or '
            wktsql += f' public.st_intersects(t.geom,public.st_transform( public.st_geomfromtext( \'{wkt}\', {wktsrid}), {tablesrid} )) '
        sql = f'select count(1)  from {arr[0]}."{arr[1]}" t where {wktsql}'
        self.cur.execute(sql)
        rows = self.cur.fetchall()
        for row in rows:
            return row[0]

    # 设置表的别名（基于postgresql comment）
    def setTableAlias(self, table, alias):
        sql = f'COMMENT ON TABLE {table} IS \'{alias}\''
        self.cur.execute(sql)
        self.conn.commit()

    # 删除备份表并创建备份表
    def restoreBackTable(self, tablename):
        arr = tablename.split(".")
        # 删除当前表
        deletesql = f'drop table if exists {arr[0]}."{arr[1]}"'
        self.cur.execute(deletesql)
        self.conn.commit()
        # 删除当前表序列
        deletesequencesql = f'drop sequence if exists {tablename}{self.Sequence}'
        self.cur.execute(deletesequencesql)
        self.conn.commit()
        # 恢复back表
        createsql = f'create table {arr[0]}."{arr[1]}" as table {arr[0]}."{arr[1]}{self.BackTable}"'
        self.cur.execute(createsql)
        self.conn.commit()
        # 查询数据表中id的最大值
        sql = f'select (max(id) + 1) from {arr[0]}."{arr[1]}"'
        self.cur.execute(sql)
        rows = self.cur.fetchall()
        max = 1
        for row in rows:
            max = row[0]
        # 创建序列
        addsequencesql = f'create sequence {tablename}{self.Sequence} start with {max} increment by 1 no minvalue no maxvalue cache 1'
        self.cur.execute(addsequencesql)
        self.conn.commit()
        # 设置id字段不能为空
        notnullsql = f'alter table {arr[0]}."{arr[1]}" alter column id set not null'
        self.cur.execute(notnullsql)
        self.conn.commit()
        # 使用序列
        usesequencesql = f'alter table {arr[0]}."{arr[1]}" alter column id set default nextval(\'{tablename}{self.Sequence}\')'
        self.cur.execute(usesequencesql)
        self.conn.commit()

    # 删除备份表并创建备份表
    def resetBackTable(self, tablename):
        arr = tablename.split(".")
        deletesql = f'drop table if exists {arr[0]}."{arr[1]}{self.BackTable}"'
        self.cur.execute(deletesql)
        self.conn.commit()
        createsql = f'create table {arr[0]}."{arr[1]}{self.BackTable}" as table {arr[0]}."{arr[1]}"'
        self.cur.execute(createsql)
        self.conn.commit()

    def getTableName(self, recordid):
        sp = recordid.split(".")
        res = f'"{sp[0]}"."{sp[1].lower()}"'
        return res

    # 查询数据库当前模式下有没有同名的表
    def checkTableName(self, schemaname, tablename):
        sql = f'select tablename from pg_tables t where t.schemaname = \'{schemaname}\' and t.tablename = \'{tablename}\''
        self.cur.execute(sql)
        rows = self.cur.fetchall()
        return len(rows) == 0

    # 判断数据是否为字符串
    def is_string(self, var):
        return isinstance(var, str)