services.py

from typing import List, Dict, Tuple, Union, Optional, Any
import os

import gradio as gr

import requests
import backoff


class OrganizationNotFound(Exception):
    "No organization(s) found"


@backoff.on_exception(
    backoff.expo,
    (requests.exceptions.Timeout, requests.exceptions.ConnectionError),
    max_tries=5
)
def _get_json(route: str, **request_kwargs) -> Dict[str, Any] | str:
    r = requests.get(
        url=f"{os.getenv('LOI_API_URL')}/{route}",
        params=request_kwargs,
        headers={"x-api-key": os.getenv("LOI_API_KEY")},
        timeout=30
    )

    r.raise_for_status()
    return r.json().get("response")


@backoff.on_exception(
    backoff.expo,
    (requests.exceptions.Timeout, requests.exceptions.ConnectionError),
    max_tries=5
)
def _post_json(route: str, *, payload: Dict[str, Any]) -> Dict[str, Any] | str | int:
    r = requests.post(
        url=f"{os.getenv('LOI_API_URL')}/{route}",
        json=payload,
        headers={"x-api-key": os.getenv("LOI_API_KEY")},
        timeout=30
    )

    r.raise_for_status()
    return r.json().get("response")


def organization_pair_autofill(
    recipient_name: str,
    recipient_ein: str,
    funder_name: str,
    funder_ein: str
):
    recip_match = _get_json("/organization/search", name=recipient_name, ein=recipient_ein)
    if len(recip_match or []) == 0:
        # raise OrganizationNotFound()
        raise gr.Error("No matching recipient could be found")
    gr.Info(f"{recipient_name} found, auto-filling fields...")

    funder_match = _get_json("/organization/search", name=funder_name, ein=funder_ein)
    if len(funder_match or []) == 0:
        # raise OrganizationNotFound()
        raise gr.Error("No matching funder could be found")
    gr.Info(f"{funder_name} found, auto-filling fields...")

    data = _get_json(
        "/organization/autofill",
        recipient_candid_entity_id=recip_match[0]["candid_entity_id"],
        funder_candid_entity_id=funder_match[0]["candid_entity_id"],
    )

    return (
        data.get("recipient_data", {}).get("projects_text"),
        data.get("recipient_data", {}).get("capacity_text"),
        data.get("recipient_data", {}).get("contact_text"),
        data.get("recipient_data", {}).get("data_text"),  # accomplishments
        data.get("recipient_data", {}).get("mission_statement_text"),
        data.get("funder_data", {}).get("mission_statement_text"),
        data.get("funding_history_text"),
        data.get("recipient_data", {}).get("org_data"),
        recip_match[0]["candid_entity_id"],
        data.get("funder_data", {}).get("org_data"),
        funder_match[0]["candid_entity_id"],
    )


def cost_estimator(
    recipient_candid_entity_id: Union[int, str],
    recipient_data: Dict[str, Any],
    funder_data: Dict[str, Any],
    program_desc: Optional[str] = None
) -> str:
    estimate: str = _post_json(
        "budget",
        payload={
            "recipient_candid_entity_id": recipient_candid_entity_id,
            "program_description": program_desc,
            "recipient_data": recipient_data,
            "funder_data": funder_data,
        }
    )
    return estimate


def identify_vague_statements(text: str) -> Tuple[List[str], List[str], List[str]]:
    data = _get_json("/editorialai/vaguestatement", input_section=text)

    statements, alternatives, reasons = [], [], []
    for record in data:
        statements.append(record["vague_statement"])
        alternatives.append(record["alternative_text"])
        reasons.append(record["reason"])

    return statements, alternatives, reasons


def help_mission_statement(recipient_name: str, recipient_mission_info: str):
    return _get_json("/writerhelper/missionstatement", info=recipient_mission_info, org_name=recipient_name)


def draft_letter(
    recipient_name: str, funder_name: str, projects: str, amount: str,
    recipient_mission_statement: str = "", funder_mission_statement: str = "",
    project_name: str = "", project_purpose: str = "",
    prior_contact: str = "", connection: str = "",
    capacity: str = "", path_to_solution: str = "", recent_accomplishments: str = "",
    recipient_history: str = "", mutual_history: str = "",
    geo_pop_targets: str = "", project_data_stats: str = "",
    desired_objectives: str = "", major_activities: str = "", key_staff: str = "", standout_features: str = "",
    success_metric: str = "",
    other_funding_sources: str = "",
    contact_information: str = ""
):
    gr.Info("Writing the letter, please scroll to the top of the page.")
    opening: str = _post_json(
        "/writer/opening",
        payload=dict(
            funder_name=funder_name,
            recipient_name=recipient_name,
            project_name=project_name,
            project_purpose=project_purpose,
            amount=amount,
            prior_contact=prior_contact,
            connection=connection
        )
    )

    recipient_description: str = _post_json(
        "/writer/org",
            payload=dict(
            opening=opening,
            recipient_mission_statement=recipient_mission_statement,
            capacity=capacity,
            history=recipient_history,
            path=path_to_solution,
            accomplishment=recent_accomplishments
        )
    )

    need_statement: str  = _post_json(
        "/writer/need",
        payload=dict(
            recipient_desc=recipient_description,
            target=geo_pop_targets,
            data=project_data_stats,
            funder_mission_statement=funder_mission_statement
        )
    )

    project_description: str = _post_json(
        "/writer/project",
        payload=dict(
            need=need_statement,
            projects=projects,
            desired_objectives=desired_objectives,
            major_activities=major_activities,
            key_staff=key_staff,
            stand_out=standout_features,
            success=success_metric
        )
    )

    funding_request: str = _post_json(
        "/writer/fund",
        payload=dict(
            project_desc=project_description,
            amount=amount,
            funding_history=mutual_history,
            other_funding=other_funding_sources
        )
    )

    conclusion: str = _post_json(
        "/writer/conclusion",
        payload=dict(
            funding_request=funding_request,
            project_desc=project_description,
            follow_up=contact_information
        )
    )

    return (
        opening,
        recipient_description,
        need_statement,
        project_description,
        funding_request,
        conclusion
    )


def send_feedback(
    context: Dict[str, Any],
    letter: Dict[str, Any],
    relevance: int,
    coherence: int,
    fluency: int,
    overall_quality: int,
    comments: Optional[str] = None,
    email: Optional[str] = None
) -> int:
    count = _post_json(
        "feedback",
        payload={
            "context": context,
            "letter": letter,
            "relevance": relevance,
            "coherence": coherence,
            "fluency": fluency,
            "overall_quality": overall_quality,
            "comments": comments,
            "email": email
        }
    )
    return count