D-Net

from pydantic import BaseModel, Schema

from typing import Dict, List

from datetime import datetime

from fastapi import HTTPException

import logging

from time import time

rels = dict(issupplementto="IsSupplementTo", issupplementedby="IsSupplementedBy", references="References",

            isreferencedby="IsReferencedBy")

class LinkProvider(BaseModel):

    name:str = Schema(None, title= "The name of the Provider that provides the links", max_length= 300)

    totalRelationships:int=Schema(None, title= "The number of links that It provides")

class LinkPublisher(BaseModel):

    name:str = Schema(None, title= "The name of the Publisher that provides the links", max_length= 300)

    totalRelationships:int=Schema(None, title= "The number of links that It provides")

class IdentifierType(BaseModel):

    ID: str = None

    IDScheme:str = None

    IDURL:str= None

class ScholixProviderType(BaseModel):

    name:str = None

    identifier:List[IdentifierType] = []

class RelationshipType(BaseModel):

    Name:str

    SubType:str = None

    SubTypeSchema:str = None

class CreatorType(BaseModel):

    Name: str

    Identifier:List[IdentifierType] = []

class ScholixItemType(BaseModel):

    Identifier:IdentifierType = None

    Title:str = None

    Type:str 

    Creator:List[CreatorType] = []

    PublicationDate:str = None

    Publisher:List[ScholixProviderType] =[]    

class ScholixType(BaseModel):

    HarvestDate:str=None

    LicenseURL:str=None

    LinkProvider:List[ScholixProviderType] =[]

    LinkPublicationDate: str = None

    RelationshipType:RelationshipType

    source:ScholixItemType

    target:ScholixItemType

class PageResultType(BaseModel):

    currentPage:int

    totalLinks:int

    totalPages:int

    result:List[ScholixType] = []

def get_scholix_resource(item):

    title = ''

    if 'title' in item:

        title = item.title

    if title is not None and len(title):

        if title[0] == '"' and title[-1] == '"':

            title = title[1:-1]

    identifier = [dict(ID=x.identifier, IDScheme=x.schema, IDURL=x.url) for x in

                  item.identifier]

    identifier = identifier[0]

    creator = []

    if 'creator' in item and item.creator is not None:        

        creator = [dict(Name=x.name) for x in item.creator]

    publicationDate = None

    if 'publicationDate' in item:

        publicationDate = item.publicationDate

    publisher = []

    if 'publisher' in item and item.publisher is not None:

        publisher = [dict(name= x.name) for x in item.publisher if x.name is not None]

    c_type = item.objectType

    if item.objectType == 'publication':

        c_type = 'literature'

    resource = dict(Title=title, Identifier=identifier, Creator=creator, PublicationDate= publicationDate, Publisher = publisher, Type= c_type)

    return resource

def convert_response(response):

    now = datetime.now()

    log = logging.getLogger("scholexplorer")

    start = time()

    for item in response.hits:

        current_item = {'LinkPublicationDate': item.publicationDate, 'HarvestDate': item.publicationDate,

                         "LinkProvider": []}

        for linkProvider in item.linkprovider:

            current_item['LinkProvider'].append(ScholixProviderType(name=linkProvider.name,

                                                     identifier=[IdentifierType(ID=x.identifier, IDScheme=x.schema) for x in

                                                                 linkProvider.identifiers]))

        rel_sub_type = rels.get(item.relationship.name.lower(), "IsRelatedTo")

        current_item['RelationshipType'] = dict(Name=rel_sub_type, SubType=item.relationship.name,

                                                SubTypeSchema=item.relationship.schema)

        current_item['source'] = get_scholix_resource(item.source)

        current_item['target'] = get_scholix_resource(item.target)

        yield current_item

    end = time()

    log.debug("response converted in {} ms".format(end-start))

def create_response(response, current_page):

    log = logging.getLogger("scholexplorer")    

    if current_page > 9999:

        raise HTTPException(status_code=400,detail="MAX NUMBER OF PAGE REACHED")

    start = time()

    result = {'totalLinks': response.hits.total.value, 'currentPage': current_page /100,

              'totalPages': 1 + response.hits.total.value / 100, 'result': []}

    result['result'] = convert_response(response)

    end = time()

    log.debug("response created in {} ms".format(end-start))

    return result

from fastapi import FastAPI, Query,HTTPException

from pydantic import BaseModel, Schema

from typing import Dict, List

