mpeltriaux
cf050a92ef
* refactors settings into env usage * adds proxy usage for schneider parcel fetching (using public web address instead of internal ip address)
65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
"""
|
|
Author: Michel Peltriaux
|
|
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
|
|
Contact: ksp-servicestelle@sgdnord.rlp.de
|
|
Created on: 14.12.22
|
|
|
|
"""
|
|
import json
|
|
from json import JSONDecodeError
|
|
|
|
import requests
|
|
|
|
from konova.sub_settings import schneider_settings
|
|
from konova.sub_settings.proxy_settings import PROXIES
|
|
|
|
|
|
class ParcelFetcher:
|
|
base_url = schneider_settings.base_url
|
|
auth_header = schneider_settings.auth_header
|
|
auth_header_token = schneider_settings.auth_header_token
|
|
|
|
geometry = None
|
|
geojson = None
|
|
results = None
|
|
|
|
def __init__(self, geometry):
|
|
if geometry is None:
|
|
raise AssertionError("Geometry must not be none")
|
|
self.geometry = geometry
|
|
|
|
# Reduce size of geometry to avoid "intersections" because of exact border matching
|
|
buffer_threshold = 0.001
|
|
geom = geometry.geom.buffer(-buffer_threshold)
|
|
if geom.area < buffer_threshold:
|
|
# Fallback for malicious geometries which are way too small and would disappear on negative buffering
|
|
geom = geometry.geom
|
|
self.geojson = geom.ewkt
|
|
self.results = []
|
|
|
|
def get_parcels(self, url: str = None):
|
|
post_url = url
|
|
if post_url is None:
|
|
post_url = f"{self.base_url}/parcel/intersect/"
|
|
|
|
response = requests.post(
|
|
url=post_url,
|
|
proxies=PROXIES,
|
|
data=self.geojson,
|
|
headers={
|
|
self.auth_header: self.auth_header_token
|
|
}
|
|
)
|
|
|
|
try:
|
|
content = json.loads(response.content.decode("utf-8"))
|
|
except JSONDecodeError:
|
|
content = {}
|
|
next = content.get("next", None)
|
|
fetched_parcels = content.get("results", [])
|
|
self.results += fetched_parcels
|
|
|
|
if next:
|
|
self.get_parcels(next)
|
|
|
|
return self.results |