Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

D42 14600 - Lenovo Warranty Check Error #31

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

D42 14600 - Lenovo Warranty Check Error #31

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e65b44a
updated lenovo, ibm warranty sync. needs testing
osanchez42 Feb 18, 2020
1ea1f07
parameter fixes
osanchez42 Mar 9, 2020
7b51471
tested and did some minor code refactoring
osanchez42 May 19, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
8 changes: 4 additions & 4 deletions Files/shared.py
View file
Open in desktop
Expand Up @@ -118,19 +118,19 @@ def __get_hp_cfg(self):
def __get_ibm_cfg(self):
# IBM ---------------------------------------------
ibm_url = self.cc.get('ibm', 'url')
ibm_url2 = self.cc.get('ibm', 'url2')
client_id = self.cc.get('ibm', 'client_id')
return {
'url': ibm_url,
'url2': ibm_url2
'client_id': client_id
}

def __get_lenovo_cfg(self):
# lenovo -------------------------------------------
lenovo_url = self.cc.get('lenovo', 'url')
lenovo_url2 = self.cc.get('lenovo', 'url2')
client_id = self.cc.get('lenovo', 'client_id')
return {
'url': lenovo_url,
'url2': lenovo_url2
'client_id': client_id
}

def __get_meraki_cfg(self):
Expand Down
10 changes: 5 additions & 5 deletions Files/warranty.cfg.example
View file
Open in desktop
Expand Up @@ -33,15 +33,15 @@ api_secret =
url = https://css.api.hp.com

[ibm]
url = http://support.lenovo.com/services/us/en/ContentService/GetProducts
url2 = http://support.lenovo.com/us/en/products
url = https://SupportAPI.lenovo.com/v2.5/Warranty
client_id =

[lenovo]
url = http://support.lenovo.com/services/us/en/ContentService/GetProducts
url2 = http://support.lenovo.com/us/en/products
url = https://SupportAPI.lenovo.com/v2.5/Warranty
client_id =

[meraki]
api_key =
api_key =
url = https://api.meraki.com/api/v0

[other]
Expand Down
115 changes: 58 additions & 57 deletions Files/warranty_ibm_lenovo.py
View file
Open in desktop
Expand Up @@ -6,7 +6,7 @@
import random
import requests

from shared import DEBUG, RETRY, ORDER_NO_TYPE, left
from shared import DEBUG, RETRY, ORDER_NO_TYPE, left, Device42rest
from warranty_abstract import WarrantyBase

try:
Expand All @@ -19,7 +19,7 @@ class IbmLenovo(WarrantyBase, object):
def __init__(self, vendor, params):
super(IbmLenovo, self).__init__()
self.url = params['url']
self.url2 = params['url2']
self.client_id = params['client_id']
self.debug = DEBUG
self.retry = RETRY
self.order_no = ORDER_NO_TYPE
Expand All @@ -31,12 +31,23 @@ def __init__(self, vendor, params):
if self.order_no == 'common':
self.common = self.generate_random_order_no()

def get_product_info(self, serial, retry=True):
def get_product_info(self, serials, retry=True):
if self.debug:
print '\t[+] Checking possible product "%s"' % serial
timeout = 10
print '\t[+] Checking possible product "%s"' % serials

timeout = 30

headers = {
'Content-Type': 'application/x-www-form-urlencoded',
'ClientID': self.client_id
}

params = {
'Serial': serials
}

try:
resp = self.requests.get(self.url + '?productId=' + serial, verify=True, timeout=timeout)
resp = self.requests.get(self.url, params=params, headers=headers, verify=True, timeout=timeout)
msg = 'Status code: %s' % str(resp.status_code)
if str(resp.status_code) == '401':
print '\t[!] HTTP error. Message was: %s' % msg
Expand All @@ -45,70 +56,57 @@ def get_product_info(self, serial, retry=True):
time.sleep(30)
if retry:
print '\n[!] Retry'
self.get_product_info(serial, False)
self.get_product_info(serials, False)
else:
return None
else:
if self.debug:
print
print resp.json()
print
return resp.json()
except requests.RequestException as e:
self.error_msg(e)
return None

