734576817c
- Extracted Email functions to app/lib_email.py. - Extracted Export functions to app/lib_export.py. - Extracted JWT utilities to app/lib_jwt.py. - Extracted Hash utilities to app/lib_hash.py. - Updated app/lib_general.py to import from these new modules for backward compatibility. - Updated V3 Frontend API Guide with latest security and site_domain exception details.
117 lines
4.1 KiB
Python
117 lines
4.1 KiB
Python
import os
|
|
import pandas
|
|
import pathlib
|
|
import logging
|
|
from typing import Optional, Union
|
|
|
|
from app.log import logger_reset
|
|
from app.config import settings
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
# ### BEGIN ### API Lib Export ### create_export_file() ###
|
|
# Moved from lib_general.py 2026-01-07
|
|
@logger_reset
|
|
def create_export_file(
|
|
data_dict_list: list,
|
|
subdir_path: str,
|
|
filename: str,
|
|
column_name_li: list = [],
|
|
rm_id: bool = True,
|
|
export_type: str = 'CSV', # CSV, Excel
|
|
) -> Union[bool, str]:
|
|
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
log.debug(locals())
|
|
|
|
hosted_tmp_path = settings.FILES_PATH['hosted_tmp_root']
|
|
log.info(f'Hosted Temp Path: {hosted_tmp_path}')
|
|
|
|
subdirectory_dest = os.path.join(hosted_tmp_path, subdir_path)
|
|
log.debug(subdirectory_dest)
|
|
pathlib.Path(subdirectory_dest).mkdir(parents=True, exist_ok=True)
|
|
file_dest_w_subdir = os.path.join(subdirectory_dest, filename)
|
|
log.info(f'File Dest With Subdir: {file_dest_w_subdir}')
|
|
|
|
if column_name_li:
|
|
log.info('Using column name list passed')
|
|
else:
|
|
log.info('Using an auto generated column name list')
|
|
column_name_li = list(data_dict_list[0].keys())
|
|
log.debug(column_name_li)
|
|
|
|
if rm_id:
|
|
for column_name in list(column_name_li):
|
|
if column_name.endswith('_id'):
|
|
column_name_li.remove(column_name)
|
|
log.info(f'Removing column name: {column_name}')
|
|
log.info(column_name_li)
|
|
|
|
data_dataframe = pandas.DataFrame(data_dict_list)
|
|
log.debug(data_dataframe)
|
|
|
|
missing_cols = [col for col in column_name_li if col not in data_dataframe.columns]
|
|
if missing_cols:
|
|
column_name_li = [col for col in column_name_li if col not in missing_cols]
|
|
|
|
try:
|
|
if export_type == 'CSV':
|
|
log.info('Saving dataframe to CSV file')
|
|
full_dest_path = file_dest_w_subdir+'.csv'
|
|
filename_w_ext = filename+'.csv'
|
|
tmp_file_path = os.path.join(subdir_path,filename_w_ext)
|
|
data_dataframe.to_csv(
|
|
full_dest_path,
|
|
na_rep='NULL',
|
|
columns=column_name_li,
|
|
index=False,
|
|
)
|
|
elif export_type == 'Excel':
|
|
log.info('Saving dataframe to Excel file')
|
|
full_dest_path = file_dest_w_subdir+'.xlsx'
|
|
filename_w_ext = filename+'.xlsx'
|
|
tmp_file_path = os.path.join(subdir_path,filename_w_ext)
|
|
data_dataframe.to_excel(
|
|
full_dest_path,
|
|
na_rep='NULL',
|
|
columns=column_name_li,
|
|
index=False,
|
|
)
|
|
except:
|
|
log.exception('Something went wrong while trying to save the export file.')
|
|
return False
|
|
|
|
log.info(f'Temp File Path: {tmp_file_path}')
|
|
|
|
return tmp_file_path
|
|
# ### END ### API Lib Export ### create_export_file() ###
|
|
|
|
# ### BEGIN ### API Lib Export ### return_full_tmp_path() ###
|
|
# This is for using with return FileResponse(path=full_tmp_path, filename=filename)
|
|
# Moved from lib_general.py 2026-01-07
|
|
@logger_reset
|
|
def return_full_tmp_path(
|
|
full_tmp_path: str = None,
|
|
subdir_path: str = None,
|
|
filename: str = None,
|
|
) -> Union[bool, str]:
|
|
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
log.debug(locals())
|
|
|
|
hosted_tmp_path = settings.FILES_PATH['hosted_tmp_root']
|
|
log.info(f'Hosted Temp Path: {hosted_tmp_path}')
|
|
|
|
if full_tmp_path:
|
|
file_dest = os.path.join(hosted_tmp_path, full_tmp_path)
|
|
return file_dest
|
|
elif subdir_path and filename:
|
|
subdirectory_dest = os.path.join(hosted_tmp_path, subdir_path)
|
|
log.debug(subdirectory_dest)
|
|
pathlib.Path(subdirectory_dest).mkdir(parents=True, exist_ok=True)
|
|
file_dest_w_subdir = os.path.join(subdirectory_dest, filename)
|
|
log.info(f'File Dest With Subdir: {file_dest_w_subdir}')
|
|
|
|
return file_dest_w_subdir
|
|
else:
|
|
return False
|
|
# ### END ### API Lib Export ### return_full_tmp_path() ###
|