feat: 70 write documents to kafka

.env.example

@@ -3,7 +3,30 @@
 #################
 DEBUG=
 
-# Proxy
+
+############
+# Postgres #
+############
+RISKI__DB__HOSTNAME=db
+RISKI__DB__NAME=example_db
+RISKI__DB__PASSWORD=password
+RISKI__DB__USER=postgres
+RISKI__DB__PORT=5432
+RISKI__DB__BATCH_SIZE=100
+
+#########
+# Kafka #
+#########
+RISKI__KAFKA__SERVER=<kafka-broker-url>
+RISKI__KAFKA__SECURITY=true
+RISKI__KAFKA__TOPIC=<kafka-topic>
+RISKI__KAFKA__CA_B64=<ca_in_base64>
+RISKI__KAFKA__PKCS12_DATA=<auth_pkcs12_in_base64>
+RISKI__KAFKA__PKCS12_PW=<auth_pkcs12_pw_in_base64>
+
+#########
+# Proxy #
+#########
 HTTP_PROXY=
 HTTPS_PROXY=
 NO_PROXY=

.github/workflows/extractor-tests.yml

@@ -38,6 +38,16 @@ jobs:
       OPENAI_API_BASE: https://example.de
       OPENAI_API_VERSION: 2024-08-01-preview
 
+      TIKTOKEN_CACHE_DIR: tiktoken_cache
+      RISKI_OPENAI_EMBEDDING_MODEL: text-embedding-3-large
+      RISKI_EXTRACTOR__TEST__DB_NAME: test_db
+      RISKI_EXTRACTOR__TEST__DB_USER: postgres
+      RISKI_EXTRACTOR__TEST__DB_PASSWORD: password
+      RISKI_EXTRACTOR__TEST__DB_HOSTNAME: 127.0.0.1
+      RISKI_EXTRACTOR__TEST__DB_PORT: 5432
+      RISKI__KAFKA__SERVER: <no-server>
+      RISKI__KAFKA__SECURITY: false
+
     defaults:
       run:
         working-directory: riski-extractor

riski-core/src/core/kafka/security.py

@@ -0,0 +1,81 @@
+"""Security helpers for configuring Kafka TLS/mTLS."""
+
+from base64 import b64decode
+from logging import Logger
+from pathlib import Path
+from ssl import SSLContext, create_default_context
+from tempfile import TemporaryDirectory
+
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.serialization import pkcs12
+from faststream.security import BaseSecurity
+
+from src.logtools import getLogger
+
+logger: Logger = getLogger(__name__)
+
+
+def setup_security(kafka_pkcs12_data: str, kafka_pkcs12_pw: str, kafka_ca_b64: str) -> BaseSecurity:
+    """Set up Kafka security configuration based on application settings.
+
+    Returns:
+        BaseSecurity: The configured security object for Kafka.
+
+    Raises:
+        ValueError: If required environment variables are not set or if there are issues
+            loading the PKCS#12 file.
+    """
+    with TemporaryDirectory() as tempdir:
+        tempdir_path: Path = Path(tempdir)
+        logger.debug("Setting up Kafka mTLS security using environment variables.")
+
+        # Unpack CA file
+        ca_data: str | None = kafka_ca_b64
+        ca_data = b64decode(s=ca_data).decode(encoding="utf-8")
+
+        # Unpack PKCS#12 file
+        pkcs12_data: str | None = kafka_pkcs12_data
+        pkcs12_bytes = b64decode(s=pkcs12_data)
+
+        # Unpack PKCS#12 password
+        pkcs12_pw: str | None = kafka_pkcs12_pw
+        pkcs12_pw_bytes: bytes = b64decode(s=pkcs12_pw)
+
+        # Extract the private key and certificate from the PKCS#12 file
+        try:
+            private_key, certificate, _ = pkcs12.load_key_and_certificates(
+                data=pkcs12_bytes,
+                password=pkcs12_pw_bytes,
+            )
+        except Exception as e:
+            raise ValueError(f"Failed to load PKCS#12 file: {e}")
+
+        if private_key is None or certificate is None:
+            raise ValueError("PKCS#12 file does not contain a private key and certificate.")
+
+        # Write cert and key to temporary files
+        cert_file: Path = tempdir_path / "kafka.cert"
+        key_file: Path = tempdir_path / "kafka.key"
+        with open(file=cert_file, mode="wb") as f:
+            f.write(certificate.public_bytes(encoding=serialization.Encoding.PEM))
+        with open(file=key_file, mode="wb") as f:
+            f.write(
+                private_key.private_bytes(
+                    encoding=serialization.Encoding.PEM,
+                    format=serialization.PrivateFormat.TraditionalOpenSSL,
+                    encryption_algorithm=serialization.NoEncryption(),
+                )
+            )
+
+        # Create SSL context
+        ssl_context: SSLContext = create_default_context(
+            cadata=ca_data,
+        )
+
+        # Load client cert and key
+        ssl_context.load_cert_chain(
+            certfile=cert_file,
+            keyfile=key_file,
+        )
+
+    return BaseSecurity(ssl_context=ssl_context, use_ssl=True)

