Function _send_chunks_to_es

backend/data_process/tasks.py:613–701

(
    chunks: List[Dict[str, Any]],
    index_name: str,
    authorization: str | None,
    task_id: Optional[str] = None,
    source: str = "",
    original_filename: str = "",
    large_mode: bool = False,
)

def _send_chunks_to_es(
    chunks: List[Dict[str, Any]],
    index_name: str,
    authorization: str | None,
    task_id: Optional[str] = None,
    source: str = "",
    original_filename: str = "",
    large_mode: bool = False,
) -> Dict[str, Any]:
    async def _post():
        elasticsearch_url = ELASTICSEARCH_SERVICE
        if not elasticsearch_url:
            raise _build_forward_error(
                message="ELASTICSEARCH_SERVICE env is not set",
                index_name=index_name,
                source=source,
                original_filename=original_filename,
            )
        route_url = f"/indices/{index_name}/documents"
        full_url = elasticsearch_url + route_url
        headers = {"Content-Type": "application/json"}
        if authorization:
            headers["Authorization"] = authorization
        if task_id:
            headers["X-Task-Id"] = task_id
        try:
            connector = aiohttp.TCPConnector(verify_ssl=False)
            timeout = aiohttp.ClientTimeout(total=600)

            request_params: Dict[str, str] = {}

            if large_mode:
                request_params["large_mode"] = "true"

            async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
                async with session.post(
                    full_url,
                    headers=headers,
                    json=chunks,
                    params=request_params,
                    raise_for_status=False
                ) as response:
                    text = await response.text()
                    status = response.status
                    parsed_body = _parse_json_or_none(text)

                    if status >= 400:
                        error_code = _extract_error_code_from_es_response(
                            parsed_body, text)
                        if error_code:
                            raise Exception(json.dumps({
                                "error_code": error_code
                            }, ensure_ascii=False))

                        raise Exception(
                            f"ElasticSearch service returned HTTP {status}")

                    result = parsed_body if isinstance(parsed_body, dict) else await response.json()