Skip to content

B
brandkbs2
Commit 23029ed9 by shenjunjie
Options

Merge branch 'feature' into 'release' 

Feature

See merge request !415
parents 97a76eb0 5761750e
Showing with 391 additions and 115 deletions
src/main/java/com/zhiwei/brandkbs2/controller/app/AppSearchController.java
......@@ -11,9 +11,11 @@ import com.zhiwei.brandkbs2.controller.BaseController;
import com.zhiwei.brandkbs2.easyexcel.EasyExcelUtil;
import com.zhiwei.brandkbs2.enmus.RoleEnum;
import com.zhiwei.brandkbs2.model.ResponseResult;
import com.zhiwei.brandkbs2.pojo.MarkFlowEntity;
import com.zhiwei.brandkbs2.pojo.WholeSearchRecord;
import com.zhiwei.brandkbs2.pojo.dto.*;
import com.zhiwei.brandkbs2.pojo.vo.ChannelListVO;
import com.zhiwei.brandkbs2.pojo.vo.PageVO;
import com.zhiwei.brandkbs2.service.*;
import com.zhiwei.brandkbs2.util.Tools;
import com.zhiwei.middleware.event.pojo.dto.BrandkbsEventSearchDTO;
......@@ -199,14 +201,28 @@ public class AppSearchController extends BaseController {
@LogRecord(values = "keyword", description = "查舆情", arguments = true, entity = true)
@PostMapping("/mark/list")
public ResponseResult getYuqingMarkList(@RequestBody MarkSearchDTO markSearchDTO) {
return ResponseResult.success(markDataService.getYuqingMarkList(markSearchDTO));
PageVO<MarkFlowEntity> yuqingMarkList = markDataService.getYuqingMarkList(markSearchDTO);
// 仅第一页增加平台进量（声量）统计
if (1 == markSearchDTO.getPage()) {
JSONObject info = yuqingMarkList.getInfo();
info.put("counts", markDataService.countMarkPlatform(markSearchDTO));
yuqingMarkList.setInfo(info);
}
return ResponseResult.success(yuqingMarkList);
}
@ApiOperation("原始数据列表")
@LogRecord(values = "keyword", description = "查原始数据", arguments = true, entity = true)
@PostMapping("/origin/list")
public ResponseResult getOriginList(@RequestBody MarkSearchDTO markSearchDTO) {
return ResponseResult.success(markDataService.getOriginList(markSearchDTO));
PageVO<MarkFlowEntity> originList = markDataService.getOriginList(markSearchDTO);
// 仅第一页增加平台进量（声量）统计
if (1 == markSearchDTO.getPage()) {
JSONObject info = originList.getInfo();
info.put("counts", markDataService.countOriginPlatform(markSearchDTO));
originList.setInfo(info);
}
return ResponseResult.success(originList);
}
@ApiOperation("搜索-舆情列表-搜索条件")
......
src/main/java/com/zhiwei/brandkbs2/pojo/BytedanceCustomEventUpdateTask.java
......@@ -60,6 +60,14 @@ public class BytedanceCustomEventUpdateTask extends AbstractBaseMongo{
*/
private Boolean cancel;
/**
* 已处理的数据量
*/
private int schedule;
/**
* 待处理的数据总量
*/
private int totalSchedule;
/**
* 项目id
*/
private String projectId;
......@@ -139,6 +147,8 @@ public class BytedanceCustomEventUpdateTask extends AbstractBaseMongo{
task.setEventInfo(new ArrayList<>());
task.setErrorData(new ArrayList<>());
task.setCancel(false);
task.setSchedule(0);
task.setTotalSchedule(0);
task.setProjectId(UserThreadLocal.getProjectId());
task.setUserId(UserThreadLocal.getUserId());
task.setNickName(UserThreadLocal.getNickname());
......@@ -157,6 +167,8 @@ public class BytedanceCustomEventUpdateTask extends AbstractBaseMongo{
task.setEventInfo(null);
task.setErrorData(null);
task.setCancel(false);
task.setSchedule(0);
task.setTotalSchedule(0);
task.setProjectId(UserThreadLocal.getProjectId());
task.setUserId(UserThreadLocal.getUserId());
task.setNickName(UserThreadLocal.getNickname());
......
src/main/java/com/zhiwei/brandkbs2/pojo/BytedanceCustomPlatformWeight.java
......@@ -21,4 +21,8 @@ public class BytedanceCustomPlatformWeight extends AbstractBaseMongo{
* 权重
*/
private double weight;
/**
* 常量
*/
private double constant;
}
src/main/java/com/zhiwei/brandkbs2/service/EsSearchService.java
package com.zhiwei.brandkbs2.service;
import com.alibaba.fastjson.JSONObject;
import com.zhiwei.brandkbs2.pojo.dto.MarkSearchDTO;
import org.apache.commons.lang3.tuple.Pair;
import org.elasticsearch.search.SearchHits;
......@@ -18,7 +19,7 @@ public interface EsSearchService {
/**
* 搜索标注数据
*
* @param dto 搜素实体类
* @param dto 搜索实体类
* @param scroll 滚动查询
* @return 搜索结果
*/
......@@ -27,7 +28,7 @@ public interface EsSearchService {
/**
* 搜索竞品标注数据
*
* @param dto 搜素实体类
* @param dto 搜索实体类
* @param scroll 滚动查询
* @return 搜索结果
*/
......@@ -36,10 +37,25 @@ public interface EsSearchService {
/**
* 搜索原始数据
*
* @param dto 搜素实体类
* @param dto 搜索实体类
* @param scroll 滚动查询
* @return 搜索结果
*/
Pair<SearchHits[], Map<String, Long>> searchOriginHitsAndCount(MarkSearchDTO dto, boolean scroll) throws IOException;
/**
* 查舆情舆情库-平台进量（声量）统计
* @param dto 搜索实体类
* @return
* @throws IOException
*/
JSONObject countMarkPlatform(MarkSearchDTO dto) throws IOException;
/**
* 查舆情原始数据-平台进量（声量）统计
* @param dto 搜索实体类
* @return
* @throws IOException
*/
JSONObject countOriginPlatform(MarkSearchDTO dto) throws IOException;
}
src/main/java/com/zhiwei/brandkbs2/service/MarkDataService.java
......@@ -378,4 +378,18 @@ public interface MarkDataService {
* @return
*/
Long interactionUpdatePrediction(MarkSearchDTO dto);
/**
* 查舆情舆情库-平台进量（声量）统计
* @param dto 标注数据搜索传输类
* @return
*/
JSONObject countMarkPlatform(MarkSearchDTO dto);
/**
* 查舆情原始数据-平台进量（声量）统计
* @param dto 标注数据搜索传输类
* @return
*/
JSONObject countOriginPlatform(MarkSearchDTO dto);
}
src/main/java/com/zhiwei/brandkbs2/service/impl/EsSearchServiceImpl.java
package com.zhiwei.brandkbs2.service.impl;
import com.alibaba.fastjson.JSONObject;
import com.zhiwei.brandkbs2.common.GenericAttribute;
import com.zhiwei.brandkbs2.common.GlobalPojo;
import com.zhiwei.brandkbs2.es.EsClientDao;
import com.zhiwei.brandkbs2.es.EsQueryTools;
import com.zhiwei.brandkbs2.function.ArticleFunction;
import com.zhiwei.brandkbs2.pojo.dto.MarkSearchDTO;
import com.zhiwei.brandkbs2.service.EsSearchService;
import com.zhiwei.brandkbs2.util.Tools;
import com.zhiwei.qbjc.bean.pojo.common.MessagePlatform;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
......@@ -21,9 +24,8 @@ import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.*;
import java.util.stream.Collectors;
/**
* @ClassName: EsSearchServiceImpl
......@@ -55,6 +57,18 @@ public class EsSearchServiceImpl implements EsSearchService {
return searchHitsAndCount(dto, this::createOriginSearchHelper, hotEsDao, scroll);
}
@Override
public JSONObject countMarkPlatform(MarkSearchDTO dto) throws IOException {
EsClientDao.SearchHelper helper = createSearchHelper(dto);
return platformCount(helper.getPostFilter(), esClientDao);
}
@Override
public JSONObject countOriginPlatform(MarkSearchDTO dto) throws IOException {
EsClientDao.SearchHelper helper = createOriginSearchHelper(dto);
return platformCount(helper.getPostFilter(), hotEsDao);
}
private Pair<SearchHits[], Map<String, Long>> searchHitsAndCount(MarkSearchDTO dto, ArticleFunction.CreateSearchHelper createSearchHelper,
EsClientDao clientDao, boolean scroll) throws IOException {
EsClientDao.SearchHelper helper = createSearchHelper.create(dto);
......@@ -63,8 +77,8 @@ public class EsSearchServiceImpl implements EsSearchService {
return Pair.of(clientDao.searchScrollResponse(helper).stream().map(SearchResponse::getHits).toArray(SearchHits[]::new), null);
}
SearchHits searchHits = clientDao.searchHits(helper);
Map<String, Long> platformCount = platformCount(dto.getPage(), dto.getPlatforms(), helper.getPostFilter(), clientDao);
return Pair.of(new SearchHits[]{searchHits}, platformCount);
// Map<String, Long> platformCount = platformCount(dto.getPage(), dto.getPlatforms(), helper.getPostFilter(), clientDao);
return Pair.of(new SearchHits[]{searchHits}, new HashMap<>());
}
private EsClientDao.SearchHelper createSearchHelper(MarkSearchDTO dto) {
......@@ -250,18 +264,23 @@ public class EsSearchServiceImpl implements EsSearchService {
*
* @return
*/
private Map<String, Long> platformCount(int page, List<String> platforms, BoolQueryBuilder postFilter, EsClientDao esClientDao) throws IOException {
Map<String, Long> counts = new HashMap<>();
// TODO 平台进量暂不统计
private JSONObject platformCount(BoolQueryBuilder postFilter, EsClientDao esClientDao) throws IOException {
List<JSONObject> list = new ArrayList<>();
//平台计量统计 仅第一页且无平台限制
// if (1 == page && CollectionUtils.isEmpty(platforms)) {
// for (MessagePlatform platform : GlobalPojo.PLATFORMS) {
// BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery().must(postFilter).must(EsQueryTools.assemblePlatformQuery(Collections.singletonList(platform)));
// Long count = esClientDao.count(esClientDao.getIndexes(), queryBuilder, null);
// counts.put(platform.getName(), count);
// }
// }
return counts;
for (MessagePlatform platform : GlobalPojo.PLATFORMS.stream().filter(messagePlatform -> !"外媒".equals(messagePlatform.getName())).collect(Collectors.toList())) {
BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery().must(postFilter).must(EsQueryTools.assemblePlatformQuery(Collections.singletonList(platform)));
Long count = esClientDao.count(esClientDao.getIndexes(), queryBuilder, null);
JSONObject jsonObject = new JSONObject();
jsonObject.put("id", platform.getId());
jsonObject.put("name", platform.getName());
jsonObject.put("count", count);
list.add(jsonObject);
}
Long total = list.stream().map(jsonObject -> jsonObject.getLongValue("count")).reduce(Long::sum).orElse(0L);
JSONObject res = new JSONObject();
res.put("count", list);
res.put("total", total);
return res;
}
}
src/main/java/com/zhiwei/brandkbs2/service/impl/MarkDataServiceImpl.java
......@@ -184,10 +184,12 @@ public class MarkDataServiceImpl implements MarkDataService {
long total = hitsAndCounts.getLeft()[0].getTotalHits().value;
// 消息列表
List<MarkFlowEntity> flowEntityList = getMarkFlowEntity4Origin(markSearchDTO, hitsAndCounts.getLeft()[0]);
JSONObject jsonObject = new JSONObject();
jsonObject.put("highlightWord", highlightWordDao.findList(Query.query(Criteria.where("projectId").is(UserThreadLocal.getProjectId()))).stream().map(HighlightWord::getKeyword).collect(Collectors.toList()));
// 返回分页结果并设置平台count
return PageVO.createPageVo(total, markSearchDTO.getPage(), markSearchDTO.getPageSize(), flowEntityList)
// 各平台计量
.setInfo(new JSONObject(ImmutableMap.of("highlightWord", highlightWordDao.findList(Query.query(Criteria.where("projectId").is(UserThreadLocal.getProjectId()))).stream().map(HighlightWord::getKeyword).collect(Collectors.toList()))));
.setInfo(jsonObject);
} catch (IOException e) {
ExceptionCast.cast(CommonCodeEnum.FAIL, "es查询异常");
}
......@@ -266,10 +268,12 @@ public class MarkDataServiceImpl implements MarkDataService {
long total = hitsAndCounts.getLeft()[0].getTotalHits().value;
// 消息列表
List<MarkFlowEntity> flowEntityList = getMarkFlowEntity(markSearchDTO, hitsAndCounts.getLeft()[0]);
JSONObject jsonObject = new JSONObject();
// 高亮关键词
jsonObject.put("highlightWord", GlobalPojo.HIGHLIGHT_MAP.get(UserThreadLocal.getProjectId()));
// 返回分页结果并返回高亮关键词
return PageVO.createPageVo(total, markSearchDTO.getPage(), markSearchDTO.getPageSize(), flowEntityList)
// 高亮关键词
.setInfo(new JSONObject(ImmutableMap.of("highlightWord", GlobalPojo.HIGHLIGHT_MAP.get(UserThreadLocal.getProjectId()))));
.setInfo(jsonObject);
} catch (IOException e) {
ExceptionCast.cast(CommonCodeEnum.FAIL, "es查询异常");
}
......@@ -1150,17 +1154,12 @@ public class MarkDataServiceImpl implements MarkDataService {
public Pair<JSONObject, Integer> searchWholeNetworkWithBalance(SearchFilterDTO dto) {
if (dto.isExternalDataSource()) { // 查商业数据库
try {
JSONObject params = externalTransform(dto);
JSONObject taskEntity = new JSONObject();
taskEntity.put("appId", "6183571e0d710000f6003a12"); // 应用id, 由张志伟提供给使用者
taskEntity.put("taskType", null); // 可为null, 任务类型
taskEntity.put("taskDescription", null); // 可为null, 任务描述
taskEntity.put("taskName", UserThreadLocal.getProjectId() + "_" + UserThreadLocal.getUserId());
taskEntity.put("userName", "品见"); // 提交任务的人
taskEntity.put("searchParams", params); // 任务详情参数
ResponseEntity<JSONObject> responseEntity = restTemplate.postForEntity(addIStarShineKSDataUrl, taskEntity, JSONObject.class);
JSONObject result = Objects.requireNonNull(responseEntity.getBody()).getJSONObject("data");
JSONObject json = searchWholeBackUp(result.get("id").toString());
JSONObject data = getSearchWholeOpinionJson(dto);
JSONObject json = packageExternalDataSourceResult(data);
// 仅当第一页时统计平台进量（声量）
if (1 == dto.getPage()){
json.put("info", getExternalDataSourcePlatformCount(dto));
}
return Pair.of(json, json.getInteger("pageLimit"));
} catch (Exception e) {
log.error("error searchWholeNetwork - ", e);
......@@ -1172,6 +1171,36 @@ public class MarkDataServiceImpl implements MarkDataService {
}
}
/**
* 获取外部数据源搜索结果json
* @param dto
* @return
*/
private JSONObject getSearchWholeOpinionJson(SearchFilterDTO dto){
// 组装请求参数
JSONObject taskEntity = packageExternalDataSourceParams(dto);
ResponseEntity<JSONObject> responseEntity = restTemplate.postForEntity(addIStarShineKSDataUrl, taskEntity, JSONObject.class);
JSONObject result = Objects.requireNonNull(responseEntity.getBody()).getJSONObject("data");
return searchWholeBackUp(result.get("id").toString());
}
/**
* 外部数据源组装请求参数
* @param dto
* @return
*/
private JSONObject packageExternalDataSourceParams(SearchFilterDTO dto){
JSONObject params = externalTransform(dto);
JSONObject taskEntity = new JSONObject();
taskEntity.put("appId", "6183571e0d710000f6003a12"); // 应用id, 由张志伟提供给使用者
taskEntity.put("taskType", null); // 可为null, 任务类型
taskEntity.put("taskDescription", null); // 可为null, 任务描述
taskEntity.put("taskName", UserThreadLocal.getProjectId() + "_" + UserThreadLocal.getUserId());
taskEntity.put("userName", "品见"); // 提交任务的人
taskEntity.put("searchParams", params); // 任务详情参数
return taskEntity;
}
@Override
public List<ExportSearchWholeDTO> exportSearchWhole(SearchFilterDTO dto) {
Integer exportAmount = userService.queryUserInfo(UserThreadLocal.getUserId(), UserThreadLocal.getProjectId()).getExportAmount();
......@@ -1195,6 +1224,11 @@ public class MarkDataServiceImpl implements MarkDataService {
return jsonArray.stream().map(json -> ExportSearchWholeDTO.creatExportSearchWholeDTO((JSONObject) json)).collect(Collectors.toList());
}
/**
* 舆情库数据结果组装成标准分页返回格式
* @param responseEntity
* @return
*/
private JSONObject searchWholeOpinion(ResponseEntity<JSONObject> responseEntity) {
JSONObject jsonObject = new JSONObject();
JSONObject bodyData = Objects.requireNonNull(responseEntity.getBody()).getJSONObject("data");
......@@ -1227,6 +1261,20 @@ public class MarkDataServiceImpl implements MarkDataService {
result.put("ipLocation", list.get("ip_location"));
resultList.add(result);
}
// 仅第一页返回平台进量（声量）统计
if (1 == data.getInteger("page")) {
JSONObject count = bodyData.getJSONObject("count");
List<JSONObject> platformCounts = count.getJSONArray("count").toJavaList(JSONObject.class);
// 去除外媒和脉脉两个平台
List<JSONObject> list = platformCounts.stream().filter(json -> !Objects.equals(json.getString("name"), "脉脉") &&
!Objects.equals(json.getString("name"), "外媒")).collect(Collectors.toList());
// 重新计算总数
Long total = list.stream().map(json -> json.getLongValue("count")).reduce(Long::sum).orElse(0L);
JSONObject counts = new JSONObject();
counts.put("count", list);
counts.put("total", total);
jsonObject.put("info", counts);
}
jsonObject.put("list", resultList);
jsonObject.put("page", data.get("page"));
jsonObject.put("pageLimit", data.get("pageLimit"));
......@@ -1236,10 +1284,13 @@ public class MarkDataServiceImpl implements MarkDataService {
return jsonObject;
}
/**
* 获取外部数据源搜索结果JSONObject
* @param taskId
* @return
*/
private JSONObject searchWholeBackUp(String taskId) {
JSONObject jsonObject = new JSONObject();
List<JSONObject> resultList = new ArrayList<>();
for (int i = 0; i < 30; i++) {
ResponseEntity<JSONObject> responseEntity = restTemplate.getForEntity(getIStarShineKSInfoDataUrl, JSONObject.class, taskId);
JSONObject data = Objects.requireNonNull(responseEntity.getBody()).getJSONObject("data");
......@@ -1248,43 +1299,78 @@ public class MarkDataServiceImpl implements MarkDataService {
throw new IllegalArgumentException(task.getString("searchResultMsg"));
}
if (1 == task.getInteger("crawlerFinish")) {
//pageSize
Integer receiveCount = task.getInteger("receiveCount");
Integer searchCount = task.getInteger("searchCount");
int totalPage;
if (0 == receiveCount || 0 == searchCount){
totalPage = 0;
}else {
totalPage = searchCount % receiveCount == 0 ? searchCount / receiveCount : (searchCount / receiveCount + 1);
}
List<JSONObject> listArray = data.getJSONArray("data").toJavaList(JSONObject.class);
for (JSONObject json : listArray) {
JSONObject result = new JSONObject();
String platform = getPlatformExternal(json.getJSONObject("gather"));
result.put("platform", platform);
result.put("title", json.get("title"));
result.put("content", json.get("content"));
// 渠道获取分平台
result.put("source", getSourceExternal(json, getExternalType(json, platform)));
result.put("url", getUrl(json));
result.put("time", json.getString("ctime").length() == 10 ? json.getLong("ctime") * 1000L : json.getLong("ctime"));
resultList.add(result);
}
jsonObject.put("list", resultList);
//jsonObject.put("page",data1.get("page"));
jsonObject.put("pageLimit", receiveCount);
jsonObject.put("total", searchCount);
jsonObject.put("totalPage", totalPage);
jsonObject.put("balance", projectService.getProjectById(UserThreadLocal.getProjectId()).getWholeSearchBalance() - receiveCount);
jsonObject.put("isBackUp", true);
return jsonObject;
return data;
}
Tools.sleep(500L);
}
return jsonObject;
}
/**
* 将外部数据源JSONObject组装成标准的返回结果
* @param data
* @return
*/
private JSONObject packageExternalDataSourceResult(JSONObject data){
JSONObject jsonObject = new JSONObject();
List<JSONObject> resultList = new ArrayList<>();
JSONObject task = data.getJSONObject("task");
Integer receiveCount = task.getInteger("receiveCount");
Integer searchCount = task.getInteger("searchCount");
int totalPage;
if (0 == receiveCount || 0 == searchCount){
totalPage = 0;
}else {
totalPage = searchCount % receiveCount == 0 ? searchCount / receiveCount : (searchCount / receiveCount + 1);
}
List<JSONObject> listArray = data.getJSONArray("data").toJavaList(JSONObject.class);
for (JSONObject json : listArray) {
JSONObject result = new JSONObject();
String platform = getPlatformExternal(json.getJSONObject("gather"));
result.put("platform", platform);
result.put("title", json.get("title"));
result.put("content", json.get("content"));
// 渠道获取分平台
result.put("source", getSourceExternal(json, getExternalType(json, platform)));
result.put("url", getUrl(json));
result.put("time", json.getString("ctime").length() == 10 ? json.getLong("ctime") * 1000L : json.getLong("ctime"));
resultList.add(result);
}
jsonObject.put("list", resultList);
//jsonObject.put("page",data1.get("page"));
jsonObject.put("pageLimit", receiveCount);
jsonObject.put("total", searchCount);
jsonObject.put("totalPage", totalPage);
jsonObject.put("balance", projectService.getProjectById(UserThreadLocal.getProjectId()).getWholeSearchBalance() - receiveCount);
jsonObject.put("isBackUp", true);
return jsonObject;
}
/**
* 获取实时采集（外部数据源）平台进量（声量）
* @param dto
* @return
*/
private JSONObject getExternalDataSourcePlatformCount(SearchFilterDTO dto){
JSONObject res = new JSONObject();
List<JSONObject> list = new ArrayList<>();
List<String> platforms = Arrays.asList("App", "博客", "平媒", "新闻", "微博", "微信", "论坛", "视频", "短视频");
for (String platform : platforms) {
// 重置平台条件
dto.setPlatforms(Collections.singletonList(platform));
JSONObject data = getSearchWholeOpinionJson(dto);
JSONObject task = data.getJSONObject("task");
JSONObject jsonObject = new JSONObject();
jsonObject.put("count", task.getLongValue("searchCount"));
jsonObject.put("name", platform);
list.add(jsonObject);
}
Long total = list.stream().map(jsonObject -> jsonObject.getLongValue("count")).reduce(Long::sum).orElse(0L);
res.put("count", list);
res.put("total", total);
return res;
}
private ClassB.TypeB getExternalType(JSONObject tJson, String platform) {
int wtype = tJson.getIntValue("wtype");
if (platform.equals("微博") && 2 == wtype) {
......@@ -1580,6 +1666,34 @@ public class MarkDataServiceImpl implements MarkDataService {
return null;
}
@Override
public JSONObject countMarkPlatform(MarkSearchDTO dto) {
defaultMarkSearch(dto);
try {
return esSearchService.countMarkPlatform(dto);
}catch (Exception e){
ExceptionCast.cast(CommonCodeEnum.FAIL, "es查询异常", e);
}
return null;
}
@Override
public JSONObject countOriginPlatform(MarkSearchDTO dto) {
// 时间范围默认当天
if (Objects.isNull(dto.getStartTime()) || Objects.isNull(dto.getEndTime())) {
Date now = new Date();
dto.setEndTime(now.getTime());
dto.setStartTime(DateUtils.addDays(now, -1).getTime());
}
defaultMarkSearch(dto);
try {
return esSearchService.countOriginPlatform(dto);
}catch (Exception e){
ExceptionCast.cast(CommonCodeEnum.FAIL, "es查询异常", e);
}
return null;
}
private static JSONObject getTitleAndUrl(Map<String, Object> map) {
JSONObject json = new JSONObject();
BaseMap baseMap = Tools.getBaseFromEsMap(map);
......
src/main/java/com/zhiwei/brandkbs2/service/impl/ToolsetServiceImpl.java
......@@ -264,6 +264,10 @@ public class ToolsetServiceImpl implements ToolsetService {
* @return
*/
private List<JSONObject> urlInteractionUpdate(List<String> urls){
return urlInteractionUpdate(urls, null);
}
private List<JSONObject> urlInteractionUpdate(List<String> urls, String taskId){
// 通过url获取域名进而获取任务类型
Map<String, String> map = new HashMap<>();
urls.forEach(url -> map.compute(url, (key, value) -> {
......@@ -287,6 +291,10 @@ public class ToolsetServiceImpl implements ToolsetService {
for (List<String> ids : ListUtils.partition(taskIdList, 50)) {
result.addAll(getInteractionResult(ids));
}
if (Objects.nonNull(taskId)) {
// 更新处理进度
addProcessSchedule(taskId, taskIdList.size());
}
}
return result;
}
......@@ -329,13 +337,18 @@ public class ToolsetServiceImpl implements ToolsetService {
try {
ResponseEntity<JSONObject> response = restTemplate.postForEntity(interactionResultUrl, request, JSONObject.class);
JSONObject body = response.getBody();
if (Objects.nonNull(body) && 200 == body.getIntValue("code") && !body.isEmpty() && !body.getJSONArray("data").toJavaList(JSONObject.class).isEmpty() &&
body.getJSONArray("data").toJavaList(JSONObject.class).stream().map(m -> m.getIntValue("code")).allMatch(code -> 200 == code)) {
if (Objects.nonNull(body) && !body.isEmpty() && 200 == body.getIntValue("code") && CollectionUtils.isNotEmpty(body.getJSONArray("data").toJavaList(JSONObject.class)) &&
body.getJSONArray("data").toJavaList(JSONObject.class).stream().map(m -> m.getIntValue("code")).allMatch(code -> 200 == code || 403 == code) &&
taskIds.size() == body.getJSONArray("data").toJavaList(JSONObject.class).size()) {
res = body.getJSONArray("data").toJavaList(JSONObject.class);
break;
} else {
Thread.sleep(10000L);
}
// 最后一轮即使不满足条件也该返回
if (29 == i && Objects.nonNull(body) && !body.isEmpty() && 200 == body.getIntValue("code") && CollectionUtils.isNotEmpty(body.getJSONArray("data").toJavaList(JSONObject.class))){
res = body.getJSONArray("data").toJavaList(JSONObject.class);
}
}catch (Exception e){
ExceptionCast.cast(CommonCodeEnum.FAIL, "轮询互动量更新任务异常,taskId:" + taskIds, e);
}
......@@ -362,7 +375,8 @@ public class ToolsetServiceImpl implements ToolsetService {
readExcel.setClazz(UploadBytedanceEventDTO.class);
readExcel.setAnalysisEventListener(new BytedanceEventListener(task.getId(), data));
EasyExcelUtil.read(file, readExcel);
// 更新处理进度
resetProcessSchedule(task.getId(), data.size());
ApplicationProjectListener.getThreadPool().execute(() -> processEventUpdate(data, task, extraCompute));
}
......@@ -386,6 +400,10 @@ public class ToolsetServiceImpl implements ToolsetService {
// 将旧数据与新数据合并
List<BytedanceCustomEventUpdateTaskData> taskData = bytedanceCustomEventUpdateTaskDataDao.findList(new Query(Criteria.where("taskId").is(taskId)));
taskData.addAll(data);
// 清空错误旧数据
task.getErrorData().clear();
// 更新进度
resetProcessSchedule(task.getId(), taskData.size());
ApplicationProjectListener.getThreadPool().execute(() -> processEventUpdate(taskData, task, task.getExtraCompute()));
}
......@@ -405,6 +423,8 @@ public class ToolsetServiceImpl implements ToolsetService {
jsonObject.put("taskId", task.getId());
jsonObject.put("extraCompute", task.getExtraCompute());
jsonObject.put("cTime", task.getCTime());
jsonObject.put("schedule", task.getSchedule());
jsonObject.put("totalSchedule", task.getTotalSchedule());
if ((Objects.equals(task.getTaskStatus(), BytedanceCustomEventUpdateTask.TaskStatus.FINISH.getStatus()) &&
Objects.equals(task.getProcessStatus(), BytedanceCustomEventUpdateTask.ProcessStatus.FINISH.getStatus())) ||
Objects.equals(task.getTaskStatus(), BytedanceCustomEventUpdateTask.TaskStatus.ERROR.getStatus()) ||
......@@ -527,8 +547,7 @@ public class ToolsetServiceImpl implements ToolsetService {
String taskId = task.getId();
try {
List<BytedanceCustomEventUpdateTask.EventInfo> eventInfos = new ArrayList<>();
Map<String, List<BytedanceCustomEventUpdateTaskData>> map =
taskData.stream().collect(Collectors.groupingBy(BytedanceCustomEventUpdateTaskData::getEventName));
Map<String, List<BytedanceCustomEventUpdateTaskData>> map = taskData.stream().collect(Collectors.groupingBy(BytedanceCustomEventUpdateTaskData::getEventName));
for (Map.Entry<String, List<BytedanceCustomEventUpdateTaskData>> entry : map.entrySet()) {
List<BytedanceCustomEventUpdateTaskData> data = entry.getValue();
BytedanceCustomEventUpdateTask.EventInfo eventInfo = new BytedanceCustomEventUpdateTask.EventInfo();
......@@ -536,11 +555,17 @@ public class ToolsetServiceImpl implements ToolsetService {
// 计算影响力指数
eventInfo.setInfluence(computeInf(data));
eventInfos.add(eventInfo);
// 更新渠道匹配处理进度
addProcessSchedule(taskId, data.size());
}
// 若需要计算实时传播力与传播影响力
if (extraCompute) {
// 更新为互动量更新处理状态
updateStatus(taskId, BytedanceCustomEventUpdateTask.TaskStatus.CALCULATING, BytedanceCustomEventUpdateTask.ProcessStatus.INTERACTION_UPDATE);
long count = taskData.stream().filter(data -> Objects.isNull(data.getWechatRead()) && Objects.isNull(data.getWechatReading()) && Objects.isNull(data.getWeiboForward())
&& Objects.isNull(data.getWeiboComment()) && Objects.isNull(data.getWeiboLike())).count();
// 更新互动量更新处理进度
resetProcessSchedule(taskId, (int) count);
for (BytedanceCustomEventUpdateTask.EventInfo eventInfo : eventInfos) {
List<BytedanceCustomEventUpdateTaskData> data = map.get(eventInfo.getName());
// 计算实时传播力
......@@ -550,12 +575,6 @@ public class ToolsetServiceImpl implements ToolsetService {
}
updateStatus(taskId, BytedanceCustomEventUpdateTask.TaskStatus.CALCULATING, BytedanceCustomEventUpdateTask.ProcessStatus.INFLUENCE_COMPUTE);
}
// 移除数据中的错误数据
if (CollectionUtils.isNotEmpty(task.getErrorData())){
taskData.removeAll(task.getErrorData());
}
bytedanceCustomEventUpdateTaskDataDao.deleteOneByQuery(new Query(Criteria.where("taskId").is(taskId)));
bytedanceCustomEventUpdateTaskDataDao.insertMany(taskData);
task.setEventInfo(eventInfos);
updateStatus(task, BytedanceCustomEventUpdateTask.TaskStatus.FINISH, BytedanceCustomEventUpdateTask.ProcessStatus.FINISH);
}catch (Exception e){
......@@ -609,8 +628,10 @@ public class ToolsetServiceImpl implements ToolsetService {
}
}
updateStatus(taskId, null, BytedanceCustomEventUpdateTask.ProcessStatus.INFLUENCE_COMPUTE);
resetProcessSchedule(taskId, data.size());
// 计算影响力指数
task.setContendInfluence(computeInf(data));
addProcessSchedule(taskId, data.size());
updateStatus(task, BytedanceCustomEventUpdateTask.TaskStatus.FINISH, BytedanceCustomEventUpdateTask.ProcessStatus.FINISH);
}catch (Exception e){
log.info("竞品字节事件影响力补充计算出错-taskId：{}", taskId, e);
......@@ -656,8 +677,22 @@ public class ToolsetServiceImpl implements ToolsetService {
private void updateStatus(String taskId, BytedanceCustomEventUpdateTask.TaskStatus taskStatus){
BytedanceCustomEventUpdateTask task = bytedanceCustomEventUpdateTaskDao.findOneById(taskId);
updateStatus(task, taskStatus, null);
}
private void addProcessSchedule(String taskId, Integer schedule){
BytedanceCustomEventUpdateTask task = bytedanceCustomEventUpdateTaskDao.findOneById(taskId);
task.setUTime(System.currentTimeMillis());
task.setTaskStatus(taskStatus.getStatus());
int taskSchedule = task.getSchedule();
task.setSchedule(taskSchedule + schedule);
bytedanceCustomEventUpdateTaskDao.updateOne(task);
}
private void resetProcessSchedule(String taskId, Integer total){
BytedanceCustomEventUpdateTask task = bytedanceCustomEventUpdateTaskDao.findOneById(taskId);
task.setUTime(System.currentTimeMillis());
task.setSchedule(0);
task.setTotalSchedule(total);
bytedanceCustomEventUpdateTaskDao.updateOne(task);
}
......@@ -725,9 +760,9 @@ public class ToolsetServiceImpl implements ToolsetService {
// 微信文章数
int wechatArticle = (int) data.stream().filter(article -> Objects.equals("微信", article.getPlatform())).count();
// 网媒文章数
int media = (int) data.stream().filter(article -> Objects.equals("网媒", article.getPlatform())).count();
int normalMediaArticle = (int) data.stream().filter(article -> Objects.equals("网媒", article.getPlatform())).count();
// 其他平台文章数
int otherPlatform = (int) data.stream().filter(article -> !Objects.equals("网媒", article.getPlatform())
int otherPlatformArticle = (int) data.stream().filter(article -> !Objects.equals("网媒", article.getPlatform())
&& !Objects.equals("微博", article.getPlatform()) && !Objects.equals("微信", article.getPlatform())).count();
for (BytedanceCustomEventUpdateTaskData datum : data) {
// 选填字段全部未填，后续走互动量更新
......@@ -742,48 +777,94 @@ public class ToolsetServiceImpl implements ToolsetService {
wechatRead = wechatRead + datum.getWechatRead();
wechatReading = wechatReading + datum.getWechatReading();
}
data.removeAll(willBeInteractionUpdateData);
// 选填字段全部未填的链接互动量更新结果
try {
List<String> successUrl = new ArrayList<>();
List<String> urls = willBeInteractionUpdateData.stream().map(BytedanceCustomEventUpdateTaskData::getUrl).collect(Collectors.toList());
List<JSONObject> jsonObjects = urlInteractionUpdate(urls);
Map<String, BytedanceCustomEventUpdateTaskData> urlMap = willBeInteractionUpdateData.stream().collect(Collectors.toMap(BytedanceCustomEventUpdateTaskData::getUrl, o -> o, (v1, v2) -> v1));
// 互动量更新
List<JSONObject> jsonObjects = urlInteractionUpdate(urls, task.getId());
for (JSONObject jsonObject : jsonObjects) {
// 转发数
int repostCount = Objects.isNull(jsonObject.getInteger("repostCount")) ? 0 : jsonObject.getIntValue("repostCount");
// 评论数
int commentCount = Objects.isNull(jsonObject.getInteger("commentCount")) ? 0 : jsonObject.getIntValue("commentCount");
// 点赞数
int likeCount = Objects.isNull(jsonObject.getInteger("likeCount")) ? 0 : jsonObject.getIntValue("likeCount");
// 阅读数
int readCount = Objects.isNull(jsonObject.getInteger("readCount")) ? 0 : jsonObject.getIntValue("readCount");
// 在看数
int kanCount = Objects.isNull(jsonObject.getInteger("kanCount")) ? 0 : jsonObject.getIntValue("kanCount");
weiboForward = weiboForward + repostCount;
weiboComment = weiboComment + commentCount;
weiboLike = weiboLike + likeCount;
wechatRead = wechatRead + readCount;
wechatReading = wechatReading + kanCount;
if (200 == jsonObject.getInteger("code")) {
// 成功的链接
String url = jsonObject.getString("url");
successUrl.add(url);
// 转发数
int repostCount = Objects.isNull(jsonObject.getInteger("repostCount")) ? 0 : jsonObject.getIntValue("repostCount");
urlMap.get(url).setWeiboForward(repostCount);
// 评论数
int commentCount = Objects.isNull(jsonObject.getInteger("commentCount")) ? 0 : jsonObject.getIntValue("commentCount");
urlMap.get(url).setWeiboComment(commentCount);
// 点赞数
int likeCount = Objects.isNull(jsonObject.getInteger("likeCount")) ? 0 : jsonObject.getIntValue("likeCount");
urlMap.get(url).setWeiboLike(likeCount);
// 阅读数
int readCount = Objects.isNull(jsonObject.getInteger("readCount")) ? 0 : jsonObject.getIntValue("readCount");
urlMap.get(url).setWechatRead(readCount);
// 在看数，实际使用的是返回结果中的分享数
int shareCount = Objects.isNull(jsonObject.getInteger("shareCount")) ? 0 : jsonObject.getIntValue("shareCount");
urlMap.get(url).setWechatReading(shareCount);
// 累加
weiboForward = weiboForward + repostCount;
weiboComment = weiboComment + commentCount;
weiboLike = weiboLike + likeCount;
wechatRead = wechatRead + readCount;
wechatReading = wechatReading + shareCount;
}
}
// 报错数据
urls.removeAll(successUrl); // 去除成功数据，剩余即为失败数据
List<BytedanceCustomEventUpdateTaskData> errorData = urls.stream().map(urlMap::get).collect(Collectors.toList());
List<BytedanceCustomEventUpdateTaskData> error = task.getErrorData();
error.addAll(errorData);
task.setErrorData(error);
// 互动量更新成功数据
List<BytedanceCustomEventUpdateTaskData> successData = successUrl.stream().map(urlMap::get).collect(Collectors.toList());
data.addAll(successData);
}catch (Exception e){
// 报错数据：为互动量更新失败数据，更新失败数据不参与计算
log.info("字节事件影响力更新-互动量更新出错-id:{}", task.getId(), e);
List<BytedanceCustomEventUpdateTaskData> errorData = task.getErrorData();
errorData.addAll(willBeInteractionUpdateData);
task.setErrorData(errorData);
}
// 平台指标权重
Map<String, Double> weightMap = bytedanceCustomPlatformWeightDao.findList(new Query())
// 存储数据以用于补充计算
bytedanceCustomEventUpdateTaskDataDao.deleteOneByQuery(new Query(Criteria.where("taskId").is(task.getId())));
bytedanceCustomEventUpdateTaskDataDao.insertMany(data);
// 平台指标权重，常量
Map<String, BytedanceCustomPlatformWeight> weightMap = bytedanceCustomPlatformWeightDao.findList(new Query())
.stream()
.collect(Collectors.toMap(BytedanceCustomPlatformWeight::getType, BytedanceCustomPlatformWeight::getWeight));
// 各平台数据指标与权重相乘求和
return BigDecimal.valueOf(media).multiply(BigDecimal.valueOf(weightMap.get("网媒文章数")))
.add(BigDecimal.valueOf(otherPlatform).multiply(BigDecimal.valueOf(weightMap.get("其他平台文章数"))))
.add(BigDecimal.valueOf(wechatArticle).multiply(BigDecimal.valueOf(weightMap.get("微信文章数"))))
.add(BigDecimal.valueOf(wechatRead).multiply(BigDecimal.valueOf(weightMap.get("微信阅读数"))))
.add(BigDecimal.valueOf(wechatReading).multiply(BigDecimal.valueOf(weightMap.get("微信在看数"))))
.add(BigDecimal.valueOf(weiboArticle).multiply(BigDecimal.valueOf(weightMap.get("微博消息数"))))
.add(BigDecimal.valueOf(weiboForward).multiply(BigDecimal.valueOf(weightMap.get("微博转发数"))))
.add(BigDecimal.valueOf(weiboComment).multiply(BigDecimal.valueOf(weightMap.get("微博评论数"))))
.add(BigDecimal.valueOf(weiboLike).multiply(BigDecimal.valueOf(weightMap.get("微博点赞数"))))
.collect(Collectors.toMap(BytedanceCustomPlatformWeight::getType, o -> o));
BytedanceCustomPlatformWeight normalMediaArticleWeight = weightMap.get("网媒文章数");
BytedanceCustomPlatformWeight otherPlatformArticleWeight = weightMap.get("其他平台文章数");
BytedanceCustomPlatformWeight wechatArticleWeight = weightMap.get("微信文章数");
BytedanceCustomPlatformWeight wechatReadWeight = weightMap.get("微信阅读数");
BytedanceCustomPlatformWeight wechatReadingWeight = weightMap.get("微信在看数");
BytedanceCustomPlatformWeight weiboArticleWeight = weightMap.get("微博消息数");
BytedanceCustomPlatformWeight weiboForwardWeight = weightMap.get("微博转发数");
BytedanceCustomPlatformWeight weiboCommentWeight = weightMap.get("微博评论数");
BytedanceCustomPlatformWeight weiboLikeWeight = weightMap.get("微博点赞数");
// 标准值计算
BigDecimal bigDecimal100 = BigDecimal.valueOf(100);
BigDecimal NM = BigDecimal.valueOf(normalMediaArticle).divide(BigDecimal.valueOf(normalMediaArticleWeight.getConstant()), 9, RoundingMode.HALF_UP).multiply(bigDecimal100);
BigDecimal OP = BigDecimal.valueOf(otherPlatformArticle).divide(BigDecimal.valueOf(otherPlatformArticleWeight.getConstant()), 9, RoundingMode.HALF_UP).multiply(bigDecimal100);
BigDecimal WC = BigDecimal.valueOf(wechatArticle).divide(BigDecimal.valueOf(wechatArticleWeight.getConstant()), 9, RoundingMode.HALF_UP).multiply(bigDecimal100);
BigDecimal WCR = BigDecimal.valueOf(Math.log(wechatRead + 1)).divide(BigDecimal.valueOf(Math.log(wechatReadWeight.getConstant() + 1)), 9, RoundingMode.HALF_UP).multiply(bigDecimal100);
BigDecimal WCL = BigDecimal.valueOf(Math.log(wechatReading + 1)).divide(BigDecimal.valueOf(Math.log(wechatReadingWeight.getConstant() + 1)), 9, RoundingMode.HALF_UP).multiply(bigDecimal100);
BigDecimal WB = BigDecimal.valueOf(weiboArticle).divide(BigDecimal.valueOf(weiboArticleWeight.getConstant()), 9, RoundingMode.HALF_UP).multiply(bigDecimal100);
BigDecimal WBF = BigDecimal.valueOf(Math.log(weiboForward + 1)).divide(BigDecimal.valueOf(Math.log(weiboForwardWeight.getConstant() + 1)), 9, RoundingMode.HALF_UP).multiply(bigDecimal100);
BigDecimal WBC = BigDecimal.valueOf(Math.log(weiboComment + 1)).divide(BigDecimal.valueOf(Math.log(weiboCommentWeight.getConstant() + 1)), 9, RoundingMode.HALF_UP).multiply(bigDecimal100);
BigDecimal WBL = BigDecimal.valueOf(Math.log(weiboLike + 1)).divide(BigDecimal.valueOf(Math.log(weiboLikeWeight.getConstant() + 1)), 9, RoundingMode.HALF_UP).multiply(bigDecimal100);
// 标准值与权重计算得RMI
return NM.multiply(BigDecimal.valueOf(normalMediaArticleWeight.getWeight()))
.add(OP.multiply(BigDecimal.valueOf(otherPlatformArticleWeight.getWeight())))
.add(WC.multiply(BigDecimal.valueOf(wechatArticleWeight.getWeight())))
.add(WCR.multiply(BigDecimal.valueOf(wechatReadWeight.getWeight())))
.add(WCL.multiply(BigDecimal.valueOf(wechatReadingWeight.getWeight())))
.add(WB.multiply(BigDecimal.valueOf(weiboArticleWeight.getWeight())))
.add(WBF.multiply(BigDecimal.valueOf(weiboForwardWeight.getWeight())))
.add(WBC.multiply(BigDecimal.valueOf(weiboCommentWeight.getWeight())))
.add(WBL.multiply(BigDecimal.valueOf(weiboLikeWeight.getWeight())))
.setScale(7, RoundingMode.HALF_UP)
.doubleValue();
}
......
src/main/resources/application-dev.properties
......@@ -77,7 +77,7 @@ qbjc.platform.url=https://yuqing.zhiweidata.com/qbjcbackPhoenix/interface/platfo
qbjc.userCenter.url=https://login.zhiweidata.com/plogin/center
qbjc.userCenter.token=AoJ0ooy3HV1EElWnvQw9YTS9b5Y+fmtkbM6DdpPgDO6D/OhNqH4qrJKarzMr
qbjc.channel.application.url=https://yuqing.zhiweidata.com/qbjcbackPhoenix/interface/middleware/channel-application
qbjc.channel.influence=https://yuqing.test.zhiweidata.com/qbjcbackPhoenix/interface/middleware/channel-influence?platform={1}&channel={2}
qbjc.channel.influence=https://yuqing.zhiweidata.com/qbjcbackPhoenix/interface/middleware/channel-influence?platform={1}&channel={2}
#\u5371\u673A\u5E93\u5916\u90E8\u63A5\u53E3
crisis.search.url=https://crisis.zhiweidata.com/app/brandkbs/crisisSearch?page={1}&size={2}&keyword={3}
crisis.searchTags.url=https://crisis.zhiweidata.com/app/brandkbs/searchCrisisByTags?page={1}&size={2}&brand={3}&category={4}
......
src/main/resources/application-local.properties
......@@ -80,7 +80,7 @@ qbjc.platform.url=https://yuqing.zhiweidata.com/qbjcbackPhoenix/interface/platfo
qbjc.userCenter.url=http://login.zhiweidata.top/plogin/center
qbjc.userCenter.token=AoJ0ooy3HV1EElWnvQw9YTS9b5Y+fmtkbM6DdpPgDO6D/OhNqH4qrJKarzMr
qbjc.channel.application.url=http://192.168.0.79:11000/qbjcbackPhoenix/interface/middleware/channel-application
qbjc.channel.influence=https://yuqing.test.zhiweidata.com/qbjcbackPhoenix/interface/middleware/channel-influence?platform={1}&channel={2}
qbjc.channel.influence=https://yuqing.zhiweidata.com/qbjcbackPhoenix/interface/middleware/channel-influence?platform={1}&channel={2}
#\u5371\u673A\u5E93\u5916\u90E8\u63A5\u53E3
crisis.search.url=https://crisis.zhiweidata.com/app/brandkbs/crisisSearch?page={1}&size={2}&keyword={3}
crisis.searchTags.url=https://crisis.zhiweidata.com/app/brandkbs/searchCrisisByTags?page={1}&size={2}&brand={3}&category={4}
......
src/main/resources/application-prod.properties
......@@ -77,7 +77,7 @@ qbjc.platform.url=https://yuqing.zhiweidata.com/qbjcbackPhoenix/interface/platfo
qbjc.userCenter.url=https://login.zhiweidata.com/plogin/center
qbjc.userCenter.token=AoJ0ooy3HV1EElWnvQw9YTS9b5Y+fmtkbM6DdpPgDO6D/OhNqH4qrJKarzMr
qbjc.channel.application.url=https://yuqing.zhiweidata.com/qbjcbackPhoenix/interface/middleware/channel-application
qbjc.channel.influence=https://yuqing.test.zhiweidata.com/qbjcbackPhoenix/interface/middleware/channel-influence?platform={1}&channel={2}
qbjc.channel.influence=https://yuqing.zhiweidata.com/qbjcbackPhoenix/interface/middleware/channel-influence?platform={1}&channel={2}
#\u5371\u673A\u5E93\u5916\u90E8\u63A5\u53E3
crisis.search.url=https://crisis.zhiweidata.com/app/brandkbs/crisisSearch?page={1}&size={2}&keyword={3}
crisis.searchTags.url=https://crisis.zhiweidata.com/app/brandkbs/searchCrisisByTags?page={1}&size={2}&brand={3}&category={4}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment