app.py

from typing import List, Literal, Tuple, TypedDict
import os

import gradio as gr

try:
    from common import org_search_component as oss
    from formatting import process_reasons, parse_pcs_descriptions, parse_geo_descriptions
    from services import RfpRecommend, RfpFeedback
except ImportError:
    from ..common import org_search_component as oss
    from .formatting import process_reasons, parse_pcs_descriptions, parse_geo_descriptions
    from .services import RfpRecommend, RfpFeedback

api = RfpRecommend()
reporting = RfpFeedback()

class LoggedComponents(TypedDict):
    recommendations: gr.components.Component
    ratings: List[gr.components.Component]
    correctness: gr.components.Component
    sufficiency: gr.components.Component
    comments: gr.components.Component
    email: gr.components.Component


def single_recommendation_response(
    item_number: int,
    rec_type: Literal["RFP"] = "RFP"
) -> gr.Radio:
    """Generates a radio button group to provide feedback for single recommendation indexed by `item_number`.
    Since the index values start from `0` we add `1` to indicate the ordinal value in the info text.

    Parameters
    ----------
    item_number : int
        Recommendation index starting from 0

    Returns
    -------
    gr.Radio
    """

    ordinal = str(item_number + 1)

    suffix = "th"
    if ordinal.endswith('1') and not ordinal.endswith('11'):
        suffix = "st"
    elif ordinal.endswith('2') and not ordinal.endswith('12'):
        suffix = "nd"
    elif ordinal.endswith('3') and not ordinal.endswith('13'):
        suffix = "rd"

    elem = gr.Radio(
        choices=[
            "Not relevant and not useful",
            "Relevant but not useful",
            "Relevant and useful"
        ],
        label=f"Recommendation #{ordinal}",
        info=f"Evaluate the {ordinal}{suffix} {rec_type} (if applicable)"
    )
    return elem


def recommend_invoke(recipient: gr.State):
    response = api(candid_entity_id=recipient[0])
    output = []
    for rfp in (response.get("recommendations", []) or []):
        output.append([
            rfp["funder_id"],
            rfp["funder_name"],
            rfp["funder_address"],
            rfp["amount"],
            (
                f"<a href='{rfp['application_url']}' target='_blank' rel='noopener noreferrer'>"
                f"{rfp['application_url']}</a>"
            ),
            rfp["deadline"],
            rfp["description"],
            parse_pcs_descriptions(rfp["taxonomy"]),
            parse_geo_descriptions(rfp["area_served"])
        ])

    if len(output) == 0:
        raise gr.Error("No relevant RFPs were found, please try again in the future as new RFPs become available.")

    return output, process_reasons(response.get("meta", {}) or {}), response


def build_recommender() -> Tuple[LoggedComponents, gr.Blocks]:
    with gr.Blocks(theme=gr.themes.Soft(), title="RFP recommendations") as demo:
        gr.Markdown(
            """
            <h1>RFP recommendations</h1>

            <p>Receive recommendations for funding opportunities relevant to your work.</p>
            <p>To get started lookup your nonprofit and then click <b>Get recommendations</b>.</p>
            """
        )

        with gr.Row():
            with gr.Column():
                _, selected_org_state = oss.render()
        with gr.Row():
            recommend = gr.Button("Get recommendations", scale=5, variant="primary")
        with gr.Row():
            with gr.Accordion(label="Parameters used for recommendations", open=False):
                reasons_output = gr.DataFrame(
                    col_count=3,
                    headers=["Reason category", "Reason value", "Reason description"],
                    interactive=False
                )

        rec_outputs = gr.DataFrame(
            label="Recommended RFPs",
            type="array",
            headers=[
                "Funder ID", "Name", "Address",
                "Amount", "Application URL", "Deadline",
                "Description", "About", "Where"
            ],
            col_count=(9, "fixed"),
            datatype=[
                "number", "str", "str",
                "str", "markdown", "date",
                "str", "markdown", "markdown"
            ],
            wrap=True,
            max_height=1000,
            column_widths=[
                "5%", "10%", "20%",
                "5", "15%", "5%",
                "10%", "10%", "20%"
            ],
            interactive=False
        )

        recommendations_json = gr.JSON(label="Recommended RFPs JSON", visible=False)

        # pylint: disable=no-member
        recommend.click(
            fn=recommend_invoke,
            inputs=[selected_org_state],
            outputs=[rec_outputs, reasons_output, recommendations_json]
        )

    logged = LoggedComponents(
        recommendations=recommendations_json
    )

    return logged, demo


