Skip to content

Commit 1dddab0

Browse files
letonghanletonghan
authored
Fix Dataprep ingest PPT and async issues (opea-project#1504)
Signed-off-by: letonghan <letong.han@intel.com>
1 parent 981faf5 commit 1dddab0

8 files changed

Lines changed: 47 additions & 13 deletions

File tree

‎comps/dataprep/src/integrations/redis.py‎

Lines changed: 17 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -146,11 +146,11 @@ async def store_by_id(client, key, value):
146146
return True
147147

148148

149-
def search_by_id(client, doc_id):
149+
async def search_by_id(client, doc_id):
150150
if logflag:
151151
logger.info(f"[ search by id ] searching docs of {doc_id}")
152152
try:
153-
results = client.load_document(doc_id)
153+
results = await client.load_document(doc_id)
154154
if logflag:
155155
logger.info(f"[ search by id ] search success of {doc_id}: {results}")
156156
return results
@@ -174,9 +174,10 @@ def drop_index(index_name, redis_url=REDIS_URL):
174174
return True
175175

176176

177-
def delete_by_id(client, id):
177+
async def delete_by_id(client, id):
178178
try:
179-
assert client.delete_document(id)
179+
res = await client.delete_document(id)
180+
assert res
180181
if logflag:
181182
logger.info(f"[ delete by id ] delete id success: {id}")
182183
except Exception as e:
@@ -286,7 +287,7 @@ def __init__(self, name: str, description: str, config: dict = None):
286287
self.client = redis.Redis(connection_pool=redis_pool)
287288
self.data_index_client, self.key_index_client = asyncio.run(self._initialize_client())
288289
self.embedder = asyncio.run(self._initialize_embedder())
289-
health_status = self.check_health()
290+
health_status = asyncio.run(self.check_health())
290291
if not health_status:
291292
logger.error("OpeaRedisDataprep health check failed.")
292293

@@ -390,7 +391,8 @@ async def ingest_files(
390391
# check whether the file already exists
391392
key_ids = None
392393
try:
393-
key_ids = search_by_id(self.key_index_client, doc_id).key_ids
394+
result = await search_by_id(self.key_index_client, doc_id)
395+
key_ids = result.key_ids
394396
if logflag:
395397
logger.info(f"[ redis ingest] File {file.filename} already exists.")
396398
except Exception as e:
@@ -435,7 +437,8 @@ async def ingest_files(
435437
# check whether the link file already exists
436438
key_ids = None
437439
try:
438-
key_ids = search_by_id(self.key_index_client, doc_id).key_ids
440+
result = await search_by_id(self.key_index_client, doc_id)
441+
key_ids = result.key_ids
439442
if logflag:
440443
logger.info(f"[ redis ingest] Link {link} already exists.")
441444
except Exception as e:
@@ -565,7 +568,8 @@ async def delete_files(self, file_path: str = Body(..., embed=True)):
565568

566569
# determine whether this file exists in db KEY_INDEX_NAME
567570
try:
568-
key_ids = search_by_id(self.key_index_client, doc_id).key_ids
571+
result = await search_by_id(self.key_index_client, doc_id)
572+
key_ids = result.key_ids
569573
except Exception as e:
570574
if logflag:
571575
logger.info(f"[ redis delete ] {e}, File {file_path} does not exists.")
@@ -576,7 +580,8 @@ async def delete_files(self, file_path: str = Body(..., embed=True)):
576580

577581
# delete file keys id in db KEY_INDEX_NAME
578582
try:
579-
assert delete_by_id(self.key_index_client, doc_id)
583+
res = await delete_by_id(self.key_index_client, doc_id)
584+
assert res
580585
except Exception as e:
581586
if logflag:
582587
logger.info(f"[ redis delete ] {e}. File {file_path} delete failed for db {KEY_INDEX_NAME}.")
@@ -586,7 +591,7 @@ async def delete_files(self, file_path: str = Body(..., embed=True)):
586591
for file_id in file_ids:
587592
# determine whether this file exists in db INDEX_NAME
588593
try:
589-
search_by_id(self.data_index_client, file_id)
594+
await search_by_id(self.data_index_client, file_id)
590595
except Exception as e:
591596
if logflag:
592597
logger.info(f"[ redis delete ] {e}. File {file_path} does not exists.")
@@ -596,7 +601,8 @@ async def delete_files(self, file_path: str = Body(..., embed=True)):
596601

597602
# delete file content
598603
try:
599-
assert delete_by_id(self.data_index_client, file_id)
604+
res = await delete_by_id(self.data_index_client, file_id)
605+
assert res
600606
except Exception as e:
601607
if logflag:
602608
logger.info(f"[ redis delete ] {e}. File {file_path} delete failed for db {INDEX_NAME}")

‎comps/dataprep/src/utils.py‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
import timeit
1717
import unicodedata
1818
import urllib.parse
19+
import uuid
1920
from concurrent.futures import ThreadPoolExecutor, as_completed
2021
from pathlib import Path
2122
from typing import Dict, List, Union
@@ -233,20 +234,28 @@ async def load_docx(docx_path):
233234
async def load_ppt(ppt_path):
234235
"""Load ppt file."""
235236
print("Converting ppt file to pptx file...")
236-
pptx_path = ppt_path + "x"
237+
238+
temp_dir = tempfile.gettempdir()
239+
base_name = os.path.splitext(os.path.basename(ppt_path))[0]
240+
pptx_path = os.path.join(temp_dir, base_name + ".pptx")
241+
237242
subprocess.run(
238243
[
239244
"libreoffice",
240245
"--headless",
241246
"--invisible",
242247
"--convert-to",
243-
"docx",
248+
"pptx",
244249
"--outdir",
245250
os.path.dirname(pptx_path),
246251
ppt_path,
247252
],
248253
check=True,
254+
stdout=subprocess.PIPE,
255+
stderr=subprocess.DEVNULL,
249256
)
257+
if not os.path.exists(pptx_path):
258+
raise FileNotFoundError(f"pptx file not created: {pptx_path}")
250259
print("Converted ppt file to pptx file.")
251260
text = await load_pptx(pptx_path)
252261
os.remove(pptx_path)

‎tests/dataprep/dataprep_utils.sh‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,13 @@ function ingest_pdf() {
6161
_invoke_curl $fqdn $port ingest -F "files=@${SCRIPT_DIR}/ingest_dataprep.pdf" $@
6262
}
6363

64+
function ingest_ppt() {
65+
local fqdn=$1
66+
local port=$2
67+
shift 2
68+
_invoke_curl $fqdn $port ingest -F "files=@${SCRIPT_DIR}/ingest_dataprep.ppt" $@
69+
}
70+
6471
function ingest_pptx() {
6572
local fqdn=$1
6673
local port=$2

‎tests/dataprep/ingest_dataprep.ppt‎

243 KB
Binary file not shown.

‎tests/dataprep/test_dataprep_milvus.sh‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,9 @@ function validate_microservice() {
5252
ingest_pdf ${ip_address} ${DATAPREP_PORT}
5353
check_result "dataprep - upload - pdf" "Data preparation succeeded" dataprep-milvus-server ${LOG_PATH}/dataprep_milvus.log
5454

55+
ingest_ppt ${ip_address} ${DATAPREP_PORT}
56+
check_result "dataprep - upload - ppt" "Data preparation succeeded" dataprep-milvus-server ${LOG_PATH}/dataprep_upload_file.log
57+
5558
ingest_pptx ${ip_address} ${DATAPREP_PORT}
5659
check_result "dataprep - upload - pptx" "Data preparation succeeded" dataprep-milvus-server ${LOG_PATH}/dataprep_milvus.log
5760

‎tests/dataprep/test_dataprep_pgvector.sh‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,9 @@ function validate_microservice() {
5454
ingest_pdf ${ip_address} ${DATAPREP_PORT}
5555
check_result "dataprep - upload - pdf" "Data preparation succeeded" dataprep-pgvector-server ${LOG_PATH}/dataprep_pgvector.log
5656

57+
ingest_ppt ${ip_address} ${DATAPREP_PORT}
58+
check_result "dataprep - upload - ppt" "Data preparation succeeded" dataprep-pgvector-server ${LOG_PATH}/dataprep_upload_file.log
59+
5760
ingest_pptx ${ip_address} ${DATAPREP_PORT}
5861
check_result "dataprep - upload - pptx" "Data preparation succeeded" dataprep-pgvector-server ${LOG_PATH}/dataprep_pgvector.log
5962

‎tests/dataprep/test_dataprep_qdrant.sh‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,9 @@ function validate_microservice() {
5454
ingest_pdf ${ip_address} ${DATAPREP_PORT}
5555
check_result "dataprep - upload - pdf" "Data preparation succeeded" dataprep-qdrant-server ${LOG_PATH}/dataprep-qdrant.log
5656

57+
ingest_ppt ${ip_address} ${DATAPREP_PORT}
58+
check_result "dataprep - upload - ppt" "Data preparation succeeded" dataprep-qdrant-server ${LOG_PATH}/dataprep_upload_file.log
59+
5760
ingest_pptx ${ip_address} ${DATAPREP_PORT}
5861
check_result "dataprep - upload - pptx" "Data preparation succeeded" dataprep-qdrant-server ${LOG_PATH}/dataprep-qdrant.log
5962

‎tests/dataprep/test_dataprep_redis.sh‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,9 @@ function validate_microservice() {
6060
ingest_pdf ${ip_address} ${DATAPREP_PORT}
6161
check_result "dataprep - upload - pdf" "Data preparation succeeded" dataprep-redis-server ${LOG_PATH}/dataprep_upload_file.log
6262

63+
ingest_ppt ${ip_address} ${DATAPREP_PORT}
64+
check_result "dataprep - upload - ppt" "Data preparation succeeded" dataprep-redis-server ${LOG_PATH}/dataprep_upload_file.log
65+
6366
ingest_pptx ${ip_address} ${DATAPREP_PORT}
6467
check_result "dataprep - upload - pptx" "Data preparation succeeded" dataprep-redis-server ${LOG_PATH}/dataprep_upload_file.log
6568

0 commit comments

Comments
 (0)