fahm7787/EnrollmentData

1"""Streamlit Customer Management System."""2 3import streamlit as st4import pandas as pd5 6from api_interface import CustomerAPI7from exceptions import (8    CustomerDuplicateError,9    CustomerManagementError,10    CustomerNotFoundError,11    CustomerStorageError,12    CustomerValidationError,13)14 15st.set_page_config(16    page_title="Customer Management System",17    page_icon="👥",18    layout="wide",19)20 21EMPTY_FORM = {22    "name": "",23    "email": "",24    "phone": "",25    "address": "",26    "self": "",27}28 29 30def get_api() -> CustomerAPI:31    if "api" not in st.session_state:32        st.session_state.api = CustomerAPI()33    return st.session_state.api34 35 36def init_form_state(form_key: str) -> None:37    values_key = f"{form_key}_values"38    errors_key = f"{form_key}_errors"39    if values_key not in st.session_state:40        st.session_state[values_key] = EMPTY_FORM.copy()41    if errors_key not in st.session_state:42        st.session_state[errors_key] = {}43 44 45def clear_form_state(form_key: str) -> None:46    st.session_state[f"{form_key}_values"] = EMPTY_FORM.copy()47    st.session_state[f"{form_key}_errors"] = {}48 49 50def show_field_error(errors: dict[str, str], field: str) -> None:51    if field in errors:52        st.error(errors[field])53 54 55def show_general_error(error: Exception) -> None:56    if isinstance(error, CustomerValidationError):57        if error.field:58            return59        st.error(f"Validation error: {error.message}")60    elif isinstance(error, CustomerDuplicateError):61        st.warning(f"Duplicate customer: {error.message}")62    elif isinstance(error, CustomerNotFoundError):63        st.warning(f"Not found: {error.message}")64    elif isinstance(error, CustomerStorageError):65        st.error(f"Storage error: {error.message}")66    elif isinstance(error, CustomerManagementError):67        st.error(f"Error: {error.message}")68    else:69        st.error(f"Unexpected error: {error}")70 71 72def render_customer_form(73    form_key: str,74    submit_label: str,75    on_submit,76) -> None:77    init_form_state(form_key)78    values = st.session_state[f"{form_key}_values"]79    errors = st.session_state[f"{form_key}_errors"]80 81    with st.form(form_key, clear_on_submit=False):82        name = st.text_input(83            "Name *",84            value=values["name"],85            placeholder="John Doe",86        )87        show_field_error(errors, "name")88 89        email = st.text_input(90            "Email ID *",91            value=values["email"],92            placeholder="john.doe@example.com",93        )94        show_field_error(errors, "email")95 96        phone = st.text_input(97            "Phone *",98            value=values["phone"],99            placeholder="+1 555-123-4567",100        )101        show_field_error(errors, "phone")102 103        address = st.text_area(104            "Address *",105            value=values["address"],106            placeholder="123 Main Street, City, State, ZIP",107            height=100,108        )109        show_field_error(errors, "address")110 111        self_info = st.text_input(112            "Self *",113            value=values["self"],114            placeholder="Self-employed / Individual / Company reference",115        )116        show_field_error(errors, "self")117 118        if "_general" in errors:119            st.error(errors["_general"])120 121        submitted = st.form_submit_button(122            submit_label, type="primary", use_container_width=True123        )124 125    if submitted:126        st.session_state[f"{form_key}_values"] = {127            "name": name,128            "email": email,129            "phone": phone,130            "address": address,131            "self": self_info,132        }133        st.session_state[f"{form_key}_errors"] = {}134 135        try:136            on_submit(name, email, phone, address, self_info)137            clear_form_state(form_key)138            st.rerun()139        except CustomerValidationError as exc:140            if exc.field:141                st.session_state[f"{form_key}_errors"] = {exc.field: exc.message}142            else:143                st.session_state[f"{form_key}_errors"] = {"_general": exc.message}144            st.rerun()145        except CustomerManagementError as exc:146            st.session_state[f"{form_key}_errors"] = {"_general": exc.message}147            st.rerun()148 149 150def render_new_customer_form(api: CustomerAPI) -> None:151    st.subheader("Register New Customer")152    st.caption("Fill in all fields to add a new customer to the system.")153 154    success_key = "new_customer_success"155    if success_key in st.session_state:156        customer = st.session_state.pop(success_key)157        st.success(f"Customer '{customer.name}' created successfully!")158        st.info(159            f"**Customer ID:** `{customer.customer_id}`  \n"160            f"**Created at:** {customer.created_at}"161        )162 163    def handle_create(name, email, phone, address, self_info):164        customer = api.create_customer(165            name=name,166            email=email,167            phone=phone,168            address=address,169            self_info=self_info,170        )171        st.session_state["new_customer_success"] = customer172 173    render_customer_form("new_customer_form", "Save Customer", handle_create)174 175 176def render_update_customer_form(api: CustomerAPI) -> None:177    st.subheader("Update Customer")178    st.caption("Select a customer and update their information.")179 180    success_key = "update_customer_success"181    if success_key in st.session_state:182        customer = st.session_state.pop(success_key)183        st.success(f"Customer '{customer.name}' updated successfully!")184        st.info(f"**Last updated:** {customer.updated_at}")185 186    try:187        customers = api.get_all_customers()188    except CustomerStorageError as exc:189        show_general_error(exc)190        return191 192    if not customers:193        st.info("No customers available to update. Add a customer first.")194        return195 196    customer_labels = {f"{c.name} ({c.email})": c.customer_id for c in customers}197 198    if "update_selected_id" not in st.session_state:199        st.session_state.update_selected_id = customers[0].customer_id200 201    selected_label = st.selectbox(202        "Select customer to update",203        options=list(customer_labels.keys()),204        index=next(205            (206                i207                for i, label in enumerate(customer_labels)208                if customer_labels[label] == st.session_state.update_selected_id209            ),210            0,211        ),212    )213    selected_id = customer_labels[selected_label]214 215    if selected_id != st.session_state.update_selected_id:216        st.session_state.update_selected_id = selected_id217        customer = api.get_customer(selected_id)218        st.session_state["update_customer_form_values"] = {219            "name": customer.name,220            "email": customer.email,221            "phone": customer.phone,222            "address": customer.address,223            "self": customer.self_info,224        }225        st.session_state["update_customer_form_errors"] = {}226        st.rerun()227 228    if "update_customer_form_values" not in st.session_state:229        customer = api.get_customer(selected_id)230        st.session_state["update_customer_form_values"] = {231            "name": customer.name,232            "email": customer.email,233            "phone": customer.phone,234            "address": customer.address,235            "self": customer.self_info,236        }237 238    def handle_update(name, email, phone, address, self_info):239        customer = api.update_customer(240            customer_id=selected_id,241            name=name,242            email=email,243            phone=phone,244            address=address,245            self_info=self_info,246        )247        st.session_state["update_customer_success"] = customer248        st.session_state.update_selected_id = customer.customer_id249 250    render_customer_form("update_customer_form", "Update Customer", handle_update)251 252 253def render_view_customers(api: CustomerAPI) -> None:254    st.subheader("View Customers")255    st.caption("Browse and search all registered customers.")256 257    try:258        total = api.get_customer_count()259        st.metric("Total Customers", total)260 261        search_query = st.text_input(262            "Search customers",263            placeholder="Search by name, email, phone, address, or self...",264        )265 266        customers = api.search_customers(search_query)267 268        if not customers:269            st.info("No customers found. Use **New Customer** to add one.")270            return271 272        rows = api.customers_to_dataframe_rows()273        if search_query.strip():274            search_lower = search_query.strip().lower()275            rows = [276                row277                for row in rows278                if search_lower in row["name"].lower()279                or search_lower in row["email"].lower()280                or search_lower in row["phone"].lower()281                or search_lower in row["address"].lower()282                or search_lower in str(row.get("self", "")).lower()283            ]284 285        df = pd.DataFrame(rows)286        display_df = df[287            ["name", "email", "phone", "address", "self", "created_at"]288        ].rename(289            columns={290                "name": "Name",291                "email": "Email ID",292                "phone": "Phone",293                "address": "Address",294                "self": "Self",295                "created_at": "Created At",296            }297        )298        st.dataframe(display_df, use_container_width=True, hide_index=True)299 300        st.divider()301        st.markdown("**Customer Details**")302 303        customer_labels = {304            f"{c.name} ({c.email})": c.customer_id for c in customers305        }306        selected_label = st.selectbox(307            "Select a customer to view full details",308            options=list(customer_labels.keys()),309        )310 311        if selected_label:312            customer_id = customer_labels[selected_label]313            try:314                customer = api.get_customer(customer_id)315                col1, col2 = st.columns(2)316                with col1:317                    st.markdown(f"**Name:** {customer.name}")318                    st.markdown(f"**Email ID:** {customer.email}")319                    st.markdown(f"**Phone:** {customer.phone}")320                    st.markdown(f"**Self:** {customer.self_info}")321                with col2:322                    st.markdown(f"**Customer ID:** `{customer.customer_id}`")323                    st.markdown(f"**Created At:** {customer.created_at}")324                    if customer.updated_at:325                        st.markdown(f"**Updated At:** {customer.updated_at}")326                    st.markdown(f"**Address:** {customer.address}")327 328                if st.button("Delete Customer", type="secondary"):329                    try:330                        api.delete_customer(customer.customer_id)331                        st.success(f"Customer '{customer.name}' deleted.")332                        st.rerun()333                    except CustomerManagementError as exc:334                        show_general_error(exc)335 336            except CustomerNotFoundError as exc:337                show_general_error(exc)338 339    except CustomerStorageError as exc:340        show_general_error(exc)341    except Exception as exc:342        show_general_error(exc)343 344 345def main() -> None:346    st.title("Customer Management System")347    st.markdown(348        "Manage customer records — add, update, and view customers."349    )350 351    try:352        api = get_api()353    except CustomerStorageError as exc:354        st.error(f"Failed to initialize the system: {exc.message}")355        return356 357    if "page" not in st.session_state:358        st.session_state.page = "view"359 360    col1, col2, col3 = st.columns(3)361    with col1:362        if st.button("New Customer", use_container_width=True, type="primary"):363            st.session_state.page = "new"364            st.rerun()365    with col2:366        if st.button("Update Customer", use_container_width=True):367            st.session_state.page = "update"368            st.rerun()369    with col3:370        if st.button("View Customers", use_container_width=True):371            st.session_state.page = "view"372            st.rerun()373 374    st.divider()375 376    if st.session_state.page == "new":377        render_new_customer_form(api)378    elif st.session_state.page == "update":379        render_update_customer_form(api)380    else:381        render_view_customers(api)382 383 384if __name__ == "__main__":385    main()386 387