gmf_forge_ai_data.connectors

20class BaseConnector(ABC):
21    """
22    Abstract base class for all data source connectors.
23
24    Connectors handle one concern only: sourcing raw content and converting
25    it to Documents. They do NOT chunk, embed, or index — those steps follow.
26
27    Contract for all implementations:
28    - Every returned Document must have a non-empty ``id`` and ``content``.
29    - ``embedding`` is always ``None`` — the caller is responsible for embedding.
30    - Source-specific metadata (path, URL, container, etc.) must be stored in
31      ``document.metadata`` under consistent, documented keys.
32    - Files that cannot be read should be skipped with a printed warning rather
33      than crashing the entire load.
34    """
35
36    @abstractmethod
37    def load(self) -> List[Document]:
38        """
39        Load documents from the data source.
40
41        Returns:
42            List of Document objects with ``id``, ``content``, ``timestamp``,
43            and ``metadata`` populated. ``embedding`` is always ``None``.
44        """

 59class FilesystemConnector(BaseConnector):
 60    """
 61    Loads documents from a local directory into Document objects.
 62
 63    Scans a root directory (optionally recursively) and converts each matching
 64    file into a Document. The document ``id`` is a stable MD5 hash of the
 65    absolute file path, so re-runs produce consistent IDs for upsert workflows.
 66
 67    Supported formats:
 68    - Native text: .txt, .md, .rst, .csv, .json, .yaml, .yml, .py, .js, .ts,
 69      .java, .cpp, .c, .h, .cs, .html, .htm, .xml, .toml, .ini, .cfg, .env
 70    - PDF (.pdf) — requires ``pip install pypdf``
 71    - Word Document (.docx) — requires ``pip install python-docx``
 72
 73    Metadata keys set on every returned Document:
 74        ``source``       Absolute file path as a string
 75        ``file_name``    File name including extension
 76        ``extension``    Lowercase extension including dot (e.g. ``".md"``)
 77        ``size_bytes``   File size in bytes
 78        ``modified_at``  Last modification time in ISO 8601 format
 79
 80    Example:
 81        ```python
 82        from gmf_forge_ai_data.connectors import FilesystemConnector
 83
 84        connector = FilesystemConnector(
 85            root_path="/data/docs",
 86            extensions=[".txt", ".md", ".pdf"],
 87            recursive=True,
 88        )
 89        docs = connector.load()
 90        # docs is List[Document] — pass to a chunker next
 91        ```
 92    """
 93
 94    def __init__(
 95        self,
 96        root_path: Union[str, Path],
 97        extensions: Optional[List[str]] = None,
 98        recursive: bool = True,
 99        encoding: str = "utf-8-sig",
100        skip_empty: bool = True,
101    ):
102        """
103        Args:
104            root_path:   Root directory to scan.
105            extensions:  Explicit list of extensions to include (e.g.
106                         ``[".txt", ".md"]``). Include the leading dot.
107                         If ``None``, all supported formats are loaded.
108            recursive:   If ``True`` (default), scan subdirectories recursively.
109            encoding:    Text encoding for native text files (default ``"utf-8"``).
110            skip_empty:  If ``True`` (default), skip files that produce no
111                         text content after stripping whitespace.
112        """
113        self.root_path = Path(root_path).resolve()
114        self.extensions: Optional[Set[str]] = (
115            {ext.lower() for ext in extensions}
116            if extensions is not None
117            else None  # None = all supported formats
118        )
119        self.recursive = recursive
120        self.encoding = encoding
121        self.skip_empty = skip_empty
122        self._logger = BasicLogger(__name__)
123
124    def load(self) -> List[Document]:
125        """
126        Scan the root directory and return one Document per matching file.
127
128        Files that raise an error during reading are skipped with a warning
129        printed to stdout so the rest of the load continues uninterrupted.
130
131        Returns:
132            List of Document objects, one per successfully loaded file,
133            sorted by file path for deterministic ordering.
134
135        Raises:
136            FileNotFoundError: If ``root_path`` does not exist.
137            NotADirectoryError: If ``root_path`` is not a directory.
138        """
139        if not self.root_path.exists():
140            raise FileNotFoundError(
141                f"Root path does not exist: {self.root_path}"
142            )
143        if not self.root_path.is_dir():
144            raise NotADirectoryError(
145                f"Root path is not a directory: {self.root_path}"
146            )
147
148        pattern = "**/*" if self.recursive else "*"
149        all_files = sorted(p for p in self.root_path.glob(pattern) if p.is_file())
150
151        documents: List[Document] = []
152        for file_path in all_files:
153            ext = file_path.suffix.lower()
154            if not self._is_accepted(ext):
155                continue
156
157            try:
158                content = self._read_file(file_path, ext)
159            except Exception as e:
160                self._logger.warning("Skipping file", file=file_path.name, error=str(e))
161                continue
162
163            if self.skip_empty and not content.strip():
164                continue
165
166            stat = file_path.stat()
167            doc_id = "fs_" + hashlib.md5(str(file_path).encode()).hexdigest()[:12]
168            documents.append(Document(
169                id=doc_id,
170                content=content.strip(),
171                timestamp=datetime.fromtimestamp(stat.st_mtime),
172                metadata={
173                    "source": str(file_path),
174                    "file_name": file_path.name,
175                    "extension": ext,
176                    "size_bytes": stat.st_size,
177                    "modified_at": datetime.fromtimestamp(stat.st_mtime).isoformat(),
178                },
179            ))
180
181        return documents
182
183    # ── Private helpers ──────────────────────────────────────────────────────
184
185    def _is_accepted(self, ext: str) -> bool:
186        """Return True if this extension should be loaded."""
187        if self.extensions is not None:
188            return ext in self.extensions
189        # No filter specified: accept all natively supported + optional formats
190        return ext in _NATIVE_TEXT_EXTENSIONS or ext in {".pdf", ".docx"}
191
192    def _read_file(self, path: Path, ext: str) -> str:
193        if ext == ".pdf":
194            return _read_pdf(path)
195        if ext == ".docx":
196            return _read_docx(path)
197        return path.read_text(encoding=self.encoding, errors="replace")

 34class SharePointConnector(BaseConnector):
 35    """
 36    Loads files from a SharePoint document library via the Microsoft Graph API.
 37
 38    Authenticates using OAuth2 client credentials — no user interaction or
 39    browser required. Recursively lists all files under ``folder_path``,
 40    downloads their content, and extracts text. Unsupported binary formats
 41    are skipped automatically.
 42
 43    Supported file types:
 44    - Native text: .txt, .md, .rst, .csv, .json, .yaml, .yml, .py, .html, .xml
 45    - PDF (.pdf) — requires ``pip install pypdf``
 46    - Word Document (.docx) — requires ``pip install python-docx``
 47
 48    Metadata keys set on every returned Document:
 49        ``source``         Microsoft Graph download URL
 50        ``file_name``      File name including extension
 51        ``extension``      Lowercase extension including dot
 52        ``size_bytes``     File size in bytes
 53        ``modified_at``    Last modification time (ISO 8601 string)
 54        ``sharepoint_id``  SharePoint item ID
 55
 56    Example:
 57        ```python
 58        from gmf_forge_ai_data.connectors import SharePointConnector
 59
 60        connector = SharePointConnector(
 61            tenant_id="your-tenant-id",
 62            client_id="your-client-id",
 63            client_secret="your-client-secret",
 64            site_id="your-site-id",
 65            folder_path="/Shared Documents/KnowledgeBase",
 66        )
 67        docs = connector.load()
 68        ```
 69    """
 70
 71    _GRAPH_BASE = "https://graph.microsoft.com/v1.0"
 72    _TOKEN_URL = "https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
 73    _SUPPORTED_EXTENSIONS = {
 74        ".txt", ".md", ".rst", ".csv", ".json", ".yaml", ".yml",
 75        ".py", ".html", ".htm", ".xml", ".pdf", ".docx",
 76    }
 77
 78    def __init__(
 79        self,
 80        tenant_id: str,
 81        client_id: str,
 82        client_secret: str,
 83        site_id: str,
 84        folder_path: str = "/",
 85        drive_id: Optional[str] = None,
 86        ssl_cert_path: Optional[str] = None,
 87    ):
 88        """
 89        Args:
 90            tenant_id:      Azure AD tenant ID.
 91            client_id:      App registration (service principal) client ID.
 92            client_secret:  App registration client secret.
 93            site_id:        SharePoint site ID. Use ``"root"`` for the tenant
 94                            root site or provide the full GUID from the Graph API.
 95            folder_path:    Path inside the drive to load from. Use ``"/"``
 96                            (default) for the entire drive root.
 97            drive_id:       Drive ID or ``None`` (default) to use the site's
 98                            default document library (``root`` drive).
 99            ssl_cert_path:  Optional path to a CA bundle PEM file for
100                            environments with corporate SSL inspection.
101        """
102        self.tenant_id = tenant_id
103        self.client_id = client_id
104        self.client_secret = client_secret
105        self.site_id = site_id
106        self.folder_path = folder_path.rstrip("/") or "/"
107        self.drive_id = drive_id or "root"
108        self.ssl_cert_path = ssl_cert_path
109        self._token: Optional[str] = None
110        self._logger = BasicLogger(__name__)
111
112    def load(self) -> List[Document]:
113        """
114        Authenticate and load all supported files from the configured folder.
115
116        Returns:
117            List of Document objects, one per successfully loaded file.
118
119        Raises:
120            ImportError: If the ``requests`` package is not installed.
121            requests.HTTPError: If authentication or a Graph API call fails.
122        """
123        try:
124            import requests  # type: ignore
125        except ImportError:
126            raise ImportError(
127                "requests is required for SharePointConnector. "
128                "Install it with: pip install requests"
129            )
130
131        self._token = self._acquire_token(requests)
132        items = self._list_items(requests, self.folder_path)
133
134        documents: List[Document] = []
135        for item in items:
136            name: str = item.get("name", "")
137            ext = ("." + name.rsplit(".", 1)[-1]).lower() if "." in name else ""
138            if ext not in self._SUPPORTED_EXTENSIONS:
139                continue
140
141            try:
142                content = self._download_item(requests, item, ext)
143            except Exception as e:
144                self._logger.warning("Skipping file", file=name, error=str(e))
145                continue
146
147            if not content.strip():
148                continue
149
150            modified = item.get("lastModifiedDateTime", "")
151            size = item.get("size", 0)
152            download_url = item.get("@microsoft.graph.downloadUrl", "")
153            item_id = item.get("id", "")
154            doc_id = "sp_" + hashlib.md5(item_id.encode()).hexdigest()[:12]
155
156            documents.append(Document(
157                id=doc_id,
158                content=content.strip(),
159                timestamp=(
160                    datetime.fromisoformat(modified.replace("Z", "+00:00"))
161                    if modified else datetime.now()
162                ),
163                metadata={
164                    "source": download_url,
165                    "file_name": name,
166                    "extension": ext,
167                    "size_bytes": size,
168                    "modified_at": modified,
169                    "sharepoint_id": item_id,
170                },
171            ))
172
173        return documents
174
175    # ── Private helpers ──────────────────────────────────────────────────────
176
177    def _acquire_token(self, requests) -> str:
178        """Acquire an OAuth2 access token via client credentials flow."""
179        url = self._TOKEN_URL.format(tenant_id=self.tenant_id)
180        resp = requests.post(
181            url,
182            data={
183                "grant_type": "client_credentials",
184                "client_id": self.client_id,
185                "client_secret": self.client_secret,
186                "scope": "https://graph.microsoft.com/.default",
187            },
188            verify=self.ssl_cert_path or True,
189            timeout=30,
190        )
191        resp.raise_for_status()
192        return resp.json()["access_token"]
193
194    def _headers(self) -> Dict[str, str]:
195        return {"Authorization": f"Bearer {self._token}"}
196
197    def _list_items(self, requests, folder_path: str) -> List[Dict]:
198        """Recursively list all file items under folder_path."""
199        # Graph API uses /drive/ (singular) for the default document library
200        # and /drives/{id}/ when a specific drive ID is given.
201        drive_segment = (
202            "drive" if self.drive_id == "root"
203            else f"drives/{self.drive_id}"
204        )
205        if folder_path == "/":
206            url = (
207                f"{self._GRAPH_BASE}/sites/{self.site_id}"
208                f"/{drive_segment}/root/children"
209            )
210        else:
211            encoded = folder_path.lstrip("/")
212            url = (
213                f"{self._GRAPH_BASE}/sites/{self.site_id}"
214                f"/{drive_segment}/root:/{encoded}:/children"
215            )
216
217        items: List[Dict] = []
218        while url:
219            resp = requests.get(
220                url,
221                headers=self._headers(),
222                verify=self.ssl_cert_path or True,
223                timeout=30,
224            )
225            resp.raise_for_status()
226            data = resp.json()
227            for item in data.get("value", []):
228                if "folder" in item:
229                    # Recurse into sub-folders
230                    child_path = folder_path.rstrip("/") + "/" + item["name"]
231                    self._logger.info("Scanning subfolder", path=child_path)
232                    try:
233                        items.extend(self._list_items(requests, child_path))
234                    except Exception as e:
235                        self._logger.warning("Skipping subfolder", path=child_path, error=str(e))
236                else:
237                    items.append(item)
238            url = data.get("@odata.nextLink")  # follow pagination if present
239
240        return items
241
242    def _download_item(self, requests, item: Dict, ext: str) -> str:
243        """Download a file item's raw bytes and decode to text."""
244        download_url = item.get("@microsoft.graph.downloadUrl")
245        if not download_url:
246            raise ValueError(f"No download URL for item: {item.get('name')}")
247        resp = requests.get(
248            download_url,
249            verify=self.ssl_cert_path or True,
250            timeout=60,
251        )
252        resp.raise_for_status()
253        raw: bytes = resp.content
254        if ext == ".pdf":
255            return self._extract_pdf(raw)
256        if ext == ".docx":
257            return self._extract_docx(raw)
258        return raw.decode("utf-8-sig", errors="replace")
259
260    @staticmethod
261    def _extract_pdf(raw: bytes) -> str:
262        try:
263            import pypdf  # type: ignore
264        except ImportError:
265            raise ImportError(
266                "pypdf required for PDF support: pip install pypdf"
267            )
268        import logging
269        logging.getLogger("pypdf").setLevel(logging.ERROR)
270        reader = pypdf.PdfReader(io.BytesIO(raw))
271        return "\n".join(page.extract_text() or "" for page in reader.pages)
272
273    @staticmethod
274    def _extract_docx(raw: bytes) -> str:
275        try:
276            import docx  # type: ignore
277        except ImportError:
278            raise ImportError(
279                "python-docx required for DOCX support: pip install python-docx"
280            )
281        doc = docx.Document(io.BytesIO(raw))
282        return "\n".join(p.text for p in doc.paragraphs if p.text.strip())

 35class BlobStorageConnector(BaseConnector):
 36    """
 37    Loads blobs from an Azure Blob Storage container into Document objects.
 38
 39    Lists all blobs in the container (optionally filtered by a path prefix),
 40    downloads their content, and extracts text. Blobs whose extension is not
 41    in the supported set are skipped automatically.
 42
 43    Supported file types:
 44    - Native text: .txt, .md, .rst, .csv, .json, .yaml, .yml, .py, .html, .xml
 45    - PDF (.pdf) — requires ``pip install pypdf``
 46    - Word Document (.docx) — requires ``pip install python-docx``
 47
 48    Metadata keys set on every returned Document:
 49        ``source``       Full blob URL
 50        ``file_name``    Last path segment of the blob name
 51        ``blob_name``    Full blob path inside the container
 52        ``extension``    Lowercase extension including dot
 53        ``size_bytes``   Blob content length in bytes
 54        ``modified_at``  Last modified time in ISO 8601 format
 55        ``container``    Container name
 56
 57    Example::
 58
 59        from gmf_forge_ai_data.connectors import BlobStorageConnector
 60
 61        connector = BlobStorageConnector(
 62            account_name="myaccount",
 63            access_key="your-storage-account-access-key",
 64            container_name="documents",
 65            prefix="knowledge-base/",
 66        )
 67        docs = connector.load()
 68    """
 69
 70    def __init__(
 71        self,
 72        account_name: str,
 73        access_key: str,
 74        container_name: str,
 75        prefix: str = "",
 76        ssl_cert_path: Optional[str] = None,
 77    ):
 78        """
 79        Args:
 80            account_name:  Storage account name.
 81            access_key:    Storage account access key.
 82            container_name: Name of the blob container to read from.
 83            prefix:        Optional blob name prefix for filtering, acting
 84                           like a folder path (e.g. ``"knowledge-base/"``).
 85                           Pass ``""`` (default) to list the entire container.
 86            ssl_cert_path: Optional path to a CA bundle PEM file for
 87                           environments with corporate SSL inspection.
 88        """
 89        self.account_name = account_name
 90        self.access_key = access_key
 91        self.container_name = container_name
 92        self.connection_string = (
 93            f"DefaultEndpointsProtocol=https;"
 94            f"AccountName={account_name};"
 95            f"AccountKey={access_key};"
 96            f"EndpointSuffix=core.windows.net"
 97        )
 98        self.account_url = f"https://{account_name}.blob.core.windows.net"
 99        self.prefix = prefix
