langgenius/dify
1
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-07-21 12:05:30 -04:00
Code Issues 775 Packages Projects Releases 10 Wiki Activity

Can not upload the files ,it alwasys indicate 400 error #20426

New Issue
Closed
opened 2026-02-21 20:07:24 -05:00 by yindo · 1 comment
yindo commented 2026-02-21 20:07:24 -05:00
Owner
Copy Link
Copy Source

Originally created by @007long on GitHub (Nov 19, 2025).

Self Checks

  • I have read the Contributing Guide and Language Policy.
  • This is only for bug report, if you would like to ask a question, please head to Discussions.
  • I have searched for existing issues search for existing issues, including closed ones.
  • I confirm that I am using English to submit this report, otherwise it will be closed.
  • 【中文用户 & Non English User】请使用英语提交，否则会被关闭 ：）
  • Please do not modify this template :) and fill in all the required fields.

Dify version

1.9.1

Cloud or Self Hosted

Self Hosted (Docker)

Steps to reproduce

I create the code like below ,I want to upload the files into the knowledge base
import requests
import os
import time
import logging
import json
from pathlib import Path
import mimetypes

class DifyKnowledgeBaseManager:
"""
Dify Knowledge Base Manager - Fixed authentication and doc_form issues
"""

def __init__(self):
    # === Configuration Area ===
    
    # Try using workspace API key instead of dataset API key
    self.API_KEY = "dataset-BErpWPBgS0zns950cWdkqEnD"
    
    # Dataset ID
    self.DATASET_ID = "a4de643b-7a7e-47b7-919f-e49c16aae406"
    
    # Dify base URL
    self.BASE_URL = "http://localhost/v1"
    
    # Resume folder path
    self.RESUMES_FOLDER = "./resumes"
    
    # Batch upload settings
    self.BATCH_SIZE = 3
    self.DELAY_BETWEEN_BATCHES = 3
    self.DELAY_BETWEEN_FILES = 1
    
    # === End Configuration Area ===
    
    # Setup logging
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(levelname)s - %(message)s',
        handlers=[
            logging.StreamHandler(),
            logging.FileHandler('knowledge_base_upload.log', encoding='utf-8')
        ]
    )
    self.logger = logging.getLogger(__name__)
    
    self.logger.info("Dify Knowledge Base Manager initialized")

def get_dataset_info(self):
    """Get detailed dataset information including doc_form"""
    self.logger.info("📋 Getting dataset details...")
    
    endpoint = f"{self.BASE_URL}/datasets/{self.DATASET_ID}"
    
    headers = {
        'Authorization': f'Bearer {self.API_KEY}',
        'Content-Type': 'application/json'
    }
    
    try:
        response = requests.get(endpoint, headers=headers, timeout=10)
        
        if response.status_code == 200:
            dataset_info = response.json()
            self.logger.info("✅ Successfully retrieved dataset information")
            
            # Print dataset info for debugging
            if 'data' in dataset_info:
                self.logger.info(f"Dataset name: {dataset_info['data'].get('name', 'Unknown')}")
                self.logger.info(f"Document count: {dataset_info['data'].get('document_count', 'Unknown')}")
                
                # Check for doc_form information
                if 'doc_form' in dataset_info['data']:
                    self.logger.info(f"Dataset doc_form: {dataset_info['data']['doc_form']}")
                else:
                    self.logger.info("Dataset doc_form: Not found, using default")
            
            return dataset_info
        else:
            self.logger.error(f"❌ Failed to get dataset info: {response.text}")
            return None
            
    except Exception as e:
        self.logger.error(f"❌ Exception getting dataset info: {str(e)}")
        return None

def upload_file_fixed(self, file_path):
    """Fixed file upload method"""
    filename = os.path.basename(file_path)
    self.logger.info(f"📤 Uploading file: {filename}")

    endpoint = f"{self.BASE_URL}/datasets/{self.DATASET_ID}/document/create_by_file"

    try:
        # Check file
        file_size = os.path.getsize(file_path)
        if file_size == 0:
            self.logger.error(f"❌ File is empty: {filename}")
            return False

        # Simplified request data - only include required parameters
        request_data = {
            "name": filename,
            "indexing_technique": "high_quality",
            # Remove doc_form and doc_language, let system use dataset defaults
        }

        # Read file
        with open(file_path, 'rb') as file:
            file_content = file.read()

        # Build multipart request
        files = {
            'file': (filename, file_content, 'application/octet-stream')
        }

        # Parameters in data field
        data = {
            'data': json.dumps(request_data)
        }

        headers = {
            'Authorization': f'Bearer {self.API_KEY}'
        }

        self.logger.info("Sending file upload request...")
        response = requests.post(
            endpoint,
            files=files,
            data=data,
            headers=headers,
            timeout=120
        )

        self.logger.info(f"Response status code: {response.status_code}")

        if response.status_code in [200, 201]:
            result = response.json()
            self.logger.info(f"✅ File upload successful: {filename}")
            return True
        else:
            error_msg = response.text
            self.logger.error(f"❌ File upload failed: {error_msg}")
            return False

    except Exception as e:
        self.logger.error(f"❌ File upload exception: {str(e)}")
        return False

