I’m having an issue using write(chunk). I don’t understand why the file is saved without data when the original uploaded KML file is smaller than 4KiB:
this part:
def _handleKmlFile(self, kml_file):
path = os.path.join(tempfile.gettempdir(), "temp.kml")
with open(path, "wb+") as destination:
for chunk in kml_file.chunks():
destination.write(chunk)
if os.path.getsize(path) == 0:
raise Exception("The KML file is empty after writing.")
yield path
class
import shutil
import os
import tempfile
from typing import List, Tuple
from ..models import (
LicensePoint,
LicenseLine,
LicensePolygon
)
from osgeo import ogr
from django.contrib.gis.geos import (
Polygon,
LineString,
LinearRing,
Point
)
ogr.UseExceptions()
class KmlLicense:
def __init__(self, kml_file, license_instance):
self.kml_file = kml_file
self.license_instance = license_instance
def process_kml(self):
print(f'Received file: {self.kml_file}')
handlerKml = self._handleKmlFile(self.kml_file)
print(f'handleKmlFile is {handlerKml}')
pathKml = next(handlerKml)
if not os.path.exists(pathKml):
raise Exception(f"The KML file does not exist at path: {pathKml}")
driver = ogr.GetDriverByName('KML')
dataSource = driver.Open(pathKml)
shutil.copy(pathKml, "/TempSave.kml")
if not dataSource:
raise Exception("Could not open the KML file")
print(f'DataSource created: {dataSource}')
self._deactivate_older_shape(dataSource)
for layer_index in range(dataSource.GetLayerCount()):
layer = dataSource.GetLayerByIndex(layer_index)
for feature in layer:
geometry = feature.GetGeometryRef()
geom_type = geometry.GetGeometryName()
geom_name = (
feature.GetField("name")
if feature.GetField("name") is not None
else geom_type
)
print(f'The name for {geom_type} is {geom_name}')
if geom_type == 'POLYGON':
try:
self._process_polygon(geometry, geom_name)
except Exception as e:
raise Exception(f'Error creating polygon: {e}')
elif geom_type == 'LINESTRING':
self._process_line(geometry, geom_name)
elif geom_type == 'POINT':
self._process_point(geometry, geom_name)
else:
raise Exception("Unsupported geometry")
next(handlerKml, None)
def _handleKmlFile(self, kml_file):
path = os.path.join(tempfile.gettempdir(), "temp.kml")
with open(path, "wb+") as destination:
for chunk in kml_file.chunks():
destination.write(chunk)
if os.path.getsize(path) == 0:
raise Exception("The KML file is empty after writing.")
yield path
def _deactivate_older_shape(self, dataSource):
polygon_count = 0
line_count = 0
point_count = 0
for layer_index in range(dataSource.GetLayerCount()):
layer = dataSource.GetLayerByIndex(layer_index)
for feature in layer:
geometry = feature.GetGeometryRef()
geom_type = geometry.GetGeometryName()
if geom_type == 'POLYGON':
polygon_count += 1
elif geom_type == 'LINESTRING':
line_count += 1
elif geom_type == 'POINT':
point_count += 1
if polygon_count > 0:
LicensePolygon.objects.filter(
license=self.license_instance).update(is_active=False)
if line_count > 0:
LicenseLine.objects.filter(
license=self.license_instance).update(is_active=False)
if point_count > 0:
LicensePoint.objects.filter(
license=self.license_instance).update(is_active=False)
def _process_polygon(
self,
polygon,
name: str
):
points: List[Tuple[float, float]] = []
ring = polygon.GetGeometryRef(0)
for point_index in range(ring.GetPointCount()):
lat, long, _ = ring.GetPoint(point_index)
points.append((lat, long))
linear_ring = LinearRing(points)
polygon_geometry = Polygon(linear_ring)
try:
new_polygon = LicensePolygon(
license=self.license_instance,
polygon_name=name,
polygon=polygon_geometry
)
new_polygon.save()
print(f'Polygon {name} created with data {new_polygon}')
except Exception as e:
raise Exception(f'Error saving polygon: {e}')
def _process_line(
self,
line,
name: str
):
points: List[Tuple[float, float]] = []
for point_index in range(line.GetPointCount()):
lat, long, _ = line.GetPoint(point_index)
points.append((lat, long))
line_geometry = LineString(points)
try:
new_line = LicenseLine(
license=self.license_instance,
line_name=name,
line=line_geometry
)
new_line.save()
print(f'Line {name} created with data {new_line}')
except Exception as e:
raise Exception(f'Error saving line: {e}')
def _process_point(
self,
point,
name: str
):
lat, long, _ = point.GetPoint()
point_geometry = Point(lat, long)
try:
new_point = LicensePoint(
license=self.license_instance,
point_name=name,
point=point_geometry
)
new_point.save()
print(f'Point {name} created with data {new_point}')
except Exception as e:
raise Exception(f'Error saving point: {e}')
from django.core.exceptions import ValidationError
from django.http import HttpResponseRedirect
from django.contrib.auth.decorators import login_required
from django.shortcuts import render
from django.contrib.auth.mixins import (
LoginRequiredMixin,
PermissionRequiredMixin
)
from django.views.generic import ListView, DetailView
from .models import License
from .forms import LicenseCreationForm
from .services.kmlProcessor import KmlLicense
class LicenseListView(LoginRequiredMixin, ListView):
model = License
context_object_name = "license_list"
template_name = "licenses/license_list.html"
ordering = ["-issue_date"]
class LicenseDetailView(
LoginRequiredMixin,
PermissionRequiredMixin,
DetailView
):
model = License
context_object_name = "license"
template_name = "licenses/license_detail.html"
login_url = "account_login"
permission_required = "licenses.special_status"
@login_required(redirect_field_name="")
def create_license(request):
if request.method == "POST":
form = LicenseCreationForm(request.POST, request.FILES)
if form.is_valid():
form.save()
license_instance = form.save()
if 'file_kml' in request.FILES:
kml = request.FILES['file_kml']
process_kml = KmlLicense(kml, license_instance)
try:
process_kml.process_kml()
except Exception as e:
raise ValidationError(
f"Error uploading KML file: {e}"
)
else:
# Debug: Check the contents of request.FILES
print("request.FILES:", request.FILES)
return HttpResponseRedirect("/licenses/")
else:
form = LicenseCreationForm()
return render(request, "licenses/license_create.html", {"form": form})
view
@login_required(redirect_field_name="")
def create_license(request):
if request.method == "POST":
form = LicenseCreationForm(request.POST, request.FILES)
if form.is_valid():
form.save()
license_instance = form.save()
if 'file_kml' in request.FILES:
kml = request.FILES['file_kml']
process_kml = KmlLicense(kml, license_instance)
try:
process_kml.process_kml()
except Exception as e:
raise ValidationError(
f"Error uploading KML file: {e}"
)
else:
# Debug: Check the contents of request.FILES
print("request.FILES:", request.FILES)
return HttpResponseRedirect("/licenses/")
else:
form = LicenseCreationForm()
return render(request, "licenses/license_create.html", {"form": form})