GeoPackage转PostgreSQL/PostGIS SQL工具 - 在线GPKG转PostGIS建表脚本转换器

/**
 * GeoPackage 转 PostgreSQL/PostGIS SQL
 * 需要安装: npm install sql.js
 * sql.js 的 WASM 文件需要放在 public 目录下
 */
import initSqlJs from 'sql.js';

/**
 * SQLite 类型 -> PostgreSQL 类型映射
 */
const TYPE_MAP = {
  'INTEGER': 'INTEGER', 'TEXT': 'TEXT', 'REAL': 'DOUBLE PRECISION',
  'MEDIUMINT': 'INTEGER', 'DATETIME': 'TIMESTAMP', 'BLOB': 'BYTEA',
};

/**
 * 将 GeoPackage 二进制几何数据转换为 WKT
 */
function gpkgGeomToWkt(blob) {
  if (!blob || blob.length < 8) return null;
  if (blob[0] !== 0x47 || blob[1] !== 0x50) return wkbToWkt(blob);
  const flags = blob[3];
  const envelopeType = (flags >> 1) & 0x07;
  const emptyFlag = (flags >> 4) & 0x01;
  if (emptyFlag) return null;
  const envelopeSizes = { 0: 0, 1: 32, 2: 48, 3: 48, 4: 64 };
  const wkbOffset = 8 + (envelopeSizes[envelopeType] || 0);
  return wkbToWkt(blob.slice(wkbOffset));
}

function wkbToWkt(wkb) {
  const dv = new DataView(wkb.buffer, wkb.byteOffset, wkb.byteLength);
  return parseWkb(dv, 0)[0];
}

function parseWkb(dv, offset) {
  const le = dv.getUint8(offset) === 1;
  offset += 1;
  const fullType = dv.getUint32(offset, le);
  offset += 4;
  const baseType = fullType % 1000;
  const hasZ = Math.floor(fullType / 1000) === 1 || Math.floor(fullType / 1000) === 3;
  const hasM = Math.floor(fullType / 1000) === 2 || Math.floor(fullType / 1000) === 3;
  const dims = 2 + (hasZ ? 1 : 0) + (hasM ? 1 : 0);
  const dim = hasZ && hasM ? ' ZM' : hasZ ? ' Z' : hasM ? ' M' : '';

  const readCoord = (off) => {
    const parts = [];
    for (let i = 0; i < dims; i++) {
      parts.push(dv.getFloat64(off, le)); off += 8;
    }
    return [parts.join(' '), off];
  };

  if (baseType === 1) { // Point
    const [c, o] = readCoord(offset);
    return ['POINT' + dim + ' (' + c + ')', o];
  } else if (baseType === 2) { // LineString
    const n = dv.getUint32(offset, le); offset += 4;
    const coords = [];
    for (let i = 0; i < n; i++) { const [c, o] = readCoord(offset); offset = o; coords.push(c); }
    return ['LINESTRING' + dim + ' (' + coords.join(', ') + ')', offset];
  } else if (baseType === 3) { // Polygon
    const nRings = dv.getUint32(offset, le); offset += 4;
    const rings = [];
    for (let i = 0; i < nRings; i++) {
      const n = dv.getUint32(offset, le); offset += 4;
      const coords = [];
      for (let j = 0; j < n; j++) { const [c, o] = readCoord(offset); offset = o; coords.push(c); }
      rings.push('(' + coords.join(', ') + ')');
    }
    return ['POLYGON' + dim + ' (' + rings.join(', ') + ')', offset];
  }
  // Multi* 类型类似...
  throw new Error('Unsupported type: ' + fullType);
}

/**
 * 主转换函数
 * @param {File} file - GeoPackage 文件
 * @param {number} srid - 坐标系 SRID
 * @param {string} schema - Schema 名称（可选）
 * @returns {Promise<{createSql: string, insertSqls: Object}>}
 */
