mirror of
https://github.com/cool-team-official/cool-admin-midway-packages.git
synced 2026-03-13 15:21:08 +00:00
616 lines
13 KiB
TypeScript
616 lines
13 KiB
TypeScript
import { CoolEventManager } from '@cool-midway/core';
|
||
import { Client } from '@elastic/elasticsearch';
|
||
import { WaitForActiveShards } from '@elastic/elasticsearch/lib/api/types';
|
||
import { Inject, Logger } from '@midwayjs/decorator';
|
||
import { ILogger } from '@midwayjs/logger';
|
||
import { EsConfig } from '.';
|
||
|
||
/**
|
||
* Es索引基类
|
||
*/
|
||
export class BaseEsIndex {
|
||
// 索引
|
||
public index: string;
|
||
// es客户端
|
||
public client: Client;
|
||
// 日志
|
||
@Logger()
|
||
coreLogger: ILogger;
|
||
// 事件
|
||
@Inject('cool:coolEventManager')
|
||
coolEventManager: CoolEventManager;
|
||
|
||
/**
|
||
* 设置索引
|
||
* @param index
|
||
*/
|
||
setIndex(index: string) {
|
||
this.index = index;
|
||
}
|
||
|
||
/**
|
||
* 处理es数据变更事件，主要用于同步数据
|
||
* @param method
|
||
* @param data
|
||
*/
|
||
async handleDataChange(index, method, data) {
|
||
this.index = index;
|
||
const {
|
||
id,
|
||
ids,
|
||
bodys,
|
||
body,
|
||
type,
|
||
refresh,
|
||
waitForActiveShards,
|
||
properties,
|
||
config,
|
||
} = data;
|
||
switch (method) {
|
||
case 'upsert':
|
||
await this.upsert(body, refresh, waitForActiveShards);
|
||
break;
|
||
case 'batchIndex':
|
||
await this.batchIndex(bodys, type, refresh, waitForActiveShards);
|
||
break;
|
||
case 'deleteById':
|
||
await this.deleteById(id, refresh, waitForActiveShards);
|
||
break;
|
||
case 'deleteByIds':
|
||
await this.deleteByIds(ids, refresh, waitForActiveShards);
|
||
break;
|
||
case 'deleteByQuery':
|
||
await this.deleteByQuery(body, refresh, waitForActiveShards);
|
||
break;
|
||
case 'updateById':
|
||
await this.updateById(body, refresh, waitForActiveShards);
|
||
break;
|
||
case 'updateByQuery':
|
||
await this.updateByQuery(body, refresh, waitForActiveShards);
|
||
break;
|
||
case 'createIndex':
|
||
await this.updateByQuery(properties, config);
|
||
break;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 数据更新事件
|
||
* @param method
|
||
* @param data
|
||
*/
|
||
async esDataChange(method, data) {
|
||
this.coolEventManager?.emit('esDataChange', this.index, method, data);
|
||
}
|
||
|
||
/**
|
||
* 设置客户端
|
||
* @param client
|
||
*/
|
||
setClient(client: Client) {
|
||
this.client = client;
|
||
}
|
||
|
||
/**
|
||
* 对象转body
|
||
* @param condition
|
||
*/
|
||
async objToBody(condition: any){
|
||
const body = {
|
||
query: {
|
||
bool: {
|
||
must: [],
|
||
},
|
||
}
|
||
}
|
||
for(const key in condition){
|
||
body.query.bool.must.push({
|
||
term: {
|
||
[key]: condition[key]
|
||
}
|
||
})
|
||
}
|
||
return body;
|
||
}
|
||
|
||
/**
|
||
* 按字段值查找
|
||
* @param condition
|
||
* @param size
|
||
* @returns
|
||
*/
|
||
async findBy(condition: any, size?: number){
|
||
const body = await this.objToBody(condition)
|
||
return this.find(body, size);
|
||
}
|
||
|
||
/**
|
||
* 按字段值分页查找
|
||
* @param condition
|
||
* @param page
|
||
* @param size
|
||
*/
|
||
async findPageBy(condition: any, page?: number, size?: number){
|
||
const body = await this.objToBody(condition)
|
||
return this.findPage(body, page, size);
|
||
}
|
||
|
||
/**
|
||
* 查询
|
||
* @param body
|
||
*/
|
||
async find(body?: any, size?: number) {
|
||
if (!body) {
|
||
body = {};
|
||
}
|
||
if(!body.size){
|
||
body.size = size ? size : 10000;
|
||
}
|
||
return this.client
|
||
.search({
|
||
index: this.index,
|
||
body,
|
||
})
|
||
.then(res => {
|
||
return (
|
||
res.hits.hits.map(e => {
|
||
e._source['id'] = e._id;
|
||
const _source: any = e._source;
|
||
['_id', '_index', '_score', '_source'].forEach(key => {
|
||
delete e[key];
|
||
});
|
||
return {
|
||
..._source,
|
||
...e,
|
||
};
|
||
}) || []
|
||
);
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 分页查询
|
||
* @param body
|
||
* @param page
|
||
* @param size
|
||
*/
|
||
async findPage(body?: any, page?: number, size?: number) {
|
||
if (!page) {
|
||
page = 1;
|
||
}
|
||
if (!body) {
|
||
body = {};
|
||
}
|
||
if (!size && !body.size) {
|
||
size = 20;
|
||
body.size = size;
|
||
}
|
||
const total = await this.findCount(body);
|
||
body.from = (page - 1) * size;
|
||
return this.client.search({ index: this.index, body }).then(res => {
|
||
const result =
|
||
res.hits.hits.map(e => {
|
||
e._source['id'] = e._id;
|
||
const _source: any = e._source;
|
||
['_id', '_index', '_score', '_source'].forEach(key => {
|
||
delete e[key];
|
||
});
|
||
return {
|
||
..._source,
|
||
...e,
|
||
};
|
||
}) || [];
|
||
return {
|
||
list: result,
|
||
pagination: {
|
||
page,
|
||
size,
|
||
total,
|
||
},
|
||
};
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 根据ID查询
|
||
* @param id
|
||
* @returns
|
||
*/
|
||
async findById(id) {
|
||
return this.client
|
||
.get({
|
||
index: this.index,
|
||
id,
|
||
})
|
||
.then(res => {
|
||
res._source['id'] = res._id;
|
||
return res._source || undefined;
|
||
})
|
||
.catch(e => {
|
||
return undefined;
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 根据多个ID查询
|
||
* @param ids
|
||
* @returns
|
||
*/
|
||
async findByIds(ids: string[]) {
|
||
return this.client
|
||
.mget({ index: this.index, body: { ids } })
|
||
.then(res => {
|
||
const result = res.docs.map((e: any) => {
|
||
e._source.id = e._id;
|
||
return e._source || 'undefined';
|
||
});
|
||
return result.filter(e => {
|
||
return e !== 'undefined';
|
||
});
|
||
})
|
||
.catch(e => {
|
||
return undefined;
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 插入与更新
|
||
* @param body
|
||
* @param refresh
|
||
* @param waitForActiveShards
|
||
* @returns
|
||
*/
|
||
async upsert(
|
||
body: any,
|
||
refresh?: boolean | 'wait_for',
|
||
waitForActiveShards?: WaitForActiveShards
|
||
) {
|
||
if (refresh == undefined) {
|
||
refresh = true;
|
||
}
|
||
if (body.id) {
|
||
this.esDataChange('upsert', {
|
||
body,
|
||
refresh,
|
||
waitForActiveShards,
|
||
});
|
||
const id = body.id;
|
||
delete body.id;
|
||
return this.client.index({
|
||
id,
|
||
index: this.index,
|
||
wait_for_active_shards: waitForActiveShards,
|
||
refresh,
|
||
body,
|
||
});
|
||
} else {
|
||
return this.client
|
||
.index({
|
||
index: this.index,
|
||
wait_for_active_shards: waitForActiveShards,
|
||
refresh,
|
||
body,
|
||
})
|
||
.then(res => {
|
||
this.esDataChange('upsert', {
|
||
body: {
|
||
...body,
|
||
id: res._id,
|
||
},
|
||
refresh,
|
||
waitForActiveShards,
|
||
});
|
||
return res;
|
||
});
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 批量插入更新
|
||
* @param bodys
|
||
* @param type
|
||
* @param refresh
|
||
* @param waitForActiveShards
|
||
* @returns
|
||
*/
|
||
async batchIndex(
|
||
bodys: any[],
|
||
type: 'index' | 'create' | 'delete' | 'update',
|
||
refresh?: boolean | 'wait_for',
|
||
waitForActiveShards?: WaitForActiveShards
|
||
) {
|
||
this.esDataChange('batchIndex', {
|
||
bodys,
|
||
type,
|
||
refresh,
|
||
waitForActiveShards,
|
||
});
|
||
if (refresh == undefined) {
|
||
refresh = true;
|
||
}
|
||
const list = [];
|
||
for (const body of bodys) {
|
||
const typeO = {};
|
||
typeO[type] = { _index: this.index, _id: body.id };
|
||
if (body.id) {
|
||
delete body.id;
|
||
}
|
||
list.push(typeO);
|
||
if (type !== 'delete') {
|
||
if (type == 'update') {
|
||
list.push({ doc: body });
|
||
} else {
|
||
list.push(body);
|
||
}
|
||
}
|
||
}
|
||
return this.client.bulk({
|
||
wait_for_active_shards: waitForActiveShards,
|
||
index: this.index,
|
||
refresh,
|
||
body: list,
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 删除索引
|
||
* @param id
|
||
* @param refresh
|
||
* @param waitForActiveShards
|
||
* @returns
|
||
*/
|
||
async deleteById(
|
||
id,
|
||
refresh?: boolean | 'wait_for',
|
||
waitForActiveShards?: WaitForActiveShards
|
||
) {
|
||
this.esDataChange('deleteById', {
|
||
id,
|
||
refresh,
|
||
waitForActiveShards,
|
||
});
|
||
if (refresh == undefined) {
|
||
refresh = true;
|
||
}
|
||
try {
|
||
return this.client.delete({
|
||
index: this.index,
|
||
refresh,
|
||
wait_for_active_shards: waitForActiveShards,
|
||
id,
|
||
});
|
||
} catch {}
|
||
}
|
||
|
||
/**
|
||
* 删除文档
|
||
* @param ids
|
||
* @param refresh
|
||
* @param waitForActiveShards
|
||
* @returns
|
||
*/
|
||
async deleteByIds(
|
||
ids: string[],
|
||
refresh?: boolean,
|
||
waitForActiveShards?: WaitForActiveShards
|
||
) {
|
||
this.esDataChange('deleteByIds', {
|
||
ids,
|
||
refresh,
|
||
waitForActiveShards,
|
||
});
|
||
if (refresh == undefined) {
|
||
refresh = true;
|
||
}
|
||
const body = {
|
||
query: {
|
||
bool: {
|
||
must: [
|
||
{
|
||
terms: {
|
||
_id: ids,
|
||
},
|
||
},
|
||
],
|
||
},
|
||
},
|
||
};
|
||
return this.client.deleteByQuery({
|
||
index: this.index,
|
||
refresh,
|
||
wait_for_active_shards: waitForActiveShards,
|
||
body,
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 根据条件批量删除
|
||
* @param body
|
||
* @param refresh
|
||
* @param waitForActiveShards
|
||
* @returns
|
||
*/
|
||
async deleteByQuery(
|
||
body,
|
||
refresh?: boolean,
|
||
waitForActiveShards?: WaitForActiveShards
|
||
) {
|
||
this.esDataChange('deleteByQuery', {
|
||
body,
|
||
refresh,
|
||
waitForActiveShards,
|
||
});
|
||
if (refresh == undefined) {
|
||
refresh = true;
|
||
}
|
||
return this.client.deleteByQuery({
|
||
index: this.index,
|
||
refresh,
|
||
wait_for_active_shards: waitForActiveShards,
|
||
body,
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 更新索引
|
||
* @param body
|
||
* @param refresh
|
||
* @param waitForActiveShards
|
||
* @returns
|
||
*/
|
||
async updateById(
|
||
body,
|
||
refresh?: boolean | 'wait_for',
|
||
waitForActiveShards?: WaitForActiveShards
|
||
) {
|
||
this.esDataChange('updateById', {
|
||
body,
|
||
refresh,
|
||
waitForActiveShards,
|
||
});
|
||
if (refresh == undefined) {
|
||
refresh = true;
|
||
}
|
||
const id = body.id;
|
||
delete body.id;
|
||
return this.client.update({
|
||
wait_for_active_shards: waitForActiveShards,
|
||
index: this.index,
|
||
id: id,
|
||
refresh,
|
||
body: {
|
||
doc: body,
|
||
},
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 根据条件更新
|
||
* @param body
|
||
* @param refresh
|
||
* @param waitForActiveShards
|
||
*/
|
||
async updateByQuery(
|
||
body,
|
||
refresh?: boolean,
|
||
waitForActiveShards?: WaitForActiveShards
|
||
) {
|
||
this.esDataChange('updateByQuery', {
|
||
body,
|
||
refresh,
|
||
waitForActiveShards,
|
||
});
|
||
if (refresh == undefined) {
|
||
refresh = true;
|
||
}
|
||
return this.client.updateByQuery({
|
||
index: this.index,
|
||
refresh,
|
||
wait_for_active_shards: waitForActiveShards,
|
||
body,
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 查询条数
|
||
* @param body
|
||
*/
|
||
async findCount(body?: any) {
|
||
let _body = Object.assign({}, body || {});
|
||
delete _body.from;
|
||
delete _body.size;
|
||
delete _body.sort;
|
||
return this.client
|
||
.count({
|
||
index: this.index,
|
||
body: _body,
|
||
})
|
||
.then(res => {
|
||
return res.count;
|
||
})
|
||
.catch(e => {
|
||
return undefined;
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 创建更新索引
|
||
* @param config 配置
|
||
*/
|
||
async createIndex(
|
||
properties: {},
|
||
config: EsConfig = {
|
||
name: '',
|
||
replicas: 1,
|
||
shards: 8,
|
||
analyzers: [],
|
||
}
|
||
) {
|
||
this.esDataChange('createIndex', {
|
||
properties,
|
||
config,
|
||
});
|
||
const body = {
|
||
settings: {
|
||
number_of_shards: config.shards,
|
||
number_of_replicas: config.replicas,
|
||
analysis: {
|
||
analyzer: {
|
||
comma: { type: 'pattern', pattern: ',' },
|
||
blank: { type: 'pattern', pattern: ' ' },
|
||
},
|
||
},
|
||
mapping: {
|
||
nested_fields: {
|
||
limit: 100,
|
||
},
|
||
},
|
||
},
|
||
mappings: {
|
||
properties: {},
|
||
},
|
||
};
|
||
if (config.analyzers) {
|
||
for (const analyzer of config.analyzers) {
|
||
for (const key in analyzer) {
|
||
body.settings.analysis.analyzer[key] = analyzer[key];
|
||
}
|
||
}
|
||
}
|
||
const param = {
|
||
index: this.index,
|
||
body,
|
||
};
|
||
param.body = body;
|
||
param.body.mappings.properties = properties;
|
||
|
||
this.client.indices.exists({ index: this.index }).then(async res => {
|
||
if (!res) {
|
||
await this.client.indices.create(param).then(res => {
|
||
if (res.acknowledged) {
|
||
console.info(
|
||
'\x1B[36m [cool:core] midwayjs cool elasticsearch ES索引创建成功: ' +
|
||
this.index +
|
||
' \x1B[0m'
|
||
);
|
||
}
|
||
});
|
||
} else {
|
||
const updateParam = {
|
||
index: this.index,
|
||
body: param.body.mappings,
|
||
};
|
||
await this.client.indices.putMapping(updateParam).then(res => {
|
||
if (res.acknowledged) {
|
||
console.info(
|
||
'\x1B[36m [cool:core] midwayjs cool elasticsearch ES索引更新成功: ' +
|
||
this.index +
|
||
' \x1B[0m'
|
||
);
|
||
}
|
||
});
|
||
}
|
||
});
|
||
}
|
||
}
|