import os.path from dotenv import dotenv_values import requests from enum import Enum import json import base64 class Cloud: HEADERS = { "Content-Type": "application/json", } def __init__(self): config = dotenv_values("../.env") self.authApiUrl = config['BIGSCREEN_API_URL'] self.adminApiUrl = config['BIGSCREEN_ADMIN_API_URL'] self.apiBearerToken = config['FAB_API_KEY'] self.email = config['FAB_EMAIL'] self.password = config['FAB_PASSWORD'] self.accessToken = None self.__authenticate() def _check_access_token(self): headers = {**self.HEADERS, "Authorization": f"Bearer {self.apiBearerToken}", "x-access-token": self.accessToken} r = requests.get(f"{self.authApiUrl}/verify", headers=headers) if r.status_code == 401: nonce = r.headers["x-bigscreen-nonce"] systemInfo = { "deviceUniqueIdentifier": "python notebook", "version": "001", "deviceName": "python notebook", "deviceModel": "python notebook", "operatingSystem": "whatever" } headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.apiBearerToken}", "x-refresh-token": self.refreshToken, "x-bigscreen-nonce": nonce, "x-bigscreen-system-info": base64.b64encode(bytes(json.dumps(systemInfo), "utf-8")) } renew = requests.get(f"{self.authApiUrl}/auth/renew", headers=headers) if renew.status_code == 200: self.accessToken = renew.headers["x-access-token"] print("Renewed the access token!") elif r.status_code == 200: print("Access token ok.") def __authenticate(self): payload = { "email": self.email, "password": self.password } headers = {**self.HEADERS, "Authorization": f"Bearer {self.apiBearerToken}"} r = requests.post(f"{self.authApiUrl}/login", headers=headers, json=payload) r.raise_for_status() self.accessToken = r.headers["x-access-token"] self.refreshToken = r.headers["x-refresh-token"] return self.accessToken def _get_admin_headers(self): self._check_access_token() return {**self.HEADERS, "Authorization": f"Bearer {self.apiBearerToken}", "x-access-token": self.accessToken} def request_job_using_id(self, job_id: str): headers = self._get_admin_headers() r = requests.get(f"{self.adminApiUrl}/admin/fabricator/job/{job_id}", headers=headers) r.raise_for_status() job_data = r.json() return job_data def reject_scan(self, job_id: str): headers = self._get_admin_headers() r = requests.get(f"{self.adminApiUrl}/admin/fabricator/schemas", headers=headers) assert r.status_code == 200, f"status code should be 200 (actual: {r.status_code})" schemas = r.json() JobStates = Enum("JobStates", dict([(v, int(k)) for k, v in schemas["JobStates"].items()])) JobActions = Enum("JobActions", dict([(v, int(k)) for k, v in schemas["JobActions"].items()])) self._check_access_token() payload = {"action": JobActions.VisualInspectionFailed.value} r = requests.put(f"{self.adminApiUrl}/admin/fabricator/job/{job_id}", headers=headers, json=payload) assert r.status_code == 200, f"status code should be 200 (actual: {r.status_code})" def accept_scan(self, job_id: str, process_dir_path: str, pulled_from_failed=False): headers = self._get_admin_headers() assert os.path.exists( f"{process_dir_path}/mesh.stl"), "manifest.json not found" assert os.path.exists( f"{process_dir_path}/manifest.json"), "manifest.json not found" with open(f"{process_dir_path}/manifest.json", "r") as f: manifest = json.load(f) r = requests.get(f"{self.adminApiUrl}/admin/fabricator/schemas", headers=headers) assert r.status_code == 200, f"status code should be 200 (actual: {r.status_code})" schemas = r.json() JobStates = Enum("JobStates", dict([(v, int(k)) for k, v in schemas["JobStates"].items()])) JobActions = Enum("JobActions", dict([(v, int(k)) for k, v in schemas["JobActions"].items()])) if pulled_from_failed: payload = {"action": JobActions.ResetToAwaitingScanVerification.value} r = requests.put(f"{self.adminApiUrl}/admin/fabricator/job/{job_id}",headers=headers, json=payload) assert r.status_code == 200, f"status code should be 200 (actual: {r.status_code})" payload = { "action": JobActions.TryExternalPreprocessorUpload.value, # <---------- note this new action "ipd": manifest["ipd"], "left_pupil": [0, 0], "right_pupil": [0, 0], "mesh": base64.b64encode( open(f"{process_dir_path}/mesh.stl", "rb").read()).decode( "utf-8") } self._check_access_token() r = requests.put(f"{self.adminApiUrl}/admin/fabricator/job/{job_id}", headers=headers, json=payload) assert r.status_code == 200, f"status code should be 200 (actual: {r.status_code} {r.text})" def request_scan(self, job_data, save_dir="./"): headers = self._get_admin_headers() if 'scanRequestId' in job_data: r = requests.get( f"{self.adminApiUrl}/admin/fabricator/scan_request/{job_data['scanRequestId']}", headers=headers) assert r.status_code == 200, f"status code should be 200 (actual: {r.status_code})" scan_request_data = r.json() r = requests.get( f"{self.adminApiUrl}/admin/fabricator/scan_request/{job_data['scanRequestId']}/scan", headers=headers) assert r.status_code == 200, f"status code should be 200 (actual: {r.status_code})" else: job_id = job_data['id'] r = requests.get( f"{self.adminApiUrl}/admin/fabricator/job/{job_id}/scan", headers=headers) assert r.status_code == 200, f"status code should be 200 (actual: {r.status_code})" scan_path = os.path.abspath(os.path.join(save_dir, "tmp_scan.zip")) with open(scan_path, "wb") as f: r.raw.decode_content = True f.write(r.content) return scan_path, scan_request_data def count_jobs_from_manufacturing_to_awaiting_nfc(self): headers = self._get_admin_headers() r = requests.get(f"{self.adminApiUrl}/admin/fabricator/schemas", headers=headers) assert r.status_code == 200, f"status code should be 200 (actual: {r.status_code})" schemas = r.json() JobStates = Enum("JobStates", dict([(v, int(k)) for k, v in schemas["JobStates"].items()])) count = 0 for job_state in [JobStates.AwaitingManufacture.value, JobStates.ManufacturingInProgress.value, JobStates.AwaitingNFC.value]: r = requests.get(f"{self.adminApiUrl}/admin/fabricator/jobs?limit=100&jobState={job_state}", headers=headers) assert r.status_code == 200, f"status code should be 200 (actual: {r.status_code})" json_data = r.json() jobs = json_data['items'] count += len(jobs) return count def get_order_contry_code(self, job_id): headers = self._get_admin_headers() r = requests.get(f"{self.adminApiUrl}/admin/fabricator/job/{job_id}", headers=headers) assert r.status_code == 200, f"status code should be 200 (actual: {r.status_code})" job = r.json() big_order_id = job['bigOrderId'] r = requests.get(f"{self.adminApiUrl}/admin/shop/order/{big_order_id}", headers=headers) assert r.status_code == 200, f"status code should be 200 (actual: {r.status_code})" big_order = r.json() return big_order['shopifyOrderSnapshot']['shipping_address']['country_code'] def is_japan_order(self, job_id): try: return self.get_order_contry_code(job_id) == 'JP' except: raise "Failed to get order country code" def query_jobs_needing_validation(self, pull_from_failed=False, ipd_whitelist = list(range(0, 100))): headers = self._get_admin_headers() r = requests.get(f"{self.adminApiUrl}/admin/fabricator/schemas", headers=headers) assert r.status_code == 200, f"status code should be 200 (actual: {r.status_code})" schemas = r.json() JobStates = Enum("JobStates", dict( [(v, int(k)) for k, v in schemas["JobStates"].items()])) jobState = JobStates.ScanVerificationFailed.value if pull_from_failed else JobStates.AwaitingScanVerification.value r = requests.get(f"{self.adminApiUrl}/admin/fabricator/jobs?limit=10000&jobState={jobState}", headers=headers) assert r.status_code == 200, f"status code should be 200 (actual: {r.status_code})" json_data = r.json() jobs = json_data['items'] jobs = sorted(jobs, key=lambda x: x['shopifyOrderId']) # filter out jobs that were made between Nov 28th 2023 and Dec 5th 2023 #jobs = [job for job in jobs if job['createdAt'] < 1701158400000] #jobs = [job for job in jobs if job['createdAt'] > 1701158400000] filtered_jobs = [] if ipd_whitelist: for job in jobs: if len(filtered_jobs) == 100: # Config to amount of jobs to run in a batch (default 100) break if job["priority"] < 50: continue if job["bigOrderId"] is None: continue try: r = requests.get(f"{self.adminApiUrl}/admin/shop/order/{job['bigOrderId']}", headers=headers) assert r.status_code == 200, f"status code should be 200 (actual: {r.status_code})" except: print(f"Failed to get bigOrder {job['bigOrderId']}") continue bigOrder = r.json() # if bigOrder["shopifyOrderSnapshot"]["shipping_address"]["country_code"] == "US": # continue # if bigOrder["shopifyOrderSnapshot"]["shipping_address"]["country_code"] == "JP": # continue # if any(lineItem["type"] == "PrescriptionLenses" for lineItem # in bigOrder["currentLineItems"]): # continue # matches with any of the prefered ipds if bigOrder['ipd'] in ipd_whitelist: filtered_jobs.append(job) # sort by shopifyOrderID return filtered_jobs