GS!
from fastapi import Depends, FastAPI, HTTPException, Request, Body, APIRouter import requests import datetime import psycopg2 from decouple import config from datetime import datetime from datetime import timedelta import json from .database import connectToDB from dateutil.relativedelta import relativedelta import pytz, re, urllib from .log_file import add_api_log from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm router = APIRouter() oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") def generate_gsbrand_data(brand_name): data_list = [] data_list_cat = [] if brand_name: sign_in_url = config("sign_in_url_gs") email = config("gs_email") password = config("gs_pass") payload = { "email": email, "password": password } res = requests.post(url=sign_in_url, data=payload) print(res, 'res user', res.json()) header_data = res.headers return_res = res.json() if return_res['message'] == 'logged in successfully': for num in range(101,200): product_URL = config("product_URL") product_URL = product_URL + "?page="+str(num) token = header_data['token'] client = header_data['client'] uid = header_data['uid'] headers = { "Content-Type": "application/json", "Accept": "application/json", "token": token, "client": client, "uid": uid } data = requests.get(url=product_URL, headers=headers) print(data, 'dataaaaaaaa' , data.json()) res_Data = data.json() for pro in res_Data['products']: record_id = 'Null' # if pro['brand']['name'] == brand_name: date = datetime.now(pytz.timezone('Asia/Kolkata')).strftime('%Y-%m-%d') if pro['category'] != None: record_id = pro['category']['id'] or 'Null' code = pro['category']['code'] or 'Null' name_english = pro['category']['name'] or 'Null' arabic_name = pro['category']['arabic_name'] or 'Null' data_list_cat.append((record_id, name_english, arabic_name, code)) brand_name = pro['brand']['name'] brand_name_arabic = pro['brand']['arabic_name'] or 'Null' product_description_arabic = pro['product_description_arabic'] or 'Null' product_description_english = pro['product_description_english'] or 'Null' net_content = pro['net_content'] or 'Null' net_content_arabic = pro['net_content_arabic'] or 'Null' image_url = pro['image_url'] or 'Null' tax_registration_number = pro['tax_registration_number'] or 'Null' source = pro['source'] or 'Null' company_name = pro['company_name'] or 'Null' target_market = pro['target_market'] or 'Null' active_from = pro['active_from'] or 'Null' active_to = pro['active_to'] or 'Null' gtin_record_status = pro['gtin_record_status'] or 'Null' variant_name_english = pro['variant_name'] or 'Null' variant_name_arabic = pro['variant_name_arabic'] or 'Null' sku_part_number = pro['sku_part_number'] or 'Null' hs_code = pro['hs_code'] or 'Null' supplier_code = pro['supplier_code'] or 'Null' gtin = pro['gtin'] or 'Null' if record_id == 'Null': data_list.append((brand_name, brand_name_arabic, product_description_arabic, product_description_english, net_content, net_content_arabic, image_url, tax_registration_number, source, company_name, target_market, active_from, active_to, gtin_record_status, variant_name_english, variant_name_arabic, sku_part_number, hs_code, supplier_code,date, record_id, gtin)) if record_id != 'Null': data_list.append((brand_name, brand_name_arabic, product_description_arabic, product_description_english, net_content, net_content_arabic, image_url, tax_registration_number, source, company_name, target_market, active_from, active_to, gtin_record_status, variant_name_english, variant_name_arabic, sku_part_number, hs_code, supplier_code, date, str(record_id), gtin)) data_list_cat = list(set([i for i in data_list_cat])) print('data_list_cat', data_list_cat) print('_____________***********_____', num) return data_list_cat, data_list @router.post("/gs1-product-data-dump", tags=["GS1-Integration"]) def brand_promotion(brand_name: str = Body(..., Embed=True)): # try: a = True if a: if not brand_name: return { "status": 200, "message": "Please provide Brand Name !!!" } gs_category_list, gs_product_data = generate_gsbrand_data(brand_name) print(gs_product_data, 'GSSSSSSSSSSSSSSSSSSSS') if len(gs_product_data) > 0: engine = connectToDB() cur = engine.cursor() mycategoryquery = f"""insert into gs_category (category_id,english_name,arabic_name,category_code) values(%s,%s,%s,%s)""" cat_api_query = cur.executemany(mycategoryquery, gs_category_list) myquery = f"""insert into gs_products (brand_name,arabic_brand_name,product_description_arabic,product_description_english,net_content,net_content_arabic,image_url,tax_registration_number,source,company_name,target_market,active_from,active_to,gtin_record_status,variant_name_english,variant_name_arabic,sku_part_number,hs_code,supplier_code,created_at, category_id,gtin) values(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""" api_query = cur.executemany(myquery, gs_product_data) engine.commit() response = { "status": 200, "Message": 'Success' } return response else: return { "status": 200, "message": "GS1 Data in not Provided or Something worng with the GS1-API !!!" } # except (Exception, psycopg2.DatabaseError) as error: # print("**********error****error********", error) # response = { # "status": 403, # "message": str(error), # } # return response
Leave a Comment