def upload_text_fixed(self, file_path):
    """Fixed text upload method"""
    filename = os.path.basename(file_path)
    self.logger.info(f"📝 Uploading via text API: {filename}")

    # Extract text content
    text_content = self.extract_text_from_file(file_path)
    if text_content is None:
        self.logger.error(f"❌ Cannot extract text content: {filename}")
        return False
    
    if len(text_content.strip()) == 0:
        self.logger.error(f"❌ Extracted text content is empty: {filename}")
        return False

    endpoint = f"{self.BASE_URL}/datasets/{self.DATASET_ID}/document/create_by_text"

    try:
        # Get dataset info to determine correct doc_form
        dataset_info = self.get_dataset_info()
        doc_form = "text_model"  # Default value
        
        if dataset_info and 'data' in dataset_info and 'doc_form' in dataset_info['data']:
            doc_form = dataset_info['data']['doc_form']
            self.logger.info(f"Using dataset's doc_form: {doc_form}")
        else:
            self.logger.info("Using default doc_form: text_model")

        # Build request body - use correct doc_form
        payload = {
            "name": filename,
            "text": text_content,
            "indexing_technique": "high_quality",
            "process_rule": {
                "mode": "automatic",
                "rules": {
                    "segmentation": {
                        "separator": "\n\n",
                        "max_tokens": 500
                    }
                }
            },
            "doc_form": doc_form,
            # Remove doc_language or use dataset default
        }

        headers = {
            'Authorization': f'Bearer {self.API_KEY}',
            'Content-Type': 'application/json'
        }

        self.logger.info("Sending text upload request...")
        response = requests.post(
            endpoint,
            json=payload,
            headers=headers,
            timeout=120
        )

        self.logger.info(f"Response status code: {response.status_code}")

        if response.status_code in [200, 201]:
            self.logger.info(f"✅ Text upload successful: {filename}")
            return True
        else:
            error_msg = response.text
            self.logger.error(f"❌ Text upload failed: {error_msg}")
            return False

    except Exception as e:
        self.logger.error(f"❌ Text upload exception: {str(e)}")
        return False

def extract_text_from_file(self, file_path):
    """Extract text content from file"""
    filename = os.path.basename(file_path)
    file_ext = os.path.splitext(filename)[1].lower()
    
    try:
        if file_ext == '.txt':
            with open(file_path, 'r', encoding='utf-8') as f:
                return f.read()
        
        elif file_ext == '.pdf':
            try:
                from PyPDF2 import PdfReader
                reader = PdfReader(file_path)
                text = ""
                for page in reader.pages:
                    page_text = page.extract_text()
                    if page_text:
                        text += page_text + "\n"
                return text if text.strip() else None
            except ImportError:
                self.logger.error("❌ Please install PyPDF2: pip install pypdf2")
                return None
            except Exception as e:
                self.logger.error(f"❌ PDF parsing failed: {str(e)}")
                return None
        
        elif file_ext == '.docx':
            try:
                from docx import Document
                doc = Document(file_path)
                text = ""
                for paragraph in doc.paragraphs:
                    if paragraph.text:
                        text += paragraph.text + "\n"
                return text if text.strip() else None
            except ImportError:
                self.logger.error("❌ Please install python-docx: pip install python-docx")
                return None
            except Exception as e:
                self.logger.error(f"❌ DOCX parsing failed: {str(e)}")
                return None
        
        else:
            self.logger.warning(f"⚠️ Unsupported file format: {filename}")
            return None
            
    except Exception as e:
        self.logger.error(f"❌ Text extraction failed: {str(e)}")
        return None

def try_different_api_keys(self):
    """Try using different types of API keys"""
    self.logger.info("🔑 Trying different API key types...")
    
    # Common API key prefixes
    api_key_types = {
        "app-": "Application API Key",
        "wk-": "Workspace API Key", 
        "dataset-": "Dataset API Key",
        "worker-": "Worker API Key"
    }
    
    current_key = self.API_KEY
    current_prefix = current_key.split('-')[0] + '-'
    
    self.logger.info(f"Current API key type: {api_key_types.get(current_prefix, 'Unknown')}")
    
    # Test current key
    if self.test_api_connectivity():
        self.logger.info("✅ Current API key is valid")
        return True
    else:
        self.logger.error("❌ Current API key is invalid, please try to get correct API key")
        return False