async function convertGpkgToPostgisSql(file, srid = 3857, schema = '') {
  const SQL = await initSqlJs({ locateFile: f => '/' + f });
  const buffer = await file.arrayBuffer();
  const db = new SQL.Database(new Uint8Array(buffer));

  const fmtTable = (name) => schema ? '"' + schema + '"."' + name + '"' : '"' + name + '"';

  // 获取所有业务表
  const tables = db.exec(
    "SELECT name FROM sqlite_master WHERE type='table' " +
    "AND name NOT LIKE 'gpkg_%' AND name NOT LIKE 'sqlite_%' AND name NOT LIKE 'rtree_%' " +
    "ORDER BY name"
  )[0].values.map(r => r[0]);

  // 获取几何列信息
  const geomMap = {};
  const geomResult = db.exec('SELECT table_name, column_name, geometry_type_name FROM gpkg_geometry_columns');
  if (geomResult.length) {
    geomResult[0].values.forEach(r => {
      geomMap[r[0]] = { col: r[1], type: r[2] };
    });
  }

  const createSqls = [
    'CREATE EXTENSION IF NOT EXISTS postgis;'
  ];
  if (schema) {
    createSqls.push('CREATE SCHEMA IF NOT EXISTS "' + schema + '";');
  }
  const insertSqls = {};

  for (const tableName of tables) {
    const columns = db.exec('PRAGMA table_info("' + tableName + '")')[0].values;
    const geomInfo = geomMap[tableName];
    const geomCol = geomInfo ? geomInfo.col : null;
    const geomType = geomInfo ? geomInfo.type : null;

    // 生成建表 SQL
    const colDefs = columns.map(col => {
      const [cid, name, type, notnull, dflt, pk] = col;
      let pgType;
      if (name === geomCol) {
        pgType = 'geometry(' + (geomType || 'Geometry') + ', ' + srid + ')';
      } else if (type.toUpperCase().startsWith('TEXT(')) {
        pgType = 'VARCHAR(' + type.match(/\d+/)[0] + ')';
      } else if (type.toUpperCase() === 'TEXT') {
        pgType = 'TEXT';
      } else {
        pgType = TYPE_MAP[type.toUpperCase()] || 'TEXT';
      }
      let def = '    "' + name + '" ' + pgType;
      if (pk) def += ' PRIMARY KEY';
      if (notnull && !pk) def += ' NOT NULL';
      return def;
    });
    createSqls.push('DROP TABLE IF EXISTS ' + fmtTable(tableName) + ';\n' +
      'CREATE TABLE ' + fmtTable(tableName) + ' (\n' + colDefs.join(',\n') + '\n);');

    // 生成插入 SQL
    const count = db.exec('SELECT COUNT(*) FROM "' + tableName + '"')[0].values[0][0];
    if (count > 0) {
      const data = db.exec('SELECT * FROM "' + tableName + '"')[0];
      const geomIdx = columns.findIndex(c => c[1] === geomCol);
      const lines = [];
      for (const row of data.values) {
        const vals = row.map((val, i) => {
          if (i === geomIdx && val instanceof Uint8Array) {
            const wkt = gpkgGeomToWkt(val);
            return wkt ? 'ST_GeomFromText(\'' + wkt + '\', ' + srid + ')' : 'NULL';
          }
          if (val === null) return 'NULL';
          if (typeof val === 'number') return String(val);
          if (val instanceof Uint8Array) return "'\\x" + Array.from(val).map(b => b.toString(16).padStart(2, '0')).join('') + "'";
          return "'" + String(val).replace(/'/g, "''") + "'";
        });
        lines.push('INSERT INTO ' + fmtTable(tableName) + ' VALUES (' + vals.join(', ') + ');');
      }
      insertSqls[tableName] = lines.join('\n');
    }
  }

  db.close();
  return { createSql: createSqls.join('\n\n'), insertSqls };
}

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
从 GeoPackage 文件读取所有表，生成 PostgreSQL/PostGIS 的建表 SQL 和插入数据 SQL。
几何列使用 PostGIS 的 ST_GeomFromText 函数插入 WKT 格式数据。
"""

import sqlite3
import struct
import os

GPKG_PATH = r'./input.gpkg'
OUTPUT_DIR = r'./postgis_sql'
SRID = 3857
SCHEMA = ''  # 留空使用 public schema

# SQLite 类型 -> PostgreSQL 类型映射
TYPE_MAP = {
    'INTEGER': 'INTEGER', 'TEXT': 'TEXT', 'REAL': 'DOUBLE PRECISION',
    'MEDIUMINT': 'INTEGER', 'DATETIME': 'TIMESTAMP', 'BLOB': 'BYTEA',
    'BOOLEAN': 'BOOLEAN', 'FLOAT': 'REAL',
}

def fmt_table(table_name):
    """格式化表名（带 Schema 前缀）"""
    if SCHEMA:
        return f'"{SCHEMA}"."{table_name}"'
    return f'"{table_name}"'

def parse_text_length(col_type):
    """从 TEXT(255) 中提取长度"""
    if col_type.startswith('TEXT'):
        if '(' in col_type:
            return int(col_type.split('(')[1].rstrip(')'))
        return None
    return None

def map_pg_type(col_name, col_type, geom_col_name=None, geom_type_name=None):
    """将 SQLite 列类型映射为 PostgreSQL 类型"""
    upper_type = col_type.upper().strip()
    if geom_col_name and col_name == geom_col_name:
        pg_geom_type = (geom_type_name or 'Geometry').upper()
        return f'geometry({pg_geom_type}, {SRID})'
    if upper_type.startswith('TEXT'):
        length = parse_text_length(upper_type)
        return f'VARCHAR({length})' if length else 'TEXT'
    base_type = upper_type.split('(')[0].strip()
    return TYPE_MAP.get(base_type, 'TEXT')

def gpkg_geom_to_wkt(blob):
    """将 GeoPackage 二进制几何数据转换为 WKT 格式"""
    if blob is None or len(blob) < 8:
        return None
    if blob[0:2] != b'GP':
        return None
    flags = blob[3]
    envelope_type = (flags >> 1) & 0x07
    if (flags >> 4) & 0x01:
        return None
    envelope_sizes = {0: 0, 1: 32, 2: 48, 3: 48, 4: 64}
    wkb_offset = 8 + envelope_sizes.get(envelope_type, 0)
    return wkb_to_wkt(blob[wkb_offset:])

def wkb_to_wkt(wkb):
    """将 WKB 二进制数据转换为 WKT 文本"""
    if not wkb or len(wkb) < 5:
        return None
    try:
        return parse_wkb_geometry(wkb, 0)[0]
    except Exception:
        return None

def read_double(data, offset, bo):
    fmt = '<d' if bo == 1 else '>d'
    return struct.unpack_from(fmt, data, offset)[0], offset + 8

def read_uint32(data, offset, bo):
    fmt = '<I' if bo == 1 else '>I'
    return struct.unpack_from(fmt, data, offset)[0], offset + 4

def parse_wkb_geometry(data, offset):
    """递归解析 WKB 几何体"""
    bo = data[offset]
    offset += 1
    geom_type, offset = read_uint32(data, offset, bo)
    if geom_type == 1:  # Point
        x, offset = read_double(data, offset, bo)
        y, offset = read_double(data, offset, bo)
        return f'POINT ({x} {y})', offset
    elif geom_type == 2:  # LineString
        n, offset = read_uint32(data, offset, bo)
        coords = []
        for _ in range(n):
            x, offset = read_double(data, offset, bo)
            y, offset = read_double(data, offset, bo)
            coords.append(f'{x} {y}')
        return f'LINESTRING ({", ".join(coords)})', offset
    elif geom_type == 3:  # Polygon
        n_rings, offset = read_uint32(data, offset, bo)
        rings = []
        for _ in range(n_rings):
            n, offset = read_uint32(data, offset, bo)
            coords = []
            for _ in range(n):
                x, offset = read_double(data, offset, bo)
                y, offset = read_double(data, offset, bo)
                coords.append(f'{x} {y}')
            rings.append(f'({", ".join(coords)})')
        return f'POLYGON ({", ".join(rings)})', offset
    elif geom_type == 4:  # MultiPoint
        n, offset = read_uint32(data, offset, bo)
        points = []
        for _ in range(n):
            wkt, offset = parse_wkb_geometry(data, offset)
            points.append(wkt.replace('POINT ', ''))
        return f'MULTIPOINT ({", ".join(points)})', offset
    elif geom_type == 5:  # MultiLineString
        n, offset = read_uint32(data, offset, bo)
        lines = []
        for _ in range(n):
            wkt, offset = parse_wkb_geometry(data, offset)
            lines.append(wkt.replace('LINESTRING ', ''))
        return f'MULTILINESTRING ({", ".join(lines)})', offset
    elif geom_type == 6:  # MultiPolygon
        n, offset = read_uint32(data, offset, bo)
        polys = []
        for _ in range(n):
            wkt, offset = parse_wkb_geometry(data, offset)
            polys.append(wkt.replace('POLYGON ', ''))
        return f'MULTIPOLYGON ({", ".join(polys)})', offset
    else:
        raise ValueError(f'Unsupported geometry type: {geom_type}')

def escape_sql_value(val):
    """转义 SQL 字符串值"""
    if val is None:
        return 'NULL'
    if isinstance(val, float):
        if val == int(val) and abs(val) < 1e15:
            return str(int(val))
        return str(val)
    if isinstance(val, int):
        return str(val)
    if isinstance(val, bytes):
        # 非 BLOB 列中的 bytes 数据跳过
        return 'NULL'
    return "'" + str(val).replace("'", "''") + "'"

def main():
    os.makedirs(OUTPUT_DIR, exist_ok=True)
    conn = sqlite3.connect(GPKG_PATH)
    cursor = conn.cursor()

    # 获取所有业务表
    cursor.execute(
        "SELECT name FROM sqlite_master WHERE type='table' "
        "AND name NOT LIKE 'gpkg_%' AND name NOT LIKE 'sqlite_%' "
        "AND name NOT LIKE 'rtree_%' ORDER BY name"
    )
    tables = [t[0] for t in cursor.fetchall()]

    # 获取几何列信息
    geom_map = {}
    cursor.execute('SELECT table_name, column_name, geometry_type_name FROM gpkg_geometry_columns')
    for row in cursor.fetchall():
        geom_map[row[0]] = {'col': row[1], 'type': row[2]}

    all_create_sqls = [
        '-- GeoPackage -> PostgreSQL/PostGIS 空间数据库',
        f'-- 坐标系: EPSG:{SRID}',
        '',
        'CREATE EXTENSION IF NOT EXISTS postgis;',
        '',
    ]
    if SCHEMA:
        all_create_sqls.append(f'CREATE SCHEMA IF NOT EXISTS "{SCHEMA}";')
        all_create_sqls.append('')

    for table_name in tables:
        cursor.execute(f'PRAGMA table_info("{table_name}")')
        columns = cursor.fetchall()
        geom_info = geom_map.get(table_name, {})
        geom_col = geom_info.get('col')
        geom_type = geom_info.get('type')

        # 生成建表 SQL
        col_defs = []
        for col in columns:
            cid, name, col_type, notnull, default, pk = col
            pg_type = map_pg_type(name, col_type, geom_col, geom_type)
            defn = f'    "{name}" {pg_type}'
            if pk: defn += ' PRIMARY KEY'
            if notnull and not pk: defn += ' NOT NULL'
            col_defs.append(defn)
        create_sql = f'DROP TABLE IF EXISTS {fmt_table(table_name)};\nCREATE TABLE {fmt_table(table_name)} (\n{",".join(col_defs)}\n);'
        all_create_sqls.append(create_sql)

        # 生成插入 SQL
        cursor.execute(f'SELECT COUNT(*) FROM "{table_name}"')
        count = cursor.fetchone()[0]
        if count > 0:
            geom_idx = next((i for i, c in enumerate(columns) if c[1] == geom_col), -1)
            cursor.execute(f'SELECT * FROM "{table_name}"')
            for row in cursor.fetchall():
                values = []
                for i, val in enumerate(row):
                    if i == geom_idx and val is not None:
                        wkt = gpkg_geom_to_wkt(val)
                        values.append(f"ST_GeomFromText('{wkt}', {SRID})" if wkt else 'NULL')
                    else:
                        values.append(escape_sql_value(val))
                col_list = ', '.join(f'"{c[1]}"' for c in columns)
                print(f'INSERT INTO {fmt_table(table_name)} ({col_list}) VALUES ({", ".join(values)});')

    # 写入建表 SQL 文件
    with open(os.path.join(OUTPUT_DIR, '00_create_tables.sql'), 'w', encoding='utf-8') as f:
        f.write('\n\n'.join(all_create_sqls))

    conn.close()
    print('完成!')

if __name__ == '__main__':
    main()