not working

geospatial-data-converter/kml_tricks.py

@@ -1,19 +1,24 @@
 import zipfile
+from io import StringIO
 from typing import Any
 
 import bs4
-import fiona
 import geopandas as gpd
+import lxml  # nosec
 import pandas as pd
 
-fiona.drvsupport.supported_drivers["KML"] = "rw"
-
 
 def parse_description_to_gdf(gdf: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
-    parsed_dataframes = [
-        pd.read_html(desc)[1 if len(pd.read_html(desc)) > 1 else 0].T
-        for desc in gdf["Description"]
-    ]
+    def _gen():
+        for desc in gdf["Description"]:
+            try:
+                html_df = pd.read_html(StringIO(desc), flavor="lxml")
+                yield html_df[-1].T
+            except (lxml.etree.ParserError, lxml.etree.XMLSyntaxError) as e:
+                print(desc)
+                raise pd.errors.ParserError from e
+
+    parsed_dataframes = list(_gen())
 
     for df in parsed_dataframes:
         df.columns = df.iloc[0]
@@ -34,12 +39,18 @@ def read_kml_file(path: str) -> Any:
             "KMZ contains more than one KML. Extract or convert to multiple KMLs.",
         )
 
-    return gpd.read_file(f"zip://{path}\\{kml_files[0]}", driver="KML")
+    return gpd.read_file(
+        f"zip://{path}\\{kml_files[0]}",
+        driver="KML",
+        engine="pyogrio",
+    )
 
 
 def parse_file_to_gdf(path: str) -> gpd.GeoDataFrame:
     if path.endswith(".kml"):
-        return parse_description_to_gdf(gpd.read_file(path, driver="KML"))
+        return parse_description_to_gdf(
+            gpd.read_file(path, driver="KML", engine="pyogrio"),
+        )
 
     if path.endswith(".kmz"):
         return parse_description_to_gdf(read_kml_file(path))
@@ -48,7 +59,7 @@ def parse_file_to_gdf(path: str) -> gpd.GeoDataFrame:
 
 
 def extract_data_from_kml_code(kml_code: str) -> pd.DataFrame:
-    soup = bs4.BeautifulSoup(kml_code, "html.parser")
+    soup = bs4.BeautifulSoup(kml_code, features="xml")
     rows = soup.find_all("schemadata")
 
     data = (
@@ -61,7 +72,6 @@ def extract_data_from_kml_code(kml_code: str) -> pd.DataFrame:
 
 def extract_kml_from_file(file_path: str) -> str:
     file_extension = file_path.lower().split(".")[-1]
-    kml_files = None
 
     if file_extension == "kml":
         with open(file_path, "r") as kml:
@@ -70,28 +80,28 @@ def extract_kml_from_file(file_path: str) -> str:
     if file_extension == "kmz":
         with zipfile.ZipFile(file_path) as kmz:
             kml_files = [f for f in kmz.namelist() if f.lower().endswith(".kml")]
-        if len(kml_files) != 1:
-            raise IndexError(
-                "KMZ contains more than one KML. Extract or convert to multiple KMLs.",
-            )
-        with kmz.open(kml_files[0]) as kml:
-            return kml.read().decode()
+            if len(kml_files) != 1:
+                raise IndexError(
+                    "KMZ contains more than one KML. Extract or convert to multiple KMLs.",
+                )
+            with kmz.open(kml_files[0]) as kml:
+                return kml.read().decode()
 
     raise ValueError("File path must end with .kml or .kmz")
 
 
-def extract_data_from_file(file_path: str) -> pd.DataFrame:
+def extract_data_from_file(file_path: str) -> gpd.GeoDataFrame:
     df = extract_data_from_kml_code(extract_kml_from_file(file_path))
 
     if file_path.endswith(".kmz"):
         file_gdf = read_kml_file(file_path)
     else:
-        file_gdf = gpd.read_file(file_path, driver="KML")
+        file_gdf = gpd.read_file(file_path, driver="KML", engine="pyogrio")
 
     return gpd.GeoDataFrame(df, geometry=file_gdf["geometry"], crs=file_gdf.crs)
 
 
-def read_ge_file(file_path: str) -> pd.DataFrame:
+def read_ge_file(file_path: str) -> gpd.GeoDataFrame:
     try:
         return parse_file_to_gdf(file_path)
     except (pd.errors.ParserError, ValueError):

geospatial-data-converter/utils.py

@@ -6,6 +6,8 @@ from typing import BinaryIO
 
 import geopandas as gpd
 
+from kml_tricks import read_ge_file
+
 output_format_dict = {
     "ESRI Shapefile": ("shp", "zip", "application/zip"),  # must be zipped
     "OpenFileGDB": ("gdb", "zip", "application/zip"),  # must be zipped
@@ -17,7 +19,9 @@ output_format_dict = {
 
 def read_file(file: BinaryIO, *args, **kwargs) -> gpd.GeoDataFrame:
     """Read a file and return a GeoDataFrame"""
-    if file.name.lower().endswith(".zip"):
+    basename, ext = os.path.splitext(os.path.basename(file.name))
+    ext = ext.lower().strip(".")
+    if ext == "zip":
         with TemporaryDirectory() as tmp_dir:
             tmp_file_path = os.path.join(tmp_dir, file.name)
             with open(tmp_file_path, "wb") as tmp_file:
@@ -28,6 +32,12 @@ def read_file(file: BinaryIO, *args, **kwargs) -> gpd.GeoDataFrame:
                 engine="pyogrio",
                 **kwargs,
             )
+    elif ext in ("kml", "kmz"):
+        with TemporaryDirectory() as tmp_dir:
+            tmp_file_path = os.path.join(tmp_dir, file.name)
+            with open(tmp_file_path, "wb") as tmp_file:
+                tmp_file.write(file.read())
+            return read_ge_file(tmp_file_path)
     return gpd.read_file(file, *args, engine="pyogrio", **kwargs)