def run_warranty_check(self, inline_serials, retry=True):
global full_serials
full_serials = {}

if self.debug:
print '\t[+] Checking warranty "%s"' % inline_serials
timeout = 10
result = []

for serial in inline_serials.split(','):
product_info = self.get_product_info(serial)
current_product = None

if len(product_info) > 1:
for product in product_info:
current_product = self.get_product_info(product['Name'])[0]
if current_product['Name'] is not None:
break
else:
try:
current_product = product_info[0]
except IndexError:
print '\t[+] Unable to find "%s" orders' % serial
continue

if current_product is not None:

url = self.url2 + '/' + current_product['Id'] + '?tabName=Warranty&beta=false'

resp = self.requests.post(
url,
data={'SERIALNUMBERKEY': current_product['Serial']},
verify=True,
timeout=timeout
)

# possible redirects
if resp.url != url:
resp = self.requests.post(
resp.url,
data={'SERIALNUMBERKEY': current_product['Serial']},
verify=True,
timeout=timeout
)

data_object = re.search(r"ds_warranties=(.*?});", resp.text)
json_object = json.loads(data_object.group(1))
result.append(json_object)
incoming_serials = inline_serials.split(',')
inline_serials = []

for d42_serial in incoming_serials:
d42_serial = d42_serial.upper()
if '_' in d42_serial:
full_serials.update({d42_serial.split('_')[0]: d42_serial})
d42_serial = d42_serial.split('_')[0]
elif '(' in d42_serial:
full_serials.update({d42_serial.split('(')[0]: d42_serial})
d42_serial = d42_serial.split('(')[0]
else:
full_serials.update({d42_serial: d42_serial})
inline_serials.append(d42_serial)

result = self.get_product_info(inline_serials, retry)
return result

def process_result(self, result, purchases):
global full_serials
data = {}

for item in result:
# The API returns results for single devices in a different format than multiple devices, this keeps everything
# returned from the API in the same list format
if 'Warranty' in result:
result = [result]

if 'BaseWarranties' in item and len(item['BaseWarranties']) > 0:
warranties = item['BaseWarranties']
for item in result:
# Warranties
if 'Warranty' in item and len(item['Warranty']) > 0:
warranties = item['Warranty']
else:
continue

Expand All @@ -135,24 +133,27 @@ def process_result(self, result, purchases):

lines = []

# process warranty line items for a device
for warranty in warranties:

start_date = warranty['Start']['UTC'].split('T')[0]
end_date = warranty['End']['UTC'].split('T')[0]
try:
start_date = warranty['Start'].split('T')[0]
end_date = warranty['End'].split('T')[0]
except (KeyError, AttributeError):
continue

data.update({'line_start_date': start_date})
data.update({'line_end_date': end_date})
data.update({'line_contract_type': warranty['Origin']})
data.update({'line_contract_type': warranty['Type']})

try:
# There's a max 64 character limit on the line service type field in Device42 (version 13.1.0)
service_level_description = left(sub_item['ServiceLevelDescription'], 64)
service_level_description = left(warranty['Name'], 64)
data.update({'line_service_type': service_level_description})
except:
pass

# update or duplicate? Compare warranty dates by serial, contract_id and end date
hasher = serial.split('.')[0] + start_date + end_date
hasher = serial + start_date + end_date

try:
d_purchase_id, d_order_no, d_line_no, d_contractid, d_start, d_end, forcedupdate = purchases[hasher]
Expand Down
2 changes: 1 addition & 1 deletion starter.py
View file
Open in desktop
Expand Up @@ -61,7 +61,7 @@ def get_vendor_api(name):
elif vendor == 'ibm' or vendor == 'lenovo':
ibm_lenovo_params = {
'url': current_cfg['url'],
'url2': current_cfg['url2'],
'client_id': current_cfg['client_id'],
'd42_rest': d42_rest
}
api = IbmLenovo(vendor, ibm_lenovo_params)
Expand Down