Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/bones/address.py: 87%

1import hashlib 

2import logging 

3import typing as t 

4import urllib.parse 

5 

6import requests 

7 

8from .record import RecordBone 

9from .string import StringBone 

10from .selectcountry import SelectCountryBone 

11from .spatial import SpatialBone 

12from ..skeleton.relskel import RelSkel 

13from .. import db, i18n 

14from ..data.zip_patterns import ZIP_CODE_PATTERNS 

15 

16CACHE_KIND = "viur-addressbone-geocache" 

17 

18 

19class AddressRelSkel(RelSkel): 

20 street_name = StringBone(descr="Street", required=True) 

21 street_number = StringBone(descr="House number") 

22 address_addition = StringBone(descr="Address addition") 

23 zip_code = StringBone( 

24 descr="ZIP / Postal code", 

25 params={ 

26 "pattern": ZIP_CODE_PATTERNS, 

27 "pattern-error": i18n.translate( 

28 "viur.core.bones.address.zip_code.invalid", 

29 defaultText="Invalid ZIP code", 

30 ), 

31 }, 

32 ) 

33 city = StringBone(descr="City", required=True) 

34 country = SelectCountryBone(descr="Country") 

35 coordinates = SpatialBone( 

36 descr="Coordinates", 

37 boundsLat=(-90, 90), 

38 boundsLng=(-180, 180), 

39 gridDimensions=(10, 10), 

40 ) 

41 

42 

43class AddressBone(RecordBone): 

44 type = RecordBone.type + ".address" 

45 

46 def __init__( 

47 self, 

48 *, 

49 using: t.Type[RelSkel] = AddressRelSkel, 

50 format: str = "$(street_name) $(street_number), $(zip_code) $(city)", 

51 **kwargs, 

52 ): 

53 super().__init__(using=using, format=format, **kwargs) 

54 

55 def after_from_client(self, skel, name, errors): 

56 value = skel[name] 

57 if value is not None: 

58 coords = self.geocode(value) 

59 if coords: 

60 value["coordinates"] = coords 

61 

62 @staticmethod 

63 def _cache_key(params: str) -> db.Key: 

64 digest = hashlib.sha256(params.encode()).hexdigest() 

65 return db.Key(CACHE_KIND, digest) 

66 

67 @staticmethod 

68 def geocode(skel: RelSkel) -> tuple[float, float] | None: 

69 street = f"{skel['street_name'] or ''} {skel['street_number'] or ''}".strip() 

70 params = { 

71 "street": street, 

72 "postalcode": skel["zip_code"] or "", 

73 "city": skel["city"] or "", 

74 "country": skel["country"] or "", 

75 "format": "json", 

76 "limit": 1, 

77 } 

78 cache_key = AddressBone._cache_key(urllib.parse.urlencode(params)) 

79 try: 

80 cached = db.get(cache_key) 

81 if cached is not None: 

82 return cached["lat"], cached["lng"] 

83 response = requests.get( 

84 "https://nominatim.openstreetmap.org/search", 

85 params=params, 

86 headers={"User-Agent": "viur-addressbone/1.0 (viur.dev)"}, 

87 timeout=5, 

88 ) 

89 if response.status_code != 200: 

90 logging.error( 

91 f"AddressBone: Nominatim returned {response.status_code=}" 

92 ) 

93 return None 

94 data = response.json() 

95 if not data: 

96 return None 

97 lat, lng = float(data[0]["lat"]), float(data[0]["lon"]) 

98 entity = db.Entity(cache_key) 

99 entity["lat"] = lat 

100 entity["lng"] = lng 

101 db.put(entity) 

102 return lat, lng 

103 except Exception as e: 

104 logging.error(f"AddressBone: Nominatim geocoding failed with {e=}") 

105 return None