|
|
|
@ -57,7 +57,7 @@ client = None
|
|
|
|
|
# with strings in the field.
|
|
|
|
|
keyword_fields = ["nick_id", "user_id", "net_id"]
|
|
|
|
|
|
|
|
|
|
mapping = {
|
|
|
|
|
mapping_int = {
|
|
|
|
|
"mappings": {
|
|
|
|
|
"properties": {
|
|
|
|
|
"ts": {"type": "date", "format": "epoch_second"},
|
|
|
|
@ -65,24 +65,32 @@ mapping = {
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
mapping = dict(mapping_int)
|
|
|
|
|
for field in keyword_fields:
|
|
|
|
|
mapping["mappings"]["properties"][field] = {"type": "text"}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
del mapping_int["mappings"]["properties"]["file_tim"]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
async def initialise_elasticsearch():
|
|
|
|
|
"""
|
|
|
|
|
Initialise the Elasticsearch client.
|
|
|
|
|
"""
|
|
|
|
|
auth = (ELASTICSEARCH_USERNAME, ELASTICSEARCH_PASSWORD)
|
|
|
|
|
client = AsyncElasticsearch(ELASTICSEARCH_HOST, http_auth=auth, verify_certs=False)
|
|
|
|
|
for index in ("main", "restricted"):
|
|
|
|
|
for index in ("main", "restricted", "internal"):
|
|
|
|
|
if index == "internal":
|
|
|
|
|
map_dict = mapping_int
|
|
|
|
|
else:
|
|
|
|
|
map_dict = mapping
|
|
|
|
|
if await client.indices.exists(index=index):
|
|
|
|
|
# update index with mapping
|
|
|
|
|
await client.indices.put_mapping(
|
|
|
|
|
index=index, properties=mapping["mappings"]["properties"]
|
|
|
|
|
index=index, properties=map_dict["mappings"]["properties"]
|
|
|
|
|
)
|
|
|
|
|
else:
|
|
|
|
|
await client.indices.create(index=index, mappings=mapping["mappings"])
|
|
|
|
|
await client.indices.create(index=index, mappings=map_dict["mappings"])
|
|
|
|
|
return client
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@ -119,9 +127,13 @@ async def store_batch(data):
|
|
|
|
|
else:
|
|
|
|
|
indexmap[INDEX].append(msg)
|
|
|
|
|
|
|
|
|
|
print("KEYS", indexmap.keys())
|
|
|
|
|
|
|
|
|
|
for index, index_messages in indexmap.items():
|
|
|
|
|
for message in index_messages:
|
|
|
|
|
result = await client.index(index=index, body=message)
|
|
|
|
|
if index == "internal":
|
|
|
|
|
print("INTERNAL RES", result)
|
|
|
|
|
if not result["result"] == "created":
|
|
|
|
|
log.error(f"Indexing failed: {result}")
|
|
|
|
|
log.debug(f"Indexed {len(data)} messages in ES")
|
|
|
|
|