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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
|
from hashlib import md5
from .helper import _tree_generator, _url_generator
from .session import PicnicAPISession, PicnicAuthError
DEFAULT_URL = "https://storefront-prod.{}.picnicinternational.com/api/{}"
DEFAULT_COUNTRY_CODE = "NL"
DEFAULT_API_VERSION = "15"
class PicnicAPI:
def __init__(
self, username: str = None, password: str = None,
country_code: str = DEFAULT_COUNTRY_CODE, auth_token: str = None
):
self._country_code = country_code
self._base_url = _url_generator(
DEFAULT_URL, self._country_code, DEFAULT_API_VERSION
)
self.session = PicnicAPISession(auth_token=auth_token)
# Login if not authenticated
if not self.session.authenticated and username and password:
self.login(username, password)
def _get(self, path: str, add_picnic_headers=False):
url = self._base_url + path
# Make the request, add special picnic headers if needed
headers = {
"x-picnic-agent": "30100;1.15.77-10293",
"x-picnic-did": "3C417201548B2E3B"
} if add_picnic_headers else None
response = self.session.get(url, headers=headers).json()
if self._contains_auth_error(response):
raise PicnicAuthError("Picnic authentication error")
return response
def _post(self, path: str, data=None):
url = self._base_url + path
response = self.session.post(url, json=data).json()
if self._contains_auth_error(response):
raise PicnicAuthError(f"Picnic authentication error: {response['error'].get('message')}")
return response
@staticmethod
def _contains_auth_error(response):
if not isinstance(response, dict):
return False
error_code = response.setdefault("error", {}).get("code")
return error_code == "AUTH_ERROR" or error_code == "AUTH_INVALID_CRED"
def login(self, username: str, password: str):
path = "/user/login"
secret = md5(password.encode("utf-8")).hexdigest()
data = {"key": username, "secret": secret, "client_id": 1}
return self._post(path, data)
def logged_in(self):
return self.session.authenticated
def get_user(self):
return self._get("/user")
def search(self, term: str):
path = "/search?search_term=" + term
return self._get(path)
def get_lists(self, list_id: str = None):
if list_id:
path = "/lists/" + list_id
else:
path = "/lists"
return self._get(path)
def get_sublist(self, list_id: str, sublist_id: str) -> list:
"""Get sublist.
Args:
list_id (str): ID of list, corresponding to requested sublist.
sublist_id (str): ID of sublist.
Returns:
list: Sublist result.
"""
return self._get(f"/lists/{list_id}?sublist={sublist_id}")
def get_cart(self):
return self._get("/cart")
def add_product(self, product_id: str, count: int = 1):
data = {"product_id": product_id, "count": count}
return self._post("/cart/add_product", data)
def remove_product(self, product_id: str, count: int = 1):
data = {"product_id": product_id, "count": count}
return self._post("/cart/remove_product", data)
def clear_cart(self):
return self._post("/cart/clear")
def get_delivery_slots(self):
return self._get("/cart/delivery_slots")
def get_delivery(self, delivery_id: str):
path = "/deliveries/" + delivery_id
return self._get(path)
def get_delivery_scenario(self, delivery_id: str):
path = "/deliveries/" + delivery_id + "/scenario"
return self._get(path, add_picnic_headers=True)
def get_delivery_position(self, delivery_id: str):
path = "/deliveries/" + delivery_id + "/position"
return self._get(path, add_picnic_headers=True)
def get_deliveries(self, summary: bool = False, data=None):
data = [] if data is None else data
if summary:
return self._post("/deliveries/summary", data=data)
return self._post("/deliveries", data=data)
def get_current_deliveries(self):
return self.get_deliveries(data=["CURRENT"])
def get_categories(self, depth: int = 0):
return self._get(f"/my_store?depth={depth}")["catalog"]
def print_categories(self, depth: int = 0):
tree = "\n".join(_tree_generator(self.get_categories(depth=depth)))
print(tree)
__all__ = ["PicnicAPI"]
|