def build_feedback(
    components: LoggedComponents,
    N: int = 5,
    rec_type: Literal["RFP"] = "RFP",
) -> gr.Blocks:

    def handle_feedback(*args):
        try:
            reporting(
                recommendation_data=args[0],
                ratings=list(args[1: (N + 1)]),
                info_is_correct=args[N + 1],
                info_is_sufficient=args[N + 2],
                comments=args[N + 3],
                email=args[N + 4]
            )
            gr.Info("Thank you for providing feedback!")
        except Exception as ex:
            if hasattr(ex, "response"):
                error_msg = ex.response.json().get("response", {}).get("error")
                raise gr.Error(f"Failed to submit feedback: {error_msg}")
            raise gr.Error("Failed to submit feedback")

    feedback_components = []
    with gr.Blocks(theme=gr.themes.Soft(), title="Candid AI demo") as demo:
        gr.Markdown("""
            <h1>Help us improve this tool with your valuable feedback</h1>

            Please provide feedback for the recommendations on the previous tab.

            It is not required to provide feedback on all recommendations before submitting.
            """
        )

        with gr.Row():
            with gr.Column():
                with gr.Group():
                    for i in range(N):
                        f = single_recommendation_response(i, rec_type=rec_type)
                        feedback_components.append(f)
                        if "ratings" not in components:
                            components["ratings"] = [f]
                        else:
                            components["ratings"].append(f)

                correctness = gr.Radio(
                    choices=["True", "False"],
                    label="Information is correct?",
                    info="Are the displayed RFP details correct?"
                )
                sufficiency = gr.Radio(
                    choices=["True", "False"],
                    label="Sufficient data?",
                    info="Is enough RFP data available to provide meaningful recommendations?"
                )

                comment = gr.Textbox(label="Additional comments (optional)", lines=4)
                email = gr.Textbox(label="Your email (optional)", lines=1)

                components["correctness"] = correctness
                components["sufficiency"] = sufficiency
                components["comments"] = comment
                components["email"] = email

                with gr.Row():
                    submit = gr.Button("Submit Feedback", variant='primary', scale=5)
                    gr.ClearButton(components=feedback_components, variant="stop")

        # pylint: disable=no-member
        submit.click(
            fn=handle_feedback,
            inputs=[comp for k, cl in components.items() for comp in (cl if isinstance(cl, list) else [cl])],
            outputs=None,
            show_api=False,
            api_name=False,
            preprocess=False,
        )
    return demo



def build_demo():
    logger, recommender = build_recommender()
    feedback = build_feedback(logger)
    return gr.TabbedInterface(
        interface_list=[recommender, feedback],
        tab_names=["RFP recommendations", "Feedback"],
        title="Candid's RFP recommendation engine",
        theme=gr.themes.Soft()
    )


if __name__ == '__main__':
    app = build_demo()
    app.queue(max_size=5).launch(
        show_api=False,
        auth=[
            (os.getenv("APP_USERNAME"), os.getenv("APP_PASSWORD")),
            (os.getenv("APP_PUBLIC_USERNAME"), os.getenv("APP_PUBLIC_PASSWORD")),
        ],
        auth_message="Login to Candid's letter of intent demo",
        ssr_mode=False
    )