Add job_search input stream

Author: Rashid Schami

job_url_scraper/config.py

@@ -1,3 +1,4 @@
 KAFKA = 'kafka:9092'
+KAFKA_INPUT_TOPIC = 'job_search'
 KAFKA_OUTPUT_TOPIC = 'job_urls'
 KAFKA_ERROR_TOPIC = 'errors'

job_url_scraper/helper.py

@@ -3,7 +3,8 @@
 from contextlib import closing
 from bs4 import BeautifulSoup
 from retry import retry
-from kafka import KafkaProducer
+import json
+from kafka import KafkaProducer, KafkaConsumer
 from config import KAFKA
 
 
@@ -15,6 +16,15 @@ def get_producer():
                          )
 
 
+@retry(tries=5, delay=30)
+def get_consumer(topic):
+    return KafkaConsumer(
+        topic,
+        bootstrap_servers=[KAFKA],
+        value_deserializer=lambda x: json.loads(x.decode('ascii'))
+    )
+
+
 @retry(tries=3, delay=5)
 def get_html(url):
     with closing(get(url, stream=True)) as response:

job_url_scraper/job_url_scraper.py

@@ -1,11 +1,13 @@
 import re
 from helper import get_html
 from time import sleep
+from urllib.parse import quote_plus
 
 
 class JobUrlScraper:
-    def __init__(self, query='software+developer', throttle_seconds=5):
-        self.query = query
+    def __init__(self, query, city, throttle_seconds=5):
+        self.query = quote_plus(query)
+        self.city = quote_plus(city)
         self.search_count = self._get_search_count()
         self.throttle_seconds = throttle_seconds
 
@@ -15,7 +17,7 @@ def _get_search_count(self):
         return int(re.findall(r'(\d+.?\d*) Jobs', search_count_text)[0].replace('.', ''))
 
     def _get_search_page_url(self, start=0):
-        return f'https://de.indeed.com/Jobs?q={self.query}&l=Berlin&sort=date&limit=50&radius=25&filter=0&start={start}'
+        return f'https://de.indeed.com/Jobs?q={self.query}&l={self.city}&sort=date&limit=50&radius=25&filter=0&start={start}'
 
     def get_all_job_urls(self):
         for url in self._get_all_search_page_urls():

job_url_scraper/main.py

@@ -1,21 +1,24 @@
-from helper import get_producer
-from config import KAFKA_ERROR_TOPIC, KAFKA_OUTPUT_TOPIC
+from helper import get_producer, get_consumer
+from config import KAFKA_ERROR_TOPIC, KAFKA_OUTPUT_TOPIC, KAFKA_INPUT_TOPIC
 from job_url_scraper import JobUrlScraper
 import traceback
 
 
 if __name__ == '__main__':
     producer = get_producer()
+    consumer = get_consumer(KAFKA_INPUT_TOPIC)
 
-    try:
-        for url in JobUrlScraper().get_all_job_urls():
-            producer.send(KAFKA_OUTPUT_TOPIC, url)
-    except Exception as e:
-        tb = traceback.format_exc()
-        producer.send(KAFKA_ERROR_TOPIC,
-                      value=f'ERROR: {e}\n{tb}\n',
-                      key='JobUrlScraper'
-                      )
-        raise
-    finally:
-        producer.flush()
+    for message in consumer:
+        try:
+            for url in JobUrlScraper(query=message.value['query'],
+                                     city=message.value['city']).get_all_job_urls():
+                producer.send(KAFKA_OUTPUT_TOPIC, url)
+        except Exception as e:
+            tb = traceback.format_exc()
+            producer.send(KAFKA_ERROR_TOPIC,
+                          value=f'ERROR: {e}\n{tb}\n',
+                          key='JobUrlScraper'
+                          )
+            raise
+        finally:
+            producer.flush()

job_url_scraper/test/test_job_url_scraper.py

@@ -14,12 +14,14 @@ def setUp(self):
 
     def test_get_all_job_urls(self):
         def get_html(url):
-            if url == 'https://de.indeed.com/Jobs?q=software+developer&l=Berlin&sort=date&limit=50&radius=25&filter=0&start=0':
+            if url == 'https://de.indeed.com/Jobs?q=software+developer+c%23&l=Berlin&sort=date&limit=50&radius=25&filter=0&start=0':
                 return self.main_page
             raise RequestException(f'Error getting {url}')
 
         with patch('job_url_scraper.get_html') as mocked_get:
             mocked_get.side_effect = get_html
             job_urls = [url for url in JobUrlScraper(
+                query='software developer c#',
+                city='Berlin',
                 throttle_seconds=0).get_all_job_urls()]
             self.assertEqual(job_urls, self.expected_job_urls)