back part pre-relize

Author: EEnota

main.py

@@ -1,7 +1,18 @@
 import argparse
-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
 
+from src import (
+    init_database,
+    create_app_users_table,
+    create_search_requests_table,
+    create_articles_table,
+    create_user_news_table,
+    create_request_stats_table,
+    create_news_tables,
+    run_pipeline_for_web_user, 
+    run_debug_pipeline
+)
+
 logging.basicConfig(
     level=logging.INFO,
     format = "%(asctime)s | %(levelname)s | %(name)s | %(message)s"
@@ -39,46 +50,42 @@ def parse_args():
         help="Amount of articels on 1 page"
 
     )
+    parser.add_argument(
+        "--debug",
+        action="store_true"
+    )
+    parser.add_argument(
+        "--user_id", 
+        type=int,
+        default=1
+        )
+    parser.add_argument(
+        "--serch_request_id",
+        type=int,
+        default=1
+    )
     return parser.parse_args()
 
-def pipeline_for_web_user(user_id: int, search_request_id: int, key_word: str, limit: int, page_size: int, debug:bool = False) -> int:
+def init_all_tables(debug: bool) -> None:
+    init_database()
+    create_app_users_table()
+    create_articles_table()
+    create_request_stats_table()
+    create_search_requests_table()
+    create_user_news_table()
     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
-        if debug:
-            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)
-        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
-    logger.info(f"{num_of_news} news on key word {key_word} already aploaded")
-    return num_of_news
+        create_news_tables()
+
 
 def main()-> None:
+    args = parse_args()
     logger.info("Starting pipeline, init database, build table..")
     try:
-        init_database()
-        create_news_tables()
-        loaded = pipeline_for_web_user()
+        init_all_tables(debug=args.debug)
+        if args.debug:
+            loaded = run_debug_pipeline(args.keyword, args.limit, args.page_size)
+        else:
+            loaded = run_pipeline_for_web_user(args.user_id, args.search_request_id, args.ketword, args.limit, args.page_size)
         logger.info("Pipline finished. loaded rows: %s", loaded)
     except Exception as e:
         logger.exception("pipeline failed: %s", e)

src/__init__.py

@@ -1,4 +1,5 @@
 from .extract import make_extract_debug, make_extract_web
 from .transform import transform_article_web, transform_article_debug
 from .load import load_news, load_web_pipeline
-from .db import init_database, create_news_tables
+from .db import init_database, create_news_tables, create_app_users_table, create_search_requests_table, create_articles_table, create_user_news_table, create_request_stats_table
+from .pipeline import run_pipeline_for_web_user, run_debug_pipeline

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_debug( key_word: str, page: int = 1, page_size: int = 100) -> str:
+def make_extract_debug( key_word: str, page: int = 1, page_size: int = 100) -> tuple[str, int]:
     params = {
     "apiKey": settings.KEY_API,
     "language":settings.langueage,

src/pipeline.py

@@ -0,0 +1,42 @@
+import logging
+from src import (
+    make_extract_web,
+    make_extract_debug,
+    transform_article_web,
+    transform_article_debug,
+    load_web_pipeline,
+    load_news,
+)
+
+logger = logging.getLogger(__name__)
+
+def run_pipeline_for_web_user(user_id: int, search_request_id: int, key_word: str, limit: int, page_size: int) -> int:
+    num_of_news = 0
+    page = 1
+    while num_of_news < limit:
+        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
+    logger.info(f"{num_of_news} news on key word {key_word} already aploaded")
+    return num_of_news
+
+def run_debug_pipeline(keyword: str, limit:int, page_size: int) -> int:
+    num_of_news = 0
+    page = 1
+    while num_of_news < limit:
+        max_rows = limit - num_of_news
+        new_file_name, raw_articles_count = make_extract_debug(keyword, page, page_size)
+        if raw_articles_count == 0:
+            logger.warning("there is no more artical")
+            break
+        clean_data = transform_article_debug(new_file_name,keyword,page)
+        result_num_of_news = load_news(clean_data, max_rows)
+        num_of_news += result_num_of_news
+        page += 1
+    logger.info(f"{num_of_news} news on key word {keyword} already aploaded")
+    return num_of_news

src/worker.py

@@ -0,0 +1,88 @@
+import logging
+import time
+
+from src.db import get_cursor
+from config.config import settings
+from src import run_pipeline_for_web_user
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s | %(levelname)s |%(name)s | %(message)s"
+)
+logger = logging.getLogger(__name__)
+
+def get_nearest_queue() -> dict | None:
+    query = """ SELECT
+      id, user_id, keyword, language, limit_count, page_size
+       FROM search_requests
+        WHERE status = 'queued'
+        ORDER BY created_at
+         LIMIT 1 """
+    with get_cursor(settings.db_news) as (_, cur):
+        cur.execute(query)
+        return cur.fetchone()
+
+def mark_as_running(search_request_id: int) -> None:
+    query = """ UPDATE search_requests
+                SET
+                status = 'running',
+                started_at = NOW(),
+                error_text = NULL
+                WHERE id = %s
+            """
+    with get_cursor(settings.db_news) as (conn, cur):
+        cur.execute(query, (search_request_id,))
+        conn.commit()
+
+def mark_as_success(search_request_id: int) -> None:
+    query = """UPDATE search_requests
+                SET 
+                status = 'success',
+                finished_at = NOW()
+                WHERE id = %s """
+    with get_cursor(settings.db_news) as (conn, cur):
+        cur.execute(query, (search_request_id,))
+        conn.commit()
+def mark_as_error(search_request_id: int, error_text: str) -> None:
+    query = """UPDATE search_requests
+                SET
+                status = 'ERROR',
+                finished_at = NOW(),
+                error_text = %s
+                WHERE id = %s"""
+    with get_cursor(settings.db_news) as (conn,cur):
+        cur.execute(query, (error_text, search_request_id))
+        conn.commit()
+
+def one_request() -> bool:
+    request_row = get_nearest_queue()
+    if not request_row:
+        return False
+    search_request_id = request_row["id"]
+    user_id = request_row["user_id"]
+    keyword = request_row["keyword"]
+    limit_count = request_row["limit_count"]
+    page_size = request_row["page_size"]
+    logger.info(f"Pipeline starts for {user_id} on {search_request_id} by keyword {keyword}")
+    mark_as_running(search_request_id)
+
+    try:
+        amount_of_articles = run_pipeline_for_web_user(user_id, search_request_id, keyword, limit_count, page_size)
+
+        mark_as_success(search_request_id)
+        logger.info(f"Pipeline finished successfully for {user_id} on {search_request_id} by {keyword}")
+        logger.info(f"pushed {amount_of_articles} articles")
+    except Exception as e:
+        logger.exception(f"Pipeline for {user_id} on {search_request_id} by {keyword} failed: {e}")
+        mark_as_error(search_request_id, e)
+    return True
+
+def run_worker_loop(pull_interval: int = 3) -> None:
+    logger.info("Worker started")
+    while True:
+        processed = one_request()
+        if not processed:
+            time.sleep(pull_interval)
+
+if __name__ == "__main__":
+    run_worker_loop()