merge afetharita-kafka-consumer-address -> deprem-yardim-address-api

Author: Veli Eroglu

.dockerignore

@@ -0,0 +1,4 @@
+venv/
+data.csv
+lab.py
+.idea/

.env.example

@@ -0,0 +1,23 @@
+CLIENT_ID="Consumer Test"
+
+BOOTSTRAP_SERVERS=<test-kafka>
+KAFKA_ADDRESS_RESOLVE_TOPIC="<test-topic>"
+KAFKA_PROCESSED_TOPIC="<test-topic>"
+INTENT_TARGET_TOPIC="<test-topic>"
+KAFKA_INTENT_TOPIC="<test-topic>"
+HF_HUB_TOKEN=<hub_token>
+
+GROUP_ID=
+
+# Max message pool count
+MAX_POOL_RECORDS=10
+MESSAGE_TIMEOUT_MS=200
+SECURITY_PROTOCOL=
+SASL_MECHANISMS=
+SASL_PLAIN_USERNAME=
+SASL_PLAIN_PASSWORD=
+ENABLE_AUTO_COMMIT=True
+OPENAI_API_KEY=<API_KEY>
+GOOGLE_API_KEY=<GAPI_KEY>
+NER_API_KEY=<NER_API_KEY>
+SENTRY_DSN=<SENTRY_DSN>

Dockerfile

@@ -0,0 +1,17 @@
+# Use Python 3.11 as the base image
+FROM python:3.11
+
+# Set the working directory
+WORKDIR /projectKafka
+ENV PYTHONPATH "."
+
+# Install the required packages
+COPY requirements.txt /projectKafka/
+RUN pip install -r requirements.txt
+
+# Copy the publisher and consumer scripts
+COPY ./ /projectKafka/
+# Set the entrypoint
+ENTRYPOINT ["python"]
+# Set the default command to run the publisher and consumer
+CMD ["/projectKafka/consumers/address_resolve.py"]

README.md

@@ -1 +1,41 @@
-# deprem-yardim-address-api
+# Mysterious Project
+
+- Stream Project using Apache Kafka  A Python project that uses Apache Kafka as the stream processing framework.
+
+## Prerequisites
+- Python 3.10+
+- Apache Kafka
+
+## Installation
+- Clone the repository to your local machine.
+```
+git clone https://github.com/[username]/stream-project-kafka.git
+```
+
+- Install the required packages mentioned in the requirements.txt file.
+```
+pip install -r requirements.txt
+```
+
+## Running
+
+- Start the address_resolve consumer by running the following command:
+
+```
+python consumers/address_resolve.py
+```
+
+- Start the processed consumer by running the following command:
+
+```
+python consumers/processed.py
+```
+
+## Contributing
+- Fork the repository.
+- Create a new branch for your changes.
+- Commit and push your changes to the new branch.
+- Create a pull request to the master branch.
+
+## License
+This project is licensed under the MIT License.

address_resolver.py

@@ -23,20 +23,20 @@ def __init__(self, GOOGLE_API_KEY, OPENAI_API_KEY, NER_API_KEY):
 
     def google_geocode_api_request(self, address_text: str, entry_id: int):
         result = self.google_api.request(address_text)
-        result['entry_id'] = entry_id
+        result['id'] = entry_id
         return result
 
     def regex_api_request(self, address_text: str, entry_id: int):
         result = self.regex_api.extract(address_text)
-        result['entry_id'] = entry_id
+        result['id'] = entry_id
         return result
 
     def ner_api_request(self, address_text: str, entry_id: int):
         result = self.ner_api.query(address_text)
-        result['entry_id'] = entry_id
+        result['id'] = entry_id
         return result
 
     def openai_api_request(self, address_text: str, entry_id: int):
         result = self.open_api.single_request(address_text)
-        result['entry_id'] = entry_id
+        result['id'] = entry_id
         return result

config.py