def test_api_connectivity(self):
    """Test API connectivity"""
    endpoint = f"{self.BASE_URL}/datasets"
    
    headers = {
        'Authorization': f'Bearer {self.API_KEY}',
        'Content-Type': 'application/json'
    }
    
    try:
        response = requests.get(endpoint, headers=headers, timeout=10)
        return response.status_code == 200
    except:
        return False

def comprehensive_test_fixed(self):
    """Fixed comprehensive test"""
    self.logger.info("🧪 Starting fixed comprehensive test...")
    
    # 1. Get dataset information
    self.logger.info("1. Getting dataset information...")
    dataset_info = self.get_dataset_info()
    if not dataset_info:
        self.logger.error("❌ Cannot get dataset information")
        return False
    
    # 2. Test API key
    self.logger.info("2. Testing API key...")
    if not self.try_different_api_keys():
        self.logger.error("❌ API key test failed")
        return False
    
    # 3. Create test file
    self.logger.info("3. Creating test file...")
    test_content = "Dify Knowledge Base API Test Document - Fixed Version"
    test_filepath = os.path.join(self.RESUMES_FOLDER, "api_test_fixed.txt")
    
    os.makedirs(self.RESUMES_FOLDER, exist_ok=True)
    with open(test_filepath, 'w', encoding='utf-8') as f:
        f.write(test_content)
    
    # 4. Priority test text upload (clearer error messages)
    self.logger.info("4. Testing text upload...")
    text_success = self.upload_text_fixed(test_filepath)
    
    # 5. Test file upload
    self.logger.info("5. Testing file upload...")
    file_success = self.upload_file_fixed(test_filepath)
    
    # Cleanup
    try:
        os.remove(test_filepath)
    except:
        pass
    
    if file_success or text_success:
        self.logger.info("✅ Comprehensive test passed")
        return True
    else:
        self.logger.error("❌ Comprehensive test failed")
        
        # Provide detailed solutions
        self.logger.info("\n🔧 Possible solutions:")
        self.logger.info("1. Check API access permissions in dataset settings")
        self.logger.info("2. Confirm the API key has correct permissions")
        self.logger.info("3. Check Dify service user authentication configuration")
        self.logger.info("4. Try using workspace API key instead of dataset API key")
        
        return False

def batch_upload_files_fixed(self):
    """
    Fixed batch upload method
    """
    self.logger.info("🚀 Starting fixed batch upload process...")
    
    # Display configuration
    self.logger.info("⚙️ Current configuration:")
    self.logger.info(f"  API Key: {self.API_KEY}")
    self.logger.info(f"  Dataset ID: {self.DATASET_ID}")
    self.logger.info(f"  Base URL: {self.BASE_URL}")
    
    # First run fixed comprehensive test
    self.logger.info("Running fixed comprehensive test first...")
    if not self.comprehensive_test_fixed():
        self.logger.error("❌ Fixed comprehensive test failed, cannot continue batch upload")
        
        # Provide guidance for getting correct API key
        self.logger.info("\n📋 Steps to get correct API key:")
        self.logger.info("1. Login to Dify backend")
        self.logger.info("2. Go to 'Workspace Settings' -> 'API Keys'")
        self.logger.info("3. Create new API key (select appropriate permissions)")
        self.logger.info("4. Use API key starting with 'app-' or 'wk-'")
        
        return
    
    self.logger.info("✅ Fixed comprehensive test passed, starting batch upload...")
    
    # Check folder
    if not os.path.exists(self.RESUMES_FOLDER):
        self.logger.error(f"❌ Folder does not exist: {self.RESUMES_FOLDER}")
        return
    
    # Collect files
    document_files = []
    for filename in os.listdir(self.RESUMES_FOLDER):
        file_path = os.path.join(self.RESUMES_FOLDER, filename)
        if os.path.isfile(file_path):
            document_files.append(file_path)
    
    total_files = len(document_files)
    self.logger.info(f"Found {total_files} document files")
    
    if total_files == 0:
        self.logger.warning("⚠️ No document files found")
        return
    
    # Display file list
    self.logger.info("File list:")
    for i, file_path in enumerate(document_files, 1):
        file_size = os.path.getsize(file_path)
        self.logger.info(f"  {i}. {os.path.basename(file_path)} ({file_size/1024:.1f} KB)")
    
    success_count = 0
    failed_files = []
    
    # Start batch upload - priority to text upload (clearer error messages)
    for i, file_path in enumerate(document_files, 1):
        filename = os.path.basename(file_path)
        self.logger.info(f"\n📎 Processing file {i}/{total_files}: {filename}")
        
        # Priority to text upload
        if self.upload_text_fixed(file_path):
            success_count += 1
            self.logger.info(f"✅ File {i} uploaded successfully via text")
        else:
            # If text upload fails, try file upload
            self.logger.info("Text upload failed, trying file upload...")
            if self.upload_file_fixed(file_path):
                success_count += 1
                self.logger.info(f"✅ File {i} uploaded successfully via file")
            else:
                failed_files.append(filename)
                self.logger.error(f"❌ File {i} upload failed")
        
        # Delay
        if i < total_files:
            time.sleep(self.DELAY_BETWEEN_FILES)
        if i % self.BATCH_SIZE == 0 and i < total_files:
            time.sleep(self.DELAY_BETWEEN_BATCHES)
    
    # Output results
    self.logger.info(f"\n{'='*60}")
    self.logger.info("📊 Batch upload completed!")
    self.logger.info(f"✅ Success: {success_count}/{total_files}")
    
    if failed_files:
        self.logger.info(f"❌ Failed: {len(failed_files)}/{total_files}")
        for failed_file in failed_files:
            self.logger.info(f"  - {failed_file}")
    else:
        self.logger.info("🎉 All files uploaded successfully!")

