fix: prepend bucket prefix in Azure Blob (SAS/SPN) to prevent cross-dataset file overwrites (#14174)

Fixes #14159

## Problem

The `put()`, `get()`, `rm()`, and `obj_exist()` methods in both
`azure_spn_conn.py` and `azure_sas_conn.py` ignore the `bucket`
parameter entirely, storing all files flat using only the filename. This
causes files from different datasets to overwrite each other when they
share the same filename.

By contrast, the MinIO and S3 implementations correctly use the bucket
(typically the knowledge base ID) as a path prefix, creating logical
folder isolation like `{kb_id}/{filename}`.

## Solution

Prepend the `bucket` parameter as a path prefix to all file operations
in both Azure storage implementations:

- `azure_spn_conn.py`: `create_file`, `delete_file`, `get_file_client`
now use `f"{bucket}/{fnm}"`
- `azure_sas_conn.py`: `upload_blob`, `delete_blob`, `download_blob`,
`get_blob_client` now use `f"{bucket}/{fnm}"`

This matches the behavior of all other storage backends (MinIO, S3) and
prevents filename collisions across knowledge bases.

## Testing

- Verified the fix aligns with how MinIO/S3 connectors handle the bucket
parameter
- The `health()` method is left unchanged as it uses a fixed test path
for connectivity checks only

Co-authored-by: octo-patch <octo-patch@github.com>
Co-authored-by: Jin Hai <haijin.chn@gmail.com>

rag/utils/azure_sas_conn.py

@@ -54,7 +54,7 @@ class RAGFlowAzureSasBlob:
     def put(self, bucket, fnm, binary, tenant_id=None):
         for _ in range(3):
             try:
-                return self.conn.upload_blob(name=fnm, data=BytesIO(binary), length=len(binary))
+                return self.conn.upload_blob(name=f"{bucket}/{fnm}", data=BytesIO(binary), length=len(binary))
             except Exception:
                 logging.exception(f"Fail put {bucket}/{fnm}")
                 self.__open__()
@@ -62,14 +62,14 @@ class RAGFlowAzureSasBlob:
 
     def rm(self, bucket, fnm):
         try:
-            self.conn.delete_blob(fnm)
+            self.conn.delete_blob(f"{bucket}/{fnm}")
         except Exception:
             logging.exception(f"Fail rm {bucket}/{fnm}")
 
     def get(self, bucket, fnm):
         for _ in range(1):
             try:
-                r = self.conn.download_blob(fnm)
+                r = self.conn.download_blob(f"{bucket}/{fnm}")
                 return r.read()
             except Exception:
                 logging.exception(f"fail get {bucket}/{fnm}")
@@ -79,7 +79,7 @@ class RAGFlowAzureSasBlob:
 
     def obj_exist(self, bucket, fnm):
         try:
-            return self.conn.get_blob_client(fnm).exists()
+            return self.conn.get_blob_client(f"{bucket}/{fnm}").exists()
         except Exception:
             logging.exception(f"Fail put {bucket}/{fnm}")
         return False

rag/utils/azure_spn_conn.py

@@ -71,7 +71,7 @@ class RAGFlowAzureSpnBlob:
     def put(self, bucket, fnm, binary, tenant_id=None):
         for _ in range(3):
             try:
-                f = self.conn.create_file(fnm)
+                f = self.conn.create_file(f"{bucket}/{fnm}")
                 f.append_data(binary, offset=0, length=len(binary))
                 return f.flush_data(len(binary))
             except Exception:
@@ -83,14 +83,14 @@ class RAGFlowAzureSpnBlob:
 
     def rm(self, bucket, fnm):
         try:
-            self.conn.delete_file(fnm)
+            self.conn.delete_file(f"{bucket}/{fnm}")
         except Exception:
             logging.exception(f"Fail rm {bucket}/{fnm}")
 
     def get(self, bucket, fnm):
         for _ in range(1):
             try:
-                client = self.conn.get_file_client(fnm)
+                client = self.conn.get_file_client(f"{bucket}/{fnm}")
                 r = client.download_file()
                 return r.read()
             except Exception:
@@ -101,7 +101,7 @@ class RAGFlowAzureSpnBlob:
 
     def obj_exist(self, bucket, fnm):
         try:
-            client = self.conn.get_file_client(fnm)
+            client = self.conn.get_file_client(f"{bucket}/{fnm}")
             return client.exists()
         except Exception:
             logging.exception(f"Fail put {bucket}/{fnm}")