@@ -0,0 +1,31 @@
+from environs import Env
+
+env = Env()
+# Read .env into os.environ
+env.read_env()
+
+
+# KAFKA SETTINGS
+CLIENT_ID = env.str("CLIENT_ID")
+BOOTSTRAP_SERVERS = env.list("BOOTSTRAP_SERVERS")
+KAFKA_ADDRESS_RESOLVE_TOPIC=env.str("KAFKA_ADDRESS_RESOLVE_TOPIC")
+KAFKA_PROCESSED_TOPIC=env.str("KAFKA_PROCESSED_TOPIC")
+KAFKA_INTENT_TOPIC=env.str("KAFKA_INTENT_TOPIC")
+INTENT_TARGET_TOPIC=env.str("INTENT_TARGET_TOPIC")
+HF_HUB_TOKEN = env.str("HF_HUB_TOKEN")
+# GROUP_ID=env.int("GROUP_ID", None)
+
+# Max message pool count
+MAX_POOL_RECORDS=env.int("MAX_POOL_RECORDS")
+MESSAGE_TIMEOUT_MS=env.int("MESSAGE_TIMEOUT_MS")
+SECURITY_PROTOCOL=env.str("SECURITY_PROTOCOL")
+SASL_MECHANISMS=env.str("SASL_MECHANISMS")
+SASL_PLAIN_USERNAME=env.str("SASL_PLAIN_USERNAME")
+SASL_PLAIN_PASSWORD=env.str("SASL_PLAIN_PASSWORD")
+ENABLE_AUTO_COMMIT=env.bool("ENABLE_AUTO_COMMIT")
+
+# Third party api key
+SENTRY_DSN = env.str("SENTRY_DSN")
+GOOGLE_API_KEY = env.str("GOOGLE_API_KEY")
+OPENAI_API_KEY = env.str("OPENAI_API_KEY")
+NER_API_KEY = env.str("NER_API_KEY")

consumers/__init__.py

@@ -0,0 +1,51 @@
+import logging
+from sentry_sdk import capture_exception
+from models.conf import KafkaSettings, KafkaConsumerSettings
+import aiokafka
+
+class BaseKafkaClient:
+    def __init__(self, server_settings: KafkaSettings, topic: str):
+        self.settings = server_settings
+        consumer_settings = KafkaConsumerSettings(**server_settings.dict())
+        # init consumer & producer
+        self.consumer = aiokafka.AIOKafkaConsumer(topic,
+                                                  **consumer_settings.dict(
+                                                      exclude_none=True))
+        self.producer = aiokafka.AIOKafkaProducer(
+            bootstrap_servers=server_settings.bootstrap_servers,
+            loop=server_settings.loop)
+
+    async def _process_message(self, record: aiokafka.ConsumerRecord):
+        try:
+            await self.process_message(record)
+        except Exception as exc:
+            logging.error(exc)
+            capture_exception(exc)
+
+    async def process_message(self, record: aiokafka.ConsumerRecord):
+        ...
+
+    async def process(self):
+        data = await self.consumer.getmany(
+            timeout_ms=self.settings.message_timeout_ms,
+            max_records=self.settings.max_pool_records)
+        for topic, records in data.items():
+            for record in records:
+                try:
+                    await self._process_message(record)
+                except Exception as exc:
+                    capture_exception(exc)
+
+    async def run(self):
+        self.running = True
+        try:
+            logging.info("Kafka Consumer Started...")
+            await self.consumer.start()
+            await self.producer.start()
+            while self.running:
+                await self.process()
+        finally:
+            logging.info("Kafka Consumer Stopped...")
+            await self.consumer.stop()
+            await self.producer.stop()
+

consumers/address_resolve.py