riski-core/src/core/settings/core.py

@@ -1,5 +1,6 @@
 from core.settings.db import DatabaseSettings
 from core.settings.genai import GenAISettings
+from core.settings.kafka import KafkaSettings
 from core.settings.testdb import TestDBSettings
 from pydantic import BaseModel, Field
 
@@ -15,6 +16,8 @@ class CoreSettings(BaseModel):
         default_factory=lambda: GenAISettings(),
     )
 
+    kafka: KafkaSettings = Field(default_factory=lambda: KafkaSettings(), description="Kafka related settings")
+
     testdb: TestDBSettings = Field(
         description="Test DB related settings",
         default_factory=lambda: TestDBSettings(),

riski-core/src/core/settings/kafka.py

@@ -0,0 +1,31 @@
+from pydantic import BaseModel, Field
+
+
+class KafkaSettings(BaseModel):
+    """
+    Kafka configuration settings.
+    """
+
+    # === Kafka Settings ===
+    server: str = Field(
+        description="Kafka Server URL",
+    )
+    topic: str = Field(
+        default="lhm-riski-parse",
+        description="Kafka Topic Name",
+    )
+    security: bool = Field(
+        description="Enable mTLS security for accessing Kafka Server",
+    )
+    ca_b64: str | None = Field(
+        default=None,
+        description="Kafka Server CA (B64 Encoded)",
+    )
+    pkcs12_data: str | None = Field(
+        default=None,
+        description="Kafka P12 (B64 Encoded)",
+    )
+    pkcs12_pw: str | None = Field(
+        default=None,
+        description="Kafka P12 Password (B64 Encoded)",
+    )

riski-extractor/app.py

@@ -4,6 +4,8 @@
 
 from config.config import Config, get_config
 from core.db.db import create_db_and_tables, init_db
+from core.kafka.security import setup_security
+from faststream.kafka import KafkaBroker
 from src.extractor.city_council_faction_extractor import CityCouncilFactionExtractor
 from src.extractor.city_council_meeting_extractor import CityCouncilMeetingExtractor
 from src.extractor.city_council_meeting_template_extractor import CityCouncilMeetingTemplateExtractor
@@ -23,7 +25,7 @@
 async def main():
     config = get_config()
     config.print_config()
-    logger = getLogger()
+    logger = getLogger(__name__)
     version = get_version()
 
     init_db(config.core.db.database_url)
@@ -63,8 +65,13 @@ async def main():
     city_council_motion_extractor.run()
     logger.info("Extracted City Council Motions")
 
-    async with Filehandler() as filehandler:
-        await filehandler.download_and_persist_files(batch_size=config.core.db.batch_size)
+    broker = await createKafkaBroker(config, logger)
+    try:
+        async with Filehandler(kafkaBroker=broker) as filehandler:
+            await filehandler.download_and_persist_files(batch_size=config.core.db.batch_size)
+    finally:
+        await broker.stop()
+        logger.info("Broker closed.")
 
     confidential_file_deleter = ConfidentialFileDeleter()
     confidential_file_deleter.delete_confidential_files()
@@ -75,5 +82,22 @@ async def main():
     return 0
 
 
+async def createKafkaBroker(config: Config, logger: Logger) -> KafkaBroker:
+    security = None
+    if config.core.kafka.security:
+        security = setup_security(config.core.kafka.pkcs12_data, config.core.kafka.pkcs12_pw, config.core.kafka.ca_b64)
+
+    # Kafka Broker and FastStream app setup
+    broker = KafkaBroker(bootstrap_servers=config.core.kafka.server, security=security)
+    logger.debug("Connecting to Broker...")
+    try:
+        await broker.connect()
+        logger.info("Broker connected.")
+        return broker
+    except Exception as e:
+        logger.exception(f"Failed to connect to broker. - {e}")
+        raise
+
+
 if __name__ == "__main__":
     sys.exit(asyncio.run(main()))

riski-extractor/logconf.yaml

@@ -27,7 +27,7 @@ loggers:
     handlers: [ console ]
     propagate: no
   riski-extractor:
-    level: INFO
+    level: DEBUG
     handlers: [ console ]
     propagate: no
   httpx:

riski-extractor/pyproject.toml

@@ -15,6 +15,8 @@ dependencies = [
     "tokenizers==0.21.2",
     "torch==2.8.0",
     "torchvision==0.23.0",
+    "cryptography>=46.0.5",
+    "faststream[kafka]>=0.6.5",
     "pytest-asyncio>=1.3.0",
     "truststore>=0.10.4",
     "gitpython>=3.1.45",
@@ -42,7 +44,7 @@ dev = [
     "langchain-docling>=1.1.0",
     "langchain-openai>=0.3.32",
     "pytest==8.4.2",
-    "faststream[kafka]==0.6.1",
+    "faststream[kafka]==0.6.5",    
     "ruff>=0.14.11",
     "python-dotenv>=1.2.1",
 ]

riski-extractor/src/extractor/base_extractor.py

@@ -37,7 +37,7 @@ def __init__(
         else:
             self.client = Client(timeout=config.request_timeout)
 
-        self.logger = getLogger()
+        self.logger = getLogger(__name__)
         self.base_url = base_url
         self.base_path = base_path
         self.parser = parser
@@ -78,7 +78,7 @@ def _get_object_html(self, link: str) -> str:
     def _get_sanitized_url(self, unsanitized_path: str) -> str:
         return f"{self.base_url}/{unsanitized_path.lstrip('./')}"
 
-    def run(self) -> list[T]:
+    def run(self):
         try:
             # Initial request for cookies, sessionID etc.
             self._initial_request()
@@ -105,8 +105,9 @@ def run(self) -> list[T]:
                     break
 
                 self._get_next_page(path=results_per_page_redirect_path, next_page_link=nav_top_next_link)
-        except Exception:
-            self.logger.exception("Error extracting objects")
+
+        except Exception as e:
+            self.logger.exception(f"Error extracting objects. - {e}")
 
     def _parse_objects_from_links(self, object_links: list[str]):
         for link in object_links:

riski-extractor/src/filehandler/filehandler.py

@@ -7,7 +7,9 @@
 from config.config import Config, get_config
 from core.db.db_access import request_batch, update_file_content
 from core.model.data_models import File
+from faststream.kafka import KafkaBroker
 from httpx import AsyncClient
+from src.kafka.message import Message
 
 from src.logtools import getLogger
 
@@ -18,14 +20,17 @@ class Filehandler:
     logger: Logger
     client: AsyncClient
 
-    def __init__(self) -> None:
-        self.logger = getLogger()
+    def __init__(self, kafkaBroker: KafkaBroker) -> None:
+        self.logger = getLogger(__name__)
         if config.https_proxy or config.http_proxy:
             limits = httpx.Limits(max_keepalive_connections=5, max_connections=250)
             self.client = AsyncClient(proxy=config.https_proxy or config.http_proxy, timeout=config.request_timeout, limits=limits)
         else:
             self.client = AsyncClient(timeout=config.request_timeout)
 
+        self.broker = kafkaBroker
+        self.logger.info("Filehandler created.")
+
     async def __aenter__(self):
         return self
 
@@ -34,10 +39,11 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
 
     async def download_and_persist_files(self, batch_size: int = 100):
         self.logger.info("Persisting content of all scraped files to database.")
-
+        tasks = []
         offset = 0
         while True:
             files: list[File] = request_batch(File, offset=offset, limit=batch_size)
+
             if not files or len(files) < 1:
                 break
             semaphore = asyncio.Semaphore(batch_size)
@@ -64,6 +70,7 @@ async def download_and_persist_file(self, file: File):
         response = await self.client.get(url=file.id)
         response.raise_for_status()
         content = response.content
+        self.logger.debug(f"Checking necessity of inserting/updating file {file.name} to database.")
         if file.content is None or content != file.content:
             content_disposition = response.headers.get("content-disposition")
             if content_disposition:
@@ -83,3 +90,14 @@ async def download_and_persist_file(self, file: File):
                 fileName = file.name
             self.logger.debug(f"Saving content of file {file.name} to database.")
             update_file_content(file.db_id, content, fileName)
+            self.logger.debug(f"Saved content of file {file.name} to database.")
+            msg = Message(content=str(file.db_id))
+            self.logger.debug(f"Publishing: {msg}.")
+            try:
+                await self.broker.publish(msg, topic=config.core.kafka.topic)
+                self.logger.debug(f"Published: {msg}.")
+            except Exception as e:
+                # If Kafka Broker is unavailable rollback the file download to ensure
+                # All Documents that have content, are published to the Kafka Queue
+                update_file_content(file.db_id, b"", "")
+                self.logger.error(f"Publishing failed. Rolled file download back: {file.db_id}. - {e}")

riski-extractor/src/kafka/message.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel
+
+
+class Message(BaseModel):
+    content: str
+    republished: bool = False

riski-extractor/src/logtools.py

@@ -5,8 +5,10 @@
 
 from yaml import safe_load
 
+defautlName: str = "riski-extractor"
 
-def getLogger(name: str = "riski-extractor") -> logging.Logger:
+
+def getLogger(name: str = None) -> logging.Logger:
     """Configures logging and returns a logger with the specified name.
 
     Parameters:
@@ -19,7 +21,9 @@ def getLogger(name: str = "riski-extractor") -> logging.Logger:
         log_config = safe_load(file)
 
     logging.config.dictConfig(log_config)
-    return logging.getLogger(name)
+
+    logger_name = f"{defautlName}.{name}" if name else defautlName
+    return logging.getLogger(logger_name)
 
 
 class JsonFormatter(logging.Formatter):

riski-extractor/src/parser/base_parser.py

@@ -13,7 +13,7 @@
 
 class BaseParser(ABC, Generic[T]):
     def __init__(self) -> None:
-        self.logger = getLogger()
+        self.logger = getLogger(__name__)
         if platform.system() == "Windows":
             # For Windows, use the specific code page that works
             locale.setlocale(locale.LC_TIME, "German_Germany.1252")