from v3.responseModel import *

from eu.dnetlib.ScholixConnector import ScholixConnector

from datetime import datetime

import time

import logging

subapi_v3 = FastAPI(title="Scholexplorer API 3.0",

    version="3.0.0",

    description="scholexplorer API version 3.0",openapi_prefix="/v3", docs_url="/ui/")

req_fields = ['sourcePid', 'targetPid', 'sourcePublisher', 'targetPublisher', 'linkProvider']

@subapi_v3.get("/LinkProvider", response_model=List[LinkProvider], tags=['LinkProvider : Operation related to the Link Provider'], description="Return a list of link provider and relative number of relations",summary="Get All Link Providers")

def linkProvider(name: str = Query(None, title="name", description="Filter the link provider by a name")):

        s = ScholixConnector()

        return s.list_datasources(name)

@subapi_v3.get("/LinkPublisher/inSource", response_model=List[LinkPublisher], 

            tags=['LinkPublisher : Operation related to the Link Publisher'], 

            description="Return a List of all Publishers that provide source objects in Scholix links and the total number of links where the source object comes from this publisher",

            summary="Get All Publishers that provide source object")

def linkPublisherInSource(name: str = Query(None, title="name", description="Filter the link publisher by a name")):

        s = ScholixConnector()                

        return s.list_publisher('source',name)

@subapi_v3.get("/LinkPublisher/inTarget", response_model=List[LinkPublisher], 

            tags=['LinkPublisher : Operation related to the Link Publisher'], 

            description="Return a List of all Publishers that provide target objects in Scholix links and the total number of links where the target object comes from this publisher",

            summary="Get All Publishers that provide target object")

def linkPublisherInTarget(name: str = Query(None, title="name", description="Filter the link publisher by a name")):

        s = ScholixConnector()

        return s.list_publisher('target',name)

@subapi_v3.get("/Links", response_model=PageResultType, tags=['Links : Operation related to the Scholix Links'], 

response_description="A list of List of scholix Links following the schema <a href='https://github.com/scholix/schema/tree/master/json/v3/schema.json'>https://github.com/scholix/schema/tree/master/json/v3/schema.json</a>",

description="""Return a List of scholix Links, this method <b>require one of the following parameters </b> (<b>sourcePid, targetPid, sourcePublisher,\

        \ targetPublisher, linkProvider</b>) all parameters can be combined""",summary="Get Scholix Links")

def links(

    linkProvider: str = Query(None, title="linkProvider", description="Filter Scholix relationships collected from a LinkProvider"),

    targetPid: str = Query(None, title="targetPid", description="Filter Scholix relationships having a target pid"),

    targetPublisher: str = Query(None, title="targetPublisher", description="Filter Scholix relationships having a target published in a Publisher named targetPublisher"),

    targetPidType: str = Query(None, title="targetPidType", description="Filter Scholix relationships having a target pid type"),

    targetType: str = Query(None, title="targetType", description="Filter Scholix relationships having a target type (literature, dataset, unknown)"),  

    sourceType: str = Query(None, title="sourceType", description="Filter Scholix relationships having a source type (literature, dataset, unknown)"),  

    sourcePid: str = Query(None, title="sourcePid", description="Filter Scholix relationships having a source pid"),

    sourcePublisher: str = Query(None, title="sourcePublisher", description="Filter Scholix relationships having a source published in a Publisher named sourcePublisher"),

    sourcePidType: str = Query(None, title="sourcePidType", description="Filter Scholix relationships having a source pid type"),

    harvestedAfter: str = Query(None, title="harvestedAfter", description="Filter scholix Links having collected after this date"),

        page:int= Query(None, title="page", description="select page of result"),

    ):

        log = logging.getLogger("scholexplorer")

        log.debug("request Link page")

        if not (sourcePid or targetPid or sourcePublisher or targetPublisher or linkProvider):

                raise HTTPException(status_code=400,detail='The method requires one of the following parameters: sourcePid, targetPid, sourcePublisher, targetPublisher, linkProvider')

        s = ScholixConnector()    

        m_page = 0

        if page:

                m_page= page*100

        return create_response(s.links(provider = linkProvider, s_pid=sourcePid, t_pid=targetPid, s_publisher=sourcePublisher,

                                           t_publisher=targetPublisher, s_pid_type=sourcePidType,

                                           t_pid_type=targetPidType, page=m_page, target_Type=targetType, source_Type=sourceType),m_page)