sanya-dbms/processing/tools/topology/spatialite.py

import sqlite3

# Path to your SpatiaLite database
db_path = "D:\\temp\\output.sqlite"

# Connect to the SQLite database
conn = sqlite3.connect(db_path)

# Enable loading of extensions
conn.enable_load_extension(True)

# Load the SpatiaLite extension
try:
    conn.execute("SELECT load_extension('mod_spatialite');")
    print("SpatiaLite extension loaded successfully!")
except sqlite3.OperationalError as e:
    print(f"Error loading SpatiaLite extension: {e}")

# Create a cursor to execute queries
cursor = conn.cursor()

# Example: Check SpatiaLite version
cursor.execute("drop table if exists topology_table_temp;")
cursor.execute("create table topology_table_temp(PIEID INTEGER PRIMARY KEY not null,PIEUUID VARCHAR(128),PIEVALIDED BOOLEAN,sid,idx,rulecode,geometry MULTILINESTRING);")
cursor.execute("with t as( select 0 sid, 0 did, numgeometries(linesfromrings(gunion(geometry))) cnt, linesfromrings(gunion(geometry)) geometry from topology_table ), cte(idx, sid, did, cnt, geometry, subgeom) AS ( SELECT 1 idx, sid, did, cnt, geometry, geometryN(geometry, 1) subgeom from t UNION ALL SELECT idx + 1 idx, sid, did, cnt, geometry, geometryN(geometry, idx + 1) subgeom FROM cte WHERE idx < cnt ) insert into topology_table_temp SELECT null,createuuid(),isvalid(casttomulti(subgeom)), sid, idx, 'A03' rulecode, casttomulti(subgeom) geometry FROM cte; ")
cursor.execute("select recovergeometrycolumn('topology_table_temp', 'geometry',  (select srid from geometry_columns where f_table_name = 'topology_table'), 'MULTILINESTRING', 'XY');")
# version = cursor.fetchone()
# print(f"SpatiaLite version: {version[0]}")

# Close the connection when done
conn.close()