path: root/fuzzycat/matching.py

import os
import re
import sys
from typing import List, Optional, Type, Union

import elasticsearch
import elasticsearch_dsl
import fatcat_openapi_client
import requests
from fatcat_openapi_client import ContainerEntity, DefaultApi, ReleaseEntity
from fatcat_openapi_client.rest import ApiException

from fuzzycat.config import settings
from fuzzycat.entities import entity_from_dict, entity_from_json
from fuzzycat.utils import es_compat_hits_total

FATCAT_API_URL = settings.get("FATCAT_API_URL", "https://api.fatcat.wiki/v0")


def match_release_fuzzy(
    release: ReleaseEntity,
    size: int = 5,
    es: Optional[Union[str, Type[elasticsearch.client.Elasticsearch]]] = None,
    api: DefaultApi = None,
) -> List[ReleaseEntity]:
    """
    Given a release entity, return a number similar release entities from
    fatcat using Elasticsearch.

    TODO: rename "es" parameter to "es_client", which would be clearer
    """
    assert isinstance(release, ReleaseEntity)

    if size is None or size == 0:
        size = 10000  # or any large number

    if isinstance(es, str):
        es = elasticsearch.Elasticsearch([es])
    if es is None:
        es = elasticsearch.Elasticsearch()
    if api is None:
        api = public_api(FATCAT_API_URL)

    # Try to match by external identifier.
    # TODO: use api, ability to disable; benchmark
    ext_ids = release.ext_ids
    attrs = (
        "doi",
        "wikidata_qid",
        "isbn13",
        "pmid",
        "pmcid",
        "core",
        "arxiv",
        "jstor",
        "ark",
        "mag",
        "doaj",
        "dblp",
        "oai",
    )
    for attr in attrs:
        value = getattr(ext_ids, attr)
        if not value:
            continue
        try:
            r = api.lookup_release(**{attr: value})
        except fatcat_openapi_client.rest.ApiException as err:
            if err.status in [404, 400]:
                r = None
            else:
                raise err
        if r:
            return [r]

    if release.title is not None and release.contribs is not None:
        names = " ".join([c.raw_name for c in release.contribs])
        body = {
            "track_total_hits": True,
            "query": {
                "bool": {
                    "must": [
                        {
                            "match": {
                                "title": {
                                    "query": release.title,
                                    "operator": "AND",
                                    "fuzziness": "AUTO",
                                },
                            }
                        },
                        {
                            "match": {
                                "contrib_names": {
                                    "query": names,
                                    "operator": "AND",
                                    "fuzziness": "AUTO",
                                }
                            }
                        },
                    ],
                },
            },
            "size": size,
        }
        resp = es.search(body=body, index="fatcat_release")
        if es_compat_hits_total(resp) > 0:
            return response_to_entity_list(resp, entity_type=ReleaseEntity, size=size, api=api)

        body = {
            "track_total_hits": True,
            "query": {
                "bool": {
                    "should": [
                        {
                            "match": {
                                "title": {
                                    "query": release.title,
                                    "operator": "AND",
                                    "fuzziness": "AUTO",
                                },
                            }
                        },
                        {
                            "match": {
                                "contrib_names": {
                                    "query": names,
                                    "operator": "AND",
                                    "fuzziness": "AUTO",
                                }
                            }
                        },
                    ],
                },
            },
            "size": size,
        }
        resp = es.search(body=body, index="fatcat_release")
        if es_compat_hits_total(resp) > 0:
            return response_to_entity_list(resp, entity_type=ReleaseEntity, size=size, api=api)

    # Note: If the title is short, we will get lots of results here; do we need
    # to check for title length or result set length length or result set
    # length here?
    body = {
        "track_total_hits": True,
        "query": {
            "match": {
                "title": {
                    "query": release.title,
                    "operator": "AND",
                }
            }
        },
        "size": size,
    }
    resp = es.search(body=body, index="fatcat_release")
    if es_compat_hits_total(resp) > 0:
        return response_to_entity_list(resp, entity_type=ReleaseEntity, size=size, api=api)

    # Get fuzzy.
    # https://www.elastic.co/guide/en/elasticsearch/reference/current/common-options.html#fuzziness
    body = {
        "track_total_hits": True,
        "query": {
            "match": {
                "title": {
                    "query": release.title,
                    "operator": "AND",
                    "fuzziness": "AUTO",
                }
            }
        },
        "size": size,
    }
    resp = es.search(body=body, index="fatcat_release")
    if es_compat_hits_total(resp) > 0:
        return response_to_entity_list(resp, entity_type=ReleaseEntity, size=size, api=api)

    # TODO: perform more queries on other fields.
    return []


