Source code for inatcog.search

"""Module to search iNat site."""

from dronefly.core.formatters.constants import WWW_BASE_URL
from dronefly.core.formatters.generic import format_taxon_name, format_user_name
from pyinaturalist.models import Place, Project, Taxon, User

from .utils import get_home


[docs]def get_place(result):
    """Get place result."""
    place = Place.from_json(result.get("record"))
    return f":round_pushpin: [{place.display_name}]({place.url})"


[docs]def get_project(result):
    """Get project result."""
    project = Project.from_json(result.get("record"))
    return f":briefcase: [{project.title}]({WWW_BASE_URL}/projects/{project.id})"


[docs]def get_user(result):
    """Get user result."""
    user = User.from_json(result.get("record"))
    return f":bust_in_silhouette: {format_user_name(user)}"


[docs]def get_taxon(result):
    """Get taxon result (v1/search)."""
    taxon = Taxon.from_json(result.get("record"))
    return (
        f":green_circle: [{format_taxon_name(taxon, with_term=True)}]"
        f"({WWW_BASE_URL}/taxa/{taxon.id})"
    )


[docs]def get_taxon2(result):
    """Get taxon result (/v1/taxa)."""
    taxon = Taxon.from_json(result)
    return (
        f":green_circle: [{format_taxon_name(taxon, with_term=True)}]"
        f"({WWW_BASE_URL}/taxa/{taxon.id})"
    )


# pylint: disable=invalid-name
get_result_type = {
    "Place": get_place,
    "Project": get_project,
    "User": get_user,
    "Taxon": get_taxon,
    "Inactive": get_taxon2,
}


[docs]def get_result(result, result_type: str = None):
    """Get fields for any result type."""
    res_type = result_type or result.get("type")
    return get_result_type[res_type](result)


# pylint: disable=too-few-public-methods
[docs]class INatSiteSearch:
    """Lookup helper for site search."""

    def __init__(self, cog):
        self.cog = cog

[docs]    async def search(self, ctx, query, **kwargs):
        """Search iNat site."""

        # Through experimentation on May 25, 2020, I've determined a smaller
        # number than 500 will usually be returned:
        # - for /v1/taxa it will be 500
        # - for /v1/search?source=x (for any single source) it will be 100
        # - for /v1/search without a source it will be 30
        # If more than the maximum per_page is specified, however, it defaults
        # back to 30, so we try to intelligently adjust the per_page in our
        # request.
        result_type = "Inactive" if "is_active" in kwargs else None
        if result_type == "Inactive":
            per_page = 500
        elif "sources" in kwargs:
            per_page = 100
        else:
            per_page = 30
        home = await get_home(ctx)
        api_kwargs = {"q": query, "per_page": per_page, "preferred_place_id": home}
        api_kwargs.update(kwargs)
        search_results = await self.cog.api.get_search_results(**api_kwargs)
        results = [
            get_result(result, result_type) for result in search_results["results"]
        ]
        # But we return the actual per_page so the pager can accurately report
        # how many results were not shown.
        return (results, search_results["total_results"], search_results["per_page"])