Spaces:
Sleeping
Sleeping
File size: 2,643 Bytes
cffeaa1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 |
from crew.tools.db_utils import run_query
INTERNAL_DB_PATH = "database/internal.db"
def get_customer_id_by_name(name: str):
# Use LIKE and wildcard to be more flexible with names
query = "SELECT customer_id FROM customer_details WHERE name LIKE ? COLLATE NOCASE"
try:
# This will match "Andy", "Andy Lau", or "Andy "
result = run_query(INTERNAL_DB_PATH, query, (f"{name.strip()}%",))
if result and len(result) > 0:
return result[0][0]
return f"NOT_FOUND: Customer '{name}' does not exist."
except Exception as e:
return f"ERROR: {str(e)}"
def get_customer_details(customer_id: str):
"""
Fetch basic customer details (name, email, nationality).
"""
query = """
SELECT name, email, nationality
FROM customer_details
WHERE customer_id = ?
"""
try:
result = run_query(INTERNAL_DB_PATH, query, (customer_id,))
if result and len(result) > 0:
name, email, nationality = result[0]
return {
"customer_id": customer_id,
"name": name,
"email": email,
"nationality": nationality
}
else:
return f"NOT_FOUND: No details found for customer_id '{customer_id}'."
except Exception as e:
return f"ERROR: Failed to fetch details for id '{customer_id}': {str(e)}"
def get_credit_score(customer_id: str):
"""
Fetch customer credit score.
"""
query = """
SELECT credit_score
FROM customer_credit_score
WHERE customer_id = ?
"""
try:
result = run_query(INTERNAL_DB_PATH, query, (customer_id,))
if result and len(result) > 0:
return result[0][0]
else:
return f"NOT_FOUND: Credit score unavailable for customer_id '{customer_id}'."
except Exception as e:
return f"ERROR: Failed to fetch credit score for id '{customer_id}': {str(e)}"
def get_account_status(customer_id: str):
"""
Fetch account status (Active, Closed, Delinquent).
"""
query = """
SELECT account_status
FROM customer_account_status
WHERE customer_id = ?
"""
try:
result = run_query(INTERNAL_DB_PATH, query, (customer_id,))
if result and len(result) > 0:
return result[0][0]
else:
return f"NOT_FOUND: Account status unavailable for customer_id '{customer_id}'."
except Exception as e:
return f"ERROR: Failed to fetch account status for id '{customer_id}': {str(e)}" |