@@ -0,0 +1,115 @@
+# built-in python
+import asyncio
+from io import BytesIO
+from multiprocessing.pool import ThreadPool
+import logging
+
+# third-party
+import sentry_sdk
+import orjson
+import pandas as pd
+import aiokafka
+
+# in-house
+from consumers import BaseKafkaClient
+from models.conf import KafkaSettings
+from config import *
+from address_resolver import AddressAPI
+
+# set logger level
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
+sentry_sdk.init(dsn=SENTRY_DSN)
+
+address_api = AddressAPI(GOOGLE_API_KEY, OPENAI_API_KEY, NER_API_KEY)
+
+class AddressResolve(BaseKafkaClient):
+
+    async def process_message(self, record: aiokafka.ConsumerRecord):
+        message = record.value
+
+        messageIo = BytesIO(message)
+        address_df = pd.read_json(messageIo)
+        address_df_replica = address_df.copy()
+
+        regex_results = pd.DataFrame(
+            [address_api.regex_api_request(raw_text, entry_id) for
+             raw_text, entry_id in
+             zip(address_df.raw_text.values, address_df.id.values)])
+        regex_to_geocode = regex_results[regex_results.ws >= 0.7]
+        del regex_results
+
+        # Ner Process
+        address_df = address_df[~address_df.id.isin(regex_to_geocode.id.values)]
+        with ThreadPool(60) as executor:
+            ner_results = executor.map(
+                lambda p: address_api.ner_api_request(*p),
+                zip(address_df.raw_text.values,
+                    address_df.id.values))
+        ner_results = pd.DataFrame(ner_results)
+        ner_to_geocode = ner_results[ner_results.ws >= 0.5]
+        del ner_results
+
+        geocode_data = pd.concat([regex_to_geocode[['address', 'id']],
+                                  ner_to_geocode[['address', 'id']]], axis=0)
+        del regex_to_geocode, ner_to_geocode
+
+        with ThreadPool(60) as executor:
+            geocode_data = executor.map(
+                lambda p: address_api.google_geocode_api_request(*p),
+                zip(geocode_data.address.values, geocode_data.id.values))
+
+        geocode_data = pd.DataFrame(geocode_data)
+        geocode_data = pd.merge(geocode_data[geocode_data.is_resolved == True],
+                                address_df_replica, on='id', how='left')
+        del address_df_replica
+
+        final_data = []
+        for d in geocode_data.iterrows():
+            d = d[1]
+            final_data.append(
+                {
+                    'location': {
+                        "formatted_address": d.get('formatted_address', ''),
+                        "latitude": d.get('latitude', 0.0),
+                        "longitude": d.get('longitude', 0.0),
+                        "northeast_lat": d.get('northeast_lat', 0.0),
+                        "northeast_lng": d.get('northeast_lng', 0.0),
+                        "southwest_lat": d.get('southwest_lat', 0.0),
+                        "southwest_lng": d.get('southwest_lng', 0.0),
+                        "entry_id": d.get('id'),
+                        "epoch": d.get('epoch'),
+                        "channel": d.get('channel')},
+                    'feed': {
+                        "id": d.get('id'),
+                        "raw_text": d.get('raw_text'),
+                        "channel": d.get('channel'),
+                        "extra_parameters": d.get('extra_parameters', {}),
+                        "epoch": d.get('epoch')}
+                }
+            )
+
+        await self.producer.send_and_wait(KAFKA_PROCESSED_TOPIC,
+                                          orjson.dumps(final_data))
+        logger.info("Message Processed.")
+
+
+if __name__ == '__main__':
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+
+    kafka_settings = KafkaSettings(
+        loop=loop,
+        client_id=CLIENT_ID,
+        bootstrap_servers=BOOTSTRAP_SERVERS,
+        max_pool_records=MAX_POOL_RECORDS,
+        message_timeout_ms=MESSAGE_TIMEOUT_MS
+    )
+
+    try:
+        server = AddressResolve(topic=KAFKA_ADDRESS_RESOLVE_TOPIC, server_settings=kafka_settings)
+        loop.run_until_complete(server.run())
+    finally:
+        loop.run_until_complete(loop.shutdown_asyncgens())
+        loop.close()

consumers/intent.py

@@ -0,0 +1,59 @@
+# built-in python
+import asyncio
+
+# third-party
+import logging
+import aiokafka
+import sentry_sdk
+
+# in-house
+from consumers import BaseKafkaClient
+from models.conf import KafkaSettings
+from config import *
+from orjson import loads
+from helpers.intent import batch_query
+
+# set logger level
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
+sentry_sdk.init(dsn=SENTRY_DSN)
+
+class Intent(BaseKafkaClient):
+
+    async def process_message(self, record: aiokafka.ConsumerRecord):
+        message = record.value
+        message_dict = loads(message)
+        full_text_list = [i.get("full_text") for i in message_dict if i]
+
+        if not full_text_list:
+            logging.warning(f"Raw text is empty, Message: {message_dict}")
+            return
+
+        response = batch_query(full_text_list, None)
+
+        if not response:
+            logging.warning(f"No response from hugging face endpoint message: {message_dict}")
+            return
+
+        await self.producer.send_and_wait(INTENT_TARGET_TOPIC,
+                                          response)
+
+if __name__ == '__main__':
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+
+    kafka_settings = KafkaSettings(
+        loop=loop,
+        client_id=CLIENT_ID,
+        bootstrap_servers=BOOTSTRAP_SERVERS,
+        max_pool_records=MAX_POOL_RECORDS,
+        message_timeout_ms=MESSAGE_TIMEOUT_MS
+    )
+
+    try:
+        server = Intent(topic=KAFKA_INTENT_TOPIC, server_settings=kafka_settings)
+        loop.run_until_complete(server.run())
+    finally:
+        loop.run_until_complete(loop.shutdown_asyncgens())
+        loop.close()

helpers/google_geocode_api.py

@@ -1,10 +1,4 @@
-import json
 import requests
-import pandas as pd
-from googlemaps import Client as GoogleMaps
-import googlemaps
-import gmaps
-import urllib.parse
 
 class GoogleGeocodeAPI:
     def __init__(self, api_key):