# Source code for opentelemetry.exporter.otlp.json.file._log_exporter

    # Copyright The OpenTelemetry Authors
    # SPDX-License-Identifier: Apache-2.0

    import logging
    from collections.abc import Sequence
    from os import PathLike
    from typing import IO, Any, overload

    from opentelemetry.exporter.otlp.json.common._log_encoder import encode_logs
    from opentelemetry.exporter.otlp.json.file._internal import _FileExporter
    from opentelemetry.sdk._logs import ReadableLogRecord
    from opentelemetry.sdk._logs.export import (
        LogRecordExporter,
        LogRecordExportResult,
    )
    from opentelemetry.sdk._shared_internal import DuplicateFilter

    _logger = logging.getLogger(__name__)
    _logger.addFilter(DuplicateFilter())

    def _encode_logs_to_dict(
        batch: Sequence[ReadableLogRecord],
    ) -> dict[str, Any] | None:
        data = encode_logs(batch)
        return data.to_dict() if data.resource_logs else None

    class FileLogExporter(LogRecordExporter):
        @overload
        def __init__(
            self,
            path: str | PathLike[str],
        ) -> None: ...

        @overload
        def __init__(
            self,
            *,
            stream: IO[str],
        ) -> None: ...

        @overload
        def __init__(
            self,
        ) -> None: ...

        def __init__(
            self,
            path: str | PathLike[str] | None = None,
            *,
            stream: IO[str] | None = None,
        ) -> None:
            self._exporter: _FileExporter[Sequence[ReadableLogRecord]] = (
                _FileExporter(
                    encode=_encode_logs_to_dict,
                    kind="logs",
                    logger=_logger,
                    path=path,
                    stream=stream,
                )
            )

        def export(
            self, batch: Sequence[ReadableLogRecord]
        ) -> LogRecordExportResult:
            return (
                LogRecordExportResult.SUCCESS
                if self._exporter.export(batch)
                else LogRecordExportResult.FAILURE
            )

        def shutdown(self) -> None:
            self._exporter.shutdown()

        # pylint: disable-next=no-self-use

        def force_flush(self, timeout_millis: float = 10_000) -> bool:
            return True