def main():
"""Main function"""
manager = DifyKnowledgeBaseManager()
manager.batch_upload_files_fixed()

if name == "main":
main()

✔️ Expected Behavior

upload the files sucessfully

❌ Actual Behavior

It failed and indicate 400 error
{"code":"invalid_param","message":"doc_form is different from the dataset doc_form.","status":40
0}

Originally created by @007long on GitHub (Nov 19, 2025). ### Self Checks - [x] I have read the [Contributing Guide](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md) and [Language Policy](https://github.com/langgenius/dify/issues/1542). - [x] This is only for bug report, if you would like to ask a question, please head to [Discussions](https://github.com/langgenius/dify/discussions/categories/general). - [x] I have searched for existing issues [search for existing issues](https://github.com/langgenius/dify/issues), including closed ones. - [x] I confirm that I am using English to submit this report, otherwise it will be closed. - [x] 【中文用户 & Non English User】请使用英语提交，否则会被关闭 ：） - [x] Please do not modify this template :) and fill in all the required fields. ### Dify version 1.9.1 ### Cloud or Self Hosted Self Hosted (Docker) ### Steps to reproduce I create the code like below ,I want to upload the files into the knowledge base import requests import os import time import logging import json from pathlib import Path import mimetypes class DifyKnowledgeBaseManager: """ Dify Knowledge Base Manager - Fixed authentication and doc_form issues """ def __init__(self): # === Configuration Area === # Try using workspace API key instead of dataset API key self.API_KEY = "dataset-BErpWPBgS0zns950cWdkqEnD" # Dataset ID self.DATASET_ID = "a4de643b-7a7e-47b7-919f-e49c16aae406" # Dify base URL self.BASE_URL = "http://localhost/v1" # Resume folder path self.RESUMES_FOLDER = "./resumes" # Batch upload settings self.BATCH_SIZE = 3 self.DELAY_BETWEEN_BATCHES = 3 self.DELAY_BETWEEN_FILES = 1 # === End Configuration Area === # Setup logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler(), logging.FileHandler('knowledge_base_upload.log', encoding='utf-8') ] ) self.logger = logging.getLogger(__name__) self.logger.info("Dify Knowledge Base Manager initialized") def get_dataset_info(self): """Get detailed dataset information including doc_form""" self.logger.info("📋 Getting dataset details...") endpoint = f"{self.BASE_URL}/datasets/{self.DATASET_ID}" headers = { 'Authorization': f'Bearer {self.API_KEY}', 'Content-Type': 'application/json' } try: response = requests.get(endpoint, headers=headers, timeout=10) if response.status_code == 200: dataset_info = response.json() self.logger.info("✅ Successfully retrieved dataset information") # Print dataset info for debugging if 'data' in dataset_info: self.logger.info(f"Dataset name: {dataset_info['data'].get('name', 'Unknown')}") self.logger.info(f"Document count: {dataset_info['data'].get('document_count', 'Unknown')}") # Check for doc_form information if 'doc_form' in dataset_info['data']: self.logger.info(f"Dataset doc_form: {dataset_info['data']['doc_form']}") else: self.logger.info("Dataset doc_form: Not found, using default") return dataset_info else: self.logger.error(f"❌ Failed to get dataset info: {response.text}") return None except Exception as e: self.logger.error(f"❌ Exception getting dataset info: {str(e)}") return None def upload_file_fixed(self, file_path): """Fixed file upload method""" filename = os.path.basename(file_path) self.logger.info(f"📤 Uploading file: {filename}") endpoint = f"{self.BASE_URL}/datasets/{self.DATASET_ID}/document/create_by_file" try: # Check file file_size = os.path.getsize(file_path) if file_size == 0: self.logger.error(f"❌ File is empty: {filename}") return False # Simplified request data - only include required parameters request_data = { "name": filename, "indexing_technique": "high_quality", # Remove doc_form and doc_language, let system use dataset defaults } # Read file with open(file_path, 'rb') as file: file_content = file.read() # Build multipart request files = { 'file': (filename, file_content, 'application/octet-stream') } # Parameters in data field data = { 'data': json.dumps(request_data) } headers = { 'Authorization': f'Bearer {self.API_KEY}' } self.logger.info("Sending file upload request...") response = requests.post( endpoint, files=files, data=data, headers=headers, timeout=120 ) self.logger.info(f"Response status code: {response.status_code}") if response.status_code in [200, 201]: result = response.json() self.logger.info(f"✅ File upload successful: {filename}") return True else: error_msg = response.text self.logger.error(f"❌ File upload failed: {error_msg}") return False except Exception as e: self.logger.error(f"❌ File upload exception: {str(e)}") return False def upload_text_fixed(self, file_path): """Fixed text upload method""" filename = os.path.basename(file_path) self.logger.info(f"📝 Uploading via text API: {filename}") # Extract text content text_content = self.extract_text_from_file(file_path) if text_content is None: self.logger.error(f"❌ Cannot extract text content: {filename}") return False if len(text_content.strip()) == 0: self.logger.error(f"❌ Extracted text content is empty: {filename}") return False endpoint = f"{self.BASE_URL}/datasets/{self.DATASET_ID}/document/create_by_text" try: # Get dataset info to determine correct doc_form dataset_info = self.get_dataset_info() doc_form = "text_model" # Default value if dataset_info and 'data' in dataset_info and 'doc_form' in dataset_info['data']: doc_form = dataset_info['data']['doc_form'] self.logger.info(f"Using dataset's doc_form: {doc_form}") else: self.logger.info("Using default doc_form: text_model") # Build request body - use correct doc_form payload = { "name": filename, "text": text_content, "indexing_technique": "high_quality", "process_rule": { "mode": "automatic", "rules": { "segmentation": { "separator": "\n\n", "max_tokens": 500 } } }, "doc_form": doc_form, # Remove doc_language or use dataset default } headers = { 'Authorization': f'Bearer {self.API_KEY}', 'Content-Type': 'application/json' } self.logger.info("Sending text upload request...") response = requests.post( endpoint, json=payload, headers=headers, timeout=120 ) self.logger.info(f"Response status code: {response.status_code}") if response.status_code in [200, 201]: self.logger.info(f"✅ Text upload successful: {filename}") return True else: error_msg = response.text self.logger.error(f"❌ Text upload failed: {error_msg}") return False except Exception as e: self.logger.error(f"❌ Text upload exception: {str(e)}") return False def extract_text_from_file(self, file_path): """Extract text content from file""" filename = os.path.basename(file_path) file_ext = os.path.splitext(filename)[1].lower() try: if file_ext == '.txt': with open(file_path, 'r', encoding='utf-8') as f: return f.read() elif file_ext == '.pdf': try: from PyPDF2 import PdfReader reader = PdfReader(file_path) text = "" for page in reader.pages: page_text = page.extract_text() if page_text: text += page_text + "\n" return text if text.strip() else None except ImportError: self.logger.error("❌ Please install PyPDF2: pip install pypdf2") return None except Exception as e: self.logger.error(f"❌ PDF parsing failed: {str(e)}") return None elif file_ext == '.docx': try: from docx import Document doc = Document(file_path) text = "" for paragraph in doc.paragraphs: if paragraph.text: text += paragraph.text + "\n" return text if text.strip() else None except ImportError: self.logger.error("❌ Please install python-docx: pip install python-docx") return None except Exception as e: self.logger.error(f"❌ DOCX parsing failed: {str(e)}") return None else: self.logger.warning(f"⚠️ Unsupported file format: {filename}") return None except Exception as e: self.logger.error(f"❌ Text extraction failed: {str(e)}") return None def try_different_api_keys(self): """Try using different types of API keys""" self.logger.info("🔑 Trying different API key types...") # Common API key prefixes api_key_types = { "app-": "Application API Key", "wk-": "Workspace API Key", "dataset-": "Dataset API Key", "worker-": "Worker API Key" } current_key = self.API_KEY current_prefix = current_key.split('-')[0] + '-' self.logger.info(f"Current API key type: {api_key_types.get(current_prefix, 'Unknown')}") # Test current key if self.test_api_connectivity(): self.logger.info("✅ Current API key is valid") return True else: self.logger.error("❌ Current API key is invalid, please try to get correct API key") return False def test_api_connectivity(self): """Test API connectivity""" endpoint = f"{self.BASE_URL}/datasets" headers = { 'Authorization': f'Bearer {self.API_KEY}', 'Content-Type': 'application/json' } try: response = requests.get(endpoint, headers=headers, timeout=10) return response.status_code == 200 except: return False def comprehensive_test_fixed(self): """Fixed comprehensive test""" self.logger.info("🧪 Starting fixed comprehensive test...") # 1. Get dataset information self.logger.info("1. Getting dataset information...") dataset_info = self.get_dataset_info() if not dataset_info: self.logger.error("❌ Cannot get dataset information") return False # 2. Test API key self.logger.info("2. Testing API key...") if not self.try_different_api_keys(): self.logger.error("❌ API key test failed") return False # 3. Create test file self.logger.info("3. Creating test file...") test_content = "Dify Knowledge Base API Test Document - Fixed Version" test_filepath = os.path.join(self.RESUMES_FOLDER, "api_test_fixed.txt") os.makedirs(self.RESUMES_FOLDER, exist_ok=True) with open(test_filepath, 'w', encoding='utf-8') as f: f.write(test_content) # 4. Priority test text upload (clearer error messages) self.logger.info("4. Testing text upload...") text_success = self.upload_text_fixed(test_filepath) # 5. Test file upload self.logger.info("5. Testing file upload...") file_success = self.upload_file_fixed(test_filepath) # Cleanup try: os.remove(test_filepath) except: pass if file_success or text_success: self.logger.info("✅ Comprehensive test passed") return True else: self.logger.error("❌ Comprehensive test failed") # Provide detailed solutions self.logger.info("\n🔧 Possible solutions:") self.logger.info("1. Check API access permissions in dataset settings") self.logger.info("2. Confirm the API key has correct permissions") self.logger.info("3. Check Dify service user authentication configuration") self.logger.info("4. Try using workspace API key instead of dataset API key") return False def batch_upload_files_fixed(self): """ Fixed batch upload method """ self.logger.info("🚀 Starting fixed batch upload process...") # Display configuration self.logger.info("⚙️ Current configuration:") self.logger.info(f" API Key: {self.API_KEY}") self.logger.info(f" Dataset ID: {self.DATASET_ID}") self.logger.info(f" Base URL: {self.BASE_URL}") # First run fixed comprehensive test self.logger.info("Running fixed comprehensive test first...") if not self.comprehensive_test_fixed(): self.logger.error("❌ Fixed comprehensive test failed, cannot continue batch upload") # Provide guidance for getting correct API key self.logger.info("\n📋 Steps to get correct API key:") self.logger.info("1. Login to Dify backend") self.logger.info("2. Go to 'Workspace Settings' -> 'API Keys'") self.logger.info("3. Create new API key (select appropriate permissions)") self.logger.info("4. Use API key starting with 'app-' or 'wk-'") return self.logger.info("✅ Fixed comprehensive test passed, starting batch upload...") # Check folder if not os.path.exists(self.RESUMES_FOLDER): self.logger.error(f"❌ Folder does not exist: {self.RESUMES_FOLDER}") return # Collect files document_files = [] for filename in os.listdir(self.RESUMES_FOLDER): file_path = os.path.join(self.RESUMES_FOLDER, filename) if os.path.isfile(file_path): document_files.append(file_path) total_files = len(document_files) self.logger.info(f"Found {total_files} document files") if total_files == 0: self.logger.warning("⚠️ No document files found") return # Display file list self.logger.info("File list:") for i, file_path in enumerate(document_files, 1): file_size = os.path.getsize(file_path) self.logger.info(f" {i}. {os.path.basename(file_path)} ({file_size/1024:.1f} KB)") success_count = 0 failed_files = [] # Start batch upload - priority to text upload (clearer error messages) for i, file_path in enumerate(document_files, 1): filename = os.path.basename(file_path) self.logger.info(f"\n📎 Processing file {i}/{total_files}: {filename}") # Priority to text upload if self.upload_text_fixed(file_path): success_count += 1 self.logger.info(f"✅ File {i} uploaded successfully via text") else: # If text upload fails, try file upload self.logger.info("Text upload failed, trying file upload...") if self.upload_file_fixed(file_path): success_count += 1 self.logger.info(f"✅ File {i} uploaded successfully via file") else: failed_files.append(filename) self.logger.error(f"❌ File {i} upload failed") # Delay if i < total_files: time.sleep(self.DELAY_BETWEEN_FILES) if i % self.BATCH_SIZE == 0 and i < total_files: time.sleep(self.DELAY_BETWEEN_BATCHES) # Output results self.logger.info(f"\n{'='*60}") self.logger.info("📊 Batch upload completed!") self.logger.info(f"✅ Success: {success_count}/{total_files}") if failed_files: self.logger.info(f"❌ Failed: {len(failed_files)}/{total_files}") for failed_file in failed_files: self.logger.info(f" - {failed_file}") else: self.logger.info("🎉 All files uploaded successfully!") def main(): """Main function""" manager = DifyKnowledgeBaseManager() manager.batch_upload_files_fixed() if __name__ == "__main__": main() ### ✔️ Expected Behavior upload the files sucessfully ### ❌ Actual Behavior It failed and indicate 400 error {"code":"invalid_param","message":"doc_form is different from the dataset doc_form.","status":40 0}
yindo added the 🐞 bug label 2026-02-21 20:07:24 -05:00
yindo closed this issue 2026-02-21 20:07:24 -05:00
yindo commented 2026-02-21 20:07:25 -05:00
Author
Owner
Copy Link
Copy Source

