Переделано на everything

Author: EEnota

README.md

@@ -15,7 +15,7 @@ db: создается база данных и таблица если их е
 Pipeline: программа запускается скриптом разделенным на 5 основных модулей (db.py, extract.py, load.py, transform.py, main.py)
 
 ### Схема ETL
-top-headlines из NewsApi (extract) -> Филтр на наличие автора, заголовка, описания не меньше 20 символов, наличие url (transform) -> загрузка статей чей url отсутствует в базе данных (load)
+Everything из NewsApi (extract) -> Филтр на наличие автора, заголовка, описания не меньше 20 символов, наличие url (transform) -> загрузка статей чей url отсутствует в базе данных (load)
 
 ### Структура папок
 ```text

config/config.py

@@ -31,9 +31,10 @@ class Settings:
     db_news: str = os.getenv("DB_NEWS","news_db")
 
     KEY_API: str = os.getenv("NEWSAPI_KEY", "NO")
-    NEWS_URL = "https://newsapi.org/v2/top-headlines"
+    NEWS_URL = "https://newsapi.org/v2/everything"
 
-    COUNTRY = "US"
+    langueage = "ru"
+    sortBy = "publishedAt"
 
 settings = Settings()
 

main.py

@@ -24,12 +24,6 @@ def parse_args():
         required=True,
         help="Keyword for news search"
     )
-
-    parser.add_argument(
-        "--category",
-        required=True,
-        help="Category for news search"
-    )
     parser.add_argument(
         "--limit",
         type=positive_int,
@@ -48,23 +42,22 @@ def parse_args():
 def pipeline() -> None:
     args = parse_args()
     key_word = args.keyword
-    category = args.category
     limit = args.limit
     page_size = args.page_size
     num_of_news = 0
     page = 1
     while num_of_news < limit:
         remaining = limit - num_of_news
-        raw_file_name,raw_articles_count = make_extract(category, key_word, page, page_size)
+        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(raw_file_name, category, key_word, page)
+        clean_file_name = transform_article(raw_file_name, key_word, page)
         result_num_of_news = load_news(clean_file_name, max_rows=remaining)
 
         num_of_news += result_num_of_news
         page += 1
-    logger.info(f"{num_of_news} news on category {category} already aploaded")
+    logger.info(f"{num_of_news} news on key word {key_word} already aploaded")
     return num_of_news
 
 def main()-> None:

notebooks/01_eda.ipynb

@@ -15,8 +15,26 @@
     "import os\n",
     "from pathlib import Path \n",
     "import pandas as pd \n",
-    "import mathplotlib.pyplot as plt \n",
-    "3"
+    "import mathplotlib.pyplot as plt \n"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "8f9db0d1",
+   "metadata": {},
+   "source": [
+    "В чем поинт происходящего. Помимо очевидного сбора новостей, мне захотелось собрать некую статистику. \n",
+    "Как много статей от общего числа не проходят фильтр?\n",
+    "По каким критериям они не проходят?\n",
+    "Что является первичной причиной отброса?\n",
+    "Как часто по каждому ключевому слову попадаются одинаковые ссылки?\n",
+    "и главный вопрос, кто такой E_nota?\n",
+    "\n",
+    "Что будет происходить у меня далее?\n",
+    "Я буду считывать статистику из data/clean/stats скорее всего я буду после этого чистить папки data/clean и data/raw чтобы избегать использования лишней памяти. После чего я буду строить графики на тему:\n",
+    "А что является чаще всего причиной отброса в целом?\n",
+    "А что явяется чаще всего причиной по каждой теме?\n",
+    "А какое ключивое слово имеет больше всего одинаковых ссылок?"
    ]
   }
  ],

src/db.py

@@ -74,8 +74,7 @@ def create_news_tables() -> None:
     query = """
             CREATE TABLE IF NOT EXISTS bad_news_bears (
             id BIGSERIAL PRIMARY KEY,
-            country VARCHAR NOT NULL,
-            category VARCHAR NOT NULL,
+            language VARCHAR NOT NULL,
             key_word VARCHAR NOT NULL,
             author VARCHAR,
             title TEXT NOT NULL,

src/extract.py

@@ -9,11 +9,11 @@
 logger = logging.getLogger(__name__)
 BASE_DIR = Path(__file__).resolve().parent.parent
 
-def import_to_raw_json(data:dict[str, Any], category: str, key_word: str, page: int) -> str:
+def import_to_raw_json(data:dict[str, Any], key_word: str, page: int) -> str:
     raw_dir = BASE_DIR / "data" / "raw"
     raw_dir.mkdir(parents=True, exist_ok=True)
     timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
-    create_data = f"{timestamp}_{category}_{key_word}_page_{page}.json"
+    create_data = f"{timestamp}_{key_word}_page_{page}.json"
     file_path = raw_dir / create_data
 
     with open(file_path, "w", encoding="utf-8") as f:
@@ -23,29 +23,28 @@ def import_to_raw_json(data:dict[str, Any], category: str, key_word: str, page:
 
 
 
-def make_extract(category: str, 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) -> tuple[str,int]:
     params = {
     "apiKey": settings.KEY_API,
-    "country":settings.COUNTRY,
-    "category":category,
+    "language":settings.langueage,
     "q": key_word,
     "pageSize" : page_size,
-    "page" : page
+    "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["country"] = settings.COUNTRY
-        payload["category"] = category
+        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, category, key_word, page)
+        new_file_name = import_to_raw_json(payload, key_word, page)
         return new_file_name, articles_count
 
     except r.exceptions.Timeout:

src/load.py

@@ -15,8 +15,7 @@ def load_news(clean_news:str, max_rows: Optional[int] = None)-> int:
 
     query = """
             INSERT INTO bad_news_bears (
-            country, 
-            category, 
+            language, 
             key_word, 
             author, 
             title, 
@@ -25,7 +24,7 @@ def load_news(clean_news:str, max_rows: Optional[int] = None)-> int:
             published_at,
             fetched_at
             )
-            VALUES (%s, %s, %s, %s,%s, %s, %s, %s,%s)
+            VALUES (%s, %s, %s,%s, %s, %s, %s,%s)
             ON CONFLICT (url) DO NOTHING
             """
 
@@ -34,8 +33,7 @@ def load_news(clean_news:str, max_rows: Optional[int] = None)-> int:
             if max_rows is not None and num_of_news >= max_rows:
                 break
             cur.execute(query, 
-                        (new["country"],
-                         new["category"],
+                        (new["language"],
                          new["key_word"],
                          new["author"],
                          new["title"],

src/transform.py

@@ -58,8 +58,7 @@ def clean_article(data) -> tuple[list[dict[str,str]], dict]:
         statistic["accepted_articles"] += 1
         clean_data.append(
             {
-                "country": data["country"],
-                "category": data["category"],
+                "language": data["language"],
                 "key_word": data["key_word"],
                 "author": element["author"],
                 "title": element["title"],
@@ -71,7 +70,7 @@ def clean_article(data) -> tuple[list[dict[str,str]], dict]:
         )
     return clean_data, statistic
 
-def transform_article(new_file_name:str,category: str, hot_pot: str, page: int) -> str:
+def transform_article(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)
@@ -80,8 +79,8 @@ def transform_article(new_file_name:str,category: str, hot_pot: str, page: int)
     load_dir = BASE_DIR / "clean"
     load_dir.mkdir(parents=True, exist_ok=True)
     timestamp = datetime.now().strftime("%Y_%m_%d-%H-%M-%S")
-    stats_clean_data = f"stats_{category}_{hot_pot}_page_{page}_{timestamp}.json"
-    create_clean_data = f"cleaned_{category}_{hot_pot}_page_{page}_{timestamp}.json"
+    stats_clean_data = f"stats_{key_word}_page_{page}_{timestamp}.json"
+    create_clean_data = f"cleaned_{key_word}_page_{page}_{timestamp}.json"
     cleaned_file_path = load_dir / create_clean_data
     stats_dir = load_dir/ "stats" 
     stats_dir.mkdir(parents=True, exist_ok=True)