Untitled
unknown
plain_text
a month ago
5.7 kB
3
Indexable
import logging from io import BytesIO from typing import List import openpyxl import pandas as pd import strawberry from sqlalchemy.sql import text from strawberry import mutation, type from strawberry.file_uploads import Upload from app.actions.import_functionality_helper.import_functionality_mutation_helper import ( get_category_id, get_full_group_list, get_full_product_category_list, get_full_product_country_list, get_last_version_id, get_lov_values, get_mdm_id, get_subcategory_ids, is_existing_record, is_valid_data, save_product_categories, save_topics_from_excel, truncate_tables, update_product_category_version, ) from app.graphql.schema import Info logger = logging.getLogger(__name__) @type class ImportMutationPayload: success: bool message: str = "" @type class ImportFunctionalityMutation: @mutation async def read_file(self, file: Upload, import_type: str, info: Info) -> ImportMutationPayload: session = info.context.session try: file_content = await file.read() workbook = openpyxl.load_workbook(filename=BytesIO(file_content)) sheet = workbook.active data = [] for row in sheet.iter_rows(values_only=True): data.append(row) df = pd.DataFrame(data[1:], columns=data[0]) headers = df.columns.tolist() expected_headers = [ "FUNCTIONAL GROUP", "COUNTRY", "PRODUCT", "CATEGORY", "TOPIC", "TA SUB TYPE", ] if headers != expected_headers: return ImportMutationPayload(success = False, message="This is not a correct excel to upload") data_error_for_all = "" prd_cat_list = [] last_version_id = ( 1 if import_type.lower() == "truncateandproceed" else get_last_version_id(session) ) product_country_list = get_full_product_country_list(session) product_category_list = get_full_product_category_list(session) group_list = get_full_group_list(session) func_group_list = get_lov_values(session, "PROD CAT GRP") sub_ta_list = get_lov_values(session, "therapeutic") for i, row in df.iterrows(): prd_group = row["FUNCTIONAL GROUP"].strip() country = row["COUNTRY"].strip() product_name = row["PRODUCT"].strip() cat_name = row["CATEGORY"].strip() ta_sub_type = row["TA SUB TYPE"].strip() topic_names = row["TOPIC"].strip() mdm_id = await get_mdm_id( info, session, product_name, country, product_country_list ) cat_id = get_category_id(cat_name, group_list) prdt_sbctgries = get_subcategory_ids(topic_names, group_list) if not is_valid_data( prd_group, mdm_id, cat_id, ta_sub_type, prdt_sbctgries, func_group_list, sub_ta_list, ): data_error_for_all += f"Error in data at line {i+2}: Invalid data\n" continue if ( is_existing_record( mdm_id, cat_id, prd_group, ta_sub_type, product_category_list ) and import_type.lower() != "truncateandproceed" ): data_error_for_all += f"The record at line {i+2} already exists.\n" continue prd_cat = { "version_id": last_version_id, "prod_cat_grp": prd_group, "mdm_id": mdm_id, "category_id": cat_id, "sd": prdt_sbctgries, "ta_sub_type": ta_sub_type, "last_active": 1, "is_deleted": 0, "created_date": pd.Timestamp.now(), "created_by_id": "SYSTEM", "updated_date": pd.Timestamp.now(), "updated_by_id": "SYSTEM", } prd_cat_list.append(prd_cat) last_version_id += 1 if data_error_for_all: return ImportMutationPayload(success = False, message={data_error_for_all}) # Save data to the database if import_type.lower() == "truncateandproceed": truncate_tables(session) save_product_categories(session, prd_cat_list) update_product_category_version(session, last_version_id) save_topics_from_excel(session, df, prd_cat_list, group_list) else: save_product_categories(session, prd_cat_list) update_product_category_version(session, last_version_id) save_topics_from_excel(session, df, prd_cat_list, group_list) return ImportMutationPayload(success = True, message="Import completed successfully") except Exception as e: session.rollback() return ImportMutationPayload(success = False, message=f"Error: {str(e)}") finally: session.close()
Editor is loading...
Leave a Comment