@dosubot[bot] commented on GitHub (Nov 19, 2025):

The 400 error is happening because the doc_form parameter in your file upload request is missing or doesn't match the dataset's expected value. When you upload via /document/create_by_file, if you don't explicitly set doc_form, it defaults to "text_model", but your dataset likely expects a different value (for example, "file_model" or something else) — and Dify strictly checks for a match between the upload and the dataset config (source, source).

To fix this, update your upload_file_fixed method so it always includes the correct doc_form in the request data. You can retrieve the expected value from your dataset info API (as you already do for text uploads). Here’s how you can modify your code:

def upload_file_fixed(self, file_path):
    filename = os.path.basename(file_path)
    self.logger.info(f"📤 Uploading file: {filename}")

    endpoint = f"{self.BASE_URL}/datasets/{self.DATASET_ID}/document/create_by_file"

    # Get dataset info to determine correct doc_form
    dataset_info = self.get_dataset_info()
    doc_form = "text_model"  # Default value
    if dataset_info and 'data' in dataset_info and 'doc_form' in dataset_info['data']:
        doc_form = dataset_info['data']['doc_form']
        self.logger.info(f"Using dataset's doc_form: {doc_form}")
    else:
        self.logger.info("Using default doc_form: text_model")

    request_data = {
        "name": filename,
        "indexing_technique": "high_quality",
        "doc_form": doc_form,  # <-- This is the key fix!
    }

    with open(file_path, 'rb') as file:
        file_content = file.read()

    files = {
        'file': (filename, file_content, 'application/octet-stream')
    }
    data = {
        'data': json.dumps(request_data)
    }
    headers = {
        'Authorization': f'Bearer {self.API_KEY}'
    }

    self.logger.info("Sending file upload request...")
    response = requests.post(
        endpoint,
        files=files,
        data=data,
        headers=headers,
        timeout=120
    )

    self.logger.info(f"Response status code: {response.status_code}")
    if response.status_code in [200, 201]:
        self.logger.info(f"✅ File upload successful: {filename}")
        return True
    else:
        error_msg = response.text
        self.logger.error(f"❌ File upload failed: {error_msg}")
        return False

