replace 'Redis' with 'diskcache'

Author: blmdxiao

README.md

@@ -84,23 +84,15 @@ Once the virtual environment is activated, you can use `pip` to install the requ
 pip install -r requirements.txt
 ```
 
-#### 4.2 Install Redis
-
-The OpenKF service relies on Redis as its caching service. If Redis is already installed, start Redis and listen on port `6379`. If not installed, refer to the following method for installation.
-
-```shell
-docker run --name redis -d -p 6379:6379 redis
-```
-
-#### 4.3 Create SQLite Database
+#### 4.2 Create SQLite Database
 
 The OpenKF service uses SQLite as its storage DB. Before starting the OpenKF service, you need to execute the following command to initialize the database and add the default configuration for admin console.
 
 ```shell
 python3 create_sqlite_db.py
 ```
 
-#### 4.4 Start the service
+#### 4.3 Start the service
 
 If you have completed the steps above, you can try to start the OpenKF service by executing the following command.
 

crawler_module/web_content_crawler.py

@@ -12,14 +12,14 @@
 
 class AsyncCrawlerSiteContent:
 
-    def __init__(self, domain_list, sqlite_db_path, max_requests, max_embedding_input, document_embedder_obj, redis_lock):
+    def __init__(self, domain_list, sqlite_db_path, max_requests, max_embedding_input, document_embedder_obj, distributed_lock):
         logger.info(f"[CRAWL_CONTENT] init, domain_list:{domain_list}")
         self.domain_list = domain_list
         self.sqlite_db_path = sqlite_db_path
         self.semaphore = asyncio.Semaphore(max_requests)
         self.max_embedding_input = max_embedding_input
         self.document_embedder_obj = document_embedder_obj
-        self.redis_lock = redis_lock
+        self.distributed_lock = distributed_lock
         self.count = 0
         self.batch_size = max_requests * 2
 
@@ -112,20 +112,19 @@ async def update_doc_status(self, doc_id_list, doc_status):
         logger.info(f"[CRAWL_CONTENT] update_doc_status, doc_id_list:{doc_id_list}, doc_status:{doc_status}")
         timestamp = int(time.time())
         async with aiosqlite.connect(self.sqlite_db_path) as db:
-            # Enable WAL mode for better concurrency
             await db.execute("PRAGMA journal_mode=WAL;")
 
-            if await self.redis_lock.aacquire_lock():
-                try:
+            try:
+                with self.distributed_lock.lock():
                     await db.execute(
                         "UPDATE t_raw_tab SET doc_status = ?, mtime = ? WHERE id IN ({placeholders})".format(
                             placeholders=','.join(['?' for _ in doc_id_list])
                         ),
                         [doc_status, timestamp] + doc_id_list
                     )
                     await db.commit()
-                finally:
-                    await self.redis_lock.arelease_lock()
+            except Exception as e:
+                logger.error(f"process distributed_lock exception:{e}")
 
     async def fetch_existing_contents(self, doc_id_list):
         """
@@ -134,7 +133,6 @@ async def fetch_existing_contents(self, doc_id_list):
         logger.info(f"[CRAWL_CONTENT] fetch_existing_contents, doc_id_list:{doc_id_list}")
         query = "SELECT id, content FROM t_raw_tab WHERE id IN ({})".format(', '.join('?' for _ in doc_id_list))
         async with aiosqlite.connect(self.sqlite_db_path) as db:
-            # Enable WAL mode for better concurrency
             await db.execute("PRAGMA journal_mode=WAL;")
 
             cursor = await db.execute(query, doc_id_list)
@@ -192,33 +190,31 @@ async def process_updated_contents(self, updated_contents, url_dict):
             content_update_queries.append((content_json, content_length, 3, timestamp, doc_id))
 
         # Lock to ensure database operations are atomic
-        if await self.redis_lock.aacquire_lock():
-            try:
-                async with aiosqlite.connect(self.sqlite_db_path) as db:
-                    # Enable WAL mode for better concurrency
-                    await db.execute("PRAGMA journal_mode=WAL;")
+        async with aiosqlite.connect(self.sqlite_db_path) as db:
+            await db.execute("PRAGMA journal_mode=WAL;")
 
+            try:
+                with self.distributed_lock.lock():
                     # Update content details in t_raw_tab
                     await db.executemany(
                         "UPDATE t_raw_tab SET content = ?, content_length = ?, doc_status = ?, mtime = ? WHERE id = ?",
                         content_update_queries
                     )
                     await db.commit()
-            finally:
-                await self.redis_lock.arelease_lock()
-
+            except Exception as e:
+                logger.error(f"process distributed_lock exception:{e}")
+    
         # Delete old embeddings
         doc_id_list = list(updated_contents.keys())
         await self.delete_embedding_doc(doc_id_list)
 
         # Prepare data for updating embeddings and database records
         data_for_embedding = [(doc_id, url_dict[doc_id], chunk_text_vec) for doc_id, chunk_text_vec in updated_contents.items()]
-        if await self.redis_lock.aacquire_lock():
-            try:
+        try:
+            with self.distributed_lock.lock():
                 records_to_add, records_to_update = await self.document_embedder_obj.aadd_content_embedding(data_for_embedding)
                 # Insert new embedding records and update t_raw_tab doc_status to 4
                 async with aiosqlite.connect(self.sqlite_db_path) as db:
-                    # Enable WAL mode for better concurrency
                     await db.execute("PRAGMA journal_mode=WAL;")
 
                     if records_to_add:
@@ -229,25 +225,24 @@ async def process_updated_contents(self, updated_contents, url_dict):
                     if records_to_update:
                         await db.executemany("UPDATE t_raw_tab SET doc_status = ?, mtime = ? WHERE id = ?", records_to_update)
                     await db.commit()
-            finally:
-                await self.redis_lock.arelease_lock()
+        except Exception as e:
+            logger.error(f"process distributed_lock exception:{e}")
 
     async def update_unchanged_contents_status(self, unchanged_doc_ids):
         """
         Update the status of unchanged contents in the database to reflect they have been processed.
         """
         logger.info(f"[CRAWL_CONTENT] update_unchanged_contents_status, unchanged_doc_ids:{unchanged_doc_ids}")
         async with aiosqlite.connect(self.sqlite_db_path) as db:
-            # Enable WAL mode for better concurrency
             await db.execute("PRAGMA journal_mode=WAL;")
 
-            if await self.redis_lock.aacquire_lock():
-                try:
+            try:
+                with self.distributed_lock.lock():
                     async with aiosqlite.connect(self.sqlite_db_path) as db:
                         await db.execute("UPDATE t_raw_tab SET doc_status = 4 WHERE id IN ({})".format(', '.join('?' for _ in unchanged_doc_ids)), unchanged_doc_ids)
                         await db.commit()
-                finally:
-                    await self.redis_lock.arelease_lock()
+            except Exception as e:
+                logger.error(f"process distributed_lock exception:{e}") 
 
     async def add_content(self, url_dict):
         """Begin processing URLs from url_dict in batches for add."""
@@ -294,26 +289,25 @@ async def delete_embedding_doc(self, doc_id_vec):
         doc_id_tuple = tuple(doc_id_vec)
         placeholder = ','.join('?' * len(doc_id_vec))  # Create placeholders
         async with aiosqlite.connect(self.sqlite_db_path) as db:
-            # Enable WAL mode for better concurrency
             await db.execute("PRAGMA journal_mode=WAL;")
 
             cursor = await db.execute(f"SELECT embedding_id_list FROM t_doc_embedding_map_tab WHERE doc_id IN ({placeholder})", doc_id_tuple)
             rows = await cursor.fetchall()
             # Parse embedding_id_list and flatten the list
             embedding_id_vec = [id for row in rows for id in json.loads(row[0])]
 
-            if await self.redis_lock.aacquire_lock():
-                try:
+            try:
+                with self.distributed_lock.lock():
                     if embedding_id_vec:
                         logger.info(f"[CRAWL_CONTENT] delete_embedding_doc, document_embedder_obj.delete_content_embedding:{embedding_id_vec}")
                         self.document_embedder_obj.delete_content_embedding(embedding_id_vec)
 
                     # Delete records from t_doc_embedding_map_tab
                     await db.execute(f"DELETE FROM t_doc_embedding_map_tab WHERE doc_id IN ({placeholder})", doc_id_tuple)
                     await db.commit()
-                finally:
-                    await self.redis_lock.arelease_lock()
-
+            except Exception as e:
+                logger.error(f"process distributed_lock exception:{e}")
+        
     async def delete_content(self, url_dict, delete_raw_table=True):
         """Begin processing URLs from url_dict in batches for deletion."""
         begin_time = int(time.time())
@@ -345,12 +339,13 @@ async def process_delete_batch(self, batch, delete_raw_table):
             # Delete records from t_raw_tab after deleting embeddings
             async with aiosqlite.connect(self.sqlite_db_path) as db:
                 await db.execute("PRAGMA journal_mode=WAL;")
-                if await self.redis_lock.aacquire_lock():
-                    try:
+
+                try:
+                    with self.distributed_lock.lock():
                         await db.execute(f"DELETE FROM t_raw_tab WHERE id IN ({placeholder})", doc_id_tuple)
                         await db.commit()
-                    finally:
-                        await self.redis_lock.arelease_lock()
+                except Exception as e:
+                    logger.error(f"process distributed_lock exception:{e}")
 
     async def update_content(self, url_dict):
         logger.info(f"[CRAWL_CONTENT] update_content begin, url_dict:{url_dict}")
@@ -361,7 +356,6 @@ async def update_content(self, url_dict):
     async def check_and_update_domain_status(self):
         logger.info(f"[CRAWL_CONTENT] check_and_update_domain_status")
         async with aiosqlite.connect(self.sqlite_db_path) as db:
-            # Enable WAL mode for better concurrency
             await db.execute("PRAGMA journal_mode=WAL;")
 
             timestamp = int(time.time())
@@ -375,13 +369,12 @@ async def check_and_update_domain_status(self):
                         "SELECT COUNT(*) FROM t_raw_tab WHERE domain = ? AND doc_status < 4", (domain,))
                     count_row = await cursor.fetchone()
                     if count_row[0] == 0:  # If no records have doc_status < 4
-                        if await self.redis_lock.aacquire_lock():
-                            try:
+                        try:
+                            with self.distributed_lock.lock():
                                 # Step 3: Update domain_status to 4 in t_domain_tab
                                 await db.execute(
                                     "UPDATE t_domain_tab SET domain_status = ?, mtime = ? WHERE domain = ?", (4, timestamp, domain))
                                 await db.commit()
-                            finally:
-                                await self.redis_lock.arelease_lock()
+                        except Exception as e:
+                            logger.error(f"process distributed_lock exception:{e}")
                         logger.info(f"[CRAWL_CONTENT] check_and_update_domain_status, Domain status updated to 4 for domain:'{domain}'")
-