Merge pull request 'broker_feedbacks' (#16) from broker_feedbacks into master
Reviewed-on: #16
This commit is contained in:
commit
e64da1e38d
|
|
@ -2,6 +2,8 @@ package eu.dnetlib.broker.oa.controllers;
|
|||
|
||||
import java.io.Serializable;
|
||||
|
||||
import eu.dnetlib.broker.common.feedbacks.FeedbackStatus;
|
||||
|
||||
public class EventFeedback implements Serializable {
|
||||
|
||||
/**
|
||||
|
|
@ -10,27 +12,22 @@ public class EventFeedback implements Serializable {
|
|||
private static final long serialVersionUID = -6967719685282712195L;
|
||||
|
||||
private String eventId;
|
||||
private String status;
|
||||
|
||||
// TOOD status should be an enum having the following values:
|
||||
// * DISCARDED: the event was not processable by the system. OpenAIRE should not interpret such status in a negative or positive sense
|
||||
// with regard to the accuracy of the notification
|
||||
// * REJECTED: a human takes the decision to reject the suggestion as it was wrong
|
||||
// * ACCEPTED: a human takes the decision to apply the suggested enrichment to the local record
|
||||
private FeedbackStatus status;
|
||||
|
||||
protected String getEventId() {
|
||||
public String getEventId() {
|
||||
return eventId;
|
||||
}
|
||||
|
||||
protected void setEventId(final String eventId) {
|
||||
public void setEventId(final String eventId) {
|
||||
this.eventId = eventId;
|
||||
}
|
||||
|
||||
protected String getStatus() {
|
||||
public FeedbackStatus getStatus() {
|
||||
return status;
|
||||
}
|
||||
|
||||
protected void setStatus(final String status) {
|
||||
public void setStatus(final FeedbackStatus status) {
|
||||
this.status = status;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -57,6 +57,8 @@ import eu.dnetlib.broker.api.ShortEventMessage;
|
|||
import eu.dnetlib.broker.common.elasticsearch.EventRepository;
|
||||
import eu.dnetlib.broker.common.elasticsearch.Notification;
|
||||
import eu.dnetlib.broker.common.elasticsearch.NotificationRepository;
|
||||
import eu.dnetlib.broker.common.feedbacks.DbEventFeedback;
|
||||
import eu.dnetlib.broker.common.feedbacks.DbEventFeedbackRepository;
|
||||
import eu.dnetlib.broker.common.properties.ElasticSearchProperties;
|
||||
import eu.dnetlib.broker.common.stats.OpenaireDsStatRepository;
|
||||
import eu.dnetlib.broker.common.subscriptions.Subscription;
|
||||
|
|
@ -87,6 +89,9 @@ public class OpenairePublicController extends AbstractDnetController {
|
|||
@Autowired
|
||||
private OpenaireDsStatRepository openaireDsStatRepository;
|
||||
|
||||
@Autowired
|
||||
private DbEventFeedbackRepository feedbackRepository;
|
||||
|
||||
@Autowired
|
||||
private ElasticSearchProperties props;
|
||||
|
||||
|
|
@ -103,31 +108,27 @@ public class OpenairePublicController extends AbstractDnetController {
|
|||
|
||||
final Optional<Subscription> optSub = subscriptionRepo.findById(subscrId);
|
||||
|
||||
if (optSub.isPresent()) {
|
||||
if (!optSub.isPresent()) {
|
||||
log.warn("Invalid subscription: " + subscrId);
|
||||
return new ScrollPage<>(null, true, new ArrayList<>());
|
||||
}
|
||||
final ElasticsearchRestTemplate esTemplate = (ElasticsearchRestTemplate) esOperations;
|
||||
|
||||
final ElasticsearchRestTemplate esTemplate = (ElasticsearchRestTemplate) esOperations;
|
||||
|
||||
final NativeSearchQuery searchQuery = new NativeSearchQueryBuilder()
|
||||
final NativeSearchQuery searchQuery = new NativeSearchQueryBuilder()
|
||||
.withQuery(QueryBuilders.termQuery("subscriptionId.keyword", subscrId))
|
||||
.withSearchType(SearchType.DEFAULT)
|
||||
.withFields("topic", "payload", "eventId")
|
||||
.withPageable(PageRequest.of(0, 100))
|
||||
.build();
|
||||
|
||||
final SearchScrollHits<Notification> scroll =
|
||||
final SearchScrollHits<Notification> scroll =
|
||||
esTemplate.searchScrollStart(SCROLL_TIMEOUT_IN_MILLIS, searchQuery, Notification.class, IndexCoordinates.of(props.getNotificationsIndexName()));
|
||||
if (scroll.hasSearchHits()) {
|
||||
final List<ShortEventMessage> values = calculateNotificationMessages(scroll);
|
||||
return new ScrollPage<>(scroll.getScrollId(), values.isEmpty() || scroll.getScrollId() == null, values);
|
||||
} else {
|
||||
esTemplate.searchScrollClear(Arrays.asList(scroll.getScrollId()));
|
||||
return new ScrollPage<>(null, true, new ArrayList<>());
|
||||
}
|
||||
|
||||
} else {
|
||||
log.warn("Invalid subscription: " + subscrId);
|
||||
return new ScrollPage<>(null, true, new ArrayList<>());
|
||||
if (scroll.hasSearchHits()) {
|
||||
final List<ShortEventMessage> values = calculateNotificationMessages(scroll);
|
||||
return new ScrollPage<>(scroll.getScrollId(), values.isEmpty() || scroll.getScrollId() == null, values);
|
||||
}
|
||||
esTemplate.searchScrollClear(Arrays.asList(scroll.getScrollId()));
|
||||
return new ScrollPage<>(null, true, new ArrayList<>());
|
||||
}
|
||||
|
||||
@Operation(summary = "Returns notifications using scrolls (other pages)")
|
||||
|
|
@ -137,14 +138,13 @@ public class OpenairePublicController extends AbstractDnetController {
|
|||
final ElasticsearchRestTemplate esTemplate = (ElasticsearchRestTemplate) esOperations;
|
||||
|
||||
final SearchScrollHits<Notification> scroll =
|
||||
esTemplate.searchScrollContinue(scrollId, SCROLL_TIMEOUT_IN_MILLIS, Notification.class, IndexCoordinates.of(props.getNotificationsIndexName()));
|
||||
esTemplate.searchScrollContinue(scrollId, SCROLL_TIMEOUT_IN_MILLIS, Notification.class, IndexCoordinates.of(props.getNotificationsIndexName()));
|
||||
if (scroll.hasSearchHits()) {
|
||||
final List<ShortEventMessage> values = calculateNotificationMessages(scroll);
|
||||
return new ScrollPage<>(scroll.getScrollId(), values.isEmpty() || scroll.getScrollId() == null, values);
|
||||
} else {
|
||||
esTemplate.searchScrollClear(Arrays.asList(scroll.getScrollId()));
|
||||
return new ScrollPage<>(null, true, new ArrayList<>());
|
||||
}
|
||||
esTemplate.searchScrollClear(Arrays.asList(scroll.getScrollId()));
|
||||
return new ScrollPage<>(null, true, new ArrayList<>());
|
||||
}
|
||||
|
||||
@Operation(summary = "Returns notifications as file")
|
||||
|
|
@ -197,8 +197,8 @@ public class OpenairePublicController extends AbstractDnetController {
|
|||
final Path pathDir = new Path(opendoarEventsPath + "/" + DigestUtils.md5Hex(id));
|
||||
|
||||
try (final FileSystem fs = FileSystem.get(conf);
|
||||
final ServletOutputStream out = res.getOutputStream();
|
||||
final GZIPOutputStream gzOut = new GZIPOutputStream(out)) {
|
||||
final ServletOutputStream out = res.getOutputStream();
|
||||
final GZIPOutputStream gzOut = new GZIPOutputStream(out)) {
|
||||
boolean first = true;
|
||||
|
||||
IOUtils.write("[\n", gzOut, StandardCharsets.UTF_8);
|
||||
|
|
@ -209,16 +209,16 @@ public class OpenairePublicController extends AbstractDnetController {
|
|||
final Path path = fileStatus.getPath();
|
||||
if (path.getName().endsWith(".json")) {
|
||||
try (final FSDataInputStream fis = fs.open(path);
|
||||
final InputStreamReader isr = new InputStreamReader(fis);
|
||||
final BufferedReader br = new BufferedReader(isr)) {
|
||||
final InputStreamReader isr = new InputStreamReader(fis);
|
||||
final BufferedReader br = new BufferedReader(isr)) {
|
||||
|
||||
first = processLine(gzOut, first, br);
|
||||
}
|
||||
} else if (path.getName().endsWith(".json.gz")) {
|
||||
try (final FSDataInputStream fis = fs.open(path);
|
||||
final GZIPInputStream gzIn = new GZIPInputStream(fis);
|
||||
final InputStreamReader isr = new InputStreamReader(gzIn);
|
||||
final BufferedReader br = new BufferedReader(isr)) {
|
||||
final GZIPInputStream gzIn = new GZIPInputStream(fis);
|
||||
final InputStreamReader isr = new InputStreamReader(gzIn);
|
||||
final BufferedReader br = new BufferedReader(isr)) {
|
||||
|
||||
first = processLine(gzOut, first, br);
|
||||
}
|
||||
|
|
@ -271,24 +271,30 @@ public class OpenairePublicController extends AbstractDnetController {
|
|||
|
||||
@Operation(summary = "Store the feedback of an event (MOCK)")
|
||||
@RequestMapping(value = "/feedback/events", method = {
|
||||
RequestMethod.POST, RequestMethod.PATCH
|
||||
RequestMethod.POST, RequestMethod.PATCH
|
||||
})
|
||||
private Map<String, String> feedbackEvent(@RequestBody final EventFeedback feedback) {
|
||||
// TOOD
|
||||
final Map<String, String> res = new HashMap<>();
|
||||
private Map<String, Object> feedbackEvent(@RequestBody final EventFeedback feedback) {
|
||||
|
||||
final DbEventFeedback dbEntry = new DbEventFeedback();
|
||||
dbEntry.setEventId(feedback.getEventId());
|
||||
dbEntry.setStatus(feedback.getStatus());
|
||||
|
||||
feedbackRepository.save(dbEntry);
|
||||
|
||||
final Map<String, Object> res = new HashMap<>();
|
||||
res.put("status", "done");
|
||||
res.put("feedback", dbEntry);
|
||||
return res;
|
||||
}
|
||||
|
||||
private List<ShortEventMessage> calculateNotificationMessages(final SearchScrollHits<Notification> scroll) {
|
||||
if (scroll.getSearchHits().size() > 0) {
|
||||
return scroll.stream()
|
||||
.map(SearchHit::getContent)
|
||||
.map(this::messageFromNotification)
|
||||
.collect(Collectors.toList());
|
||||
} else {
|
||||
return new ArrayList<>();
|
||||
.map(SearchHit::getContent)
|
||||
.map(this::messageFromNotification)
|
||||
.collect(Collectors.toList());
|
||||
}
|
||||
return new ArrayList<>();
|
||||
}
|
||||
|
||||
private ShortEventMessage messageFromNotification(final Notification n) {
|
||||
|
|
|
|||
|
|
@ -1,3 +0,0 @@
|
|||
#!/bin/bash
|
||||
|
||||
ssh -vNL 5432:10.19.65.40:5432 michele.artini@iis-cdh5-test-gw.ocean.icm.edu.pl
|
||||
|
|
@ -0,0 +1,69 @@
|
|||
package eu.dnetlib.broker.common.feedbacks;
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.util.Date;
|
||||
|
||||
import javax.persistence.Column;
|
||||
import javax.persistence.Entity;
|
||||
import javax.persistence.EnumType;
|
||||
import javax.persistence.Enumerated;
|
||||
import javax.persistence.Id;
|
||||
import javax.persistence.Table;
|
||||
|
||||
import org.springframework.data.annotation.CreatedDate;
|
||||
import org.springframework.data.annotation.LastModifiedDate;
|
||||
|
||||
@Entity(name = "feedbacks")
|
||||
@Table(name = "feedbacks")
|
||||
public class DbEventFeedback implements Serializable {
|
||||
|
||||
private static final long serialVersionUID = -1252332235554824748L;
|
||||
|
||||
@Id
|
||||
@Column(name = "eventid")
|
||||
private String eventId;
|
||||
|
||||
@Column(name = "status", nullable = false)
|
||||
@Enumerated(EnumType.STRING)
|
||||
private FeedbackStatus status;
|
||||
|
||||
@CreatedDate
|
||||
@Column(name = "creation_date", nullable = false, updatable = false, columnDefinition = "timestamp without time zone default now()'")
|
||||
private Date creationDate = new Date();
|
||||
|
||||
@LastModifiedDate
|
||||
@Column(name = "modification_date", nullable = false, columnDefinition = "timestamp without time zone default now()")
|
||||
private Date modificationDate = new Date();
|
||||
|
||||
public String getEventId() {
|
||||
return eventId;
|
||||
}
|
||||
|
||||
public void setEventId(final String eventId) {
|
||||
this.eventId = eventId;
|
||||
}
|
||||
|
||||
public FeedbackStatus getStatus() {
|
||||
return status;
|
||||
}
|
||||
|
||||
public void setStatus(final FeedbackStatus status) {
|
||||
this.status = status;
|
||||
}
|
||||
|
||||
public Date getCreationDate() {
|
||||
return creationDate;
|
||||
}
|
||||
|
||||
public void setCreationDate(final Date creationDate) {
|
||||
this.creationDate = creationDate;
|
||||
}
|
||||
|
||||
public Date getModificationDate() {
|
||||
return modificationDate;
|
||||
}
|
||||
|
||||
public void setModificationDate(final Date modificationDate) {
|
||||
this.modificationDate = modificationDate;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,7 @@
|
|||
package eu.dnetlib.broker.common.feedbacks;
|
||||
|
||||
import org.springframework.data.repository.CrudRepository;
|
||||
|
||||
public interface DbEventFeedbackRepository extends CrudRepository<DbEventFeedback, String> {
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,9 @@
|
|||
package eu.dnetlib.broker.common.feedbacks;
|
||||
|
||||
public enum FeedbackStatus {
|
||||
DISCARDED, // the event was not processable by the system. OpenAIRE should not interpret such status in a negative or positive sense
|
||||
// with regard to the accuracy of the notification
|
||||
REJECTED, // a human takes the decision to reject the suggestion as it was wrong
|
||||
ACCEPTED // a human takes the decision to apply the suggested enrichment to the local record
|
||||
|
||||
}
|
||||
|
|
@ -18,6 +18,13 @@ CREATE TABLE topic_types (
|
|||
regex text UNIQUE NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE feedbacks (
|
||||
eventid text PRIMARY KEY,
|
||||
status text NOT NULL,
|
||||
creation_date timestamp without time zone DEFAULT now(),
|
||||
modification_date timestamp without time zone DEFAULT now()
|
||||
);
|
||||
|
||||
-- curl "http://localhost:8080/api/topic-types/add" -d "name=ENRICH&expression=ENRICH%2F%3Ccond%3E%2F%3Cfield%3E&producerId=OpenAIRE&mapKeys=targetDatasourceName"
|
||||
INSERT INTO public.topic_types (id, name, expression, map_keys, producer_id, regex) VALUES ('tt-a739fa2b-fde0-4eb2-bcee-6e7f277347db', 'ENRICH', 'ENRICH/<cond>/<field>', 'targetDatasourceName', 'OpenAIRE', '^ENRICH\/[a-zA-Z0-9._-]+\/[a-zA-Z0-9._-]+$');
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue