Merge pull request #44 from damklis/develop

Develop

Author: damklis

.coveragerc

@@ -1,7 +1,7 @@
 [run]
 omit =
  *test*.py
- *main.py
+ *operator.py
 
 [report]
 exclude_lines =

airflow/dags/custom_operators/__init__.py

@@ -0,0 +1,2 @@
+from custom_operators.proxypool_operator import ProxyPoolOperator
+from custom_operators.rss_news_operator import RSSNewsOperator

airflow/dags/custom_operators/proxypool_operator.py

@@ -0,0 +1,55 @@
+import json
+from concurrent.futures import ThreadPoolExecutor
+from retry import RetryOnException as retry
+from proxypool import (
+ ProxyPoolValidator,
+ ProxyPoolScraper,
+ RedisProxyPoolClient
+)
+
+from airflow.models.baseoperator import BaseOperator
+from airflow.utils.decorators import apply_defaults
+
+
+class ProxyPoolOperator(BaseOperator):
+
+ @apply_defaults
+ def __init__(
+ self,
+ proxy_webpage,
+ number_of_proxies,
+ testing_url,
+ max_workers,
+ redis_config,
+ redis_key,
+ *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self.proxy_webpage = proxy_webpage
+ self.testing_url = testing_url
+ self.number_of_proxies = number_of_proxies
+ self.max_workers = max_workers
+ self.redis_config = redis_config
+ self.redis_key = redis_key
+
+ @retry(5)
+ def execute(self, context):
+ proxy_scraper = ProxyPoolScraper(self.proxy_webpage)
+ proxy_validator = ProxyPoolValidator(self.testing_url)
+ proxy_stream = proxy_scraper.get_proxy_stream(self.number_of_proxies)
+
+ with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
+ results = executor.map(
+ proxy_validator.validate_proxy, proxy_stream
+ )
+ valid_proxies = filter(lambda x: x.is_valid is True, results)
+ sorted_valid_proxies = sorted(
+ valid_proxies, key=lambda x: x.health, reverse=True
+ )
+
+ with RedisProxyPoolClient(self.redis_key, self.redis_config) as client:
+ client.override_existing_proxies(
+ [
+ json.dumps(record.proxy)
+ for record in sorted_valid_proxies[:5]
+ ]
+ )

airflow/dags/custom_operators/rss_news_operator.py

@@ -0,0 +1,57 @@
+from log import log
+from retry import RetryOnException as retry
+from proxypool import RedisProxyPoolClient
+from rss_news import (
+ NewsProducer,
+ NewsExporter,
+ NewsValidator
+)
+
+from airflow.models.baseoperator import BaseOperator
+from airflow.utils.decorators import apply_defaults
+
+
+@log
+class RSSNewsOperator(BaseOperator):
+
+ @apply_defaults
+ def __init__(
+ self,
+ validator_config,
+ rss_feed,
+ language,
+ redis_config,
+ redis_key,
+ bootstrap_servers,
+ topic,
+ *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self.validator_config = validator_config
+ self.rss_feed = rss_feed
+ self.language = language
+ self.redis_config = redis_config
+ self.redis_key = redis_key
+ self.bootstrap_servers = bootstrap_servers
+ self.topic = topic
+
+ @retry(5)
+ def execute(self, context):
+ validator = NewsValidator(self.validator_config)
+ producer = NewsProducer(self.rss_feed, self.language)
+ redis = RedisProxyPoolClient(self.redis_key, self.redis_config)
+
+ with NewsExporter(self.bootstrap_servers) as exporter:
+ proxy = redis.get_proxy()
+ self.logger.info(proxy)
+ try:
+ for news in producer.get_news_stream(proxy):
+ self.logger.info(news)
+ validator.validate_news(news)
+ exporter.export_news_to_broker(
+ self.topic,
+ news.as_dict()
+ )
+ except Exception as err:
+ redis.lpop_proxy()
+ self.logger.error(f"Exception: {err}")
+ raise err

airflow/dags/rss_news_dag.py

@@ -4,8 +4,10 @@
 from airflow.operators.python_operator import PythonOperator
 
 from dags_config import Config as config
-from rss_news import export_news_to_broker
-from proxypool import update_proxypool
+from custom_operators import (
+ ProxyPoolOperator,
+ RSSNewsOperator
+)
 
 
 def extract_feed_name(url):
@@ -19,12 +21,15 @@ def dummy_callable(action):
 
 def export_events(config, rss_feed, language, dag):
  feed_name = extract_feed_name(rss_feed)
- return PythonOperator(
+ return RSSNewsOperator(
  task_id=f"exporting_{feed_name}_news_to_broker",
- python_callable=export_news_to_broker,
- op_kwargs={
- "config": config, "rss_feed": rss_feed, "language": language
- },
+ validator_config=config.VALIDATOR_CONFIG,
+ rss_feed=rss_feed,
+ language=language,
+ redis_config=config.REDIS_CONFIG,
+ redis_key=config.REDIS_KEY,
+ bootstrap_servers=config.BOOTSTRAP_SERVERS,
+ topic=config.TOPIC,
  dag=dag
  )
 
@@ -46,10 +51,14 @@ def create_dag(dag_id, interval, config, language, rss_feeds):
  dag=dag
  )
 
- proxypool = PythonOperator(
+ proxypool = ProxyPoolOperator(
  task_id="updating_proxypoool",
- python_callable=update_proxypool,
- op_kwargs={"config": config},
+ proxy_webpage=config.PROXY_WEBPAGE,
+ number_of_proxies=config.NUMBER_OF_PROXIES,
+ testing_url=config.TESTING_URL,
+ max_workers=config.NUMBER_OF_PROXIES,
+ redis_config=config.REDIS_CONFIG,
+ redis_key=config.REDIS_KEY,
  dag=dag
  )
 
@@ -76,5 +85,9 @@ def create_dag(dag_id, interval, config, language, rss_feeds):
  interval = f"{n*4}-59/10 * * * *"
 
  globals()[dag_id] = create_dag(
- dag_id, interval, config, language, rss_feeds
+ dag_id,
+ interval,
+ config,
+ language,
+ rss_feeds
  )

airflow/modules/proxypool/__init__.py

@@ -1,4 +1,3 @@
 from proxypool.redis_proxypool_client import RedisProxyPoolClient
-from proxypool.main import update_proxypool
 from proxypool.proxypool_scraper import ProxyPoolScraper, ProxyRecord
 from proxypool.proxypool_validator import ProxyPoolValidator

airflow/modules/proxypool/main.py

@@ -1,4 +1,3 @@
-from rss_news.main import export_news_to_broker
 from rss_news.rss_news_producer import NewsProducer, NewsFormatter, News
 from rss_news.rss_news_exporter import NewsExporter
-from rss_news.rss_news_validator import NewsValidator
+from rss_news.rss_news_validator import NewsValidator

airflow/modules/rss_news/__init__.py

@@ -3,4 +3,4 @@
 echo "Creating default bucket $DEFAULT_BUCKET"
 
 mkdir -p /data/$DEFAULT_BUCKET \
-&& /usr/bin/minio server /data
+ && /usr/bin/minio server /data