multy client extract + main + transform

Author: EEnota

main.py

@@ -1,5 +1,5 @@
 import argparse
-from src import make_extract, transform_article, load_news, init_database, create_news_tables
+from src import make_extract, transform_article_debug, transform_article_web, load_news, init_database, create_news_tables
 import logging
 
 logging.basicConfig(
@@ -8,6 +8,8 @@
 )
 logger = logging.getLogger(__name__)
 
+   
+
 def positive_int(value: str) -> int:
     ivalue = int(value)
     if ivalue <= 0:
@@ -39,20 +41,31 @@ def parse_args():
     )
     return parser.parse_args()
 
-def pipeline() -> None:
-    args = parse_args()
-    key_word = args.keyword
-    limit = args.limit
-    page_size = args.page_size
+def pipeline_for_web_user(user_id: int, search_request_id: int, key_word: str, limit: int, page_size: int, debug:bool = False) -> int:
+    if debug:
+        args = parse_args()
+        key_word = args.keyword
+        limit = args.limit
+        page_size = args.page_size
+        user_id = 1
+        search_request_id = 1
     num_of_news = 0
     page = 1
     while num_of_news < limit:
         remaining = limit - num_of_news
-        raw_file_name,raw_articles_count = make_extract(key_word, page, page_size)
+        if debug:
+            raw_file_name, raw_articles_count = make_extract(key_word, page, page_size)
+            if raw_articles_count == 0:
+                logger.warning("there is no more artical")
+                break
+            clean_file_name = transform_article_debug(raw_file_name, key_word, page)
+            result_num_of_news = load_news(clean_file_name, max_rows=remaining)
+
+        payload, raw_articles_count = make_extract(key_word, page, page_size)
         if raw_articles_count == 0:
             logger.warning("there is no more artical")
             break
-        clean_file_name = transform_article(raw_file_name, key_word, page)
+        clean_file_name = transform_article_web(payload, key_word, page)
         result_num_of_news = load_news(clean_file_name, max_rows=remaining)
 
         num_of_news += result_num_of_news
@@ -71,6 +84,16 @@ def main()-> None:
         logger.exception("pipeline failed: %s", e)
         raise
 
+def main2(user_id, search_request_id, key_word)-> None:
+    logger.info("Starting pipeline, init database, build table..")
+    try:
+        init_database()
+        create_news_tables()
+        loaded = pipeline_for_web_user(user_id,search_request_id, key_word, 20, 50)
+        logger.info("Pipline finished. loaded rows: %s", loaded)
+    except Exception as e:
+        logger.exception("pipeline failed: %s", e)
+        raise
 
 
 if __name__ == "__main__":

src/__init__.py

@@ -1,4 +1,4 @@
 from .extract import make_extract
-from .transform import transform_article
+from .transform import transform_article_web, transform_article_debug
 from .load import load_news
 from .db import init_database, create_news_tables

src/extract.py

@@ -23,7 +23,7 @@ def import_to_raw_json(data:dict[str, Any], key_word: str, page: int) -> str:
 
 
 
-def make_extract( key_word: str, page: int = 1, page_size: int = 100) -> tuple[str,int]:
+def make_extract( key_word: str, page: int = 1, page_size: int = 100, debug_mode: bool = False) -> tuple[dict,int]:
     params = {
     "apiKey": settings.KEY_API,
     "language":settings.langueage,
@@ -43,9 +43,10 @@ def make_extract( key_word: str, page: int = 1, page_size: int = 100) -> tuple[s
         articles_count = len(payload.get("articles", []))
         if articles_count == 0:
             logger.info("There are no more articles")
-
-        new_file_name = import_to_raw_json(payload, key_word, page)
-        return new_file_name, articles_count
+        if debug_mode:
+            new_file_name = import_to_raw_json(payload, key_word, page)
+            return new_file_name, articles_count
+        return payload, articles_count
 
     except r.exceptions.Timeout:
         logger.error("Error: NewsAPI reauest time out")

src/transform.py

@@ -66,11 +66,12 @@ def clean_article(data) -> tuple[list[dict[str,str]], dict]:
                 "url": element["url"],
                 "published_at": element["publishedAt"],
                 "fetched_at": data["fetched_at"],
+                "source_name": element.get("source", {}).get("name")
             }
         )
     return clean_data, statistic
 
-def transform_article(new_file_name:str, key_word: str, page: int) -> str:
+def transform_article_debug(new_file_name:str, key_word: str, page: int) -> str:
     extract_dir = BASE_DIR / "raw" / new_file_name
     with open(extract_dir, 'r', encoding='utf-8') as f:
         data = json.load(f)
@@ -96,4 +97,16 @@ def transform_article(new_file_name:str, key_word: str, page: int) -> str:
     logger.info("rejected_articles=%s", stats['rejected_articles'])
     logger.info("reasons_counts=%s", stats['reasons_counts'])
     logger.info("prime_reason=%s", stats['prime_reason'])
-    return create_clean_data
+    return create_clean_data
+
+def transform_article_web(payload:dict)-> tuple[list[dict],dict]:
+    logger.info(f"income: {len(payload.get('articles', []))} articals")
+    clean, stats = clean_article(payload)
+    logger.info(f"outcome: {len(clean)} articals")
+    logger.info("stats collected")
+    logger.info("income_articles=%s", stats['income_articles'])
+    logger.info("accepted_articles=%s", stats['accepted_articles'])
+    logger.info("rejected_articles=%s", stats['rejected_articles'])
+    logger.info("reasons_counts=%s", stats['reasons_counts'])
+    logger.info("prime_reason=%s", stats['prime_reason'])
+    return clean, stats