Skip to content

B
brandkbs2
Commit 1b03ec32 by shenjunjie
Options

Merge branch 'feature' into 'release' 

Feature

See merge request !567
parents 56e10078 939698ae
Showing with 624 additions and 85 deletions
src/main/java/com/zhiwei/brandkbs2/common/GlobalPojo.java
......@@ -2,6 +2,7 @@ package com.zhiwei.brandkbs2.common;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.zhiwei.brandkbs2.dao.ChannelRecordRefreshTaskDao;
import com.zhiwei.brandkbs2.pojo.*;
import com.zhiwei.brandkbs2.service.ChannelService;
import com.zhiwei.brandkbs2.service.SystemInfoService;
......@@ -10,11 +11,13 @@ import com.zhiwei.middleware.automaticmark.graphs.Graphs;
import com.zhiwei.qbjc.bean.pojo.common.MessagePlatform;
import com.zhiwei.qbjc.bean.pojo.common.Tag;
import com.zhiwei.brandkbs2.dao.HighlightWordDao;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
......@@ -45,6 +48,9 @@ public class GlobalPojo {
@Resource(name = "highlightWordDao")
private HighlightWordDao highlightWordDao;
@Resource(name = "channelRecordRefreshTaskDao")
private ChannelRecordRefreshTaskDao channelRecordRefreshTaskDao;
@Resource(name = "channelServiceImpl")
private ChannelService channelService;
......@@ -143,6 +149,7 @@ public class GlobalPojo {
COMMON_SENSITIVE_CHANNEL = systemInfoService.getCommonSensitiveChannel();
BYTEDANCE_CHANNEL_INFLUENCE = systemInfoService.getByteDanceChannelInfluence();
updateHighlightGraphs();
updateIncompleteChannelRecordTask();
log.info("{}-获取PLATFORMS-size:{},TAGS-size:{},LINKED_GROUP_ID_TAGS:{},CHANNEL_TAGS:{},MEDIA_TYPE:{},PROJECT_MAP:{},YUQING-PROJECTS-size:{},PROJECT_EMOTION_CHANNEL_DATA-size:{},PROJECT_SENSITIVE_CHANNEL-size:{}, COMMON_SENSITIVE_CHANNEL-size:{},BYTEDANCE_CHANNEL_INFLUENCE-size:{}", logMsg, PLATFORMS.size(), TAGS.size(),
LINKED_GROUP_ID_TAGS.size(), CHANNEL_TAGS.size(), MEDIA_TYPE.size(), PROJECT_MAP.size(), YU_QING_PROJECTS.size(), PROJECT_EMOTION_CHANNEL_DATA.size(), PROJECT_SENSITIVE_CHANNEL.size(), COMMON_SENSITIVE_CHANNEL.size(), BYTEDANCE_CHANNEL_INFLUENCE.size());
} catch (Exception e) {
......@@ -150,6 +157,19 @@ public class GlobalPojo {
}
}
private void updateIncompleteChannelRecordTask(){
List<ChannelRecordRefreshTask> tasks = channelRecordRefreshTaskDao.findList(new Query(Criteria.where("status").is(ChannelRecordRefreshTask.TaskStatus.UPDATING.getStatus())));
if (CollectionUtils.isEmpty(tasks)){
return;
}
for (ChannelRecordRefreshTask task : tasks) {
Update update = new Update();
update.set("status", ChannelRecordRefreshTask.TaskStatus.NOT_START.getStatus());
update.set("uTime", System.currentTimeMillis());
channelRecordRefreshTaskDao.updateOneByIdWithField(task.getId(), update);
}
}
private void updateHighlightGraphs() {
PROJECT_MAP.forEach((key, project) -> {
String id = project.getId();
......
src/main/java/com/zhiwei/brandkbs2/controller/InterfaceController.java
......@@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSONObject;
import com.zhiwei.brandkbs2.model.ResponseResult;
import com.zhiwei.brandkbs2.pojo.external.*;
import com.zhiwei.brandkbs2.pojo.vo.CrisisCaseWarnVO;
import com.zhiwei.brandkbs2.service.ChannelService;
import com.zhiwei.brandkbs2.service.ProjectService;
import com.zhiwei.brandkbs2.service.ProjectWarnService;
import com.zhiwei.brandkbs2.util.TextUtil;
......@@ -32,6 +33,9 @@ public class InterfaceController {
@Resource(name = "projectWarnServiceImpl")
private ProjectWarnService projectWarnService;
@Resource(name = "channelServiceImpl")
private ChannelService channelService;
@Autowired
TextUtil textUtil;
......@@ -125,4 +129,16 @@ public class InterfaceController {
public ResponseResult getUserProjectList(String userId) {
return projectService.getUserProject(userId);
}
@ApiOperation("数据回传标注库品见数据更新-生成刷新任务")
@PostMapping("/channel/refresh-task")
public ResponseResult createChannelRecordRefreshTask(@RequestBody JSONObject json) {
long startTime = json.getLongValue("startTime");
long endTime = json.getLongValue("endTime");
List<JSONObject> brandkbsInfos = json.getJSONArray("brandkbsInfo").toJavaList(JSONObject.class);
String mgroupId = json.getString("mgroupId");
String submitter = json.getString("submitter");
channelService.createChannelRecordRefreshTask(startTime, endTime, brandkbsInfos, mgroupId, submitter);
return ResponseResult.success();
}
}
src/main/java/com/zhiwei/brandkbs2/controller/admin/UserController.java
package com.zhiwei.brandkbs2.controller.admin;
import com.alibaba.fastjson.JSONObject;
import com.zhiwei.brandkbs2.auth.Auth;
import com.zhiwei.brandkbs2.controller.BaseController;
import com.zhiwei.brandkbs2.enmus.RoleEnum;
......@@ -7,10 +8,7 @@ import com.zhiwei.brandkbs2.model.ResponseResult;
import com.zhiwei.brandkbs2.pojo.dto.UserDTO;
import com.zhiwei.brandkbs2.service.UserService;
import com.zhiwei.middleware.auth.pojo.CenterUser;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
......@@ -124,4 +122,46 @@ public class UserController extends BaseController {
return ResponseResult.success();
}
@ApiOperation("分页查询所有用户列表")
@GetMapping("/all-user/list")
@Auth(role = RoleEnum.SUPER_ADMIN)
public ResponseResult findAllUserList(@RequestParam(value = "page", defaultValue = "1") int page,
@RequestParam(value = "size", defaultValue = "10") int size,
@RequestParam(value = "keyword", defaultValue = "") String keyword) {
return ResponseResult.success(userService.findAllUserList(keyword, page, size));
}
@ApiOperation("根据用户名或手机号查询用户项目权限")
@GetMapping("/all-user/find")
@Auth(role = RoleEnum.SUPER_ADMIN)
public ResponseResult getUserByNickNameOrPhone(@RequestParam(value = "nickname", required = false) String nickname,
@RequestParam(value = "phoneNumber", required = false) String phoneNumber) {
return userService.getUserByNickNameOrPhone(nickname, phoneNumber);
}
@ApiOperation("编辑单条项目权限")
@PostMapping("/all-user/single-update")
@Auth(role = RoleEnum.SUPER_ADMIN)
public ResponseResult updateOneUserRoles(@ApiParam(name = "json",
value = "json:{id:用户id,nickname:用户名,roleId:权限,expiredTime:过期时间,exportAmount:舆情导出数量,projectId:权限项目id,key:唯一key原封不动传回来即可}",
required = true) @RequestBody JSONObject json) {
return userService.updateOneUserRoles(json);
}
@ApiOperation("批量编辑项目权限")
@PostMapping("/all-user/batch-update")
@Auth(role = RoleEnum.SUPER_ADMIN)
public ResponseResult updateBatchUserRoles(@ApiParam(name = "json", value = "json:{id:用户id,roles:[{projectId:权限项目id,roleId:权限,exportAmount:舆情导出数量,expiredTime:过期时间},{},...]}",
required = true)
@RequestBody JSONObject json) {
return userService.updateBatchUserRoles(json);
}
@ApiOperation("删除用户权限")
@DeleteMapping("/all-user/delete")
@Auth(role = RoleEnum.SUPER_ADMIN)
public ResponseResult deleteUserRole(@RequestParam(value = "id") String id,
@RequestParam(value = "key") String key) {
return userService.deleteUserRole(id, key);
}
}
\ No newline at end of file
src/main/java/com/zhiwei/brandkbs2/dao/ChannelRecordRefreshTaskDao.java 0 → 100644
package com.zhiwei.brandkbs2.dao;
import com.zhiwei.brandkbs2.pojo.ChannelRecordRefreshTask;
import org.springframework.data.mongodb.core.query.Query;
/**
* @ClassName: ChannelRecordRefreshTaskDao
* @Description 渠道记录更新任务dao
* @author: cjz
* @date: 2024-07-22 11:37
*/
public interface ChannelRecordRefreshTaskDao extends BaseMongoDao<ChannelRecordRefreshTask>{
ChannelRecordRefreshTask findOne(Query query);
}
src/main/java/com/zhiwei/brandkbs2/dao/UserDao.java
package com.zhiwei.brandkbs2.dao;
import com.zhiwei.brandkbs2.pojo.User;
import org.springframework.data.mongodb.core.query.Query;
/**
* @ClassName: UserDao
......@@ -9,5 +10,5 @@ import com.zhiwei.brandkbs2.pojo.User;
* @date: 2022-04-28 18:10
*/
public interface UserDao extends BaseMongoDao<User>{
User findOne(Query query);
}
src/main/java/com/zhiwei/brandkbs2/dao/impl/ChannelRecordRefreshTaskDaoImpl.java 0 → 100644
package com.zhiwei.brandkbs2.dao.impl;
import com.zhiwei.brandkbs2.dao.ChannelRecordRefreshTaskDao;
import com.zhiwei.brandkbs2.pojo.ChannelRecordRefreshTask;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Component;
/**
* @ClassName: ChannelRecordRefreshTaskDaoImpl
* @Description 渠道记录更新任务dao
* @author: cjz
* @date: 2024-07-22 11:37
*/
@Component("channelRecordRefreshTaskDao")
public class ChannelRecordRefreshTaskDaoImpl extends BaseMongoDaoImpl<ChannelRecordRefreshTask> implements ChannelRecordRefreshTaskDao {
private static final String COLLECTION_NAME = "brandkbs_channel_record_refresh_task";
public ChannelRecordRefreshTaskDaoImpl() {
super(COLLECTION_NAME);
}
@Override
public ChannelRecordRefreshTask findOne(Query query) {
return mongoTemplate.findOne(query, ChannelRecordRefreshTask.class, COLLECTION_NAME);
}
}
src/main/java/com/zhiwei/brandkbs2/dao/impl/UserDaoImpl.java
......@@ -2,6 +2,7 @@ package com.zhiwei.brandkbs2.dao.impl;
import com.zhiwei.brandkbs2.dao.UserDao;
import com.zhiwei.brandkbs2.pojo.User;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Component;
/**
......@@ -18,4 +19,9 @@ public class UserDaoImpl extends BaseMongoDaoImpl<User> implements UserDao {
public UserDaoImpl() {
super(COLLECTION_NAME);
}
@Override
public User findOne(Query query) {
return mongoTemplate.findOne(query, User.class, COLLECTION_NAME);
}
}
src/main/java/com/zhiwei/brandkbs2/es/ChannelEsDao.java
......@@ -28,6 +28,7 @@ import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
......@@ -114,49 +115,85 @@ public class ChannelEsDao extends EsClientDao {
}
}
public void removeChannelRecordById(List<JSONObject> list) {
List<String> idList = list.stream().map(json -> json.getString("id")).collect(Collectors.toList());
String index = getChannelRecordIndexes().get(0);
AtomicInteger count = new AtomicInteger(idList.size());
idList.forEach(id -> {
AtomicInteger searchCount = new AtomicInteger();
AtomicInteger updateCount = new AtomicInteger();
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
searchSourceBuilder.query(QueryBuilders.termQuery("record.articles.id", id));
searchSourceBuilder.size(10000);
SearchResponse response = retryTemplate.execute(context -> {
try {
return channelEsClient.search(new SearchRequest().indices(index).source(searchSourceBuilder), RequestOptions.DEFAULT);
} catch (IOException e) {
log.info("resetChannelRecordById:{},查询失败,尝试重试第{}次-", id, context.getRetryCount() + 1, e);
return null;
public Integer removeChannelRecordBatch(List<JSONObject> list) {
AtomicInteger update = new AtomicInteger();
AtomicInteger total = new AtomicInteger(list.size());
List<CompletableFuture<Boolean>> futures = new ArrayList<>(list.size());
list.forEach(json -> futures.add(CompletableFuture.supplyAsync(() -> removeChannelRecordByCondition(json, total), executor)));
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).whenComplete((r, e) -> {
futures.forEach(f -> {
if (f.join()) {
update.incrementAndGet();
}
});
if (null != response) {
searchCount.addAndGet(response.getHits().getHits().length);
for (SearchHit hit : response.getHits().getHits()) {
JSONObject record = new JSONObject((Map<String, Object>) hit.getSourceAsMap().get("record"));
Long lastTime = record.getLong("last_time");
List<ChannelIndex.Article> articles = record.getJSONArray("articles").toJavaList(JSONObject.class).stream().map(json -> {
// 移除id相同的数据
if (json.getString("id").equals(id)) {
return null;
}
return ChannelIndex.Article.fromRecordMap(json);
}).filter(Objects::nonNull).collect(Collectors.toList());
}).join();
// list.forEach(json -> {
// removeChannelRecordByCondition(json, total);
// });
return update.get();
}
private boolean removeChannelRecordByCondition(JSONObject json, AtomicInteger total) {
String index = getChannelRecordIndexes().get(0);
String id = json.getString("id");
AtomicInteger searchCount = new AtomicInteger();
AtomicInteger updateCount = new AtomicInteger();
boolean update = false;
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
boolQuery.must(QueryBuilders.termQuery("record.articles.id", id));
if (json.containsKey("projectId")) {
boolQuery.must(QueryBuilders.termQuery("project_id.keyword", json.getString("projectId")));
}
if (json.containsKey("contendIds")) {
BoolQueryBuilder contendBoolQuery = QueryBuilders.boolQuery();
json.getJSONArray("contendIds").toJavaList(String.class).forEach(contendId -> {
contendBoolQuery.should(QueryBuilders.termQuery("contend_id.keyword", contendId));
});
boolQuery.must(contendBoolQuery);
}
searchSourceBuilder.query(boolQuery);
searchSourceBuilder.size(1000);
SearchResponse response = retryTemplate.execute(context -> {
try {
return channelEsClient.search(new SearchRequest().indices(index).source(searchSourceBuilder), RequestOptions.DEFAULT);
} catch (IOException e) {
log.info("resetChannelRecordById:{},查询失败,尝试重试第{}次-", id, context.getRetryCount() + 1, e);
return null;
}
});
if (null != response) {
searchCount.addAndGet(response.getHits().getHits().length);
for (SearchHit hit : response.getHits().getHits()) {
JSONObject record = new JSONObject((Map<String, Object>) hit.getSourceAsMap().get("record"));
Long lastTime = record.getLong("last_time");
AtomicBoolean valid = new AtomicBoolean(false);
List<ChannelIndex.Article> articles = record.getJSONArray("articles").toJavaList(JSONObject.class).stream().map(article -> {
// 移除id相同的数据
if (article.getString("id").equals(id)) {
valid.set(true);
return null;
}
return ChannelIndex.Article.fromRecordMap(article);
}).filter(Objects::nonNull).collect(Collectors.toList());
// 有效更新
if (valid.get()) {
Map<String, Object> updateMap = new HashMap<>();
updateMap.put("record", new ChannelIndex.Record(lastTime, articles).toEsMap());
updateMap.put("article_count", articles.size());
try {
channelEsClient.update(new UpdateRequest().index(index).id(hit.getId()).doc(updateMap), RequestOptions.DEFAULT);
updateCount.getAndIncrement();
} catch (IOException e) {
update = true;
} catch (Exception e) {
log.info("resetChannelRecordById:{},更新失败", id, e);
}
}
}
log.info("resetChannelRecordById:{},共查询到{}条,更新{}条,剩余id数:{}", id, searchCount.get(), updateCount.get(), count.decrementAndGet());
});
}
log.info("resetChannelRecordById:{},本次查询到{}条,更新{}条,剩余id数:{}", id, searchCount.get(), updateCount.get(), total.decrementAndGet());
return update;
}
private BulkResponse upsertChannelRecordLimit(List<ChannelRecord> records, String index, int limit) {
......
src/main/java/com/zhiwei/brandkbs2/es/EsClientDao.java
......@@ -142,6 +142,73 @@ public class EsClientDao {
return res;
}
public List<JSONObject> searchRecordUnrelated(long startTime, long endTime, String mgroupId, List<JSONObject> brandkbsInfo) {
List<JSONObject> res = new ArrayList<>();
List<JSONObject> dataList = new ArrayList<>();
List<Long[]> cutTimes = Tools.cutTimeRange(startTime, endTime, ONE_HOUR * 24);
List<CompletableFuture<List<JSONObject>>> futures = new ArrayList<>(cutTimes.size());
cutTimes.forEach(times -> futures.add(CompletableFuture.supplyAsync(() -> searchRecordUnrelatedSingle(times[0], times[1], mgroupId), executor)));
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).whenComplete((r, e) -> {
futures.forEach(f -> dataList.addAll(f.join()));
}).join();
// 找到该mgroup下关联的项目品牌及竞品
Map<String, List<String>> relatedMap = brandkbsInfo.stream().collect(Collectors.toMap(json -> json.getString("projectId"), json -> {
List<String> brandIds = new ArrayList<>();
String brandId = Objects.equals(json.getString("brandId"), json.getString("projectId")) ? Constant.PRIMARY_CONTEND_ID : json.getString("brandId");
brandIds.add(brandId);
return brandIds;
}, (List<String> ls1, List<String> ls2) -> {
ls1.addAll(ls2);
return ls1;
}));
// 筛选没有关联关系需要重置的部分
dataList.forEach(data -> {
String id = data.getString("id");
List<Map<String, Object>> brandkbsCacheMaps = (List<Map<String, Object>>) data.get("brandkbs_cache_maps");
relatedMap.forEach((project, contends) -> {
List<String> unrelatedContends = new ArrayList<>();
contends.forEach(contend -> {
boolean hit = false;
for (Map<String, Object> map : brandkbsCacheMaps) {
String key = map.get("key") + "";
if (key.equals(project + "_" + contend)) {
hit = true;
break;
}
}
// 不满足关系则添加(说明需要重置)
if (!hit) {
unrelatedContends.add(contend);
}
});
if (!unrelatedContends.isEmpty()) {
JSONObject json = new JSONObject();
json.put("id", id);
json.put("projectId", project);
json.put("contendIds", unrelatedContends);
res.add(json);
}
});
});
return res;
}
public List<JSONObject> searchRecordUnrelatedSingle(long startTime, long endTime, String mgroupId) {
List<JSONObject> res = new ArrayList<>();
try {
BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
// TODO 是否需要设立标注字段限制(查询范围更小更精准)
boolQuery.must(QueryBuilders.termQuery("mgroup_id.keyword", mgroupId));
boolQuery.must(QueryBuilders.rangeQuery("time").gte(startTime).lte(endTime));
List<JSONObject> results = searchScroll(boolQuery, 10000, new String[]{"id", "brandkbs_cache_maps"});
res.addAll(results);
} catch (IOException e) {
log.error("startTime:{},endTime:{},mgroupId:{},searchRecordUnrelatedSingle-", startTime, endTime, mgroupId, e);
}
log.info("startTime:{},endTime:{},size:{}", DF.format(startTime), DF.format(endTime), res.size());
return res;
}
//
// /**
// * 搜索符合事件数据
......@@ -200,6 +267,24 @@ public class EsClientDao {
return retryTemplate.execute(context -> searchScroll(sourceBuilder));
}
private List<JSONObject> searchRecordEs(long startTime, long endTime){
List<JSONObject> results = new ArrayList<>();
if (endTime - startTime <= 10 * 60 * 1000L){
log.error("searchRecord分段查询至最小分割仍未满足-时间范围：{}-{}", startTime, endTime);
return Collections.emptyList();
}
try {
QueryBuilder queryBuilder = QueryBuilders.rangeQuery("mtime").gte(startTime).lt(endTime);
results = searchScroll(queryBuilder, 10000, CHANNEL_RECORD_FETCH_SOURCE);
}catch (Exception e){
// 时间分段查询
long midTime = startTime + (endTime - startTime) / 2;
results.addAll(searchRecordEs(startTime, midTime));
results.addAll(searchRecordEs(midTime, endTime));
}
return results;
}
private Pair<Long[], Map<ChannelIndex, ChannelIndex.Record>> searchRecord(long startTime, long endTime) {
Map<ChannelIndex, ChannelIndex.Record> res = new HashMap<>();
try {
......@@ -208,18 +293,18 @@ public class EsClientDao {
QueryBuilder queryBuilder = QueryBuilders.rangeQuery("mtime").gte(startTime).lt(endTime);
results = searchScroll(queryBuilder, 10000, CHANNEL_RECORD_FETCH_SOURCE);
}catch (Exception e){
log.error("searchRecord-搜索阶段出错-时间分段重试开始", e);
log.info("searchRecord-搜索阶段出错-时间分段重试开始", e);
// 时间分段查询
long midTime = startTime + (endTime - startTime) / 2;
try {
QueryBuilder queryBuilder1 = QueryBuilders.rangeQuery("mtime").gte(startTime).lt(midTime);
results.addAll(searchScroll(queryBuilder1, 10000, CHANNEL_RECORD_FETCH_SOURCE));
results.addAll(searchScroll(queryBuilder1, 1000, CHANNEL_RECORD_FETCH_SOURCE));
}catch (Exception e1){
log.error("searchRecord分段查询出错,时间范围:{}-{}", startTime, midTime, e1);
}
try {
QueryBuilder queryBuilder2 = QueryBuilders.rangeQuery("mtime").gte(midTime).lt(endTime);
results.addAll(searchScroll(queryBuilder2, 10000, CHANNEL_RECORD_FETCH_SOURCE));
results.addAll(searchScroll(queryBuilder2, 1000, CHANNEL_RECORD_FETCH_SOURCE));
}catch (Exception e2){
log.error("searchRecord分段查询出错,时间范围:{}-{}", midTime, endTime, e2);
}
......
src/main/java/com/zhiwei/brandkbs2/pojo/ChannelRecordRefreshTask.java 0 → 100644
package com.zhiwei.brandkbs2.pojo;
import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.Setter;
import java.util.List;
/**
* @ClassName: ChannelRecordRefreshTask
* @Description 渠道记录更新任务
* @author: cjz
* @date: 2024-07-22 10:37
*/
@Getter
@Setter
@AllArgsConstructor
public class ChannelRecordRefreshTask extends AbstractBaseMongo{
/**
* 受影响数据最小时间
*/
private Long startTime;
/**
* 受影响数据最大时间
*/
private Long endTime;
/**
* 品见信息：[{projectId:,brandId:}]
*/
List<JSONObject> brandkbsInfo;
/**
* mgroupId
*/
private String mgroupId;
/**
* 任务状态
*/
private String status;
/**
* 创建者
*/
private String submitter;
/**
* 创建时间
*/
private Long cTime;
/**
* 修改时间
*/
private Long uTime;
@Getter
public enum TaskStatus {
NOT_START("未开始"), UPDATING("进行中"), ERROR("出错"), COMPLETED("已完成");
private String status;
TaskStatus(String status) {
this.status = status;
}
}
}
src/main/java/com/zhiwei/brandkbs2/service/ChannelService.java
......@@ -333,4 +333,13 @@ public interface ChannelService {
* @return
*/
JSONObject getImportantChannelListDetail(String type, String name, String keyword);
/**
* 生成渠道记录更新任务
* @param startTime
* @param endTime
* @param yuqingProjectId
* @param submitter
*/
void createChannelRecordRefreshTask(Long startTime, Long endTime, List<JSONObject> brandkbsInfos, String yuqingProjectId, String submitter);
}
src/main/java/com/zhiwei/brandkbs2/service/TaskService.java
......@@ -64,12 +64,12 @@ public interface TaskService{
void generateDailyReport();
/**
* 每月互动量更新 时间范围近六个月
* 每日互动量更新 时间范围近10天
*/
void monthlyCustomXhsInteractionUpdate();
void dailyCustomXhsInteractionUpdate();
/**
* 每日互动量更新 时间范围近10天
* 定时拉取并进行渠道库更新任务
*/
void dailyCustomXhsInteractionUpdate();
void refreshChannelRecord();
}
src/main/java/com/zhiwei/brandkbs2/service/UserService.java
......@@ -42,7 +42,7 @@ public interface UserService {
UserInfo queryUserInfo(String userId, String pid);
/**
* 分页查询所有用户
* 分页查询所有用户（当前项目的用户）
*
* @param page 页码
* @param size 大小
......@@ -55,14 +55,14 @@ public interface UserService {
PageVO<JSONObject> findUserList(int page, int size, String keyword, String pid, int role, String sorter);
/**
* 添加用户
* 添加用户（当前项目的用户）
*
* @param userDTO
*/
ResponseResult addUser(UserDTO userDTO);
/**
* 删除用户
* 删除用户（当前项目的用户）
*
* @param userId
* @param pid
......@@ -70,7 +70,7 @@ public interface UserService {
void deleteUser(String userId, String pid);
/**
* 编辑用户
* 编辑用户（当前项目的用户）
*
* @param userDTO
*/
......@@ -135,4 +135,40 @@ public interface UserService {
*/
ResponseResult mobileCutLogin(String userId,String projectId);
/**
* 项目管理员-所有项目用户权限列表
* @param page
* @param pageSize
* @return
*/
PageVO<JSONObject> findAllUserList(String keyword, int page, int pageSize);
/**
* 项目管理员-所有项目用户权限列表-通过用户名或手机号查询品见用户项目权限
* @param nickName
* @param phoneNumber
* @return
*/
ResponseResult getUserByNickNameOrPhone(String nickName, String phoneNumber);
/**
* 项目管理员-所有项目用户权限列表-编辑单条项目权限
*
* @param json
*/
ResponseResult updateOneUserRoles(JSONObject json);
/**
* 项目管理员-所有项目用户权限列表-批量编辑项目权限
*
* @param json
*/
ResponseResult updateBatchUserRoles(JSONObject json);
/**
* 项目管理员-所有项目用户列表-删除用户权限
* @param id
* @return
*/
ResponseResult deleteUserRole(String id, String key);
}
src/main/java/com/zhiwei/brandkbs2/service/impl/ChannelServiceImpl.java
......@@ -127,6 +127,9 @@ public class ChannelServiceImpl implements ChannelService {
@Resource(name = "importantChannelOverviewDao")
private ImportantChannelOverviewDao importantChannelOverviewDao;
@Resource(name = "channelRecordRefreshTaskDao")
private ChannelRecordRefreshTaskDao channelRecordRefreshTaskDao;
@Resource(name = "mongoUtil")
MongoUtil mongoUtil;
......@@ -820,6 +823,12 @@ public class ChannelServiceImpl implements ChannelService {
}
}
@Override
public void createChannelRecordRefreshTask(Long startTime, Long endTime, List<JSONObject> brandkbsInfos, String yuqingProjectId, String submitter) {
channelRecordRefreshTaskDao.insertOne(new ChannelRecordRefreshTask(startTime, endTime, brandkbsInfos, yuqingProjectId,
ChannelRecordRefreshTask.TaskStatus.NOT_START.getStatus(), submitter, System.currentTimeMillis(), System.currentTimeMillis()));
}
private JSONObject provinceDataProcess(List<ImportantChannel> sortList){
List<JSONObject> resList = new ArrayList<>();
JSONObject jsonObject = new JSONObject(new LinkedHashMap<>());
......
src/main/java/com/zhiwei/brandkbs2/service/impl/TaskServiceImpl.java
......@@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSONObject;
import com.zhiwei.brandkbs2.auth.UserThreadLocal;
import com.zhiwei.brandkbs2.common.GenericAttribute;
import com.zhiwei.brandkbs2.common.GlobalPojo;
import com.zhiwei.brandkbs2.common.RedisKeyPrefix;
import com.zhiwei.brandkbs2.config.Constant;
import com.zhiwei.brandkbs2.dao.*;
import com.zhiwei.brandkbs2.enmus.ReportTypeEnum;
......@@ -15,6 +16,7 @@ import com.zhiwei.brandkbs2.listener.ApplicationProjectListener;
import com.zhiwei.brandkbs2.model.CommonCodeEnum;
import com.zhiwei.brandkbs2.pojo.*;
import com.zhiwei.brandkbs2.service.*;
import com.zhiwei.brandkbs2.util.RedisUtil;
import com.zhiwei.brandkbs2.util.Tools;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.ListUtils;
......@@ -24,6 +26,7 @@ import org.apache.logging.log4j.Logger;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
......@@ -83,6 +86,9 @@ public class TaskServiceImpl implements TaskService {
@Resource(name = "customInteractionUpdateRecordDao")
private CustomInteractionUpdateRecordDao customInteractionUpdateRecordDao;
@Resource(name = "channelRecordRefreshTaskDao")
private ChannelRecordRefreshTaskDao channelRecordRefreshTaskDao;
@Resource(name = "brandkbsTaskServiceImpl")
BrandkbsTaskService brandkbsTaskService;
......@@ -116,6 +122,9 @@ public class TaskServiceImpl implements TaskService {
@Resource(name = "cacheServiceExecutor")
ThreadPoolTaskExecutor cacheServiceExecutor;
@Resource(name = "redisUtil")
RedisUtil redisUtil;
@Override
public void messageFlowCount(int day) {
List<Pair<Long[], Map<ChannelIndex, ChannelIndex.Record>>> rangeTimeRecords = esClientDao.searchRecordRecentDay(day);
......@@ -278,13 +287,12 @@ public class TaskServiceImpl implements TaskService {
* 本地测试-刷新历史渠道记录用
* 不更新渠道表
*/
@Deprecated
public void messageFlowCount2(long startTime,long endTime,String mgroup) {
// 找到项目的空标签数据
List<JSONObject> list = esClientDao.searchRecordEmpty(startTime, endTime, mgroup);
public void messageFlowCountRefresh(long startTime, long endTime, String mgroupId, List<JSONObject> brandkbsInfo) {
// 找到mgroup下符合条件的数据
List<JSONObject> list = esClientDao.searchRecordUnrelated(startTime, endTime, mgroupId, brandkbsInfo);
// 移除对应数据记录
channelEsDao.removeChannelRecordById(list);
log.info("刷新历史渠道记录-统计结束");
Integer updateCount = channelEsDao.removeChannelRecordBatch(list);
log.info("刷新历史渠道记录-统计结束-受影响结果{}条,实际更新{}条", list.size(), updateCount);
}
@Override
......@@ -434,25 +442,6 @@ public class TaskServiceImpl implements TaskService {
}
@Override
public void monthlyCustomXhsInteractionUpdate() {
// // 近六个月
// long endTime = System.currentTimeMillis();
// long startTime = endTime - Constant.ONE_MONTH * 6;
// // 去除近10天区间，近10天为1天/次
// endTime = endTime - 10 * Constant.ONE_DAY;
// List<Project> projects = GlobalPojo.PROJECT_MAP.values().stream()
// .filter(project -> CollectionUtils.isNotEmpty(project.getModuleShowList()) && project.getModuleShowList().contains("xiaohongshu")).collect(Collectors.toList());
// for (Project project : projects) {
// try {
// List<Pair<String, Map<String, Object>>> res = customXhsInteractionUpdate(project.getId(), startTime, endTime, Collections.singletonList("6433c2251701316728003be4"));
// esClientDao.batchUpdate(res);
// }catch (Exception e){
// log.error("项目：{}，定制化舆情分析-互动量更新出错-时间范围：{}-{}", project.getProjectName(), startTime, endTime);
// }
// }
}
@Override
public void dailyCustomXhsInteractionUpdate() {
// 近10天
long endTime = System.currentTimeMillis();
......@@ -471,6 +460,39 @@ public class TaskServiceImpl implements TaskService {
monthlyCustomXhsInteractionUpdateNew(projects);
}
@Override
public void refreshChannelRecord() {
ChannelRecordRefreshTask task = channelRecordRefreshTaskDao.findOne(new Query(Criteria.where("status")
.is(ChannelRecordRefreshTask.TaskStatus.NOT_START.getStatus())).with(Sort.by(Sort.Order.desc("cTime"))));
if (Objects.isNull(task)){
return;
}
updateRefreshTask(task.getId(), ChannelRecordRefreshTask.TaskStatus.UPDATING.getStatus());
try {
messageFlowCountRefresh(task.getStartTime(), task.getEndTime(), task.getMgroupId(), task.getBrandkbsInfo());
}catch (Exception e){
updateRefreshTask(task.getId(), ChannelRecordRefreshTask.TaskStatus.ERROR.getStatus());
log.error("更新渠道库记录异常-taskId:{}-", task.getId(), e);
return;
}
updateRefreshTask(task.getId(), ChannelRecordRefreshTask.TaskStatus.COMPLETED.getStatus());
// 删除缓存
task.getBrandkbsInfo().forEach(info -> {
String projectId = info.getString("projectId");
String brandId = Objects.equals(info.getString("brandId"), projectId) ? Constant.PRIMARY_CONTEND_ID : info.getString("brandId");
Set<String> keys = redisUtil.keys(RedisKeyPrefix.CHANNEL_RECORD_LIST + Tools.concat(projectId, brandId) + "*");
redisUtil.delete(keys);
});
log.info("更新渠道库记录完成-taskId:{}", task.getId());
}
private void updateRefreshTask(String id, String status){
Update update = new Update();
update.set("status", status);
update.set("uTime", System.currentTimeMillis());
channelRecordRefreshTaskDao.updateOneByIdWithField(id, update);
}
private void monthlyCustomXhsInteractionUpdateNew(List<Project> projects) {
// 近六个月
long endTime = System.currentTimeMillis();
......
src/main/java/com/zhiwei/brandkbs2/service/impl/UserServiceImpl.java
......@@ -3,6 +3,7 @@ package com.zhiwei.brandkbs2.service.impl;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhiwei.brandkbs2.auth.UserThreadLocal;
import com.zhiwei.brandkbs2.common.GlobalPojo;
import com.zhiwei.brandkbs2.config.Constant;
import com.zhiwei.brandkbs2.dao.ProjectDao;
import com.zhiwei.brandkbs2.dao.UserDao;
......@@ -13,6 +14,7 @@ import com.zhiwei.brandkbs2.enmus.response.LoginCodeEnum;
import com.zhiwei.brandkbs2.exception.ExceptionCast;
import com.zhiwei.brandkbs2.model.CommonCodeEnum;
import com.zhiwei.brandkbs2.model.ResponseResult;
import com.zhiwei.brandkbs2.pojo.Project;
import com.zhiwei.brandkbs2.pojo.User;
import com.zhiwei.brandkbs2.pojo.UserInfo;
import com.zhiwei.brandkbs2.pojo.UserRole;
......@@ -418,6 +420,154 @@ public class UserServiceImpl implements UserService {
}
@Override
public PageVO<JSONObject> findAllUserList(String keyword, int page, int pageSize) {
Query query = new Query(Criteria.where("superAdmin").is(false));
userDao.addKeywordFuzz(query, keyword, "nickname");
userDao.addSort(query, "{\"cTime\":\"descend\"}");
// roles总量
List<User> list = userDao.findList(new Query(Criteria.where("superAdmin").is(false)));
long count = list.stream().map(User::getRoles).filter(Objects::nonNull).mapToLong(Collection::size).sum();
List<User> userList = userDao.findList(query);
Map<User, List<UserRole>> userRolesMap = userList.stream().collect(Collectors.toMap(o -> o, User::getRoles));
List<JSONObject> resList = new ArrayList<>();
for (Map.Entry<User, List<UserRole>> entry : userRolesMap.entrySet()) {
User user = entry.getKey();
List<UserRole> roles = entry.getValue();
for (UserRole role : roles) {
String projectId = role.getProjectId();
// 已终止/已删除的项目不返回
Project project = GlobalPojo.PROJECT_MAP.get(projectId);
project = Objects.isNull(project) ? projectDao.findOneById(projectId) : project;
if (Objects.isNull(project) || !project.isStart()){
continue;
}
JSONObject jsonObject = new JSONObject();
jsonObject.put("projectId", projectId);
jsonObject.put("projectName", project.getProjectName());
jsonObject.put("id", user.getId());
jsonObject.put("nickname", user.getNickname());
jsonObject.put("username", user.getUsername());
jsonObject.put("cTime", user.getCTime());
jsonObject.put("phoneNumber", user.getPhoneNumber());
jsonObject.put("submitter", user.getSubmitter());
jsonObject.put("roleId", role.getRoleId());
jsonObject.put("expiredTime", role.getExpiredTime());
jsonObject.put("exportAmount", role.getExportAmount());
jsonObject.put("key", role.getKey());
resList.add(jsonObject);
}
}
return PageVO.createPageVo(count, page, pageSize, resList.stream().skip((long) (page - 1) * pageSize).limit(pageSize).collect(Collectors.toList()));
}
@Override
public ResponseResult updateOneUserRoles(JSONObject json) {
String id = json.getString("id");
String nickname = json.getString("nickname");
int roleId = json.getIntValue("roleId");
long expiredTime = json.getLongValue("expiredTime");
int exportAmount = json.getIntValue("exportAmount");
String projectId = json.getString("projectId");
String key = json.getString("key");
User user = userDao.findOneById(id);
List<UserRole> roles = user.getRoles();
if (CollectionUtils.isNotEmpty(roles)){
List<String> existProject = roles.stream().filter(r -> !Objects.equals(r.getKey(), key)).map(UserRole::getProjectId).collect(Collectors.toList());
if (existProject.contains(projectId)){
return ResponseResult.failure("用户已拥有此项目的权限");
}
}
user.getRoles().stream().filter(userRoles -> userRoles.getKey().equals(key)).findAny().ifPresent(userRole -> {
// 更新原userRole
userRole.setRoleId(roleId);
userRole.setProjectId(projectId);
userRole.setKey(Tools.concat(projectId, roleId));
// 权限用户以上则清空过期时间
if (roleId < RoleEnum.CUSTOMER.getState()) {
userRole.setExpiredTime(null);
} else {
OptionalLong.of(expiredTime).ifPresent(userRole::setExpiredTime);
}
OptionalInt.of(exportAmount).ifPresent(userRole::setExportAmount);
userDao.updateOneByIdWithField(id, new Update().set("roles", roles));
});
Optional.of(nickname).ifPresent(nickName -> userDao.updateOneByIdWithField(id, new Update().set("nickname", nickName)));
return ResponseResult.success();
}
@Override
public ResponseResult getUserByNickNameOrPhone(String nickName, String phoneNumber) {
Query query = new Query();
if (Objects.isNull(nickName) && Objects.isNull(phoneNumber)){
return ResponseResult.failure("用户名和手机号为空");
}
if (Objects.nonNull(nickName)){
query.addCriteria(Criteria.where("nickname").is(nickName));
}
if (Objects.nonNull(phoneNumber)){
query.addCriteria(Criteria.where("phoneNumber").is(Long.parseLong(phoneNumber)));
}
query.addCriteria(Criteria.where("superAdmin").is(false));
User user = userDao.findOne(query);
if (Objects.isNull(user)){
return ResponseResult.failure("不存在此项目用户");
}
JSONObject jsonObject = new JSONObject();
jsonObject.put("id", user.getId());
jsonObject.put("nickname", user.getNickname());
jsonObject.put("phoneNumber", user.getPhoneNumber());
List<JSONObject> roles = user.getRoles().stream().map(role -> {
JSONObject json = new JSONObject();
String projectId = role.getProjectId();
// 已终止/已删除的项目不返回
Project project = GlobalPojo.PROJECT_MAP.get(projectId);
project = Objects.isNull(project) ? projectDao.findOneById(projectId) : project;
if (Objects.isNull(project) || !project.isStart()){
return null;
}
json.put("projectName", project.getProjectName());
json.put("key", role.getKey());
json.put("projectId", projectId);
json.put("roleId", role.getRoleId());
json.put("exportAmount", role.getExportAmount());
json.put("expiredTime", role.getExpiredTime());
return json;
}).filter(Objects::nonNull).collect(Collectors.toList());
jsonObject.put("roles", roles);
return ResponseResult.success(jsonObject);
}
@Override
public ResponseResult updateBatchUserRoles(JSONObject json) {
String id = json.getString("id");
User user = userDao.findOneById(id);
List<UserRole> newRoles = json.getJSONArray("roles").toJavaList(UserRole.class);
List<String> keys = newRoles.stream().map(UserRole::getKey).collect(Collectors.toList());
// 无需被修改的权限列表
List<UserRole> oldRoles = user.getRoles().stream().filter(role -> !keys.contains(role.getKey())).collect(Collectors.toList());
for (UserRole role : newRoles) {
if (Objects.isNull(role.getExportAmount())) {
role.setExportAmount(1000);
}
role.setKey(Tools.concat(role.getProjectId(), role.getRoleId()));
}
oldRoles.addAll(newRoles);
userDao.updateOneByIdWithField(id, new Update().set("roles", oldRoles));
return ResponseResult.success();
}
@Override
public ResponseResult deleteUserRole(String id, String key) {
User user = userDao.findOneById(id);
List<UserRole> roles = user.getRoles().stream().filter(userRole -> !userRole.getKey().equals(key)).collect(Collectors.toList());
Update update = new Update();
update.set("roles", roles);
update.set("submitter", UserThreadLocal.getNickname());
userDao.updateOneByIdWithField(id, update);
return ResponseResult.success();
}
@Override
public Map<String, Object> getLoginInfo() {
// String userId = UserThreadLocal.getUserId();
// String projectId = UserThreadLocal.getProjectId();
......
src/main/java/com/zhiwei/brandkbs2/task/ControlCenter.java
......@@ -119,19 +119,6 @@ public class ControlCenter {
}
}
// @Async("scheduledExecutor")
// @Scheduled(cron = "0 30 3 1 * ?")
// public void monthlyCustomXhsInteractionUpdate() {
// log.info("每月互动量更新-启动");
// try {
// taskService.monthlyCustomXhsInteractionUpdate();
// } catch (Exception e) {
// log.error("每月互动量更新-出错", e);
// } finally {
// log.info("每月互动量更新-结束");
// }
// }
@Async("scheduledExecutor")
@Scheduled(cron = "0 0 5 * * ?")
public void dailyCustomXhsInteractionUpdate() {
......@@ -144,4 +131,17 @@ public class ControlCenter {
log.info("每日互动量更新-结束");
}
}
@Async("scheduledExecutor")
@Scheduled(cron = "0 0/10 * * * ?")
public void refreshChannelRecord() {
log.info("每十分钟拉取并进行渠道库更新任务-启动");
try {
taskService.refreshChannelRecord();
} catch (Exception e) {
log.error("每十分钟拉取并进行渠道库更新任务-出错", e);
} finally {
log.info("每十分钟拉取并进行渠道库更新任务-结束");
}
}
}
src/main/java/com/zhiwei/brandkbs2/util/RedisUtil.java
......@@ -6,6 +6,7 @@ import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.util.Collection;
import java.util.List;
import java.util.Objects;
import java.util.Set;
......@@ -186,4 +187,8 @@ public class RedisUtil {
public void remove(String key) {
setExpire(key, "-1", 1, TimeUnit.SECONDS);
}
public void delete(Collection<String> key) {
stringRedisTemplate.delete(key);
}
}
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