path: root/fatcat_scholar/schema.py

"""
Originally wrote these as dataclasses using pydantic.dataclasses, but we don't
get serialization for free with those. This is useful for things like
auto-conversion of datetime objects.
"""

import re
import datetime
from enum import Enum
from typing import Optional, List, Any

import ftfy
from pydantic import BaseModel
from bs4 import BeautifulSoup

from fatcat_openapi_client import ReleaseEntity, ReleaseContrib
from fatcat_scholar.api_entities import entity_to_dict


class DocType(str, Enum):
    work = "work"
    sim_page = "sim_page"


class IntermediateBundle(BaseModel):
    doc_type: DocType
    releases: List[ReleaseEntity]
    biblio_release_ident: Optional[str]
    grobid_fulltext: Optional[Any]
    pdftotext_fulltext: Optional[Any]
    sim_fulltext: Optional[Any]

    class Config:
        arbitrary_types_allowed = True
        json_encoders = {
            ReleaseEntity: lambda re: entity_to_dict(re),
        }


class AccessType(str, Enum):
    ia_sim = "ia_sim"
    ia_file = "ia_file"
    wayback = "wayback"
    web = "web"
    repository = "repository"
    paywall = "paywall"
    loginwall = "loginwall"
    shadow = "shadow"


class ScholarBiblio(BaseModel):
    release_ident: Optional[str]
    title: Optional[str]
    subtitle: Optional[str]
    original_title: Optional[str]
    release_date: Optional[datetime.date]
    release_year: Optional[int]
    release_type: Optional[str]
    release_stage: Optional[str]
    withdrawn_status: Optional[str]
    lang_code: Optional[str]
    country_code: Optional[str]
    volume: Optional[str]
    volume_int: Optional[str]  # TODO: needed?
    issue: Optional[str]
    issue_int: Optional[str]  # TODO: needed?
    pages: Optional[str]
    first_page: Optional[str]
    first_page_int: Optional[str]  # TODO: needed?
    number: Optional[str]

    doi: Optional[str]
    doi_prefix: Optional[str]
    doi_registrar: Optional[str]
    pmid: Optional[str]
    pmcid: Optional[str]
    isbn13: Optional[str]
    wikidata_qid: Optional[str]
    arxiv_id: Optional[str]
    jstor_id: Optional[str]
    mag_id: Optional[str]

    license_slug: Optional[str]
    publisher: Optional[str]
    publisher_type: Optional[str]
    container_name: Optional[str]
    container_original_name: Optional[str]
    container_ident: Optional[str]
    container_issnl: Optional[str]
    container_wikidata_qid: Optional[str]
    issns: List[str]
    container_type: Optional[str]
    contrib_count: Optional[int]
    contrib_names: List[str]
    affiliations: List[str]


class ScholarFulltext(BaseModel):
    lang_code: Optional[str]
    body: str
    acknowledgement: Optional[str]
    annex: Optional[str]
    release_ident: Optional[str]
    file_ident: Optional[str]
    file_sha1: Optional[str]
    file_mimetype: Optional[str]
    thumbnail_url: Optional[str]
    access_url: Optional[str]
    access_type: Optional[AccessType]


class ScholarRelease(BaseModel):
    ident: Optional[str]
    revision: Optional[str]
    title: str
    release_date: Optional[datetime.date]
    release_year: Optional[int]
    release_type: Optional[str]
    release_stage: Optional[str]
    withdrawn_status: Optional[str]

    doi: Optional[str]
    doi_prefix: Optional[str]
    doi_registrar: Optional[str]
    pmid: Optional[str]
    pmcid: Optional[str]
    isbn13: Optional[str]
    wikidata_qid: Optional[str]
    arxiv_id: Optional[str]
    jstor_id: Optional[str]
    mag_id: Optional[str]

    license_slug: Optional[str]
    container_name: Optional[str]
    container_ident: Optional[str]
    container_issnl: Optional[str]
    container_type: Optional[str]


class ScholarSim(BaseModel):
    issue_item: str
    pub_collection: str
    sim_pubid: str
    first_page: Optional[str]


class ScholarAbstract(BaseModel):
    body: str
    lang_code: Optional[str]


class ScholarAccess(BaseModel):
    access_type: AccessType
    access_url: str
    mimetype: Optional[str]
    file_ident: Optional[str]
    release_ident: Optional[str]


class ScholarDoc(BaseModel):
    key: str
    doc_type: str  # enum: work or page
    doc_index_ts: datetime.datetime
    work_ident: Optional[str]
    tags: List[str] = []

    biblio: ScholarBiblio
    fulltext: Optional[ScholarFulltext]
    ia_sim: Optional[ScholarSim]
    abstracts: List[ScholarAbstract]
    releases: List[ScholarRelease]
    access: List[ScholarAccess]


def doi_split_prefix(doi: str) -> str:
    return doi.split("/")[0]


def release_doi_registrar(release: ReleaseEntity) -> Optional[str]:
    if not release.ext_ids.doi or not release.extra:
        return None
    for registrar in ("crossref", "datacite", "jalc"):
        if registrar in release.extra:
            return registrar
    # TODO: should we default to Crossref?
    return None


UNWANTED_ABSTRACT_PREFIXES = [
    # roughly sort this long to short
    "Abstract No Abstract ",
    "Publisher Summary ",
    "Abstract ",
    "ABSTRACT ",
    "Summary ",
    "Background: ",
    "Background ",
]


def scrub_text(raw: str, mimetype: str = None) -> str:
    """
    This function takes a mimetype-hinted string and tries to reduce it to a
    simple token-and-punctuation scheme with any and all markup removed. Eg,
    HTML tags, JATS XML tags, LaTeX, whatever.

    The output should be clean and "HTML safe" (though should still be escaped
    in HTML to get entity encoding correct).

    TODO: not using mimetype hint for latex yet
    """
    text = ftfy.fix_text(raw)

    # remove HTML
    text = BeautifulSoup(text, "html.parser").get_text()

    # TODO: for performance, compile these as globals?
    # Three regexes below adapted from Blendle cleaner.py
    # https://github.com/blendle/research-summarization/blob/master/enrichers/cleaner.py#L29
    text = re.sub(r"…", "...", text)
    text = re.sub(r"[`‘’‛⸂⸃⸌⸍⸜⸝]", "'", text)
    text = re.sub(r"[„“]|(\'\')|(,,)", '"', text)
    text = re.sub(r"\s+", " ", text).strip()

    # hack to remove abstract prefixes
    for prefix in UNWANTED_ABSTRACT_PREFIXES:
        if text.startswith(prefix):
            text = text[len(prefix) :]
            break

    assert text, "Empty abstract"
    return text


def contrib_name(contrib: ReleaseContrib) -> str:
    # TODO: support more cultural normals for name presentation
    if contrib.raw_name:
        return contrib.raw_name
    elif contrib.given_name and contrib.surname:
        return f"{contrib.given_name} {contrib.surname}"
    elif contrib.surname:
        return contrib.surname
    else:
        return contrib.given_name


def contrib_affiliation(contrib: ReleaseContrib) -> Optional[str]:
    # TODO
    return None


def es_abstracts_from_release(release: ReleaseEntity) -> List[ScholarAbstract]:

    d = dict()
    for abst in release.abstracts:
        if abst.lang not in d:
            d[abst.lang] = ScholarAbstract(
                lang_code=abst.lang, body=scrub_text(abst.content)
            )
    return list(d.values())


def es_biblio_from_release(release: ReleaseEntity) -> ScholarBiblio:

    if release.container:
        publisher = release.publisher
        container_name = release.container.name
        container_original_name = (
            release.container.extra and release.container.extra.get("original_name")
        )
        container_ident = release.container.ident
        container_type = release.container.container_type
        container_issnl = release.container.issnl
        issns = [
            container_issnl,
        ]
        if release.extra.get("issne"):
            issns.append(release.extra["issne"])
        if release.extra.get("issnp"):
            issns.append(release.extra["issnp"])
        issns = list(set(issns))
    else:
        publisher = release.extra.get("publisher")
        container_name = release.extra.get("container_name")
        container_original_name = None
        container_ident = None
        container_type = None
        container_issnl = None
        issns = []

    first_page: Optional[str] = None
    if release.pages:
        first_page = release.pages.split("-")[0]
    first_page_int: Optional[int] = None
    if first_page and first_page.isdigit():
        first_page_int = int(first_page)

    ret = ScholarBiblio(
        release_ident=release.ident,
        title=release.title,
        subtitle=release.subtitle,
        original_title=release.original_title,
        release_date=release.release_date,
        release_year=release.release_year,
        release_type=release.release_type,
        release_stage=release.release_stage,
        withdrawn_status=release.withdrawn_status,
        lang_code=release.language,
        country_code=release.extra and release.extra.get("country"),
        volume=release.volume,
        volume_int=None,
        issue=release.issue,
        issue_int=None,
        pages=release.pages,
        first_page=first_page,
        first_page_int=None,
        number=release.number,
        doi=release.ext_ids.doi,
        doi_prefix=release.ext_ids.doi and doi_split_prefix(release.ext_ids.doi),
        doi_registrar=release_doi_registrar(release),
        pmid=release.ext_ids.pmid,
        pmcid=release.ext_ids.pmcid,
        isbn13=release.ext_ids.isbn13,
        wikidata_qid=release.ext_ids.wikidata_qid,
        arxiv_id=release.ext_ids.arxiv,
        jstor_id=release.ext_ids.jstor,
        mag_id=release.ext_ids.mag,
        license_slug=release.license_slug,
        publisher=publisher,
        container_name=container_name,
        container_original_name=container_original_name,
        container_ident=container_ident,
        container_type=container_type,
        container_issnl=container_issnl,
        issns=issns,
        # TODO; these filters sort of meh. refactor to be above?
        contrib_names=list(
            filter(lambda x: bool(x), [contrib_name(c) for c in release.contribs])
        ),
        contrib_count=len([c for c in release.contribs if c.index]),
        affiliations=list(
            filter(
                lambda x: bool(x),
                [contrib_affiliation(c) for c in release.contribs if c.index],
            )
        ),
    )
    return ret


def es_release_from_release(release: ReleaseEntity) -> ScholarRelease:

    if release.container:
        container_name = release.container.name
        container_ident = release.container.ident
        container_issnl = release.container.issnl
        container_type = release.container.container_type
    else:
        container_name = release.extra.get("container_name")
        container_ident = None
        container_issnl = None
        container_type = None

    ret = ScholarRelease(
        ident=release.ident,
        revision=release.revision,
        title=release.title,
        release_date=release.release_date,
        release_year=release.release_year,
        release_type=release.release_type,
        release_stage=release.release_stage,
        withdrawn_status=release.withdrawn_status,
        doi=release.ext_ids.doi,
        doi_prefix=release.ext_ids.doi and doi_split_prefix(release.ext_ids.doi),
        doi_registrar=release_doi_registrar(release),
        pmid=release.ext_ids.pmid,
        pmcid=release.ext_ids.pmcid,
        isbn13=release.ext_ids.isbn13,
        wikidata_qid=release.ext_ids.wikidata_qid,
        arxiv_id=release.ext_ids.arxiv,
        jstor_id=release.ext_ids.jstor,
        mag_id=release.ext_ids.mag,
        license_slug=release.license_slug,
        container_name=container_name,
        container_ident=container_ident,
        container_issnl=container_issnl,
        container_type=container_type,
    )
    return ret