import datetime import json import re import time import requests import streetview from django.utils.text import slugify from geo_location.models import City, Country, Demographic, District, Iso31662, Location, LondonArea, LondonZipState, LondonZone, PostCode, PriceGrowth, PropertyDataRaw, Region from django.conf import settings from django.utils.text import slugify from django.contrib.gis.geos import GEOSGeometry, Polygon, MultiPolygon class StreetView: def __init__(self, api_key=None) -> None: self.api_key = api_key def get_panoramas(self, lat, lon): ps = streetview.search_panoramas(lat=lat, lon=lon) return ps async def get_panorama_async(self, id): image = await streetview.get_panorama_async(pano_id=id) image.save(f"downloads/{id}.jpg", "jpeg") return image def get_panorama(self, id, zoom=1): image = streetview.get_panorama(pano_id=id, zoom=zoom) image.save(f"downloads/{id}.jpg", "jpeg") class GeoLocation: from geopy.geocoders import Nominatim geo_locator = Nominatim( user_agent=f"Entralon/{datetime.datetime.now().strftime('%Y%m%d%H')}") def __init__(self) -> None: self.location = {} def get_address(self, lat, lon, language="en"): location = self.geo_locator.reverse(f"{lat:.6f}, {lon:.6f}", language=language) return location def get_location(self, address): location = self.geo_locator.geocode(address) return location def check_location_address(): locations = Location.objects.filter(address__isnull=True) print(f"location num: {locations.count()}") geo_location = GeoLocation() for location in locations: print(f"location id: {location.id}") gl = geo_location.get_address( location.coordinates.y, location.coordinates.x) if gl is not None: location.address = gl.address location.raw = gl.raw location.save() else: print(gl) def set_iso_3166(): dir = f"{settings.BASE_DIR}/geo_location/files" country = Country.objects.get(id=3) for i in range(4): with open(f'{dir}/gadm41_ARE_{i}.json', 'r', encoding="utf8") as file: data = json.load(file, ) for d in data['features']: l_properties = d.get('properties') gid_0 = l_properties.get("GID_0", "NA") gid_0 = gid_0 if gid_0 != "NA" else None gid_1 = l_properties.get("GID_1", "NA") gid_1 = gid_1 if gid_1 != "NA" else None gid_2 = l_properties.get("GID_2", "NA") gid_2 = gid_2 if gid_2 != "NA" else None gid_3 = l_properties.get("GID_3", "NA") gid_3 = gid_3 if gid_3 != "NA" else None gid_4 = l_properties.get("GID_4", "NA") gid_4 = gid_4 if gid_4 != "NA" else None name_0 = l_properties.get("NAME_0", "NA") name_0 = re.sub(r"(\w)([A-Z])", r"\1 \2",name_0) if name_0 != "NA" else None name_1 = l_properties.get("NAME_1", "NA") name_1 = re.sub(r"(\w)([A-Z])", r"\1 \2",name_1) if name_1 != "NA" else None name_2 = l_properties.get("NAME_2", "NA") name_2 = re.sub(r"(\w)([A-Z])", r"\1 \2",name_2) if name_2 != "NA" else None name_3 = l_properties.get("NAME_3", "NA") name_3 = re.sub(r"(\w)([A-Z])", r"\1 \2",name_3) if name_3 != "NA" else None name_4 = l_properties.get("NAME_4", "NA") name_4 = re.sub(r"(\w)([A-Z])", r"\1 \2",name_4) if name_4 != "NA" else None country_name = l_properties.get("COUNTRY", "NA") country_name = re.sub(r"(\w)([A-Z])", r"\1 \2",country_name) if country_name != "NA" else None nl_name_0 = l_properties.get("NL_NAME_0", "NA") nl_name_0 = nl_name_0 if nl_name_0 != "NA" else None nl_name_1 = l_properties.get("NL_NAME_1", "NA") nl_name_1 = nl_name_1 if nl_name_1 != "NA" else None nl_name_2 = l_properties.get("NL_NAME_2", "NA") nl_name_2 = nl_name_2 if nl_name_2 != "NA" else None nl_name_3 = l_properties.get("NL_NAME_3", "NA") nl_name_3 = nl_name_3 if nl_name_3 != "NA" else None nl_name_4 = l_properties.get("NL_NAME_4", "NA") nl_name_4 = nl_name_4 if nl_name_4 != "NA" else None var_name_0 = l_properties.get("VARNAME_0", "NA") var_name_0 = var_name_0 if var_name_0 != "NA" else None var_name_1 = l_properties.get("VARNAME_1", "NA") var_name_1 = var_name_1 if var_name_1 != "NA" else None var_name_2 = l_properties.get("VARNAME_2", "NA") var_name_2 = var_name_2 if var_name_2 != "NA" else None var_name_3 = l_properties.get("VARNAME_3", "NA") var_name_3 = var_name_3 if var_name_3 != "NA" else None var_name_4 = l_properties.get("VARNAME_4", "NA") var_name_4 = var_name_4 if var_name_4 != "NA" else None type_0 = l_properties.get("TYPE_0", "NA") type_0 = type_0 if type_0 != "NA" else None type_1 = l_properties.get("TYPE_1", "NA") type_1 = type_1 if type_1 != "NA" else None type_2 = l_properties.get("TYPE_2", "NA") type_2 = type_2 if type_2 != "NA" else None type_3 = l_properties.get("TYPE_3", "NA") type_3 = type_3 if type_3 != "NA" else None type_4 = l_properties.get("TYPE_4", "NA") type_4 = type_4 if type_4 != "NA" else None eng_type_0 = l_properties.get("ENGTYPE_0", "NA") eng_type_0 = re.sub(r"(\w)([A-Z])", r"\1 \2",eng_type_0) if eng_type_0 != "NA" else None eng_type_1 = l_properties.get("ENGTYPE_1", "NA") eng_type_1 = re.sub(r"(\w)([A-Z])", r"\1 \2",eng_type_1) if eng_type_1 != "NA" else None eng_type_2 = l_properties.get("ENGTYPE_2", "NA") eng_type_2 = re.sub(r"(\w)([A-Z])", r"\1 \2",eng_type_2) if eng_type_2 != "NA" else None eng_type_3 = l_properties.get("ENGTYPE_3", "NA") eng_type_3 = re.sub(r"(\w)([A-Z])", r"\1 \2",eng_type_3) if eng_type_3 != "NA" else None eng_type_4 = l_properties.get("ENGTYPE_4", "NA") eng_type_4 = re.sub(r"(\w)([A-Z])", r"\1 \2",eng_type_4) if eng_type_4 != "NA" else None cc_0 = l_properties.get("CC_0", "NA") cc_0 = cc_0 if cc_0 != "NA" else None cc_1 = l_properties.get("CC_1", "NA") cc_1 = cc_1 if cc_1 != "NA" else None cc_2 = l_properties.get("CC_2", "NA") cc_2 = cc_2 if cc_2 != "NA" else None cc_3 = l_properties.get("CC_3", "NA") cc_3 = cc_3 if cc_3 != "NA" else None cc_4 = l_properties.get("CC_4", "NA") cc_4 = cc_4 if cc_4 != "NA" else None hasc_0 = l_properties.get("HASC_0", "NA") hasc_0 = hasc_0 if hasc_0 != "NA" else None hasc_1 = l_properties.get("HASC_1", "NA") hasc_1 = hasc_1 if hasc_1 != "NA" else None hasc_2 = l_properties.get("HASC_2", "NA") hasc_2 = hasc_2 if hasc_2 != "NA" else None hasc_3 = l_properties.get("HASC_3", "NA") hasc_3 = hasc_3 if hasc_3 != "NA" else None hasc_4 = l_properties.get("HASC_4", "NA") hasc_4 = hasc_4 if hasc_4 != "NA" else None iso_1 = l_properties.get("ISO_1", "NA") iso_1 = iso_1 if iso_1 != "NA" else None code = None geo_type = None gid = None parent = None name = None slug = None if i == 4: gid = gid_4 name = name_4 try: parent = Iso31662.objects.get(gid=gid_3) except: pass elif i == 3: geo_type = eng_type_3 or "Municipality" gid = gid_3 name = name_3 or gid_3 try: parent = Iso31662.objects.get(gid=gid_2) except: pass elif i == 2: geo_type = eng_type_2 or "Municipal Region" gid = gid_2 name = name_2 or gid_2 try: parent = Iso31662.objects.get(gid=gid_1) except: pass elif i == 1: geo_type = eng_type_1 or "Emirate" code = iso_1 gid = gid_1 name = name_1 or gid_1 try: parent = Iso31662.objects.get(gid=gid_0) except: pass elif i == 0: geo_type = "Country" code = gid_0 gid = gid_0 name = country_name if name is None: print(f"ARE_{i} name error: {gid_0} {gid_1} {gid_2} {gid_3} {gid_4}") continue slug = slugify(name, allow_unicode=True) l_geometry = d.get('geometry') geom_str = json.dumps(l_geometry) geom = GEOSGeometry(geom_str) try: iso = Iso31662.objects.get(gid=gid) except: iso = Iso31662.objects.create(slug=slug, gid=gid, name=name, country=country) iso.level = i iso.code = code iso.name = name iso.coordinates = geom iso.type = geo_type iso.gid = gid iso.parent = parent iso.gid_0 = gid_0 iso.gid_1 = gid_1 iso.gid_2 = gid_2 iso.gid_3 = gid_3 iso.gid_4 = gid_4 iso.name_0 = name_0 iso.name_1 = name_1 iso.name_2 = name_2 iso.name_3 = name_3 iso.name_4 = name_4 iso.nl_name_0 = nl_name_0 iso.nl_name_1 = nl_name_1 iso.nl_name_2 = nl_name_2 iso.nl_name_3 = nl_name_3 iso.nl_name_4 = nl_name_4 iso.var_name_0 = var_name_0 iso.var_name_1 = var_name_1 iso.var_name_2 = var_name_2 iso.var_name_3 = var_name_3 iso.var_name_4 = var_name_4 iso.type_0 = type_0 iso.type_1 = type_1 iso.type_2 = type_2 iso.type_3 = type_3 iso.type_4 = type_4 iso.eng_type_0 = eng_type_0 iso.eng_type_1 = eng_type_1 iso.eng_type_2 = eng_type_2 iso.eng_type_3 = eng_type_3 iso.eng_type_4 = eng_type_4 iso.cc_0 = cc_0 iso.cc_1 = cc_1 iso.cc_2 = cc_2 iso.cc_3 = cc_3 iso.cc_4 = cc_4 iso.hasc_0 = hasc_0 iso.hasc_1 = hasc_1 iso.hasc_2 = hasc_2 iso.hasc_3 = hasc_3 iso.hasc_4 = hasc_4 iso.iso_1 = iso_1 iso.save() locations = Location.objects.filter(coordinates__intersects=iso.coordinates) if i == 0: locations.update(iso_3166_2_lvl_0=iso) elif i == 1: locations.update(iso_3166_2_lvl_1=iso) elif i == 2: locations.update(iso_3166_2_lvl_2=iso) elif i == 3: locations.update(iso_3166_2_lvl_3=iso) elif i == 4: locations.update(iso_3166_2_lvl_4=iso) country = Country.objects.get(id=2) for i in range(5): with open(f'{dir}/gadm41_GBR_{i}.json', 'r', encoding="utf8") as file: data = json.load(file) for d in data['features']: l_properties = d.get('properties') gid_0 = l_properties.get("GID_0", "NA") gid_0 = gid_0 if gid_0 != "NA" else None gid_1 = l_properties.get("GID_1", "NA") gid_1 = gid_1 if gid_1 != "NA" else None gid_2 = l_properties.get("GID_2", "NA") gid_2 = gid_2 if gid_2 != "NA" else None gid_3 = l_properties.get("GID_3", "NA") gid_3 = gid_3 if gid_3 != "NA" else None gid_4 = l_properties.get("GID_4", "NA") gid_4 = gid_4 if gid_4 != "NA" else None name_0 = l_properties.get("NAME_0", "NA") name_0 = re.sub(r"(\w)([A-Z])", r"\1 \2",name_0) if name_0 != "NA" else None name_1 = l_properties.get("NAME_1", "NA") name_1 = re.sub(r"(\w)([A-Z])", r"\1 \2",name_1) if name_1 != "NA" else None name_2 = l_properties.get("NAME_2", "NA") name_2 = re.sub(r"(\w)([A-Z])", r"\1 \2",name_2) if name_2 != "NA" else None name_3 = l_properties.get("NAME_3", "NA") name_3 = re.sub(r"(\w)([A-Z])", r"\1 \2",name_3) if name_3 != "NA" else None name_4 = l_properties.get("NAME_4", "NA") name_4 = re.sub(r"(\w)([A-Z])", r"\1 \2",name_4) if name_4 != "NA" else None country_name = l_properties.get("COUNTRY", "NA") country_name = re.sub(r"(\w)([A-Z])", r"\1 \2",country_name) if country_name != "NA" else None nl_name_0 = l_properties.get("NL_NAME_0", "NA") nl_name_0 = nl_name_0 if nl_name_0 != "NA" else None nl_name_1 = l_properties.get("NL_NAME_1", "NA") nl_name_1 = nl_name_1 if nl_name_1 != "NA" else None nl_name_2 = l_properties.get("NL_NAME_2", "NA") nl_name_2 = nl_name_2 if nl_name_2 != "NA" else None nl_name_3 = l_properties.get("NL_NAME_3", "NA") nl_name_3 = nl_name_3 if nl_name_3 != "NA" else None nl_name_4 = l_properties.get("NL_NAME_4", "NA") nl_name_4 = nl_name_4 if nl_name_4 != "NA" else None var_name_0 = l_properties.get("VARNAME_0", "NA") var_name_0 = var_name_0 if var_name_0 != "NA" else None var_name_1 = l_properties.get("VARNAME_1", "NA") var_name_1 = var_name_1 if var_name_1 != "NA" else None var_name_2 = l_properties.get("VARNAME_2", "NA") var_name_2 = var_name_2 if var_name_2 != "NA" else None var_name_3 = l_properties.get("VARNAME_3", "NA") var_name_3 = var_name_3 if var_name_3 != "NA" else None var_name_4 = l_properties.get("VARNAME_4", "NA") var_name_4 = var_name_4 if var_name_4 != "NA" else None type_0 = l_properties.get("TYPE_0", "NA") type_0 = type_0 if type_0 != "NA" else None type_1 = l_properties.get("TYPE_1", "NA") type_1 = type_1 if type_1 != "NA" else None type_2 = l_properties.get("TYPE_2", "NA") type_2 = type_2 if type_2 != "NA" else None type_3 = l_properties.get("TYPE_3", "NA") type_3 = type_3 if type_3 != "NA" else None type_4 = l_properties.get("TYPE_4", "NA") type_4 = type_4 if type_4 != "NA" else None eng_type_0 = l_properties.get("ENGTYPE_0", "NA") eng_type_0 = re.sub(r"(\w)([A-Z])", r"\1 \2",eng_type_0) if eng_type_0 != "NA" else None eng_type_1 = l_properties.get("ENGTYPE_1", "NA") eng_type_1 = re.sub(r"(\w)([A-Z])", r"\1 \2",eng_type_1) if eng_type_1 != "NA" else None eng_type_2 = l_properties.get("ENGTYPE_2", "NA") eng_type_2 = re.sub(r"(\w)([A-Z])", r"\1 \2",eng_type_2) if eng_type_2 != "NA" else None eng_type_3 = l_properties.get("ENGTYPE_3", "NA") eng_type_3 = re.sub(r"(\w)([A-Z])", r"\1 \2",eng_type_3) if eng_type_3 != "NA" else None eng_type_4 = l_properties.get("ENGTYPE_4", "NA") eng_type_4 = re.sub(r"(\w)([A-Z])", r"\1 \2",eng_type_4) if eng_type_4 != "NA" else None cc_0 = l_properties.get("CC_0", "NA") cc_0 = cc_0 if cc_0 != "NA" else None cc_1 = l_properties.get("CC_1", "NA") cc_1 = cc_1 if cc_1 != "NA" else None cc_2 = l_properties.get("CC_2", "NA") cc_2 = cc_2 if cc_2 != "NA" else None cc_3 = l_properties.get("CC_3", "NA") cc_3 = cc_3 if cc_3 != "NA" else None cc_4 = l_properties.get("CC_4", "NA") cc_4 = cc_4 if cc_4 != "NA" else None hasc_0 = l_properties.get("HASC_0", "NA") hasc_0 = hasc_0 if hasc_0 != "NA" else None hasc_1 = l_properties.get("HASC_1", "NA") hasc_1 = hasc_1 if hasc_1 != "NA" else None hasc_2 = l_properties.get("HASC_2", "NA") hasc_2 = hasc_2 if hasc_2 != "NA" else None hasc_3 = l_properties.get("HASC_3", "NA") hasc_3 = hasc_3 if hasc_3 != "NA" else None hasc_4 = l_properties.get("HASC_4", "NA") hasc_4 = hasc_4 if hasc_4 != "NA" else None iso_1 = l_properties.get("ISO_1", "NA") iso_1 = iso_1 if iso_1 != "NA" else None code = None geo_type = None gid = None parent = None name = None slug = None if i == 4: gid = gid_4 name = name_4 geo_type = eng_type_4 or "Metropolitan District Ward" try: parent = Iso31662.objects.get(gid=gid_3) except: pass elif i == 3: geo_type = eng_type_3 or "Unitary Authority" gid = gid_3 name = name_3 or gid_3 try: parent = Iso31662.objects.get(gid=gid_2) except: pass elif i == 2: geo_type = eng_type_2 or "Metropolitan Borough (City)" gid = gid_2 name = name_2 or gid_2 try: parent = Iso31662.objects.get(gid=gid_1) except: pass elif i == 1: geo_type = eng_type_1 or "Constituent Country" code = iso_1 gid = gid_1 name = name_1 or gid_1 try: parent = Iso31662.objects.get(gid=gid_0) except: pass elif i == 0: geo_type = "Country" code = gid_0 gid = gid_0 name = country_name if name is None: print(f"GBR_{i} name error: {gid_0} {gid_1} {gid_2} {gid_3} {gid_4}") continue slug = slugify(name, allow_unicode=True) l_geometry = d.get('geometry') geom_str = json.dumps(l_geometry) geom = GEOSGeometry(geom_str) try: iso = Iso31662.objects.get(gid=gid) except: iso = Iso31662.objects.create(slug=slug, gid=gid, name=name, country=country) iso.level = i iso.code = code iso.name=name iso.coordinates = geom iso.type = geo_type iso.gid = gid iso.parent = parent iso.gid_0 = gid_0 iso.gid_1 = gid_1 iso.gid_2 = gid_2 iso.gid_3 = gid_3 iso.gid_4 = gid_4 iso.name_0 = name_0 iso.name_1 = name_1 iso.name_2 = name_2 iso.name_3 = name_3 iso.name_4 = name_4 iso.nl_name_0 = nl_name_0 iso.nl_name_1 = nl_name_1 iso.nl_name_2 = nl_name_2 iso.nl_name_3 = nl_name_3 iso.nl_name_4 = nl_name_4 iso.var_name_0 = var_name_0 iso.var_name_1 = var_name_1 iso.var_name_2 = var_name_2 iso.var_name_3 = var_name_3 iso.var_name_4 = var_name_4 iso.type_0 = type_0 iso.type_1 = type_1 iso.type_2 = type_2 iso.type_3 = type_3 iso.type_4 = type_4 iso.eng_type_0 = eng_type_0 iso.eng_type_1 = eng_type_1 iso.eng_type_2 = eng_type_2 iso.eng_type_3 = eng_type_3 iso.eng_type_4 = eng_type_4 iso.cc_0 = cc_0 iso.cc_1 = cc_1 iso.cc_2 = cc_2 iso.cc_3 = cc_3 iso.cc_4 = cc_4 iso.hasc_0 = hasc_0 iso.hasc_1 = hasc_1 iso.hasc_2 = hasc_2 iso.hasc_3 = hasc_3 iso.hasc_4 = hasc_4 iso.iso_1 = iso_1 iso.save() locations = Location.objects.filter(coordinates__intersects=iso.coordinates) if i == 0: locations.update(iso_3166_2_lvl_0=iso) elif i == 1: locations.update(iso_3166_2_lvl_1=iso) elif i == 2: locations.update(iso_3166_2_lvl_2=iso) elif i == 3: locations.update(iso_3166_2_lvl_3=iso) elif i == 4: locations.update(iso_3166_2_lvl_4=iso) def set_area(): # ['city', 'zipstate', 'innerlondon', 'country', 'zone', 'area', 'borough', 'outherlondon', 'districts'] dir = f"{settings.BASE_DIR}/geo_location/files" with open(f'{dir}/area.json', 'r', encoding="utf8") as file: data = json.load(file, ) for d in data: name = d['properties'].get('Name') local_name = d['properties'].get('LocalName') if local_name is not None: country = Country.objects.get(id=3) city = City.objects.get(id=3) else: country = Country.objects.get(id=2) city = City.objects.get(id=1) slug = slugify(name, allow_unicode=True) category = d.get('Category') geometry = d.get('geometry') try: geom_str = json.dumps(geometry) geom = GEOSGeometry(geom_str) if isinstance(geom, MultiPolygon):pass elif isinstance(geom, Polygon): geom = MultiPolygon([geom]) if category == "zipstate": try: zip_state = LondonZipState.objects.get(slug=slug) except: zip_state = LondonZipState.objects.create(name=name, slug=slug) zip_state.city = city zip_state.country = country zip_state.coordinates = geom zip_state.save() elif category == "zone": try: zone = LondonZone.objects.get(slug=slug) except: zone = LondonZone.objects.create(name=name, slug=slug) zone.city = city zone.country = country zone.coordinates = geom zone.save() elif category == "area": try: area = LondonArea.objects.get(slug=slug) except: area = LondonArea.objects.create(name=name, slug=slug) area.city = city area.country = country area.coordinates = geom area.save() elif category == "districts": try: district = District.objects.get(slug=slug) except: district = District.objects.create(name=name, slug=slug) district.city = city district.country = country district.coordinates = geom district.save() elif category == "city": try: city = City.objects.get(slug=slug) except: city = City.objects.create(name=name, slug=slug) city.coordinates = geom city.save() except Exception as e: print(f"for {name} type {category} error: {e}") def get_are_locations(): for obj in Country.objects.all(): if obj.coordinates is not None: Location.objects.filter(coordinates__intersects=obj.coordinates).update(country=obj) for obj in City.objects.all(): if obj.coordinates is not None: Location.objects.filter(coordinates__intersects=obj.coordinates).update(city=obj) for obj in District.objects.all(): if obj.coordinates is not None: Location.objects.filter(coordinates__intersects=obj.coordinates).update(district=obj) for obj in LondonZipState.objects.all(): if obj.coordinates is not None: Location.objects.filter(coordinates__intersects=obj.coordinates).update(london_zip_state=obj) for obj in LondonArea.objects.all(): if obj.coordinates is not None: Location.objects.filter(coordinates__intersects=obj.coordinates).update(london_area=obj) for obj in LondonZone.objects.all(): if obj.coordinates is not None: Location.objects.filter(coordinates__intersects=obj.coordinates).update(london_zone=obj) for obj in LondonZipState.objects.all(): if obj.coordinates is not None: Location.objects.filter(coordinates__intersects=obj.coordinates).update(london_zip_state=obj) def set_price_growth(): from django.conf import settings from django.utils.text import slugify dir = f"{settings.BASE_DIR}/geo_location/files" country = Country.objects.get(id=2) with open(f'{dir}/propdata-uk-growth.json', 'r', encoding="utf8") as file: data = json.load(file, ) MONTH = {'JAN': '1', 'FEB': '2', 'MAR': '3', 'APR': '4', 'MAY': '5','JUN': '6', 'JUL': '7', 'AUG': '8', 'SEP': '9', 'OCT': '10', 'NOV': '11', 'DEC': '12'} data = data.get('growth') for g_d in data: postcode = g_d.get('postcode') postcode_type = g_d.get('postcode_type') g_data = g_d.get('data') last_prise = None price_change = None for g in g_data: year = int(g[0][-4:]) try: month = int(MONTH.get(g[0][:3])) except: month = 11 price = float(g[1]) price_change = price - last_prise if last_prise else None last_prise = price change_percent = float(g[2][:-1]) if g[2] else None date = datetime.date(year, month, 1) try: PriceGrowth.objects.get(country=country, postcode=postcode, date=date) except: PriceGrowth.objects.create(country=country, postcode=postcode, date=date, postcode_type=postcode_type, price=price, price_change=price_change, change_percent=change_percent) def set_demographics(): from django.conf import settings from django.utils.text import slugify dir = f"{settings.BASE_DIR}/geo_location/files" country = Country.objects.get(id=2) with open(f'{dir}/propdata-uk-demographics.json', 'r', encoding="utf8") as file: data = json.load(file) data = data.get('demog') for g_d in data: postcode = g_d.get('postcode') try: post_code = PostCode.objects.get(out_code=postcode) except: post_code = None postcode_type = g_d.get('postcode_type') g_data = g_d.get('data') deprivation = g_data.get('deprivation') if deprivation == "Insufficient data": deprivation = None health = g_data.get('health') if health == "Insufficient data": health = None social_grade = g_data.get('social_grade') if social_grade == "Insufficient data": social_grade = None age = g_data.get('age') if age == "Insufficient data": age = None politics = g_data.get('politics') if politics == "Insufficient data": politics = None proportion_with_degree = g_data.get('proportion_with_degree') if proportion_with_degree == "Insufficient data": proportion_with_degree = None vehicles_per_household = g_data.get('vehicles_per_household') if vehicles_per_household == "Insufficient data": vehicles_per_household = None commute_method = g_data.get('commute_method') if commute_method == "Insufficient data": commute_method = None try: demographic = Demographic.objects.get(country=country, postcode=postcode) except: demographic = Demographic.objects.create(country=country, postcode=postcode) demographic.post_code = post_code demographic.postcode_type = postcode_type demographic.deprivation = deprivation demographic.health = health demographic.social_grade = social_grade demographic.age = age demographic.politics = politics demographic.proportion_with_degree = proportion_with_degree demographic.vehicles_per_household = vehicles_per_household demographic.commute_method = commute_method demographic.save() def get_uk_postcodes(): regions = [ "north_east", "north_west", "east_midlands", "west_midlands", "east_of_england", "greater_london", "south_east", "south_west", "wales", "scotland", "northern_ireland", ] country = Country.objects.get(id=2) for region in regions: url = f"https://api.propertydata.co.uk/postcode-key-stats?key=YYRZRTTGG5®ion={region}" region_name = region.replace("_", " ").title() region_slug = slugify(region_name, allow_unicode=True) region, _ = Region.objects.get_or_create(name=region_name, country=country, slug=region_slug) pdr, _ = PropertyDataRaw.objects.get_or_create(region=region, type=2) data = {} if pdr.data: data = pdr.data else: req = requests.get(url, timeout=30) if req.status_code == 200: data = req.json() pdr.data = data pdr.checked = False pdr.save() else: print(req.status_code) continue if data.get('status') == "success": for d in data.get('data'): out_code = d.get('outcode') avg_price = d.get('avg_price') avg_rent = d.get('avg_rent') avg_yield = d.get('avg_yield') try: if avg_yield: avg_yield = float(avg_yield[:-1]) except: print(f"postcode {out_code} avg_yield {avg_yield} error") avg_yield = None growth_1y = d.get('growth_1y') try: if growth_1y: growth_1y = float(growth_1y[:-1]) except: print(f"postcode {out_code} growth_1y {growth_1y} error") growth_1y = None growth_3y = d.get('growth_3y') try: if growth_3y: growth_3y = float(growth_3y[:-1]) except: print(f"postcode {out_code} growth_3y {growth_3y} error") growth_3y = None growth_5y = d.get('growth_5y') try: if growth_5y: growth_5y = float(growth_5y[:-1]) except: print(f"postcode {out_code} growth_5y {growth_5y} error") growth_5y = None sales_per_month = d.get('sales_per_month') turnover = d.get('turnover') try: if turnover: turnover = float(turnover[:-1]) except: print(f"postcode {out_code} turnover {turnover} error") turnover = None post_code,_ = PostCode.objects.get_or_create(out_code=out_code, region=region) post_code.avg_price = avg_price post_code.avg_rent = avg_rent post_code.avg_yield = avg_yield post_code.growth_1y = growth_1y post_code.growth_3y = growth_3y post_code.growth_5y = growth_5y post_code.sales_per_month = sales_per_month post_code.turnover = turnover post_code.save() pdr.checked = True pdr.save() else: print(data.get('message')) time.sleep(3) def get_price_growth(): country = Country.objects.get(id=2) post_codes = PostCode.objects.all() MONTH = {'JAN': '1', 'FEB': '2', 'MAR': '3', 'APR': '4', 'MAY': '5','JUN': '6', 'JUL': '7', 'AUG': '8', 'SEP': '9', 'OCT': '10', 'NOV': '11', 'DEC': '12'} for post_code in post_codes: url = f"https://api.propertydata.co.uk/growth?key=RORUV4AKHU&postcode={post_code.out_code}" pdr, _ = PropertyDataRaw.objects.get_or_create(postcode=post_code, type=0) data = {} if pdr.data: data = pdr.data send_r = False else: req = requests.get(url, timeout=30) send_r = True if req.status_code == 200: data = req.json() pdr.data = data pdr.checked = False pdr.save() else: print(post_code.out_code, req.status_code) continue if data.get('status') == "success": postcode = data.get('postcode') postcode_type = data.get('postcode_type') g_data = data.get('data') last_prise = None price_change = None for g in g_data: year = int(g[0][-4:]) try: month = int(MONTH.get(g[0][:3])) except: month = 11 price = float(g[1]) price_change = price - last_prise if last_prise else None last_prise = price change_percent = float(g[2][:-1]) if g[2] else None date = datetime.date(year, month, 1) try: pg = PriceGrowth.objects.get(country=country, postcode=postcode, date=date) except: pg = PriceGrowth.objects.create(country=country, postcode=postcode, date=date) pg.post_code=post_code pg.postcode_type=postcode_type pg.price=price pg.price_change=price_change pg.change_percent=change_percent pg.save() print(f"{post_code.out_code} check data") if send_r: time.sleep(3) def get_demographics(): country = Country.objects.get(id=2) post_codes = PostCode.objects.all() MONTH = {'JAN': '1', 'FEB': '2', 'MAR': '3', 'APR': '4', 'MAY': '5','JUN': '6', 'JUL': '7', 'AUG': '8', 'SEP': '9', 'OCT': '10', 'NOV': '11', 'DEC': '12'} for post_code in post_codes: url = f"https://api.propertydata.co.uk/demographics?key=RORUV4AKHU&postcode={post_code.out_code}" pdr, _ = PropertyDataRaw.objects.get_or_create(postcode=post_code, type=3) data = {} if pdr.data: data = pdr.data send_r = False else: req = requests.get(url, timeout=30) send_r = True if req.status_code == 200: data = req.json() pdr.data = data pdr.checked = False pdr.save() else: print(post_code.out_code, req.status_code) continue if data.get('status') == "success": postcode = data.get('postcode') postcode_type = data.get('postcode_type') g_data = data.get('data') deprivation = g_data.get('deprivation') if deprivation == "Insufficient data": deprivation = None health = g_data.get('health') if health == "Insufficient data": health = None social_grade = g_data.get('social_grade') if social_grade == "Insufficient data": social_grade = None age = g_data.get('age') if age == "Insufficient data": age = None politics = g_data.get('politics') if politics == "Insufficient data": politics = None proportion_with_degree = g_data.get('proportion_with_degree') if proportion_with_degree == "Insufficient data": proportion_with_degree = None vehicles_per_household = g_data.get('vehicles_per_household') if vehicles_per_household == "Insufficient data": vehicles_per_household = None commute_method = g_data.get('commute_method') if commute_method == "Insufficient data": commute_method = None try: demographic = Demographic.objects.get(country=country, postcode=postcode) except: demographic = Demographic.objects.create(country=country, postcode=postcode) demographic.post_code = post_code demographic.postcode_type = postcode_type demographic.deprivation = deprivation demographic.health = health demographic.social_grade = social_grade demographic.age = age demographic.politics = politics demographic.proportion_with_degree = proportion_with_degree demographic.vehicles_per_household = vehicles_per_household demographic.commute_method = commute_method demographic.save() print(f"{post_code.out_code} check data") if send_r: time.sleep(3) def get_schools_in_area(south, west, north, east): """ Get schools within a bounding box using Overpass API Parameters are coordinates defining the area bounds """ overpass_url = "https://overpass-api.de/api/interpreter" query = f""" [out:json][timeout:25]; ( way["amenity"="school"]({south},{west},{north},{east}); node["amenity"="school"]({south},{west},{north},{east}); ); out body; >; out skel qt; """ response = requests.post(overpass_url, data=query, timeout=30) data = response.json() schools = [] for element in data['elements']: if 'tags' in element: school = { 'id': element['id'], 'type': element['type'], 'name': element['tags'].get('name:en'), 'isced_level': element['tags'].get('isced:level'), 'operator': element['tags'].get('operator'), 'address': element['tags'].get('addr:full'), 'lat': element.get('lat', 0), 'lon': element.get('lon', 0) } schools.append(school) return schools def get_subway_stations(south, west, north, east): """ Get subway/metro stations within a bounding box using Overpass API Parameters are coordinates defining the area bounds """ overpass_url = "https://overpass-api.de/api/interpreter" query = f""" [out:json][timeout:25]; ( node["station"="subway"]({south},{west},{north},{east}); node["railway"="station"]["subway"="yes"]({south},{west},{north},{east}); node["station"="metro"]({south},{west},{north},{east}); ); out body; >; out skel qt; """ response = requests.post(overpass_url, data=query, timeout=30) data = response.json() stations = [] for element in data['elements']: if 'tags' in element: station = { 'id': element['id'], 'name': element['tags'].get('name:en'), 'network': element['tags'].get('network'), 'operator': element['tags'].get('operator'), 'lines': element['tags'].get('line'), 'lat': element.get('lat'), 'lon': element.get('lon') } stations.append(station) return stations def calculate_distance_duration(origin_lat, origin_lng, dest_lat, dest_lng, mode="foot-walking"): """ Calculate distance and duration between two points using OpenRouteService API mode options: foot-walking, driving-car, cycling-regular """ # Get free API key from https://openrouteservice.org/ api_key = settings.OPENROUTE_API_KEY headers = { 'Accept': 'application/json, application/geo+json, application/gpx+xml, img/png; charset=utf-8', 'Authorization': api_key, 'Content-Type': 'application/json; charset=utf-8' } body = { "coordinates": [[origin_lng, origin_lat], [dest_lng, dest_lat]], "instructions": "false", } url = f"https://api.openrouteservice.org/v2/directions/{mode}" try: response = requests.post(url, json=body, headers=headers, timeout=30) data = response.json() if 'routes' in data: route = data['routes'][0] distance = route['summary']['distance'] # meters duration = route['summary']['duration'] # seconds return { 'distance': round(distance), 'duration': round(duration), 'distance_text': f"{round(distance/1000, 2)} km", 'duration_text': f"{round(duration/60)} mins" } except Exception as e: print(f"Error calculating distance/duration: {e}") return None