alltheplaces/locations/storefinders/amrest_eu.py at master · ilias52730/alltheplaces · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
from scrapy import Spider
from scrapy.http import JsonRequest, Response

from locations.categories import Extras, apply_yes_no
from locations.dict_parser import DictParser
from locations.hours import OpeningHours
from locations.items import Feature


class AmrestEUSpider(Spider):
    """
    AmRest is a European multinational casual dining, fast-food restaurant and coffee shop operator headquartered in the Spanish capital, Madrid
    https://www.wikidata.org/wiki/Q4738898
    https://en.wikipedia.org/wiki/AmRest

    This spider is specifically for the common functionality across all child brands.
    """

    api_brand_key: str = None
    api_brand_country_key: str = None
    api_source: str = None
    api_auth_source: str = None
    api_channel: str = None

    def start_requests(self):
        headers = {
            "Brand": self.api_brand_key,
        }
        if self.api_source:
            headers.update({"Source": self.api_source})
        data = {
            "deviceUuid": "FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF",
            "deviceUuidSource": "FINGERPRINT",
            "source": f"{self.api_auth_source}",
        }
        yield JsonRequest(
            url=f"https://api.amrest.eu/amdv/ordering-api/{self.api_brand_country_key}/rest/v1/auth/get-token",
            headers=headers,
            data=data,
            callback=self.parse_auth_token,
        )

    def parse_auth_token(self, response):
        auth_token = response.json()["token"]
        headers = {
            "Authorization": f"Bearer {auth_token}",
            "Brand": self.api_brand_key,
        }
        if self.api_source:
            headers.update({"Source": self.api_source})
        yield JsonRequest(
            url=f"https://api.amrest.eu/amdv/ordering-api/{self.api_brand_country_key}/rest/v3/restaurants/",
            headers=headers,
            callback=self.parse_restaurants_list,
        )

    def parse_restaurants_list(self, response):
        headers = {
            "Authorization": response.request.headers.get("Authorization"),
            "Brand": self.api_brand_key,
        }
        if self.api_source:
            headers.update({"Source": self.api_source})
        for restaurant in response.json()["restaurants"]:
            restaurant_id = restaurant["id"]
            if self.api_channel:
                yield JsonRequest(
                    url=f"https://api.amrest.eu/amdv/ordering-api/{self.api_brand_country_key}/rest/v2/restaurants/{restaurant_id}/{self.api_channel}",
                    headers=headers,
                    callback=self.parse_restaurant_details,
                )
            else:
                yield JsonRequest(
                    url=f"https://api.amrest.eu/amdv/ordering-api/{self.api_brand_country_key}/rest/v2/restaurants/details/{restaurant_id}",
                    headers=headers,
                    callback=self.parse_restaurant_details,
                )

    def parse_restaurant_details(self, response):
        location = response.json()["details"]
        item = DictParser.parse(location)
        item["ref"] = str(location["id"])
        item["postcode"] = location["addressPostalCode"]
        item["housenumber"] = location.get("addressStreetNo")
        item["street"] = location["addressStreet"]
        item["opening_hours"] = self.parse_hours(location["facilityOpenHours"])

        apply_yes_no(Extras.DELIVERY, item, location.get("delivery"), False)
        apply_yes_no(Extras.TAKEAWAY, item, location.get("takeaway"), False)
        apply_yes_no(Extras.INDOOR_SEATING, item, location.get("dineIn"), False)
        extra_features = [feature["key"] for feature in location.get("filters", {})]
        if extra_features:
            apply_yes_no(Extras.DRIVE_THROUGH, item, "driveThru" in extra_features, False)
            apply_yes_no(Extras.OUTDOOR_SEATING, item, location.get("garden"), False)
            apply_yes_no(Extras.WIFI, item, "wifi" in extra_features, False)

        yield from self.post_process_item(item, response, location)

    def post_process_item(self, item: Feature, response: Response, location: dict):
        yield item

    @staticmethod
    def parse_hours(opening_hours: dict) -> OpeningHours:
        oh = OpeningHours()
        for day in [
            "openHoursMon",
            "openHoursTue",
            "openHoursWed",
            "openHoursThu",
            "openHoursFri",
            "openHoursSat",
            "openHoursSun",
        ]:
            for times in opening_hours[day]:
                oh.add_range(day[-3:], times["openFrom"], times["openTo"])
        return oh