test_upload.py

import requests
import json

# 上传文件
pdf_path = r'C:\Users\YanBo\Desktop\ColPali_Multimodal_RAG_3pages.pdf'
with open(pdf_path, 'rb') as f:
    files = {'file': ('ColPali_Multimodal_RAG_3pages.pdf', f, 'application/pdf')}
    data = {'tenant_id': 'test-tenant'}
    response = requests.post('http://localhost:8000/api/v1/assets', files=files, data=data)
    print("Upload response:", response.status_code)
    asset_data = response.json()
    print(json.dumps(asset_data, indent=2, ensure_ascii=False))
    asset_id = asset_data['asset_id']

# 创建索引任务
job_data = {
    "tenant_id": "test-tenant",
    "asset_id": asset_id,
    "pipeline": {
        "mineru": {"enabled": True},
        "bridge": {"enabled": True},
        "kg": {"enabled": True}
    },
    "callback_url": None
}
response = requests.post('http://localhost:8000/api/v1/indexing/jobs', json=job_data)
print("\nCreate job response:", response.status_code)
job_result = response.json()
print(json.dumps(job_result, indent=2, ensure_ascii=False))
job_id = job_result['job_id']

# 查询任务状态
import time
for i in range(10):
    time.sleep(2)
    response = requests.get(f'http://localhost:8000/api/v1/indexing/jobs/{job_id}')
    job_status = response.json()
    print(f"\n[{i+1}] Job status: {job_status['status']}")
    if job_status['status'] in ['completed', 'failed']:
        print(json.dumps(job_status, indent=2, ensure_ascii=False))
        break