Multy user

Author: EEnota

main.py

@@ -1,5 +1,5 @@
 import argparse
-from src import make_extract, transform_article_debug, transform_article_web, 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, load_web_pipeline, make_extract_web, make_extract_debug
 import logging
 
 logging.basicConfig(
@@ -54,19 +54,19 @@ def pipeline_for_web_user(user_id: int, search_request_id: int, key_word: str, l
     while num_of_news < limit:
         remaining = limit - num_of_news
         if debug:
-            raw_file_name, raw_articles_count = make_extract(key_word, page, page_size)
+            raw_file_name, raw_articles_count = make_extract_debug(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_web(payload, key_word, page)
-        result_num_of_news = load_news(clean_file_name, max_rows=remaining)
+        else:
+            payload, raw_articles_count = make_extract_web(key_word, page, page_size)
+            if raw_articles_count == 0:
+                logger.warning("there is no more artical")
+                break
+            clean_data, stats = transform_article_web(payload)
+            result_num_of_news = load_web_pipeline(user_id, search_request_id, clean_data, stats)
 
         num_of_news += result_num_of_news
         page += 1
@@ -78,23 +78,11 @@ def main()-> None:
     try:
         init_database()
         create_news_tables()
-        loaded = pipeline()
-        logger.info("Pipline finished. loaded rows: %s", loaded)
-    except Exception as e:
-        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)
+        loaded = pipeline_for_web_user()
         logger.info("Pipline finished. loaded rows: %s", loaded)
     except Exception as e:
         logger.exception("pipeline failed: %s", e)
         raise
 
-
 if __name__ == "__main__":
     main()

src/__init__.py

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

src/extract.py

@@ -23,7 +23,42 @@ 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, debug_mode: bool = False) -> tuple[dict,int]:
+def make_extract_debug( key_word: str, page: int = 1, page_size: int = 100) -> str:
+    params = {
+    "apiKey": settings.KEY_API,
+    "language":settings.langueage,
+    "q": key_word,
+    "pageSize" : page_size,
+    "page" : page,
+    "sortBy": settings.sortBy
+    }
+    try:
+        data = r.get(settings.NEWS_URL, params=params, timeout=15)
+        data.raise_for_status()
+        payload = data.json()
+        logger.info(f"raise of status: {data.status_code}")
+        payload["fetched_at"] = datetime.now().isoformat()
+        payload["language"] = settings.langueage
+        payload["key_word"] = key_word
+        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
+    except r.exceptions.Timeout:
+        logger.error("Error: NewsAPI reauest time out")
+        raise
+    except r.exceptions.ConnectionError:
+        logger.error("Error: no internet connection or API is not available")
+        raise
+    except r.exceptions.HTTPError as e:
+        logger.error(f"Error HTTP: {e}")
+        raise
+    except ValueError:
+        logger.error("Error: sorry we can't parse JSON")
+        raise
+
+def make_extract_web( key_word: str, page: int = 1, page_size: int = 100) -> tuple[dict,int]:
     params = {
     "apiKey": settings.KEY_API,
     "language":settings.langueage,
@@ -43,11 +78,7 @@ def make_extract( key_word: str, page: int = 1, page_size: int = 100, debug_mode
         articles_count = len(payload.get("articles", []))
         if articles_count == 0:
             logger.info("There are no more articles")
-        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")
         raise

src/load.py

@@ -45,4 +45,88 @@ def load_news(clean_news:str, max_rows: Optional[int] = None)-> int:
                 num_of_news += 1
         conn.commit()
     return num_of_news
-            
+
+def upsert_article(cur, article: dict) -> int:
+    query = """
+            INSERT INTO articles (
+            url,
+            source_name,
+            author,
+            title,
+            description,
+            published_at
+            )
+            VALUES (%s,%s,%s,%s,%s,%s)
+            ON CONFLICT (url) DO UPDATE
+            SET
+                source_name = EXCLUDED.source_name,
+                author = EXCLUDED.author,
+                title = EXCLUDED.title,
+                description = EXCLUDED.description,
+                published_at = EXCLUDED.published_at
+            RETURNING id;
+            """
+    cur.execute(query,
+                ( 
+                    article["url"], 
+                    article["source_name"], 
+                    article["author"], 
+                    article["title"],
+                    article["description"],
+                    article["published_at"],
+                ),
+                )
+    row = cur.fetchone()
+    return row["id"]
+
+            
+def load_user_news(cur, user_id: int, search_request_id: int, article_id: int, keyword: str, fetched_at: str)-> int:
+    query = """
+            INSERT INTO user_news (
+            user_id,
+            search_request_id,
+            article_id,
+            keyword,
+            fetched_at
+            )
+            values(%s,%s,%s,%s,%s)
+            ON CONFLICT DO NOTHING;
+            """
+    cur.execute(query,(user_id, search_request_id, article_id, keyword, fetched_at))
+    return cur.rowcount
+
+def load_request_stats(cur, search_request_id: int, stats: dict) -> None:
+    query = """
+            INSERT INTO request_stats (
+            search_request_id,
+            income_articles,
+            accepted_articles,
+            rejected_articles,
+            reasons_counts,
+            prime_reason
+            )
+            VALUES (%s,%s,%s,%s,%s,%s)
+            """
+    cur.execute(query, (
+            search_request_id,
+            stats["income_articles"],
+            stats["accepted_articles"],
+            stats["rejected_articles"],
+            json.dumps(stats["reasons_counts"]),
+            json.dumps(stats["prime_reason"])
+        ))
+    return None
+
+def load_web_pipeline(user_id: int, search_request_id: int, clean_data: list[dict], stats: dict) -> int:
+    loaded_count = 0
+    with get_cursor(settings.db_news) as (conn, cur):
+        for article in clean_data:
+            keyword = article["key_word"]
+            fetched_at = article["fetched_at"]
+            article_id = upsert_article(cur,article)
+            inserted = load_user_news(cur, user_id, search_request_id, article_id, keyword, fetched_at)
+            loaded_count += inserted   
+        load_request_stats(cur, search_request_id, stats)
+        conn.commit()
+    return loaded_count
+