def public_api(host_uri):
    """
    Note: unlike the authenticated variant, this helper might get called even
    if the API isn't going to be used, so it's important that it doesn't try to
    actually connect to the API host or something.
    """
    conf = fatcat_openapi_client.Configuration()
    conf.host = host_uri
    return fatcat_openapi_client.DefaultApi(fatcat_openapi_client.ApiClient(conf))


def retrieve_entity_list(
    ids: List[str],
    api: DefaultApi = None,
    entity_type: Union[Type[ReleaseEntity], Type[ContainerEntity]] = ReleaseEntity,
) -> List[Union[Type[ReleaseEntity], Type[ContainerEntity]]]:
    """
    Retrieve a list of entities. Some entities might be missing. Return all
    that are accessible.
    """
    if api is None:
        api = public_api(FATCAT_API_URL)
    result = []
    if entity_type == ReleaseEntity:
        for id in ids:
            try:
                re = api.get_release(id, hide="refs,abstracts", expand="container,contribs,files")
                result.append(re)
            except ApiException as exc:
                if exc.status == 404:
                    print("[err] failed to retrieve release entity: {}".format(id), file=sys.stderr)
                else:
                    print("[err] api failed with {}: {}".format(exc.status, exc.message),
                          file=sys.stderr)
    elif entity_type == ContainerEntity:
        for id in ids:
            try:
                re = api.get_container(id)
                result.append(re)
            except ApiException as exc:
                if exc.status == 404:
                    print("[err] failed to retrieve container entity: {}".format(id),
                          file=sys.stderr)
                else:
                    print("[err] api failed with {}: {}".format(exc.status, exc.message),
                          file=sys.stderr)
    else:
        raise ValueError("[err] cannot retrieve ids {} of type {}".format(ids, entity_type))

    return result


def response_to_entity_list(response, size=5, entity_type=ReleaseEntity, api: DefaultApi = None):
    """
    Convert an elasticsearch result to a list of entities. Accepts both a
    dictionary and an elasticsearch_dsl.response.Response.

    We take the ids from elasticsearch and retrieve entities via API.
    """
    if isinstance(response, dict):
        ids = [hit["_source"]["ident"] for hit in response["hits"]["hits"]][:size]
        return retrieve_entity_list(ids, entity_type=entity_type, api=api)
    elif isinstance(response, elasticsearch_dsl.response.Response):
        ids = [hit.to_dict().get("ident") for hit in response]
        return retrieve_entity_list(ids, entity_type=entity_type, api=api)
    else:
        raise ValueError("cannot convert {}".format(response))


def anything_to_entity(
    s: str,
    entity_type: Union[Type[ContainerEntity], Type[ReleaseEntity]],
    api_url: str = "https://api.fatcat.wiki/v0",
    es_url: str = "https://search.fatcat.wiki",
) -> Union[ContainerEntity, ReleaseEntity]:
    """
    Convert a string to a given entity type. This function may go out to the
    fatcat API or elasticsearch and hence is expensive.
    """
    names = {
        ContainerEntity: "container",
        ReleaseEntity: "release",
    }
    if not entity_type in names:
        raise ValueError("cannot convert {}, only: {}".format(entity_type, names.keys()))
    entity_name = names[entity_type]

    if s is None:
        raise ValueError("no entity found")

    if os.path.exists(s):
        with open(s) as f:
            return entity_from_json(f.read(), entity_type)

    match = re.search("/?([a-z0-9]{26})$", s)
    if match:
        url = "{}/{}/{}".format(api_url, entity_name, match.group(1))
        resp = requests.get(url)
        if resp.status_code == 200:
            return entity_from_json(resp.text, entity_type)
        if resp.status_code == 404:
            raise ValueError("entity not found: {}".format(url))

    if re.match("[0-9]{4}(-)?[0-9]{3,3}[0-9xx]", s):
        # TODO: make index name configurable
        url = "{}/fatcat_{}/_search?track_total_hits=true&q=issns:{}".format(es_url, entity_name, s)
        doc = requests.get(url).json()
        if es_compat_hits_total(resp) == 1:
            ident = doc["hits"]["hits"][0]["_source"]["ident"]
            url = "{}/{}/{}".format(api_url, entity_name, ident)
            return entity_from_json(requests.get(url).text, entity_type)

    if entity_name == "container":
        return entity_from_dict({"name": s}, entity_type)
    elif entity_name == "release":
        return entity_from_dict({"title": s, "ext_ids": {}}, entity_type)
    else:
        raise ValueError("unhandled entity type: {}".format(entity_type))