This change should resolve the error and allow your uploads to succeed, as long as your API key has the necessary permissions.

To reply, just mention @dosu.

How did I do? Good | Irrelevant | Incorrect | Verbose | Hallucination | Report 🐛 | Other  Chat with Dosu Join Discord Share on X

@dosubot[bot] commented on GitHub (Nov 19, 2025): <!-- Answer --> The 400 error is happening because the `doc_form` parameter in your file upload request is missing or doesn't match the dataset's expected value. When you upload via `/document/create_by_file`, if you don't explicitly set `doc_form`, it defaults to `"text_model"`, but your dataset likely expects a different value (for example, `"file_model"` or something else) — and Dify strictly checks for a match between the upload and the dataset config ([source](https://github.com/langgenius/dify/blob/41bb6f310927b313e42a700a2b93d29cf7509b7e/api/controllers/service_api/dataset/document.py#L260-L344), [source](https://github.com/langgenius/dify/blob/41bb6f310927b313e42a700a2b93d29cf7509b7e/api/services/dataset_service.py#L324-L2028)). To fix this, update your `upload_file_fixed` method so it always includes the correct `doc_form` in the request data. You can retrieve the expected value from your dataset info API (as you already do for text uploads). Here’s how you can modify your code: ```python def upload_file_fixed(self, file_path): filename = os.path.basename(file_path) self.logger.info(f"📤 Uploading file: {filename}") endpoint = f"{self.BASE_URL}/datasets/{self.DATASET_ID}/document/create_by_file" # Get dataset info to determine correct doc_form dataset_info = self.get_dataset_info() doc_form = "text_model" # Default value if dataset_info and 'data' in dataset_info and 'doc_form' in dataset_info['data']: doc_form = dataset_info['data']['doc_form'] self.logger.info(f"Using dataset's doc_form: {doc_form}") else: self.logger.info("Using default doc_form: text_model") request_data = { "name": filename, "indexing_technique": "high_quality", "doc_form": doc_form, # <-- This is the key fix! } with open(file_path, 'rb') as file: file_content = file.read() files = { 'file': (filename, file_content, 'application/octet-stream') } data = { 'data': json.dumps(request_data) } headers = { 'Authorization': f'Bearer {self.API_KEY}' } self.logger.info("Sending file upload request...") response = requests.post( endpoint, files=files, data=data, headers=headers, timeout=120 ) self.logger.info(f"Response status code: {response.status_code}") if response.status_code in [200, 201]: self.logger.info(f"✅ File upload successful: {filename}") return True else: error_msg = response.text self.logger.error(f"❌ File upload failed: {error_msg}") return False ``` This change should resolve the error and allow your uploads to succeed, as long as your API key has the necessary permissions. <!-- Dosu Comment Footer --> *To reply, just mention [@dosu](https://go.dosu.dev/dosubot).* --- <sup>How did I do? [Good](https://app.dosu.dev/response-feedback/b32f2ccf-6fc6-4e36-89e5-6238ffd12952?feedback_type=great_response) | [Irrelevant](https://app.dosu.dev/response-feedback/b32f2ccf-6fc6-4e36-89e5-6238ffd12952?feedback_type=irrelevant_answer) | [Incorrect](https://app.dosu.dev/response-feedback/b32f2ccf-6fc6-4e36-89e5-6238ffd12952?feedback_type=incorrect_sources) | [Verbose](https://app.dosu.dev/response-feedback/b32f2ccf-6fc6-4e36-89e5-6238ffd12952?feedback_type=too_verbose) | [Hallucination](https://app.dosu.dev/response-feedback/b32f2ccf-6fc6-4e36-89e5-6238ffd12952?feedback_type=hallucination) | [Report 🐛](https://app.dosu.dev/response-feedback/b32f2ccf-6fc6-4e36-89e5-6238ffd12952?feedback_type=bug_report) | [Other](https://app.dosu.dev/response-feedback/b32f2ccf-6fc6-4e36-89e5-6238ffd12952?feedback_type=other)</sup>&nbsp;&nbsp;[![Chat with Dosu](https://dosu.dev/dosu-chat-badge.svg)](https://app.dosu.dev/6088a4f4-073a-43f3-b3d4-4f06fb0c05a0/ask?utm_source=github)&nbsp;[![Join Discord](https://img.shields.io/badge/join-5865F2?logo=discord&logoColor=white&label=)](https://go.dosu.dev/discord-bot)&nbsp;[![Share on X](https://img.shields.io/badge/X-share-black)](https://twitter.com/intent/tweet?text=%40dosu_ai%20helped%20me%20solve%20this%20issue!&url=https%3A//github.com/langgenius/dify/issues/28419)
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
Knowledge Pipeline
Proposal
WIP
cloud
dependencies
feat:webapp
good first issue
outdated
performance
pull-request
Mirrored from GitHub Pull Request
 python:uv
refactor
size:L
stale
status: accepting prs
status: blocked
status: in discussion
stt & tts
vulnerablity
web
wont-fix
🌊 feat:workflow
🌚 invalid
🐞 bug
👻 feat:rag
💪 enhancement
📚 documentation
🔨 feat:tools
🙋‍♂️ question
🙏 help wanted
🤔 cant-reproduce
🤖 feat:agent
No Label 🐞 bug
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: langgenius/dify#20426
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?