100        self.ssl_cert_path = ssl_cert_path
101        self._logger = BasicLogger(__name__)
102
103    def load(self) -> List[Document]:
104        """
105        List and download all supported blobs in the container under ``prefix``.
106
107        Returns:
108            List of Document objects, one per successfully loaded blob.
109
110        Raises:
111            ImportError: If the ``azure-storage-blob`` package is not installed.
112        """
113        try:
114            from azure.storage.blob import BlobServiceClient  # type: ignore
115        except ImportError:
116            raise ImportError(
117                "azure-storage-blob is required for BlobStorageConnector. "
118                "Install it with: pip install azure-storage-blob"
119            )
120
121        client = BlobServiceClient.from_connection_string(self.connection_string)
122        container_client = client.get_container_client(self.container_name)
123
124        documents: List[Document] = []
125        for blob in container_client.list_blobs(name_starts_with=self.prefix or None):
126            name: str = blob.name
127            ext = ("." + name.rsplit(".", 1)[-1]).lower() if "." in name else ""
128            if ext not in _SUPPORTED_EXTENSIONS:
129                continue
130
131            try:
132                content = self._download_blob(container_client, name, ext)
133            except Exception as e:
134                self._logger.warning("Skipping blob", blob=name, error=str(e))
135                continue
136
137            if not content.strip():
138                continue
139
140            modified = blob.last_modified
141            size = blob.size or 0
142            blob_url = f"{self.account_url}/{self.container_name}/{name}"
143            doc_id = "blob_" + hashlib.md5(blob_url.encode()).hexdigest()[:12]
144            file_name = name.split("/")[-1]
145
146            documents.append(Document(
147                id=doc_id,
148                content=content.strip(),
149                timestamp=(
150                    modified if isinstance(modified, datetime) else datetime.now()
151                ),
152                metadata={
153                    "source": blob_url,
154                    "file_name": file_name,
155                    "blob_name": name,
156                    "extension": ext,
157                    "size_bytes": size,
158                    "modified_at": modified.isoformat() if modified else "",
159                    "container": self.container_name,
160                },
161            ))
162
163        return documents
164
165    # ── Private helpers ──────────────────────────────────────────────────────
166
167    def _download_blob(self, container_client, name: str, ext: str) -> str:
168        raw: bytes = container_client.download_blob(name).readall()
169        if ext == ".pdf":
170            return self._extract_pdf(raw)
171        if ext == ".docx":
172            return self._extract_docx(raw)
173        return raw.decode("utf-8-sig", errors="replace")
174
175    @staticmethod
176    def _extract_pdf(raw: bytes) -> str:
177        try:
178            import pypdf  # type: ignore
179        except ImportError:
180            raise ImportError(
181                "pypdf required for PDF support: pip install pypdf"
182            )
183        import logging
184        logging.getLogger("pypdf").setLevel(logging.ERROR)
185        reader = pypdf.PdfReader(io.BytesIO(raw))
186        return "\n".join(page.extract_text() or "" for page in reader.pages)
187
188    @staticmethod
189    def _extract_docx(raw: bytes) -> str:
190        try:
191            import docx  # type: ignore
192        except ImportError:
193            raise ImportError(
194                "python-docx required for DOCX support: pip install python-docx"
195            )
196        doc = docx.Document(io.BytesIO(raw))
197        return "\n".join(p.text for p in doc.paragraphs if p.text.strip())

 89class SoapConnector(BaseConnector):
 90    """
 91    Generic SOAP / WSDL connector base class.
 92
 93    Manages the complete connection lifecycle for a single WSDL endpoint:
 94    authentication via HTTP Basic Auth, optional SSL certificate configuration,
 95    lazy connection initialisation, method invocation, and automatic response
 96    serialisation from zeep objects to plain Python dicts.
 97
 98    Subclass this to add service-specific method wrappers and implement
 99    ``load()`` with your document-extraction logic.
100
101    Parameters
102    ----------
103    wsdl_url:
104        Full URL to the WSDL endpoint, e.g.
105        ``"https://example.com/Api/AdminApi.svc?wsdl"``.
106    username:
107        HTTP Basic Auth username for the service.
108    password:
109        HTTP Basic Auth password for the service.
110    credentials_type_qname:
111        Fully qualified name of the SOAP ``Credentials`` complex type as
112        defined in the WSDL, e.g.
113        ``"{http://schemas.datacontract.org/2004/07/My.Api}Credentials"``.
114        When supplied, a typed credentials object is created and injected
115        automatically into every ``call()`` invocation as the
116        ``credentials`` keyword argument.
117        When ``None`` (default), no credentials object is injected — use
118        this when the service authenticates via HTTP headers only.
119    credentials_username_field:
120        Field name for the username inside the Credentials complex type.
121        Default: ``"Username"``.
122    credentials_password_field:
123        Field name for the password inside the Credentials complex type.
124        Default: ``"Password"``.
125    ssl_verify:
126        Whether to verify SSL certificates. Set to ``False`` for services
127        using self-signed certificates (issues a warning).
128        Default: ``True``.
129    ssl_cert_path:
130        Path to a CA certificate bundle PEM file for corporate SSL
131        inspection proxies. Sets ``REQUESTS_CA_BUNDLE`` and
132        ``SSL_CERT_FILE`` environment variables automatically.
133    connect_on_init:
134        If ``True``, call ``connect()`` immediately in ``__init__``.
135        If ``False`` (default), connection is established lazily on the
136        first ``call()`` invocation.
137
138    Raises
139    ------
140    ImportError
141        If the ``zeep`` package is not installed.
142    """
143
144    def __init__(
145        self,
146        wsdl_url: str,
147        username: str,
148        password: str,
149        credentials_type_qname: Optional[str] = None,
150        credentials_username_field: str = "Username",
151        credentials_password_field: str = "Password",
152        ssl_verify: Union[bool, str] = True,
153        ssl_cert_path: Optional[str] = None,
154        connect_on_init: bool = False,
155    ) -> None:
156        try:
157            import zeep  # noqa: F401
158        except ImportError as exc:
159            raise ImportError(
160                "zeep is required for SoapConnector. "
161                "Install it with:  pip install zeep"
162            ) from exc
163
164        self.wsdl_url = wsdl_url
165        self.username = username
166        self.password = password
167        self._credentials_type_qname = credentials_type_qname
168        self._credentials_username_field = credentials_username_field
169        self._credentials_password_field = credentials_password_field
170        self.ssl_verify: Union[bool, str] = ssl_verify
171        self.ssl_cert_path = ssl_cert_path
172
173        self._client: Optional[Any] = None       # zeep.Client
174        self._credentials: Optional[Any] = None  # typed Credentials object
175        self._logger = BasicLogger(__name__)
176
177        if ssl_cert_path:
178            import os as _os
179            _os.environ.setdefault("REQUESTS_CA_BUNDLE", ssl_cert_path)
180            _os.environ.setdefault("SSL_CERT_FILE", ssl_cert_path)
181            self.ssl_verify = ssl_cert_path
182
183        if not ssl_verify:
184            import urllib3
185            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
186
187        if connect_on_init:
188            self.connect()
189
190    # ------------------------------------------------------------------ #
191    # Connection management                                                #
192    # ------------------------------------------------------------------ #
193
194    def connect(self) -> None:
195        """
196        Establish the SOAP connection.
197
198        Creates a ``zeep.Client`` configured with HTTP Basic Auth and the
199        SSL settings supplied at construction. If ``credentials_type_qname``
200        was provided, also builds the typed Credentials object that will be
201        injected into every subsequent ``call()``.
202
203        This method is idempotent — calling it when already connected is a
204        no-op. It is called automatically by ``call()`` on first use, so
205        explicit calls are optional unless you want to fail fast at startup.
206
207        Raises
208        ------
209        Exception
210            Any exception raised by the zeep ``Client`` constructor (e.g.
211            network unreachable, invalid WSDL URL) is logged and re-raised.
212        """
213        if self._client is not None:
214            return
215
216        from zeep import Client
217        from zeep.transports import Transport
218        from requests import Session
219        from requests.auth import HTTPBasicAuth
220
221        try:
222            session = Session()
223            session.auth = HTTPBasicAuth(self.username, self.password)
224            session.verify = self.ssl_verify
225            transport = Transport(session=session)
226            self._client = Client(wsdl=self.wsdl_url, transport=transport)
227            self._logger.info("SOAP client connected", wsdl=self.wsdl_url)
228        except Exception as exc:
229            self._logger.error(
230                "Failed to connect SOAP client",
231                wsdl=self.wsdl_url,
232                error=str(exc),
233            )
234            raise
235
236        # Build the typed Credentials object if the WSDL uses one
237        if self._credentials_type_qname:
238            try:
239                cred_type = self._client.get_type(self._credentials_type_qname)
240                self._credentials = cred_type(**{
241                    self._credentials_username_field: self.username,
242                    self._credentials_password_field: self.password,
243                })
244                self._logger.info(
245                    "SOAP credentials object created",
246                    type_qname=self._credentials_type_qname,
247                )
248            except Exception as exc:
249                self._logger.warning(
250                    "Could not create typed credentials object — "
251                    "credentials will NOT be injected automatically",
252                    type_qname=self._credentials_type_qname,
253                    error=str(exc),
254                )
255
256    def disconnect(self) -> None:
257        """Close the underlying requests session and reset the client."""
258        if self._client is not None:
259            try:
260                self._client.transport.session.close()
261            except Exception:
262                pass
263            self._client = None
264            self._credentials = None
265            self._logger.info("SOAP client disconnected", wsdl=self.wsdl_url)
266
267    # ------------------------------------------------------------------ #
268    # Method invocation                                                    #
269    # ------------------------------------------------------------------ #
270
271    def call(self, method: str, **kwargs: Any) -> Any:
272        """
273        Invoke a SOAP service method and return a serialised plain dict.
274
275        Connects automatically on first use. If a typed Credentials object
276        was built during ``connect()``, it is injected as the ``credentials``
277        keyword argument before the call — no need to pass it manually.
278
279        Parameters
280        ----------
281        method:
282            Name of the SOAP operation as declared in the WSDL, e.g.
283            ``"GetDocumentsByFolder"``.
284        **kwargs:
285            Additional keyword arguments forwarded to the operation, e.g.
286            ``folderId="abc-123"``.
287
288        Returns
289        -------
290        Any
291            Python-native value (dict, list, str, int, etc.) produced by
292            ``zeep.helpers.serialize_object()``. All zeep-specific types
293            are stripped so callers work with plain Python objects.
294
295        Raises
296        ------
297        AttributeError
298            If ``method`` is not found in the WSDL service definition.
299        Exception
300            Any SOAP fault or transport-level error raised by zeep.
301        """
302        self.connect()  # no-op if already connected
303
304        if self._credentials is not None and "credentials" not in kwargs:
305            kwargs["credentials"] = self._credentials
306
307        operation = getattr(self._client.service, method)
308
309        self._logger.debug("Calling SOAP method", method=method)
310        try:
311            raw = operation(**kwargs)
312        except Exception as exc:
313            err = str(exc)
314            # SOAP faults that are navigational noise (folder not found,
315            # invalid id) stay at debug.  All other faults log at warning —
316            # the caller decides whether to escalate further.
317            _navigational = ("not be found", "invalid", "could not be found")
318            if any(e in err.lower() for e in _navigational):
319                self._logger.debug("SOAP method call failed", method=method, error=err)
320            else:
321                self._logger.warning("SOAP method call failed", method=method, error=err)
322            raise
323
324        from zeep import helpers
325        result = helpers.serialize_object(raw, target_cls=dict)
326        self._logger.debug("SOAP method returned", method=method)
327        return result
328
329    # ------------------------------------------------------------------ #
330    # Text extraction utility                                              #
331    # ------------------------------------------------------------------ #
332
333    def extract_text(
334        self,
335        data: Union[bytes, str],
336        filename: str,
337    ) -> str:
338        """
339        Extract plain text from raw file bytes or base64-encoded data.
340
341        Dispatches by file extension:
342        - ``.pdf``  — extracts text via ``pypdf`` (``pip install pypdf``)
343        - ``.docx`` — extracts text via ``python-docx`` (``pip install python-docx``)
344        - anything else — decodes as UTF-8
345
346        Call this inside your ``load()`` implementation after retrieving the
347        raw file content from the service::
348
349            data    = self.call("GetDocumentData", documentId=doc_id)
350            content = self.extract_text(data["Data"], "policy.pdf")
351
352        Parameters
353        ----------
354        data:
355            Raw bytes **or** a base64-encoded string as typically returned by
356            SOAP document download operations.
357        filename:
358            File name (or any string ending with the correct extension) used
359            to determine the extraction strategy, e.g. ``"policy.pdf"``.
360
361        Returns
362        -------
363        str
364            Extracted plain text. Returns an empty string when ``data`` is
365            empty or ``None``.
366        """
367        if not data:
368            return ""
369
370        # Normalise to bytes
371        if isinstance(data, str):
372            try:
373                file_bytes = base64.b64decode(data)
374            except Exception:
375                # Not base64 — treat as raw text
376                return data
377        else:
378            file_bytes = data
379
380        ext = ("." + filename.rsplit(".", 1)[-1]).lower() if "." in filename else ""
381
382        if ext == ".pdf":
383            return self._extract_pdf(file_bytes, filename)
384        if ext == ".docx":
385            return self._extract_docx(file_bytes, filename)
386
387        # Default: decode as UTF-8
388        try:
389            return file_bytes.decode("utf-8", errors="replace")
390        except Exception as exc:
391            self._logger.warning(
392                "Text decode failed, returning empty string",
393                file=filename,
394                error=str(exc),
395            )
396            return ""
397
398    # ------------------------------------------------------------------ #
399    # Abstract interface (developer must implement)                        #
400    # ------------------------------------------------------------------ #
401
402    @abstractmethod
403    def load(self) -> List[Document]:
404        """
405        Load documents from the SOAP service.
406
407        This method is entirely the developer's responsibility. Use
408        ``self.call()`` to invoke service methods and ``self.extract_text()``
409        to convert file bytes to plain text. Assemble and return
410        ``Document`` objects with the ``id``, ``content``, and ``metadata``
411        fields that matter for your use case.
412
413        Returns
414        -------
415        List[Document]
416            One Document per logical document retrieved from the service.
417            ``embedding`` should always be ``None`` — the caller embeds.
418        """
419
420    # ------------------------------------------------------------------ #
421    # Private helpers                                                      #
422    # ------------------------------------------------------------------ #
423
424    def _extract_pdf(self, file_bytes: bytes, filename: str) -> str:
425        try:
426            import pypdf  # type: ignore
427        except ImportError:
428            self._logger.warning(
429                "pypdf not installed — skipping PDF extraction",
430                file=filename,
431            )
432            return f"[PDF extraction skipped — install pypdf: {filename}]"
433        try:
434            reader = pypdf.PdfReader(io.BytesIO(file_bytes))
435            return "\n\n".join(
436                page.extract_text() or "" for page in reader.pages
437            ).strip()
438        except Exception as exc:
439            self._logger.warning(
440                "PDF extraction failed", file=filename, error=str(exc)
441            )
442            return f"[PDF extraction error: {filename}]"
443
444    def _extract_docx(self, file_bytes: bytes, filename: str) -> str:
445        try:
446            import docx  # type: ignore
447        except ImportError:
448            self._logger.warning(
449                "python-docx not installed — skipping DOCX extraction",
450                file=filename,
451            )
452            return f"[DOCX extraction skipped — install python-docx: {filename}]"
453        try:
454            doc = docx.Document(io.BytesIO(file_bytes))
455            return "\n\n".join(
456                p.text for p in doc.paragraphs if p.text.strip()
457            ).strip()
458        except Exception as exc:
459            self._logger.warning(
460                "DOCX extraction failed", file=filename, error=str(exc)
461            )
462            return f"[DOCX extraction error: {filename}]"
463
464    # ------------------------------------------------------------------ #
465    # Convenience                                                          #
466    # ------------------------------------------------------------------ #
467
468    def _make_doc_id(self, *parts: str) -> str:
469        """
470        Generate a stable document ID from one or more string parts.
471
472        Convenience helper for ``load()`` implementations to produce
473        consistent, collision-resistant IDs::
474
475            doc_id = self._make_doc_id(str(doc["DocumentId"]))
476        """
477        combined = "_".join(str(p) for p in parts)
478        return "soap_" + hashlib.md5(combined.encode()).hexdigest()[:12]
479
480    def __enter__(self) -> "SoapConnector":
481        self.connect()
482        return self
483
484    def __exit__(self, *_: Any) -> None:
485        self.disconnect()