File: //home/arjun/projects/unlimited-leads/Unlimited-Leads-Be/Admin/helper.py
from services.s3_handler import S3Handler
import logging
import hashlib
from typing import Union
import pandas as pd
import numpy as np
import re
logger = logging.getLogger(__name__)
def compute_hash(email:Union[str, None], phone:Union[str, None], lead_type:Union[str, None],
business_name:Union[str, None], first_name:Union[str, None],
last_name:Union[str, None], street_address:Union[str, None],
city:Union[str, None], state:Union[str, None], country:Union[str, None],
zipcode:Union[str, None], sic_code:Union[str, None]) -> str:
# All fields except email and phone
all_fields = [lead_type, business_name, last_name, first_name, sic_code,
street_address, city, state, country, zipcode]
available_fields = [field.lower() for field in all_fields if field != None]
# Use all available_fields if both email and phone are None
if not email and not phone:
value_to_hash = f"{' '.join(available_fields)}"
elif not email:
value_to_hash = f"{phone.lower()}"
elif not phone:
value_to_hash = f"{email.lower()}"
else:
value_to_hash = f"{email.lower()}{phone.lower()}"
# Return the hash value
return hashlib.sha256(value_to_hash.encode()).hexdigest()
def normalize_empty_to_none(value:str) -> Union[str, None]:
"""
Converts an empty string value to None for database insertion.
Otherwise, returns the original value.
"""
if value != "":
return value
return None
def convert_to_int_string(value):
"""
Converts a numeric value to a string representation of an integer if the value is an int or float.
Returns the string representation directly for other data types.
Returns an empty string if the value is null or NaN.
"""
if pd.notnull(value):
if isinstance(value, (int, float)):
return str(int(value))
return str(value)
return ""
def remove_timestamp_from_filename(filename):
"""
Removes a timestamp in the format '-YYYY-MM-DD HH:MM:SS.ssssss+HH:MM'
from the end of a filename.
"""
# Regular expression to match the timestamp at the end of the filename
timestamp_pattern = r"-\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d+\+\d{2}:\d{2}$"
# Remove the timestamp from the filename
return re.sub(timestamp_pattern, "", filename)
def download_template_csv_file(key, s3_handler = S3Handler()):
"""
Generate a presigned URL to download the template CSV file.
Args:
key (str): The S3 object key for the template CSV file.
expiration (int): Time in seconds for the presigned URL to remain valid.
Returns:
str or None: The presigned URL for downloading the template CSV file, or None if an error occurs.
"""
try:
# Generate the presigned URL for the template CSV file
presigned_url = s3_handler.generate_presigned_url(key, s3_handler.expiration)
return presigned_url
except Exception as e:
logger.error(f"Error generating download URL for template CSV file: {str(e)}")
return None