import io import os from datetime import datetime from typing import Optional import pandas as pd from fastapi import APIRouter, Depends from sqlalchemy import String, cast, func from sqlalchemy.orm import Session from src.apps.base.models.country import Country from src.apps.base.models.regions import Region, SubRegion from src.apps.export_data.schemas.export_data import DownloadFilesCreateSchema from src.apps.export_data.services.download_file import create_download_file from src.apps.files.schemas.file import CreateFileWithPathSchema from src.apps.files.services.file import create_file_from_path from src.apps.wine.appellation.models.appellation import Appellation, SubAppellation from src.apps.wine.bottle_size.models.bottle_size import BottleSize from src.apps.wine.global_noise.models.global_noise import GlobalNoise from src.apps.wine.keyword_spread.models.keyword_spread import KeywordSpread from src.apps.wine.keyword_substitution.models.keyword_substitution import ( KeywordSubstitution, ) from src.apps.wine.producer.models.producer import ( Producer, ProducerKeyword, ProducerNoise, ) from src.apps.wine.variety.models.variety import Variety from src.apps.wine.wine.models.wine import ( Wine, WineDb, WineDuplication, WineKeyword, WineNoise, ) from src.apps.wine.word_alias.models.word_alias import WordAlias from src.apps.wine.word_elimination.models.word_elimination import WordElimination from src.apps.wine.word_reserved.models.word_reserved import WordReserved from src.celery.celery_app import celery_app from src.core.dependencies import get_db from src.utils.constants import ( EXPORT_DATA_LINE_SEPARATOR, EXPORT_DATA_LINE_TERMINATOR, EXPORT_DATA_MAX_LIMIT, EXPORT_DATA_SUB_FOLDER, ) from src.utils.enums import ( ExportDataTypes, ExportDataTypesFileName, ExportDataTypesHeaders, ) from src.utils.helpers.functions import cleanup_files_by_prefix, get_description @celery_app.task(name="export_data.generate_mongo_map_export_file") def generate_mongo_map_export_file(): try: # get a new db session db = next(get_db()) # count the number of rows to be exported count = db.query(Wine).filter(Wine.mongo_id.isnot(None), Wine.mongo_id != "").count() # query to get the data to be exported data = ( db.query(Wine.literal.label("WineID"), Wine.mongo_id.label("MongoID"), Wine.last_updated) .filter(Wine.mongo_id.isnot(None), Wine.mongo_id != "") .order_by(Wine.id) .all() ) # prepare file file_name = f"{ExportDataTypesFileName.MONGO_MAP.value}{datetime.now().strftime('%Y-%m-%d-%H%M%S')}.txt" download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "masterdata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.MONGO_MAP.value) local_path = os.path.join(download_location, file_name) df = pd.DataFrame(data) # Replace NaN with empty string df = df.fillna("").replace("NaN", "") # Write to CSV file df.to_csv(local_path, index=False, sep="\t") return {"status": "success", "file_name": file_name, "total_rows": count} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.generate_wine_db_export_file") def generate_wine_db_export_file(): try: # get a new db session db = next(get_db()) # count the number of rows to be exported count = db.query(WineDb).count() # query to get the data to be exported data = ( db.query( WineDb.literal.label("winealert_id"), Producer.literal.label("WineAl"), func.concat(Producer.name_show, " ", WineDb.name, " ", WineDb.color).label( "ProdShow LabelName ColorClass" ), WineDb.wine_n_id.label("VinN"), Producer.name.label("Prod"), Producer.name_show.label("ProdShow"), WineDb.name.label("LabelName"), Variety.name.label("Variety"), WineDb.color.label("ColorClass"), WineDb.sweetness.label("Dryness"), WineDb.type.label("WineType"), Country.name.label("Country"), Region.name.label("Region"), SubRegion.name.label("Sub-region"), Appellation.name.label("Appellation"), SubAppellation.name.label("Sub-appellation"), WineDb.status.label("Status"), WineDb.last_updated.label("Last Modified"), ) .join(Producer, WineDb.producer_id == Producer.id) .outerjoin(Country, cast(Country.id, String) == WineDb.wine_country) .outerjoin(Region, cast(Region.id, String) == WineDb.wine_region) .outerjoin(Variety, Variety.id == WineDb.variety_id) .outerjoin(SubRegion, cast(SubRegion.id, String) == WineDb.wine_location) .outerjoin(Appellation, cast(Appellation.id, String) == WineDb.wine_locale) .outerjoin(SubAppellation, cast(SubAppellation.id, String) == WineDb.wine_site) .filter(WineDb.literal != "z") .filter(~WineDb.literal.like("%xxx")) .order_by(WineDb.literal) .all() ) # prepare file file_name = f"{ExportDataTypesFileName.WINE_DB.value}{datetime.now().strftime('%Y-%m-%d-%H%M%S')}.txt" download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "masterdata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_DB.value) local_path = os.path.join(download_location, file_name) df = pd.DataFrame(data) # Replace NaN with empty string df = df.fillna("").replace(["NaN", "nan", "None", None], "") # Write to CSV file df.to_csv(local_path, index=False, sep="\t") return {"status": "success", "file_name": file_name, "total_rows": count} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.generate_wine_keyword_export_file") def generate_wine_keyword_export_file(): try: # get a new db session db = next(get_db()) # count the number of rows to be exported count = db.query(WineKeyword).count() # query to get the data to be exported data = ( db.query( WineDb.literal.label("winealert_id"), WineKeyword.pattern, WineKeyword.producer_keyword, WineKeyword.base_keyword, WineKeyword.appellation_keyword, WineKeyword.color_keyword, WineKeyword.common_modifiers_keyword, WineKeyword.modifiers_keyword, WineKeyword.custom_not_keyword, WineKeyword.global_not_keyword, WineKeyword.last_updated, ) .join(WineDb, WineKeyword.wine_db_id == WineDb.id) .filter(WineDb.literal != "z") .filter(~WineDb.literal.like("%xxx")) .order_by(WineDb.literal) .all() ) # prepare file file_name = f"{ExportDataTypesFileName.WINE_KEYWORD.value}{datetime.now().strftime('%Y-%m-%d-%H%M%S')}.txt" download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "masterdata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_KEYWORD.value) local_path = os.path.join(download_location, file_name) df = pd.DataFrame(data) # Replace NaN with empty string df = df.fillna("").replace("NaN", "") # Write to CSV file df.to_csv(local_path, index=False, sep="\t") return {"status": "success", "file_name": file_name, "total_rows": count} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.generate_producer_keyword_export_file") def generate_producer_keyword_export_file(): try: # get a new db session db = next(get_db()) # count the number of rows to be exported count = db.query(ProducerKeyword).count() # query to get the data to be exported data = ( db.query( ProducerKeyword.id, Producer.literal.label("WineAl"), ProducerKeyword.keyword, ProducerKeyword.last_updated, ) .join(Producer, ProducerKeyword.producer_id == Producer.id) .order_by(ProducerKeyword.id) .all() ) # prepare file file_name = f"{ExportDataTypesFileName.PRODUCER_KEYWORD.value}{datetime.now().strftime('%Y-%m-%d-%H%M%S')}.txt" download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "masterdata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.PRODUCER_KEYWORD.value) local_path = os.path.join(download_location, file_name) df = pd.DataFrame(data) # Replace NaN with empty string df = df.fillna("").replace("NaN", "") # Write to CSV file df.to_csv(local_path, index=False, sep="\t") return {"status": "success", "file_name": file_name, "total_rows": count} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.generate_global_noise_export_file") def generate_global_noise_export_file(): try: # get a new db session db = next(get_db()) # count the number of rows to be exported count = db.query(GlobalNoise).count() # query to get the data to be exported data = ( db.query(GlobalNoise.id.label("NoiseID"), GlobalNoise.noise.label("Noise"), GlobalNoise.updated_at) .order_by(GlobalNoise.id) .all() ) # prepare file file_name = f"{ExportDataTypesFileName.GLOBAL_NOISE.value}{datetime.now().strftime('%Y-%m-%d-%H%M%S')}.txt" download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "masterdata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.GLOBAL_NOISE.value) local_path = os.path.join(download_location, file_name) df = pd.DataFrame(data) # Replace NaN with empty string df = df.fillna("").replace("NaN", "") # Write to CSV file df.to_csv(local_path, index=False, sep="\t") return {"status": "success", "file_name": file_name, "total_rows": count} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.generate_producer_noise_export_file") def generate_producer_noise_export_file(): try: # get a new db session db = next(get_db()) # count the number of rows to be exported count = db.query(ProducerNoise).count() # query to get the data to be exported data = ( db.query( ProducerNoise.id.label("Noiseid"), Producer.literal.label("WineAl"), ProducerNoise.noise.label("Noise"), ProducerNoise.last_updated, ) .join(Producer, ProducerNoise.producer_id == Producer.id) .order_by(ProducerNoise.id) .all() ) # prepare file file_name = f"{ExportDataTypesFileName.PRODUCER_NOISE.value}{datetime.now().strftime('%Y-%m-%d-%H%M%S')}.txt" download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "masterdata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.PRODUCER_NOISE.value) local_path = os.path.join(download_location, file_name) df = pd.DataFrame(data) # Replace NaN with empty string df = df.fillna("").replace("NaN", "") # Write to CSV file df.to_csv(local_path, index=False, sep="\t") return {"status": "success", "file_name": file_name, "local_path": local_path, "total_rows": count} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.get_wine_noise_export_file") def generate_wine_noise_export_file(): try: # get a new db session db = next(get_db()) # count the number of rows to be exported count = db.query(WineNoise).count() # query to get the data to be exported data = ( db.query( WineNoise.id.label("NoiseId"), WineNoise.literal.label("Wine Alert Id"), WineNoise.noise.label("Noise"), WineNoise.last_updated, ) .join(WineDb, WineNoise.wine_db_id == WineDb.id) .order_by(WineNoise.id) .all() ) # prepare file file_name = f"{ExportDataTypesFileName.WINE_NOISE.value}{datetime.now().strftime('%Y-%m-%d-%H%M%S')}.txt" download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "masterdata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) df = pd.DataFrame(data) # Replace NaN with empty string df = df.fillna("").replace("NaN", "") # Write to CSV file df.to_csv(local_path, index=False, sep="\t") return {"status": "success", "file_name": file_name, "total_rows": count} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.get_spread_table_export_file") def generate_spread_table_export_file(): try: # get a new db session db = next(get_db()) # count the number of rows to be exported count = db.query(KeywordSpread).count() # query to get the data to be exported data = db.query( KeywordSpread.id.label("SpreadTableID"), KeywordSpread.keyword.label("Keyword"), KeywordSpread.region_code.label("RegionCode"), KeywordSpread.keyword_type.label("KeywordType"), KeywordSpread.last_updated, ).all() # prepare file file_name = f"{ExportDataTypesFileName.SPREAD_TABLE.value}{datetime.now().strftime('%Y-%m-%d-%H%M%S')}.txt" download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "masterdata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) df = pd.DataFrame(data) # Replace NaN with empty string df = df.fillna("").replace("NaN", "") # Write to CSV file # df.to_csv(local_path, index=False) df.to_csv(local_path, index=False, sep="\t") return {"status": "success", "file_name": file_name, "total_rows": count} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.get_word_reserved_export_file") def generate_word_reserved_export_file(): try: # get a new db session db = next(get_db()) # count the number of rows to be exported count = db.query(WordReserved).count() # query to get the data to be exported data = db.query( WordReserved.id.label("ID"), WordReserved.word.label("Word"), WordReserved.reserved_type.label("ReservedType"), WordReserved.last_updated, ).all() # prepare file file_name = f"{ExportDataTypesFileName.WORD_RESERVED.value}{datetime.now().strftime('%Y-%m-%d-%H%M%S')}.txt" download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "masterdata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) df = pd.DataFrame(data) # Replace NaN with empty string df = df.fillna("").replace("NaN", "") # Write to CSV file # df.to_csv(local_path, index=False) df.to_csv(local_path, index=False, sep="\t") return {"status": "success", "file_name": file_name, "total_rows": count} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.get_bottle_size_export_file") def generate_bottle_size_export_file(): try: # get a new db session db = next(get_db()) # count the number of rows to be exported count = db.query(BottleSize).count() # query to get the data to be exported data = db.query( BottleSize.id.label("ID"), BottleSize.name.label("Name"), BottleSize.alias.label("Alias"), BottleSize.exclusion.label("Exclusion"), BottleSize.last_updated, ).all() # prepare file file_name = f"{ExportDataTypesFileName.BOTTLE_SIZE.value}{datetime.now().strftime('%Y-%m-%d-%H%M%S')}.txt" download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "masterdata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) df = pd.DataFrame(data) # Replace NaN with empty string df = df.fillna("").replace("NaN", "") # Write to CSV file # df.to_csv(local_path, index=False) df.to_csv(local_path, index=False, sep="\t") return {"status": "success", "file_name": file_name, "total_rows": count} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.get_wine_alias_export_file") def generate_wine_alias_export_file(): try: # get a new db session db = next(get_db()) # count the number of rows to be exported count = db.query(WordAlias).count() # query to get the data to be exported data = db.query( WordAlias.id.label("ID"), WordAlias.word.label("Word"), WordAlias.alias.label("Alias"), WordAlias.last_updated, ).all() # prepare file file_name = f"{ExportDataTypesFileName.WINE_ALIAS.value}{datetime.now().strftime('%Y-%m-%d-%H%M%S')}.txt" download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "masterdata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) df = pd.DataFrame(data) # Replace NaN with empty string df = df.fillna("").replace("NaN", "") # Write to CSV file # df.to_csv(local_path, index=False) df.to_csv(local_path, index=False, sep="\t") return {"status": "success", "file_name": file_name, "total_rows": count} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.get_varieties_export_file") def generate_varieties_export_file(): try: # get a new db session db = next(get_db()) # count the number of rows to be exported count = db.query(Variety).count() # query to get the data to be exported data = db.query( Variety.id.label("ID"), Variety.name.label("Name"), Variety.default_color.label("DefaultColor"), Variety.last_updated, ).all() # prepare file file_name = f"{ExportDataTypesFileName.VARIETIES.value}{datetime.now().strftime('%Y-%m-%d-%H%M%S')}.txt" download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "masterdata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) df = pd.DataFrame(data) # Replace NaN with empty string df = df.fillna("").replace("NaN", "") # Write to CSV file # df.to_csv(local_path, index=False) df.to_csv(local_path, index=False, sep="\t") return {"status": "success", "file_name": file_name, "total_rows": count} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.colour_substitution_export_file") def generate_colour_substitution_export_file(): try: # get a new db session db = next(get_db()) # count the number of rows to be exported count = db.query(KeywordSubstitution).count() # query to get the data to be exported data = db.query( KeywordSubstitution.id.label("ID"), KeywordSubstitution.keyword.label("Keyword"), KeywordSubstitution.substitution.label("Substitution"), KeywordSubstitution.substitution_type.label("SubstitutionType"), KeywordSubstitution.last_updated, ).all() # prepare file file_name = ( f"{ExportDataTypesFileName.COLOUR_SUBSTITUTIONS.value}{datetime.now().strftime('%Y-%m-%d-%H%M%S')}.txt" ) download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "masterdata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) df = pd.DataFrame(data) # Replace NaN with empty string df = df.fillna("").replace("NaN", "") # Write to CSV file # df.to_csv(local_path, index=False) df.to_csv(local_path, index=False, sep="\t") return {"status": "success", "file_name": file_name, "total_rows": count} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.get_word_elimination_export_file") def generate_word_elimination_export_file(): try: # get a new db session db = next(get_db()) # count the number of rows to be exported count = db.query(WordElimination).count() # query to get the data to be exported data = db.query( WordElimination.id.label("ID"), WordElimination.word.label("Word"), WordElimination.elimination_type.label("EliminationType"), WordElimination.last_updated, ).all() # prepare file file_name = f"{ExportDataTypesFileName.WORD_ELIMINATION.value}{datetime.now().strftime('%Y-%m-%d-%H%M%S')}.txt" download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "masterdata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) df = pd.DataFrame(data) # Replace NaN with empty string df = df.fillna("").replace("NaN", "") # Write to CSV file # df.to_csv(local_path, index=False) df.to_csv(local_path, index=False, sep="\t") return {"status": "success", "file_name": file_name, "total_rows": count} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.generate_appellation_export_file") def generate_appellation_export_file(): try: # get a new db session db = next(get_db()) # count the number of rows to be exported count = db.query(Appellation).count() # query to get the data to be exported data = ( db.query( Appellation.id.label("ID"), Appellation.name.label("Name"), Country.name.label("Country"), Region.name.label("Region"), SubRegion.name.label("Sub-region"), Appellation.updated_at.label("Last Modified"), ) .outerjoin(Country, Country.id == Appellation.country_id) .outerjoin(Region, Region.id == Appellation.region_id) .outerjoin(SubRegion, SubRegion.id == Appellation.sub_region_id) .order_by(Appellation.id) .all() ) # prepare file file_name = f"{ExportDataTypesFileName.APPELLATION.value}{datetime.now().strftime('%Y-%m-%d-%H%M%S')}.txt" download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "masterdata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_DB.value) local_path = os.path.join(download_location, file_name) df = pd.DataFrame(data) # Replace NaN with empty string df = df.fillna("").replace("NaN", "") # Write to CSV file df.to_csv(local_path, index=False, sep="\t") return {"status": "success", "file_name": file_name, "total_rows": count} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() ########################## Update from CSV tasks ########################## @celery_app.task(name="export_data.update_mongo_map_from_csv") def update_mongo_map_from_csv(file_name: str): db = next(get_db()) try: # Read the download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "uploaddata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) # Read the tab-separated TXT file df = pd.read_csv(local_path, sep="\t") # Example: ensure correct column names # Expected: id, name, country, price, score for _, row in df.iterrows(): wine: Wine = db.query(Wine).filter(Wine.literal == row["WineID"]).first() if wine: # Update existing wine.mongo_id = row["MongoID"] db.add(wine) db.commit() return {"status": "success", "file_name": file_name} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.update_wine_db_from_csv") def update_wine_db_from_csv(file_name: str): db = next(get_db()) try: # Read the download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "uploaddata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) # Read the tab-separated TXT file df = pd.read_csv(local_path, sep="\t") # Example: ensure correct column names # Expected: id, name, country, price, score for _, row in df.iterrows(): wine: WineDb = db.query(WineDb).filter(WineDb.literal == row["winealert_id"]).first() if wine: # Update existing wine.name = row["LabelName"] wine.color = row["ColorClass"] wine.type = row["WineType"] wine.sweetness = row["Dryness"] wine.wine_n_id = row["VinN"] wine.status = row["Status"] if row["WineAl"]: producer = ( db.query(Producer).filter(func.lower(Producer.literal) == func.lower(row["WineAl"])).first() ) if producer: producer.name_show = row["ProdShow"] producer.name = row["Prod"] db.add(producer) wine.producer_id = producer.id if row["Variety"]: variety = db.query(Variety).filter(func.lower(Variety.name) == func.lower(row["Variety"])).first() if variety: wine.variety_id = variety.id if row["Country"]: country = db.query(Country).filter(func.lower(Country.name) == func.lower(row["Country"])).first() if country: wine.wine_country = str(country.id) if row["Region"]: region = db.query(Region).filter(func.lower(Region.name) == func.lower(row["Region"])).first() if region: wine.wine_region = str(region.id) if row["Sub-region"]: sub_region = ( db.query(SubRegion).filter(func.lower(SubRegion.name) == func.lower(row["Sub-region"])).first() ) if sub_region: wine.wine_location = str(sub_region.id) if row["Appellation"]: appellation = ( db.query(Appellation) .filter(func.lower(Appellation.name) == func.lower(row["Appellation"])) .first() ) if appellation: wine.wine_locale = str(appellation.id) if row["Sub-appellation"]: sub_appellation = ( db.query(SubAppellation) .filter(func.lower(SubAppellation.name) == func.lower(row["Sub-appellation"])) .first() ) if sub_appellation: wine.wine_site = str(sub_appellation.id) db.add(wine) db.commit() return {"status": "success", "file_name": file_name} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.update_wine_keyword_from_csv") def update_wine_keyword_from_csv(file_name: str): db = next(get_db()) try: # Read the download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "uploaddata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) # Read the tab-separated TXT file df = pd.read_csv(local_path, sep="\t") # Example: ensure correct column names # Expected: id, name, country, price, score for _, row in df.iterrows(): winedb: WineDb = db.query(WineDb).filter(WineDb.literal == row["winealert_id"]).first() if winedb: wine: WineKeyword = db.query(WineKeyword).filter(WineKeyword.wine_db_id == winedb.id).first() if wine: # Update existing wine.pattern = row["pattern"] wine.producer_keyword = row["producer_keyword"] wine.base_keyword = row["base_keyword"] wine.appellation_keyword = row["appellation_keyword"] wine.color_keyword = row["color_keyword"] wine.common_modifiers_keyword = row["common_modifiers_keyword"] wine.modifiers_keyword = row["modifiers_keyword"] wine.custom_not_keyword = row["custom_not_keyword"] wine.global_not_keyword = row["global_not_keyword"] db.add(wine) db.commit() return {"status": "success", "file_name": file_name} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.update_producer_keyword_from_csv") def update_producer_keyword_from_csv(file_name: str): db = next(get_db()) try: # Read the download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "uploaddata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) # Read the tab-separated TXT file df = pd.read_csv(local_path, sep="\t") # Example: ensure correct column names # Expected: id, name, country, price, score for _, row in df.iterrows(): wine: ProducerKeyword = db.query(ProducerKeyword).filter(ProducerKeyword.id == row["id"]).first() if wine: # Update existing wine.keyword = row["keyword"] db.add(wine) db.commit() return {"status": "success", "file_name": file_name} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.update_global_noise_from_csv") def update_global_noise_from_csv(file_name: str): db = next(get_db()) try: # Read the download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "uploaddata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) # Read the tab-separated TXT file df = pd.read_csv(local_path, sep="\t") # Example: ensure correct column names # Expected: id, name, country, price, score for _, row in df.iterrows(): wine: GlobalNoise = db.query(GlobalNoise).filter(GlobalNoise.id == row["NoiseId"]).first() if wine: # Update existing wine.noise = row["Noise"] db.add(wine) db.commit() return {"status": "success", "file_name": file_name} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.update_producer_noise_from_csv") def update_producer_noise_from_csv(file_name: str): db = next(get_db()) try: # Read the download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "uploaddata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) # Read the tab-separated TXT file df = pd.read_csv(local_path, sep="\t") # Example: ensure correct column names # Expected: id, name, country, price, score for _, row in df.iterrows(): wine: ProducerNoise = db.query(ProducerNoise).filter(ProducerNoise.id == row["NoiseId"]).first() if wine: # Update existing wine.noise = row["Noise"] db.add(wine) db.commit() return {"status": "success", "file_name": file_name} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.update_wine_noise_from_csv") def update_wine_noise_from_csv(file_name: str): db = next(get_db()) try: # Read the download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "uploaddata") os.makedirs(download_location, exist_ok=True) # cleanup # # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) # Read the tab-separated TXT file df = pd.read_csv(local_path, sep="\t") # Example: ensure correct column names # Expected: id, name, country, price, score for _, row in df.iterrows(): wine: WineNoise = db.query(WineNoise).filter(WineNoise.id == row["NoiseId"]).first() if wine: # Update existing wine.noise = row["Noise"] db.add(wine) db.commit() return {"status": "success", "file_name": file_name} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.update_spread_table_from_csv") def update_spread_table_from_csv(file_name: str): db = next(get_db()) try: # Read the download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "uploaddata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) # Read the tab-separated TXT file df = pd.read_csv(local_path, sep="\t") # Example: ensure correct column names # Expected: id, name, country, price, score for _, row in df.iterrows(): wine: KeywordSpread = db.query(KeywordSpread).filter(KeywordSpread.id == row["SpreadTableID"]).first() if wine: # Update existing wine.keyword = row["Keyword"] wine.region_code = row["RegionCode"] wine.keyword_type = row["KeywordType"] db.add(wine) db.commit() return {"status": "success", "file_name": file_name} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.update_word_reserved_from_csv") def update_word_reserved_from_csv(file_name: str): db = next(get_db()) try: # Read the download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "uploaddata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) # Read the tab-separated TXT file df = pd.read_csv(local_path, sep="\t") # Example: ensure correct column names # Expected: id, name, country, price, score for _, row in df.iterrows(): wine: WordReserved = db.query(WordReserved).filter(WordReserved.id == row["ID"]).first() if wine: # Update existing wine.word = row["Word"] wine.reserved_type = row["ReservedType"] db.add(wine) db.commit() return {"status": "success", "file_name": file_name} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.update_bottle_size_from_csv") def update_bottle_size_from_csv(file_name: str): db = next(get_db()) try: # Read the download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "uploaddata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) # Read the tab-separated TXT file df = pd.read_csv(local_path, sep="\t") # Example: ensure correct column names # Expected: id, name, country, price, score for _, row in df.iterrows(): wine: BottleSize = db.query(BottleSize).filter(BottleSize.id == row["ID"]).first() if wine: # Update existing wine.name = row["Name"] wine.alias = row["Alias"] wine.exclusion = row["Exclusion"] db.add(wine) db.commit() return {"status": "success", "file_name": file_name} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.update_wine_alias_from_csv") def update_wine_alias_from_csv(file_name: str): db = next(get_db()) try: # Read the download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "uploaddata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) # Read the tab-separated TXT file df = pd.read_csv(local_path, sep="\t") # Example: ensure correct column names # Expected: id, name, country, price, score for _, row in df.iterrows(): wine: WordAlias = db.query(WordAlias).filter(WordAlias.id == row["ID"]).first() if wine: # Update existing wine.word = row["Word"] wine.alias = row["Alias"] db.add(wine) db.commit() return {"status": "success", "file_name": file_name} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.update_varieties_from_csv") def update_varieties_from_csv(file_name: str): db = next(get_db()) try: # Read the download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "uploaddata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) # Read the tab-separated TXT file df = pd.read_csv(local_path, sep="\t") # Example: ensure correct column names # Expected: id, name, country, price, score for _, row in df.iterrows(): wine: Variety = db.query(Variety).filter(Variety.id == row["ID"]).first() if wine: # Update existing wine.name = row["Name"] wine.default_color = row["DefaultColor"] db.add(wine) db.commit() return {"status": "success", "file_name": file_name} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.update_colour_substitution_from_csv") def update_colour_substitution_from_csv(file_name: str): db = next(get_db()) try: # Read the download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "uploaddata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) # Read the tab-separated TXT file df = pd.read_csv(local_path, sep="\t") # Example: ensure correct column names # Expected: id, name, country, price, score for _, row in df.iterrows(): wine: KeywordSubstitution = ( db.query(KeywordSubstitution).filter(KeywordSubstitution.id == row["ID"]).first() ) if wine: # Update existing wine.keyword = row["Keyword"] wine.substitution = row["Substitution"] wine.substitution_type = row["SubstitutionType"] db.add(wine) db.commit() return {"status": "success", "file_name": file_name} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.update_word_elimination_from_csv") def update_word_elimination_from_csv(file_name: str): db = next(get_db()) try: # Read the download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "uploaddata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) # Read the tab-separated TXT file df = pd.read_csv(local_path, sep="\t") # Example: ensure correct column names # Expected: id, name, country, price, score for _, row in df.iterrows(): wine: WordElimination = db.query(WordElimination).filter(WordElimination.id == row["ID"]).first() if wine: # Update existing wine.word = row["Word"] wine.elimination_type = row["EliminationType"] db.add(wine) db.commit() return {"status": "success", "file_name": file_name} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close() @celery_app.task(name="export_data.update_appellation_from_csv") def update_appellation_from_csv(file_name: str): db = next(get_db()) try: # Read the download_location = os.path.join(os.getcwd(), EXPORT_DATA_SUB_FOLDER, "uploaddata") os.makedirs(download_location, exist_ok=True) # cleanup # cleanup_files_by_prefix(download_location, ExportDataTypesFileName.WINE_NOISE.value) local_path = os.path.join(download_location, file_name) # Read the tab-separated TXT file df = pd.read_csv(local_path, sep="\t") # Example: ensure correct column names # Expected: id, name, country, price, score for _, row in df.iterrows(): wine: Appellation = db.query(Appellation).filter(Appellation.id == row["ID"]).first() if wine: # Update existing wine.name = row["Name"] if row["Country"]: country = db.query(Country).filter(func.lower(Country.name) == func.lower(row["Country"])).first() if country: wine.country_id = country.id if row["Region"]: region = db.query(Region).filter(func.lower(Region.name) == func.lower(row["Region"])).first() if region: wine.region_id = region.id if row["Sub-region"]: sub_region = ( db.query(SubRegion).filter(func.lower(SubRegion.name) == func.lower(row["Sub-region"])).first() ) if sub_region: wine.sub_region_id = sub_region.id db.add(wine) db.commit() return {"status": "success", "file_name": file_name} except Exception as e: return {"status": "failed", "error": str(e)} finally: db.close()