1
0
mirror of https://github.com/gsi-upm/senpy synced 2024-09-20 22:01:41 +00:00
senpy/sentiment-meaningCloud/meaningcloud_plugin.py
2018-07-24 17:28:32 +02:00

248 lines
9.2 KiB
Python

# -*- coding: utf-8 -*-
import time
import requests
import json
import string
import os
from os import path
import time
from senpy.plugins import SentimentPlugin
from senpy.models import Results, Entry, Entity, Topic, Sentiment, Error
from senpy.utils import check_template
class MeaningCloudPlugin(SentimentPlugin):
'''
Sentiment analysis with meaningCloud service.
To use this plugin, you need to obtain an API key from meaningCloud signing up here:
https://www.meaningcloud.com/developer/login
When you had obtained the meaningCloud API Key, you have to provide it to the plugin, using param apiKey.
Example request:
http://senpy.cluster.gsi.dit.upm.es/api/?algo=meaningCloud&language=en&apiKey=<API key>&input=I%20love%20Madrid.
'''
name = 'sentiment-meaningcloud'
author = 'GSI UPM'
version = "1.1"
maxPolarityValue = 1
minPolarityValue = -1
extra_params = {
"language": {
"aliases": ["language", "l"],
"required": True,
"options": ["en","es","ca","it","pt","fr","auto"],
"default": "auto"
},
"apikey":{
"aliases": ["apiKey", "meaningcloud-key", "meaningcloud-apikey"],
"required": True
}
}
def _polarity(self, value):
if 'NONE' in value:
polarity = 'marl:Neutral'
polarityValue = 0
elif 'N' in value:
polarity = 'marl:Negative'
polarityValue = -1
elif 'P' in value:
polarity = 'marl:Positive'
polarityValue = 1
return polarity, polarityValue
def analyse_entry(self, entry, params):
txt = entry['nif:isString']
api = 'http://api.meaningcloud.com/'
lang = params.get("language")
model = "general"
key = params["apikey"]
parameters = {
'key': key,
'model': model,
'lang': lang,
'of': 'json',
'txt': txt,
'tt': 'a'
}
try:
r = requests.post(
api + "sentiment-2.1", params=parameters, timeout=3)
parameters['lang'] = r.json()['model'].split('_')[1]
lang = parameters['lang']
r2 = requests.post(
api + "topics-2.0", params=parameters, timeout=3)
except requests.exceptions.Timeout:
raise Error("Meaning Cloud API does not response")
api_response = r.json()
api_response_topics = r2.json()
if not api_response.get('score_tag'):
raise Error(r.json())
entry['language_detected'] = lang
self.log.debug(api_response)
agg_polarity, agg_polarityValue = self._polarity(
api_response.get('score_tag', None))
agg_opinion = Sentiment(
id="Opinion0",
marl__hasPolarity=agg_polarity,
marl__polarityValue=agg_polarityValue,
marl__opinionCount=len(api_response['sentence_list']))
agg_opinion.prov(self)
entry.sentiments.append(agg_opinion)
self.log.debug(api_response['sentence_list'])
count = 1
for sentence in api_response['sentence_list']:
for nopinion in sentence['segment_list']:
self.log.debug(nopinion)
polarity, polarityValue = self._polarity(
nopinion.get('score_tag', None))
opinion = Sentiment(
id="Opinion{}".format(count),
marl__hasPolarity=polarity,
marl__polarityValue=polarityValue,
marl__aggregatesOpinion=agg_opinion.get('id'),
nif__anchorOf=nopinion.get('text', None),
nif__beginIndex=int(nopinion.get('inip', None)),
nif__endIndex=int(nopinion.get('endp', None)))
count += 1
opinion.prov(self)
entry.sentiments.append(opinion)
mapper = {'es': 'es.', 'en': '', 'ca': 'es.', 'it':'it.', 'fr':'fr.', 'pt':'pt.'}
for sent_entity in api_response_topics['entity_list']:
resource = "_".join(sent_entity.get('form', None).split())
entity = Entity(
id="Entity{}".format(sent_entity.get('id')),
itsrdf__taIdentRef="http://{}dbpedia.org/resource/{}".format(
mapper[lang], resource),
nif__anchorOf=sent_entity.get('form', None),
nif__beginIndex=int(sent_entity['variant_list'][0].get('inip', None)),
nif__endIndex=int(sent_entity['variant_list'][0].get('endp', None)))
sementity = sent_entity['sementity'].get('type', None).split(">")[-1]
entity['@type'] = "ODENTITY_{}".format(sementity)
entity.prov(self)
entry.entities.append(entity)
for topic in api_response_topics['concept_list']:
if 'semtheme_list' in topic:
for theme in topic['semtheme_list']:
concept = Topic()
concept.id = "Topic{}".format(topic.get('id'))
concept['@type'] = "ODTHEME_{}".format(theme['type'].split(">")[-1])
concept['fam:topic-reference'] = "http://dbpedia.org/resource/{}".format(theme['type'].split('>')[-1])
entry.prov(self)
entry.topics.append(concept)
yield entry
test_cases = [
{
'params': {
'algo': 'sentiment-meaningCloud',
'intype': 'direct',
'expanded-jsonld': 0,
'informat': 'text',
'prefix': '',
'plugin_type': 'analysisPlugin',
'urischeme': 'RFC5147String',
'outformat': 'json-ld',
'conversion': 'full',
'language': 'en',
'apikey': '00000',
'algorithm': 'sentiment-meaningCloud'
},
'input': 'Hello World Obama',
'expected': {
'sentiments': [
{'marl:hasPolarity': 'marl:Neutral'}],
'entities': [
{'itsrdf:taIdentRef': 'http://dbpedia.org/resource/Obama'}],
'topics': [
{'fam:topic-reference': 'http://dbpedia.org/resource/Astronomy'}]
},
'responses': [
{
'url': 'http://api.meaningcloud.com/sentiment-2.1',
'method': 'POST',
'json': {
'model': 'general_en',
'sentence_list': [{
'text':
'Hello World',
'endp':
'10',
'inip':
'0',
'segment_list': [{
'text':
'Hello World',
'segment_type':
'secondary',
'confidence':
'100',
'inip':
'0',
'agreement':
'AGREEMENT',
'endp':
'10',
'polarity_term_list': [],
'score_tag':
'NONE'
}],
'score_tag':
'NONE',
}],
'score_tag':
'NONE'
}
}, {
'url': 'http://api.meaningcloud.com/topics-2.0',
'method': 'POST',
'json': {
'entity_list': [{
'form':
'Obama',
'id':
'__1265958475430276310',
'variant_list': [{
'endp': '16',
'form': 'Obama',
'inip': '12'
}],
'sementity': {
'fiction': 'nonfiction',
'confidence': 'uncertain',
'class': 'instance',
'type': 'Top>Person'
}
}],
'concept_list': [{
'form':
'world',
'id':
'5c053cd39d',
'relevance':
'100',
'semtheme_list': [{
'id': 'ODTHEME_ASTRONOMY',
'type': 'Top>NaturalSciences>Astronomy'
}]
}],
}
}]
}
]
if __name__ == '__main__':
from senpy import easy_test
easy_test()