Development

.gitignore

@@ -1,4 +1,15 @@
 celerybeat-schedule.*
 src/celerybeat-schedule.*
 feed*.json
-.DS_Store
+.DS_Store
+
+# Ignore all __pycache__ directories at any level
+**/__pycache__/
+
+# Ignore all compiled Python files
+*.pyc
+*.pyo
+*.pyd
+
+#Ignore saved data
+data/*

docker-compose.yml

@@ -28,9 +28,36 @@ services:
  dockerfile: .docker/Dockerfile
  volumes:
  - .:/workspace:cached
+ - mongo_data:/data/db 
  depends_on:
  - redis
+ - mongo
  #entrypoint: ["/usr/local/bin/entrypoint.sh"]
  command: ["celery", "-A", justinsight.celery, "worker", "--loglevel=info"] #the lowercase j is actually so important
 
+ celery_beat:
+ build:
+ context: .
+ dockerfile: .docker/Dockerfile
+ volumes:
+ - .:/workspace:cached
+ depends_on:
+ - redis
+ #entrypoint: ["/usr/local/bin/entrypoint.sh"]
+ command: ["celery", "-A", justinsight.celery, "beat", "--loglevel=info"] #the lowercase j is actually so important
+
+ mongo:
+ image: mongo:latest
+ container_name: mongo
+ restart: unless-stopped
+ environment:
+ MONGO_INITDB_ROOT_USERNAME: myuser
+ MONGO_INITDB_ROOT_PASSWORD: mypassword
+ ports:
+ - "27017:27017"
+ volumes:
+ - mongo_data:/data/db 
+ - .:/workspace:cached
 
+volumes:
+ mongo_data:

readme.md

@@ -22,7 +22,7 @@ A lightweight, news intelligence service for ingesting free news sources, taggin
  cd justinsight
 
 
-## Setting up Celery Beat:
+## Setting up Celery Beat
 - On MacOS
  - run: docker compose up --build
  - OR for running in the background: docker compose up -d
@@ -34,5 +34,15 @@ A lightweight, news intelligence service for ingesting free news sources, taggin
  - use EC2 section of entrypoint.sh
  - to shut down celery: docker compose down
 
-*When running in background you can use "docker logs <container_name_or_id>" to see the log and "docker ps"
-to find the ID.
+Note: When running in background you can use "docker logs <container_name_or_id>" to see the log and "docker ps" to find the ID.
+
+## How to Create and Schedule a Task
+1) Create a method for your task wherever you are working on it that can be called and complete the task. It should be in a package (ie there should be an \_\_init\_\_.py file in the same folder as the file you are working on).
+2) Go to src/justinsight/tasks.py and create a new task following the format of the sample task. Remember to add an import statement if one is needed.
+3) Go to src/justinsight/celery.py and schedule a new task after the comment showing where to schedule them. The new task can follow the same format as those above
+
+ "sample-task-name": { 
+ "task": "justinsight.tasks.your_task_name", 
+ "schedule": x, #where x is the number of seconds between when the task should happen 
+ "args": (), #potential arguments for your task 
+ },

requirements.txt

@@ -1 +1,6 @@
-celery[redis]
+beautifulsoup4
+celery[redis]
+feedparser
+playwright
+pymongo
+requests

src/ingest/base_ingestor.py

@@ -0,0 +1,52 @@
+import datetime
+import feedparser
+import hashlib
+import re
+from src.ingest.save_to_database import save_entry
+
+class BaseIngestor:
+ RSS_URL = None #will be set by the subclasses 
+
+ def slugify(self, text):
+ # Convert title to a filesystem-friendly slug
+ text = text.lower()
+ text = re.sub(r'[^a-z0-9]+', '-', text)
+ return text.strip('-')
+
+ def format_date(self, entry):
+ # Extract and format the published date
+ try:
+ dt = datetime.datetime(*entry.published_parsed[:6])
+ return dt.strftime("%Y-%m-%d")
+ except:
+ return "unknown-date"
+
+ def fetch_full_text(self, url):
+ raise NotImplementedError(
+ "Subclass must implement fetch_full_text()"
+ )
+
+ def generate_entry_hash(self, entry):
+ hash_input = f"{entry.title}{entry.link}{entry.get('published', '')}"
+ return hashlib.sha256(hash_input.encode('utf-8')).hexdigest()
+
+ def format_entry(self, entry):
+ title_slug = self.slugify(entry.title)
+ date_str = self.format_date(entry)
+ full_text = self.fetch_full_text(entry.link)
+
+ data = {
+ "title": entry.title,
+ "link": entry.link,
+ "published": entry.get("published", ""),
+ "summary": entry.get("summary", ""),
+ "full_text": full_text,
+ "id": self.generate_entry_hash(entry)
+ }
+ return data
+
+ def check_and_save_new_entries(self):
+ feed = feedparser.parse(self.RSS_URL)
+
+ for entry in feed.entries:
+ save_entry(self.format_entry(entry))

src/ingest/bbc_ingestor.py

@@ -0,0 +1,15 @@
+from bs4 import BeautifulSoup
+import requests
+from src.ingest.base_ingestor import BaseIngestor
+
+class BBCIngestor(BaseIngestor):
+ RSS_URL = "http://feeds.bbci.co.uk/news/world/rss.xml"
+
+ def fetch_full_text(self, article_url):
+ response = requests.get(self.RSS_URL)
+ soup = BeautifulSoup(response.content, 'html.parser')
+
+ article = soup.find('article')
+
+ if article:
+ return(article.get_text())

src/ingest/save_to_database.py

@@ -0,0 +1,18 @@
+from pymongo import MongoClient
+
+# Include username, password, and authentication database
+client = MongoClient("mongodb://myuser:mypassword@mongo:27017/justinsightdb?authSource=admin")
+
+# Get (or create) a database
+db = client["justinsightdb"]
+
+# Get (or create) a collection
+collection = db["articles"]
+
+def save_entry(entry):
+ #check if the entry has already been saved and if it has not then save it
+ entry_hash = entry["id"]
+ if collection.count_documents({"id": entry_hash}) == 0:
+ collection.insert_one(entry)
+ else:
+ print(f"{entry['title']} already in use!")

src/justinsight/celery.py

@@ -0,0 +1,32 @@
+# celery.py
+from celery import Celery
+
+app = Celery(
+ "justinsight",
+ broker="redis://redis:6379/0",
+ backend="redis://redis:6379/0",
+ include=["justinsight.tasks"], 
+)
+
+# Optional beat schedule
+app.conf.beat_schedule = {
+ # "sample-task-every-5-seconds": {
+ # "task": "justinsight.tasks.sample_task",
+ # "schedule": 5.0,
+ # "args": (),
+ # },
+
+ "check-BBCfeed-every-5-minutes": {
+ "task": "justinsight.tasks.bbcLogger_task",
+ "schedule": 5.0,
+ "args": (),
+ },
+
+ # "check-NYTfeed-every-5-minutes": {
+ # "task": "justinsight.tasks.nytLogger_task",
+ # "schedule": 300.0,
+ # "args": (),
+ # },
+
+ #schedule more tasks here
+}

src/justinsight/tasks.py

@@ -0,0 +1,23 @@
+# tasks.py
+from celery import shared_task
+from ingest.bbc_ingestor import BBCIngestor
+
+
+@shared_task
+def sample_task():
+ print("Sample task!")
+ return "done!"
+
+@shared_task
+def bbcLogger_task():
+ # Create an instance of the class
+ ingestor = BBCIngestor()
+ ingestor.check_and_save_new_entries() # this will invoke the inherited logic
+ return "BBC RSS Feed checked."
+
+@shared_task
+def nytLogger_task():
+ check_and_save_nyt()
+ return "NYT RSS Feed checked."
+
+#Add more tasks here in the format of the one above

.gitignore

@@ -1,4 +1,15 @@
 celerybeat-schedule.*
 src/celerybeat-schedule.*
 feed*.json
-.DS_Store
+.DS_Store
+
+# Ignore all __pycache__ directories at any level
+**/__pycache__/
+
+# Ignore all compiled Python files
+*.pyc
+*.pyo
+*.pyd
+
+#Ignore saved data
+data/*

docker-compose.yml

@@ -28,9 +28,36 @@ services:
  dockerfile: .docker/Dockerfile
  volumes:
  - .:/workspace:cached
+ - mongo_data:/data/db 
  depends_on:
  - redis
+ - mongo
  #entrypoint: ["/usr/local/bin/entrypoint.sh"]
  command: ["celery", "-A", justinsight.celery, "worker", "--loglevel=info"] #the lowercase j is actually so important
 
+ celery_beat:
+ build:
+ context: .
+ dockerfile: .docker/Dockerfile
+ volumes:
+ - .:/workspace:cached
+ depends_on:
+ - redis
+ #entrypoint: ["/usr/local/bin/entrypoint.sh"]
+ command: ["celery", "-A", justinsight.celery, "beat", "--loglevel=info"] #the lowercase j is actually so important
+
+ mongo:
+ image: mongo:latest
+ container_name: mongo
+ restart: unless-stopped
+ environment:
+ MONGO_INITDB_ROOT_USERNAME: myuser
+ MONGO_INITDB_ROOT_PASSWORD: mypassword
+ ports:
+ - "27017:27017"
+ volumes:
+ - mongo_data:/data/db 
+ - .:/workspace:cached
 
+volumes:
+ mongo_data:

readme.md

@@ -22,7 +22,7 @@ A lightweight, news intelligence service for ingesting free news sources, taggin
  cd justinsight
 
 
-## Setting up Celery Beat:
+## Setting up Celery Beat
 - On MacOS
  - run: docker compose up --build
  - OR for running in the background: docker compose up -d
@@ -34,5 +34,15 @@ A lightweight, news intelligence service for ingesting free news sources, taggin
  - use EC2 section of entrypoint.sh
  - to shut down celery: docker compose down
 
-*When running in background you can use "docker logs <container_name_or_id>" to see the log and "docker ps"
-to find the ID.
+Note: When running in background you can use "docker logs <container_name_or_id>" to see the log and "docker ps" to find the ID.
+
+## How to Create and Schedule a Task
+1) Create a method for your task wherever you are working on it that can be called and complete the task. It should be in a package (ie there should be an \_\_init\_\_.py file in the same folder as the file you are working on).
+2) Go to src/justinsight/tasks.py and create a new task following the format of the sample task. Remember to add an import statement if one is needed.
+3) Go to src/justinsight/celery.py and schedule a new task after the comment showing where to schedule them. The new task can follow the same format as those above
+
+ "sample-task-name": { 
+ "task": "justinsight.tasks.your_task_name", 
+ "schedule": x, #where x is the number of seconds between when the task should happen 
+ "args": (), #potential arguments for your task 
+ },

requirements.txt

@@ -1 +1,6 @@
-celery[redis]
+beautifulsoup4
+celery[redis]
+feedparser
+playwright
+pymongo
+requests

src/ingest/base_ingestor.py

@@ -0,0 +1,52 @@
+import datetime
+import feedparser
+import hashlib
+import re
+from src.ingest.save_to_database import save_entry
+
+class BaseIngestor:
+ RSS_URL = None #will be set by the subclasses 
+
+ def slugify(self, text):
+ # Convert title to a filesystem-friendly slug
+ text = text.lower()
+ text = re.sub(r'[^a-z0-9]+', '-', text)
+ return text.strip('-')
+
+ def format_date(self, entry):
+ # Extract and format the published date
+ try:
+ dt = datetime.datetime(*entry.published_parsed[:6])
+ return dt.strftime("%Y-%m-%d")
+ except:
+ return "unknown-date"
+
+ def fetch_full_text(self, url):
+ raise NotImplementedError(
+ "Subclass must implement fetch_full_text()"
+ )
+
+ def generate_entry_hash(self, entry):
+ hash_input = f"{entry.title}{entry.link}{entry.get('published', '')}"
+ return hashlib.sha256(hash_input.encode('utf-8')).hexdigest()
+
+ def format_entry(self, entry):
+ title_slug = self.slugify(entry.title)
+ date_str = self.format_date(entry)
+ full_text = self.fetch_full_text(entry.link)
+
+ data = {
+ "title": entry.title,
+ "link": entry.link,
+ "published": entry.get("published", ""),
+ "summary": entry.get("summary", ""),
+ "full_text": full_text,
+ "id": self.generate_entry_hash(entry)
+ }
+ return data
+
+ def check_and_save_new_entries(self):
+ feed = feedparser.parse(self.RSS_URL)
+
+ for entry in feed.entries:
+ save_entry(self.format_entry(entry))

src/ingest/bbc_ingestor.py

@@ -0,0 +1,15 @@
+from bs4 import BeautifulSoup
+import requests
+from src.ingest.base_ingestor import BaseIngestor
+
+class BBCIngestor(BaseIngestor):
+ RSS_URL = "http://feeds.bbci.co.uk/news/world/rss.xml"
+
+ def fetch_full_text(self, article_url):
+ response = requests.get(self.RSS_URL)
+ soup = BeautifulSoup(response.content, 'html.parser')
+
+ article = soup.find('article')
+
+ if article:
+ return(article.get_text())

src/ingest/save_to_database.py

@@ -0,0 +1,18 @@
+from pymongo import MongoClient
+
+# Include username, password, and authentication database
+client = MongoClient("mongodb://myuser:mypassword@mongo:27017/justinsightdb?authSource=admin")
+
+# Get (or create) a database
+db = client["justinsightdb"]
+
+# Get (or create) a collection
+collection = db["articles"]
+
+def save_entry(entry):
+ #check if the entry has already been saved and if it has not then save it
+ entry_hash = entry["id"]
+ if collection.count_documents({"id": entry_hash}) == 0:
+ collection.insert_one(entry)
+ else:
+ print(f"{entry['title']} already in use!")

src/justinsight/celery.py

@@ -0,0 +1,32 @@
+# celery.py
+from celery import Celery
+
+app = Celery(
+ "justinsight",
+ broker="redis://redis:6379/0",
+ backend="redis://redis:6379/0",
+ include=["justinsight.tasks"], 
+)
+
+# Optional beat schedule
+app.conf.beat_schedule = {
+ # "sample-task-every-5-seconds": {
+ # "task": "justinsight.tasks.sample_task",
+ # "schedule": 5.0,
+ # "args": (),
+ # },
+
+ "check-BBCfeed-every-5-minutes": {
+ "task": "justinsight.tasks.bbcLogger_task",
+ "schedule": 5.0,
+ "args": (),
+ },
+
+ # "check-NYTfeed-every-5-minutes": {
+ # "task": "justinsight.tasks.nytLogger_task",
+ # "schedule": 300.0,
+ # "args": (),
+ # },
+
+ #schedule more tasks here
+}

src/justinsight/tasks.py

@@ -0,0 +1,23 @@
+# tasks.py
+from celery import shared_task
+from ingest.bbc_ingestor import BBCIngestor
+
+
+@shared_task
+def sample_task():
+ print("Sample task!")
+ return "done!"
+
+@shared_task
+def bbcLogger_task():
+ # Create an instance of the class
+ ingestor = BBCIngestor()
+ ingestor.check_and_save_new_entries() # this will invoke the inherited logic
+ return "BBC RSS Feed checked."
+
+@shared_task
+def nytLogger_task():
+ check_and_save_nyt()
+ return "NYT RSS Feed checked."
+
+#Add more tasks here in the format of the one above

.gitignore

@@ -1,4 +1,15 @@
 celerybeat-schedule.*
 src/celerybeat-schedule.*
 feed*.json
-.DS_Store
+.DS_Store
+
+# Ignore all __pycache__ directories at any level
+**/__pycache__/
+
+# Ignore all compiled Python files
+*.pyc
+*.pyo
+*.pyd
+
+#Ignore saved data
+data/*

docker-compose.yml

@@ -28,9 +28,36 @@ services:
  dockerfile: .docker/Dockerfile
  volumes:
  - .:/workspace:cached
+ - mongo_data:/data/db 
  depends_on:
  - redis
+ - mongo
  #entrypoint: ["/usr/local/bin/entrypoint.sh"]
  command: ["celery", "-A", justinsight.celery, "worker", "--loglevel=info"] #the lowercase j is actually so important
 
+ celery_beat:
+ build:
+ context: .
+ dockerfile: .docker/Dockerfile
+ volumes:
+ - .:/workspace:cached
+ depends_on:
+ - redis
+ #entrypoint: ["/usr/local/bin/entrypoint.sh"]
+ command: ["celery", "-A", justinsight.celery, "beat", "--loglevel=info"] #the lowercase j is actually so important
+
+ mongo:
+ image: mongo:latest
+ container_name: mongo
+ restart: unless-stopped
+ environment:
+ MONGO_INITDB_ROOT_USERNAME: myuser
+ MONGO_INITDB_ROOT_PASSWORD: mypassword
+ ports:
+ - "27017:27017"
+ volumes:
+ - mongo_data:/data/db 
+ - .:/workspace:cached
 
+volumes:
+ mongo_data:

readme.md

@@ -22,7 +22,7 @@ A lightweight, news intelligence service for ingesting free news sources, taggin
  cd justinsight
 
 
-## Setting up Celery Beat:
+## Setting up Celery Beat
 - On MacOS
  - run: docker compose up --build
  - OR for running in the background: docker compose up -d
@@ -34,5 +34,15 @@ A lightweight, news intelligence service for ingesting free news sources, taggin
  - use EC2 section of entrypoint.sh
  - to shut down celery: docker compose down
 
-*When running in background you can use "docker logs <container_name_or_id>" to see the log and "docker ps"
-to find the ID.
+Note: When running in background you can use "docker logs <container_name_or_id>" to see the log and "docker ps" to find the ID.
+
+## How to Create and Schedule a Task
+1) Create a method for your task wherever you are working on it that can be called and complete the task. It should be in a package (ie there should be an \_\_init\_\_.py file in the same folder as the file you are working on).
+2) Go to src/justinsight/tasks.py and create a new task following the format of the sample task. Remember to add an import statement if one is needed.
+3) Go to src/justinsight/celery.py and schedule a new task after the comment showing where to schedule them. The new task can follow the same format as those above
+
+ "sample-task-name": { 
+ "task": "justinsight.tasks.your_task_name", 
+ "schedule": x, #where x is the number of seconds between when the task should happen 
+ "args": (), #potential arguments for your task 
+ },

requirements.txt

@@ -1 +1,6 @@
-celery[redis]
+beautifulsoup4
+celery[redis]
+feedparser
+playwright
+pymongo
+requests

src/ingest/base_ingestor.py

@@ -0,0 +1,52 @@
+import datetime
+import feedparser
+import hashlib
+import re
+from src.ingest.save_to_database import save_entry
+
+class BaseIngestor:
+ RSS_URL = None #will be set by the subclasses 
+
+ def slugify(self, text):
+ # Convert title to a filesystem-friendly slug
+ text = text.lower()
+ text = re.sub(r'[^a-z0-9]+', '-', text)
+ return text.strip('-')
+
+ def format_date(self, entry):
+ # Extract and format the published date
+ try:
+ dt = datetime.datetime(*entry.published_parsed[:6])
+ return dt.strftime("%Y-%m-%d")
+ except:
+ return "unknown-date"
+
+ def fetch_full_text(self, url):
+ raise NotImplementedError(
+ "Subclass must implement fetch_full_text()"
+ )
+
+ def generate_entry_hash(self, entry):
+ hash_input = f"{entry.title}{entry.link}{entry.get('published', '')}"
+ return hashlib.sha256(hash_input.encode('utf-8')).hexdigest()
+
+ def format_entry(self, entry):
+ title_slug = self.slugify(entry.title)
+ date_str = self.format_date(entry)
+ full_text = self.fetch_full_text(entry.link)
+
+ data = {
+ "title": entry.title,
+ "link": entry.link,
+ "published": entry.get("published", ""),
+ "summary": entry.get("summary", ""),
+ "full_text": full_text,
+ "id": self.generate_entry_hash(entry)
+ }
+ return data
+
+ def check_and_save_new_entries(self):
+ feed = feedparser.parse(self.RSS_URL)
+
+ for entry in feed.entries:
+ save_entry(self.format_entry(entry))

src/ingest/bbc_ingestor.py

@@ -0,0 +1,15 @@
+from bs4 import BeautifulSoup
+import requests
+from src.ingest.base_ingestor import BaseIngestor
+
+class BBCIngestor(BaseIngestor):
+ RSS_URL = "http://feeds.bbci.co.uk/news/world/rss.xml"
+
+ def fetch_full_text(self, article_url):
+ response = requests.get(self.RSS_URL)
+ soup = BeautifulSoup(response.content, 'html.parser')
+
+ article = soup.find('article')
+
+ if article:
+ return(article.get_text())

src/ingest/save_to_database.py

@@ -0,0 +1,18 @@
+from pymongo import MongoClient
+
+# Include username, password, and authentication database
+client = MongoClient("mongodb://myuser:mypassword@mongo:27017/justinsightdb?authSource=admin")
+
+# Get (or create) a database
+db = client["justinsightdb"]
+
+# Get (or create) a collection
+collection = db["articles"]
+
+def save_entry(entry):
+ #check if the entry has already been saved and if it has not then save it
+ entry_hash = entry["id"]
+ if collection.count_documents({"id": entry_hash}) == 0:
+ collection.insert_one(entry)
+ else:
+ print(f"{entry['title']} already in use!")

src/justinsight/celery.py

@@ -0,0 +1,32 @@
+# celery.py
+from celery import Celery
+
+app = Celery(
+ "justinsight",
+ broker="redis://redis:6379/0",
+ backend="redis://redis:6379/0",
+ include=["justinsight.tasks"], 
+)
+
+# Optional beat schedule
+app.conf.beat_schedule = {
+ # "sample-task-every-5-seconds": {
+ # "task": "justinsight.tasks.sample_task",
+ # "schedule": 5.0,
+ # "args": (),
+ # },
+
+ "check-BBCfeed-every-5-minutes": {
+ "task": "justinsight.tasks.bbcLogger_task",
+ "schedule": 5.0,
+ "args": (),
+ },
+
+ # "check-NYTfeed-every-5-minutes": {
+ # "task": "justinsight.tasks.nytLogger_task",
+ # "schedule": 300.0,
+ # "args": (),
+ # },
+
+ #schedule more tasks here
+}

src/justinsight/tasks.py

@@ -0,0 +1,23 @@
+# tasks.py
+from celery import shared_task
+from ingest.bbc_ingestor import BBCIngestor
+
+
+@shared_task
+def sample_task():
+ print("Sample task!")
+ return "done!"
+
+@shared_task
+def bbcLogger_task():
+ # Create an instance of the class
+ ingestor = BBCIngestor()
+ ingestor.check_and_save_new_entries() # this will invoke the inherited logic
+ return "BBC RSS Feed checked."
+
+@shared_task
+def nytLogger_task():
+ check_and_save_nyt()
+ return "NYT RSS Feed checked."
+
+#Add more tasks here in the format of the one above

.gitignore

@@ -1,4 +1,15 @@
 celerybeat-schedule.*
 src/celerybeat-schedule.*
 feed*.json
-.DS_Store
+.DS_Store
+
+# Ignore all __pycache__ directories at any level
+**/__pycache__/
+
+# Ignore all compiled Python files
+*.pyc
+*.pyo
+*.pyd
+
+#Ignore saved data
+data/*

docker-compose.yml

@@ -28,9 +28,36 @@ services:
  dockerfile: .docker/Dockerfile
  volumes:
  - .:/workspace:cached
+ - mongo_data:/data/db 
  depends_on:
  - redis
+ - mongo
  #entrypoint: ["/usr/local/bin/entrypoint.sh"]
  command: ["celery", "-A", justinsight.celery, "worker", "--loglevel=info"] #the lowercase j is actually so important
 
+ celery_beat:
+ build:
+ context: .
+ dockerfile: .docker/Dockerfile
+ volumes:
+ - .:/workspace:cached
+ depends_on:
+ - redis
+ #entrypoint: ["/usr/local/bin/entrypoint.sh"]
+ command: ["celery", "-A", justinsight.celery, "beat", "--loglevel=info"] #the lowercase j is actually so important
+
+ mongo:
+ image: mongo:latest
+ container_name: mongo
+ restart: unless-stopped
+ environment:
+ MONGO_INITDB_ROOT_USERNAME: myuser
+ MONGO_INITDB_ROOT_PASSWORD: mypassword
+ ports:
+ - "27017:27017"
+ volumes:
+ - mongo_data:/data/db 
+ - .:/workspace:cached
 
+volumes:
+ mongo_data:

readme.md

@@ -22,7 +22,7 @@ A lightweight, news intelligence service for ingesting free news sources, taggin
  cd justinsight
 
 
-## Setting up Celery Beat:
+## Setting up Celery Beat
 - On MacOS
  - run: docker compose up --build
  - OR for running in the background: docker compose up -d
@@ -34,5 +34,15 @@ A lightweight, news intelligence service for ingesting free news sources, taggin
  - use EC2 section of entrypoint.sh
  - to shut down celery: docker compose down
 
-*When running in background you can use "docker logs <container_name_or_id>" to see the log and "docker ps"
-to find the ID.
+Note: When running in background you can use "docker logs <container_name_or_id>" to see the log and "docker ps" to find the ID.
+
+## How to Create and Schedule a Task
+1) Create a method for your task wherever you are working on it that can be called and complete the task. It should be in a package (ie there should be an \_\_init\_\_.py file in the same folder as the file you are working on).
+2) Go to src/justinsight/tasks.py and create a new task following the format of the sample task. Remember to add an import statement if one is needed.
+3) Go to src/justinsight/celery.py and schedule a new task after the comment showing where to schedule them. The new task can follow the same format as those above
+
+ "sample-task-name": { 
+ "task": "justinsight.tasks.your_task_name", 
+ "schedule": x, #where x is the number of seconds between when the task should happen 
+ "args": (), #potential arguments for your task 
+ },

requirements.txt

@@ -1 +1,6 @@
-celery[redis]
+beautifulsoup4
+celery[redis]
+feedparser
+playwright
+pymongo
+requests

src/ingest/base_ingestor.py

@@ -0,0 +1,52 @@
+import datetime
+import feedparser
+import hashlib
+import re
+from src.ingest.save_to_database import save_entry
+
+class BaseIngestor:
+ RSS_URL = None #will be set by the subclasses 
+
+ def slugify(self, text):
+ # Convert title to a filesystem-friendly slug
+ text = text.lower()
+ text = re.sub(r'[^a-z0-9]+', '-', text)
+ return text.strip('-')
+
+ def format_date(self, entry):
+ # Extract and format the published date
+ try:
+ dt = datetime.datetime(*entry.published_parsed[:6])
+ return dt.strftime("%Y-%m-%d")
+ except:
+ return "unknown-date"
+
+ def fetch_full_text(self, url):
+ raise NotImplementedError(
+ "Subclass must implement fetch_full_text()"
+ )
+
+ def generate_entry_hash(self, entry):
+ hash_input = f"{entry.title}{entry.link}{entry.get('published', '')}"
+ return hashlib.sha256(hash_input.encode('utf-8')).hexdigest()
+
+ def format_entry(self, entry):
+ title_slug = self.slugify(entry.title)
+ date_str = self.format_date(entry)
+ full_text = self.fetch_full_text(entry.link)
+
+ data = {
+ "title": entry.title,
+ "link": entry.link,
+ "published": entry.get("published", ""),
+ "summary": entry.get("summary", ""),
+ "full_text": full_text,
+ "id": self.generate_entry_hash(entry)
+ }
+ return data
+
+ def check_and_save_new_entries(self):
+ feed = feedparser.parse(self.RSS_URL)
+
+ for entry in feed.entries:
+ save_entry(self.format_entry(entry))

src/ingest/bbc_ingestor.py

@@ -0,0 +1,15 @@
+from bs4 import BeautifulSoup
+import requests
+from src.ingest.base_ingestor import BaseIngestor
+
+class BBCIngestor(BaseIngestor):
+ RSS_URL = "http://feeds.bbci.co.uk/news/world/rss.xml"
+
+ def fetch_full_text(self, article_url):
+ response = requests.get(self.RSS_URL)
+ soup = BeautifulSoup(response.content, 'html.parser')
+
+ article = soup.find('article')
+
+ if article:
+ return(article.get_text())

src/ingest/save_to_database.py

@@ -0,0 +1,18 @@
+from pymongo import MongoClient
+
+# Include username, password, and authentication database
+client = MongoClient("mongodb://myuser:mypassword@mongo:27017/justinsightdb?authSource=admin")
+
+# Get (or create) a database
+db = client["justinsightdb"]
+
+# Get (or create) a collection
+collection = db["articles"]
+
+def save_entry(entry):
+ #check if the entry has already been saved and if it has not then save it
+ entry_hash = entry["id"]
+ if collection.count_documents({"id": entry_hash}) == 0:
+ collection.insert_one(entry)
+ else:
+ print(f"{entry['title']} already in use!")

src/justinsight/celery.py

@@ -0,0 +1,32 @@
+# celery.py
+from celery import Celery
+
+app = Celery(
+ "justinsight",
+ broker="redis://redis:6379/0",
+ backend="redis://redis:6379/0",
+ include=["justinsight.tasks"], 
+)
+
+# Optional beat schedule
+app.conf.beat_schedule = {
+ # "sample-task-every-5-seconds": {
+ # "task": "justinsight.tasks.sample_task",
+ # "schedule": 5.0,
+ # "args": (),
+ # },
+
+ "check-BBCfeed-every-5-minutes": {
+ "task": "justinsight.tasks.bbcLogger_task",
+ "schedule": 5.0,
+ "args": (),
+ },
+
+ # "check-NYTfeed-every-5-minutes": {
+ # "task": "justinsight.tasks.nytLogger_task",
+ # "schedule": 300.0,
+ # "args": (),
+ # },
+
+ #schedule more tasks here
+}

src/justinsight/tasks.py

@@ -0,0 +1,23 @@
+# tasks.py
+from celery import shared_task
+from ingest.bbc_ingestor import BBCIngestor
+
+
+@shared_task
+def sample_task():
+ print("Sample task!")
+ return "done!"
+
+@shared_task
+def bbcLogger_task():
+ # Create an instance of the class
+ ingestor = BBCIngestor()
+ ingestor.check_and_save_new_entries() # this will invoke the inherited logic
+ return "BBC RSS Feed checked."
+
+@shared_task
+def nytLogger_task():
+ check_and_save_nyt()
+ return "NYT RSS Feed checked."
+
+#Add more tasks here in the format of the one above

.gitignore

@@ -1,4 +1,15 @@
 celerybeat-schedule.*
 src/celerybeat-schedule.*
 feed*.json
-.DS_Store
+.DS_Store
+
+# Ignore all __pycache__ directories at any level
+**/__pycache__/
+
+# Ignore all compiled Python files
+*.pyc
+*.pyo
+*.pyd
+
+#Ignore saved data
+data/*

docker-compose.yml

@@ -28,9 +28,36 @@ services:
  dockerfile: .docker/Dockerfile
  volumes:
  - .:/workspace:cached
+ - mongo_data:/data/db 
  depends_on:
  - redis
+ - mongo
  #entrypoint: ["/usr/local/bin/entrypoint.sh"]
  command: ["celery", "-A", justinsight.celery, "worker", "--loglevel=info"] #the lowercase j is actually so important
 
+ celery_beat:
+ build:
+ context: .
+ dockerfile: .docker/Dockerfile
+ volumes:
+ - .:/workspace:cached
+ depends_on:
+ - redis
+ #entrypoint: ["/usr/local/bin/entrypoint.sh"]
+ command: ["celery", "-A", justinsight.celery, "beat", "--loglevel=info"] #the lowercase j is actually so important
+
+ mongo:
+ image: mongo:latest
+ container_name: mongo
+ restart: unless-stopped
+ environment:
+ MONGO_INITDB_ROOT_USERNAME: myuser
+ MONGO_INITDB_ROOT_PASSWORD: mypassword
+ ports:
+ - "27017:27017"
+ volumes:
+ - mongo_data:/data/db 
+ - .:/workspace:cached
 
+volumes:
+ mongo_data:

readme.md

@@ -22,7 +22,7 @@ A lightweight, news intelligence service for ingesting free news sources, taggin
  cd justinsight
 
 
-## Setting up Celery Beat:
+## Setting up Celery Beat
 - On MacOS
  - run: docker compose up --build
  - OR for running in the background: docker compose up -d
@@ -34,5 +34,15 @@ A lightweight, news intelligence service for ingesting free news sources, taggin
  - use EC2 section of entrypoint.sh
  - to shut down celery: docker compose down
 
-*When running in background you can use "docker logs <container_name_or_id>" to see the log and "docker ps"
-to find the ID.
+Note: When running in background you can use "docker logs <container_name_or_id>" to see the log and "docker ps" to find the ID.
+
+## How to Create and Schedule a Task
+1) Create a method for your task wherever you are working on it that can be called and complete the task. It should be in a package (ie there should be an \_\_init\_\_.py file in the same folder as the file you are working on).
+2) Go to src/justinsight/tasks.py and create a new task following the format of the sample task. Remember to add an import statement if one is needed.
+3) Go to src/justinsight/celery.py and schedule a new task after the comment showing where to schedule them. The new task can follow the same format as those above
+
+ "sample-task-name": { 
+ "task": "justinsight.tasks.your_task_name", 
+ "schedule": x, #where x is the number of seconds between when the task should happen 
+ "args": (), #potential arguments for your task 
+ },

requirements.txt

@@ -1 +1,6 @@
-celery[redis]
+beautifulsoup4
+celery[redis]
+feedparser
+playwright
+pymongo
+requests

src/ingest/base_ingestor.py

@@ -0,0 +1,52 @@
+import datetime
+import feedparser
+import hashlib
+import re
+from src.ingest.save_to_database import save_entry
+
+class BaseIngestor:
+ RSS_URL = None #will be set by the subclasses 
+
+ def slugify(self, text):
+ # Convert title to a filesystem-friendly slug
+ text = text.lower()
+ text = re.sub(r'[^a-z0-9]+', '-', text)
+ return text.strip('-')
+
+ def format_date(self, entry):
+ # Extract and format the published date
+ try:
+ dt = datetime.datetime(*entry.published_parsed[:6])
+ return dt.strftime("%Y-%m-%d")
+ except:
+ return "unknown-date"
+
+ def fetch_full_text(self, url):
+ raise NotImplementedError(
+ "Subclass must implement fetch_full_text()"
+ )
+
+ def generate_entry_hash(self, entry):
+ hash_input = f"{entry.title}{entry.link}{entry.get('published', '')}"
+ return hashlib.sha256(hash_input.encode('utf-8')).hexdigest()
+
+ def format_entry(self, entry):
+ title_slug = self.slugify(entry.title)
+ date_str = self.format_date(entry)
+ full_text = self.fetch_full_text(entry.link)
+
+ data = {
+ "title": entry.title,
+ "link": entry.link,
+ "published": entry.get("published", ""),
+ "summary": entry.get("summary", ""),
+ "full_text": full_text,
+ "id": self.generate_entry_hash(entry)
+ }
+ return data
+
+ def check_and_save_new_entries(self):
+ feed = feedparser.parse(self.RSS_URL)
+
+ for entry in feed.entries:
+ save_entry(self.format_entry(entry))

src/ingest/bbc_ingestor.py

@@ -0,0 +1,15 @@
+from bs4 import BeautifulSoup
+import requests
+from src.ingest.base_ingestor import BaseIngestor
+
+class BBCIngestor(BaseIngestor):
+ RSS_URL = "http://feeds.bbci.co.uk/news/world/rss.xml"
+
+ def fetch_full_text(self, article_url):
+ response = requests.get(self.RSS_URL)
+ soup = BeautifulSoup(response.content, 'html.parser')
+
+ article = soup.find('article')
+
+ if article:
+ return(article.get_text())

src/ingest/save_to_database.py

@@ -0,0 +1,18 @@
+from pymongo import MongoClient
+
+# Include username, password, and authentication database
+client = MongoClient("mongodb://myuser:mypassword@mongo:27017/justinsightdb?authSource=admin")
+
+# Get (or create) a database
+db = client["justinsightdb"]
+
+# Get (or create) a collection
+collection = db["articles"]
+
+def save_entry(entry):
+ #check if the entry has already been saved and if it has not then save it
+ entry_hash = entry["id"]
+ if collection.count_documents({"id": entry_hash}) == 0:
+ collection.insert_one(entry)
+ else:
+ print(f"{entry['title']} already in use!")

src/justinsight/celery.py

@@ -0,0 +1,32 @@
+# celery.py
+from celery import Celery
+
+app = Celery(
+ "justinsight",
+ broker="redis://redis:6379/0",
+ backend="redis://redis:6379/0",
+ include=["justinsight.tasks"], 
+)
+
+# Optional beat schedule
+app.conf.beat_schedule = {
+ # "sample-task-every-5-seconds": {
+ # "task": "justinsight.tasks.sample_task",
+ # "schedule": 5.0,
+ # "args": (),
+ # },
+
+ "check-BBCfeed-every-5-minutes": {
+ "task": "justinsight.tasks.bbcLogger_task",
+ "schedule": 5.0,
+ "args": (),
+ },
+
+ # "check-NYTfeed-every-5-minutes": {
+ # "task": "justinsight.tasks.nytLogger_task",
+ # "schedule": 300.0,
+ # "args": (),
+ # },
+
+ #schedule more tasks here
+}

src/justinsight/tasks.py

@@ -0,0 +1,23 @@
+# tasks.py
+from celery import shared_task
+from ingest.bbc_ingestor import BBCIngestor
+
+
+@shared_task
+def sample_task():
+ print("Sample task!")
+ return "done!"
+
+@shared_task
+def bbcLogger_task():
+ # Create an instance of the class
+ ingestor = BBCIngestor()
+ ingestor.check_and_save_new_entries() # this will invoke the inherited logic
+ return "BBC RSS Feed checked."
+
+@shared_task
+def nytLogger_task():
+ check_and_save_nyt()
+ return "NYT RSS Feed checked."
+
+#Add more tasks here in the format of the one above

.gitignore

@@ -1,4 +1,15 @@
 celerybeat-schedule.*
 src/celerybeat-schedule.*
 feed*.json
-.DS_Store
+.DS_Store
+
+# Ignore all __pycache__ directories at any level
+**/__pycache__/
+
+# Ignore all compiled Python files
+*.pyc
+*.pyo
+*.pyd
+
+#Ignore saved data
+data/*

docker-compose.yml

@@ -28,9 +28,36 @@ services:
  dockerfile: .docker/Dockerfile
  volumes:
  - .:/workspace:cached
+ - mongo_data:/data/db 
  depends_on:
  - redis
+ - mongo
  #entrypoint: ["/usr/local/bin/entrypoint.sh"]
  command: ["celery", "-A", justinsight.celery, "worker", "--loglevel=info"] #the lowercase j is actually so important
 
+ celery_beat:
+ build:
+ context: .
+ dockerfile: .docker/Dockerfile
+ volumes:
+ - .:/workspace:cached
+ depends_on:
+ - redis
+ #entrypoint: ["/usr/local/bin/entrypoint.sh"]
+ command: ["celery", "-A", justinsight.celery, "beat", "--loglevel=info"] #the lowercase j is actually so important
+
+ mongo:
+ image: mongo:latest
+ container_name: mongo
+ restart: unless-stopped
+ environment:
+ MONGO_INITDB_ROOT_USERNAME: myuser
+ MONGO_INITDB_ROOT_PASSWORD: mypassword
+ ports:
+ - "27017:27017"
+ volumes:
+ - mongo_data:/data/db 
+ - .:/workspace:cached
 
+volumes:
+ mongo_data:

readme.md

@@ -22,7 +22,7 @@ A lightweight, news intelligence service for ingesting free news sources, taggin
  cd justinsight
 
 
-## Setting up Celery Beat:
+## Setting up Celery Beat
 - On MacOS
  - run: docker compose up --build
  - OR for running in the background: docker compose up -d
@@ -34,5 +34,15 @@ A lightweight, news intelligence service for ingesting free news sources, taggin
  - use EC2 section of entrypoint.sh
  - to shut down celery: docker compose down
 
-*When running in background you can use "docker logs <container_name_or_id>" to see the log and "docker ps"
-to find the ID.
+Note: When running in background you can use "docker logs <container_name_or_id>" to see the log and "docker ps" to find the ID.
+
+## How to Create and Schedule a Task
+1) Create a method for your task wherever you are working on it that can be called and complete the task. It should be in a package (ie there should be an \_\_init\_\_.py file in the same folder as the file you are working on).
+2) Go to src/justinsight/tasks.py and create a new task following the format of the sample task. Remember to add an import statement if one is needed.
+3) Go to src/justinsight/celery.py and schedule a new task after the comment showing where to schedule them. The new task can follow the same format as those above
+
+ "sample-task-name": { 
+ "task": "justinsight.tasks.your_task_name", 
+ "schedule": x, #where x is the number of seconds between when the task should happen 
+ "args": (), #potential arguments for your task 
+ },

requirements.txt

@@ -1 +1,6 @@
-celery[redis]
+beautifulsoup4
+celery[redis]
+feedparser
+playwright
+pymongo
+requests

src/ingest/base_ingestor.py

@@ -0,0 +1,52 @@
+import datetime
+import feedparser
+import hashlib
+import re
+from src.ingest.save_to_database import save_entry
+
+class BaseIngestor:
+ RSS_URL = None #will be set by the subclasses 
+
+ def slugify(self, text):
+ # Convert title to a filesystem-friendly slug
+ text = text.lower()
+ text = re.sub(r'[^a-z0-9]+', '-', text)
+ return text.strip('-')
+
+ def format_date(self, entry):
+ # Extract and format the published date
+ try:
+ dt = datetime.datetime(*entry.published_parsed[:6])
+ return dt.strftime("%Y-%m-%d")
+ except:
+ return "unknown-date"
+
+ def fetch_full_text(self, url):
+ raise NotImplementedError(
+ "Subclass must implement fetch_full_text()"
+ )
+
+ def generate_entry_hash(self, entry):
+ hash_input = f"{entry.title}{entry.link}{entry.get('published', '')}"
+ return hashlib.sha256(hash_input.encode('utf-8')).hexdigest()
+
+ def format_entry(self, entry):
+ title_slug = self.slugify(entry.title)
+ date_str = self.format_date(entry)
+ full_text = self.fetch_full_text(entry.link)
+
+ data = {
+ "title": entry.title,
+ "link": entry.link,
+ "published": entry.get("published", ""),
+ "summary": entry.get("summary", ""),
+ "full_text": full_text,
+ "id": self.generate_entry_hash(entry)
+ }
+ return data
+
+ def check_and_save_new_entries(self):
+ feed = feedparser.parse(self.RSS_URL)
+
+ for entry in feed.entries:
+ save_entry(self.format_entry(entry))

src/ingest/bbc_ingestor.py

@@ -0,0 +1,15 @@
+from bs4 import BeautifulSoup
+import requests
+from src.ingest.base_ingestor import BaseIngestor
+
+class BBCIngestor(BaseIngestor):
+ RSS_URL = "http://feeds.bbci.co.uk/news/world/rss.xml"
+
+ def fetch_full_text(self, article_url):
+ response = requests.get(self.RSS_URL)
+ soup = BeautifulSoup(response.content, 'html.parser')
+
+ article = soup.find('article')
+
+ if article:
+ return(article.get_text())

src/ingest/save_to_database.py

@@ -0,0 +1,18 @@
+from pymongo import MongoClient
+
+# Include username, password, and authentication database
+client = MongoClient("mongodb://myuser:mypassword@mongo:27017/justinsightdb?authSource=admin")
+
+# Get (or create) a database
+db = client["justinsightdb"]
+
+# Get (or create) a collection
+collection = db["articles"]
+
+def save_entry(entry):
+ #check if the entry has already been saved and if it has not then save it
+ entry_hash = entry["id"]
+ if collection.count_documents({"id": entry_hash}) == 0:
+ collection.insert_one(entry)
+ else:
+ print(f"{entry['title']} already in use!")

src/justinsight/celery.py

@@ -0,0 +1,32 @@
+# celery.py
+from celery import Celery
+
+app = Celery(
+ "justinsight",
+ broker="redis://redis:6379/0",
+ backend="redis://redis:6379/0",
+ include=["justinsight.tasks"], 
+)
+
+# Optional beat schedule
+app.conf.beat_schedule = {
+ # "sample-task-every-5-seconds": {
+ # "task": "justinsight.tasks.sample_task",
+ # "schedule": 5.0,
+ # "args": (),
+ # },
+
+ "check-BBCfeed-every-5-minutes": {
+ "task": "justinsight.tasks.bbcLogger_task",
+ "schedule": 5.0,
+ "args": (),
+ },
+
+ # "check-NYTfeed-every-5-minutes": {
+ # "task": "justinsight.tasks.nytLogger_task",
+ # "schedule": 300.0,
+ # "args": (),
+ # },
+
+ #schedule more tasks here
+}

src/justinsight/tasks.py

@@ -0,0 +1,23 @@
+# tasks.py
+from celery import shared_task
+from ingest.bbc_ingestor import BBCIngestor
+
+
+@shared_task
+def sample_task():
+ print("Sample task!")
+ return "done!"
+
+@shared_task
+def bbcLogger_task():
+ # Create an instance of the class
+ ingestor = BBCIngestor()
+ ingestor.check_and_save_new_entries() # this will invoke the inherited logic
+ return "BBC RSS Feed checked."
+
+@shared_task
+def nytLogger_task():
+ check_and_save_nyt()
+ return "NYT RSS Feed checked."
+
+#Add more tasks here in the format of the one above

.gitignore

@@ -1,4 +1,15 @@
 celerybeat-schedule.*
 src/celerybeat-schedule.*
 feed*.json
-.DS_Store
+.DS_Store
+
+# Ignore all __pycache__ directories at any level
+**/__pycache__/
+
+# Ignore all compiled Python files
+*.pyc
+*.pyo
+*.pyd
+
+#Ignore saved data
+data/*

docker-compose.yml

@@ -28,9 +28,36 @@ services:
  dockerfile: .docker/Dockerfile
  volumes:
  - .:/workspace:cached
+ - mongo_data:/data/db 
  depends_on:
  - redis
+ - mongo
  #entrypoint: ["/usr/local/bin/entrypoint.sh"]
  command: ["celery", "-A", justinsight.celery, "worker", "--loglevel=info"] #the lowercase j is actually so important
 
+ celery_beat:
+ build:
+ context: .
+ dockerfile: .docker/Dockerfile
+ volumes:
+ - .:/workspace:cached
+ depends_on:
+ - redis
+ #entrypoint: ["/usr/local/bin/entrypoint.sh"]
+ command: ["celery", "-A", justinsight.celery, "beat", "--loglevel=info"] #the lowercase j is actually so important
+
+ mongo:
+ image: mongo:latest
+ container_name: mongo
+ restart: unless-stopped
+ environment:
+ MONGO_INITDB_ROOT_USERNAME: myuser
+ MONGO_INITDB_ROOT_PASSWORD: mypassword
+ ports:
+ - "27017:27017"
+ volumes:
+ - mongo_data:/data/db 
+ - .:/workspace:cached
 
+volumes:
+ mongo_data:

readme.md

@@ -22,7 +22,7 @@ A lightweight, news intelligence service for ingesting free news sources, taggin
  cd justinsight
 
 
-## Setting up Celery Beat:
+## Setting up Celery Beat
 - On MacOS
  - run: docker compose up --build
  - OR for running in the background: docker compose up -d
@@ -34,5 +34,15 @@ A lightweight, news intelligence service for ingesting free news sources, taggin
  - use EC2 section of entrypoint.sh
  - to shut down celery: docker compose down
 
-*When running in background you can use "docker logs <container_name_or_id>" to see the log and "docker ps"
-to find the ID.
+Note: When running in background you can use "docker logs <container_name_or_id>" to see the log and "docker ps" to find the ID.
+
+## How to Create and Schedule a Task
+1) Create a method for your task wherever you are working on it that can be called and complete the task. It should be in a package (ie there should be an \_\_init\_\_.py file in the same folder as the file you are working on).
+2) Go to src/justinsight/tasks.py and create a new task following the format of the sample task. Remember to add an import statement if one is needed.
+3) Go to src/justinsight/celery.py and schedule a new task after the comment showing where to schedule them. The new task can follow the same format as those above
+
+ "sample-task-name": { 
+ "task": "justinsight.tasks.your_task_name", 
+ "schedule": x, #where x is the number of seconds between when the task should happen 
+ "args": (), #potential arguments for your task 
+ },

requirements.txt

@@ -1 +1,6 @@
-celery[redis]
+beautifulsoup4
+celery[redis]
+feedparser
+playwright
+pymongo
+requests

src/ingest/base_ingestor.py

@@ -0,0 +1,52 @@
+import datetime
+import feedparser
+import hashlib
+import re
+from src.ingest.save_to_database import save_entry
+
+class BaseIngestor:
+ RSS_URL = None #will be set by the subclasses 
+
+ def slugify(self, text):
+ # Convert title to a filesystem-friendly slug
+ text = text.lower()
+ text = re.sub(r'[^a-z0-9]+', '-', text)
+ return text.strip('-')
+
+ def format_date(self, entry):
+ # Extract and format the published date
+ try:
+ dt = datetime.datetime(*entry.published_parsed[:6])
+ return dt.strftime("%Y-%m-%d")
+ except:
+ return "unknown-date"
+
+ def fetch_full_text(self, url):
+ raise NotImplementedError(
+ "Subclass must implement fetch_full_text()"
+ )
+
+ def generate_entry_hash(self, entry):
+ hash_input = f"{entry.title}{entry.link}{entry.get('published', '')}"
+ return hashlib.sha256(hash_input.encode('utf-8')).hexdigest()
+
+ def format_entry(self, entry):
+ title_slug = self.slugify(entry.title)
+ date_str = self.format_date(entry)
+ full_text = self.fetch_full_text(entry.link)
+
+ data = {
+ "title": entry.title,
+ "link": entry.link,
+ "published": entry.get("published", ""),
+ "summary": entry.get("summary", ""),
+ "full_text": full_text,
+ "id": self.generate_entry_hash(entry)
+ }
+ return data
+
+ def check_and_save_new_entries(self):
+ feed = feedparser.parse(self.RSS_URL)
+
+ for entry in feed.entries:
+ save_entry(self.format_entry(entry))

src/ingest/bbc_ingestor.py

@@ -0,0 +1,15 @@
+from bs4 import BeautifulSoup
+import requests
+from src.ingest.base_ingestor import BaseIngestor
+
+class BBCIngestor(BaseIngestor):
+ RSS_URL = "http://feeds.bbci.co.uk/news/world/rss.xml"
+
+ def fetch_full_text(self, article_url):
+ response = requests.get(self.RSS_URL)
+ soup = BeautifulSoup(response.content, 'html.parser')
+
+ article = soup.find('article')
+
+ if article:
+ return(article.get_text())

src/ingest/save_to_database.py

@@ -0,0 +1,18 @@
+from pymongo import MongoClient
+
+# Include username, password, and authentication database
+client = MongoClient("mongodb://myuser:mypassword@mongo:27017/justinsightdb?authSource=admin")
+
+# Get (or create) a database
+db = client["justinsightdb"]
+
+# Get (or create) a collection
+collection = db["articles"]
+
+def save_entry(entry):
+ #check if the entry has already been saved and if it has not then save it
+ entry_hash = entry["id"]
+ if collection.count_documents({"id": entry_hash}) == 0:
+ collection.insert_one(entry)
+ else:
+ print(f"{entry['title']} already in use!")

src/justinsight/celery.py

@@ -0,0 +1,32 @@
+# celery.py
+from celery import Celery
+
+app = Celery(
+ "justinsight",
+ broker="redis://redis:6379/0",
+ backend="redis://redis:6379/0",
+ include=["justinsight.tasks"], 
+)
+
+# Optional beat schedule
+app.conf.beat_schedule = {
+ # "sample-task-every-5-seconds": {
+ # "task": "justinsight.tasks.sample_task",
+ # "schedule": 5.0,
+ # "args": (),
+ # },
+
+ "check-BBCfeed-every-5-minutes": {
+ "task": "justinsight.tasks.bbcLogger_task",
+ "schedule": 5.0,
+ "args": (),
+ },
+
+ # "check-NYTfeed-every-5-minutes": {
+ # "task": "justinsight.tasks.nytLogger_task",
+ # "schedule": 300.0,
+ # "args": (),
+ # },
+
+ #schedule more tasks here
+}

src/justinsight/tasks.py

@@ -0,0 +1,23 @@
+# tasks.py
+from celery import shared_task
+from ingest.bbc_ingestor import BBCIngestor
+
+
+@shared_task
+def sample_task():
+ print("Sample task!")
+ return "done!"
+
+@shared_task
+def bbcLogger_task():
+ # Create an instance of the class
+ ingestor = BBCIngestor()
+ ingestor.check_and_save_new_entries() # this will invoke the inherited logic
+ return "BBC RSS Feed checked."
+
+@shared_task
+def nytLogger_task():
+ check_and_save_nyt()
+ return "NYT RSS Feed checked."
+
+#Add more tasks here in the format of the one above

.gitignore

@@ -1,4 +1,15 @@
 celerybeat-schedule.*
 src/celerybeat-schedule.*
 feed*.json
-.DS_Store
+.DS_Store
+
+# Ignore all __pycache__ directories at any level
+**/__pycache__/
+
+# Ignore all compiled Python files
+*.pyc
+*.pyo
+*.pyd
+
+#Ignore saved data
+data/*

docker-compose.yml

@@ -28,9 +28,36 @@ services:
  dockerfile: .docker/Dockerfile
  volumes:
  - .:/workspace:cached
+ - mongo_data:/data/db 
  depends_on:
  - redis
+ - mongo
  #entrypoint: ["/usr/local/bin/entrypoint.sh"]
  command: ["celery", "-A", justinsight.celery, "worker", "--loglevel=info"] #the lowercase j is actually so important
 
+ celery_beat:
+ build:
+ context: .
+ dockerfile: .docker/Dockerfile
+ volumes:
+ - .:/workspace:cached
+ depends_on:
+ - redis
+ #entrypoint: ["/usr/local/bin/entrypoint.sh"]
+ command: ["celery", "-A", justinsight.celery, "beat", "--loglevel=info"] #the lowercase j is actually so important
+
+ mongo:
+ image: mongo:latest
+ container_name: mongo
+ restart: unless-stopped
+ environment:
+ MONGO_INITDB_ROOT_USERNAME: myuser
+ MONGO_INITDB_ROOT_PASSWORD: mypassword
+ ports:
+ - "27017:27017"
+ volumes:
+ - mongo_data:/data/db 
+ - .:/workspace:cached
 
+volumes:
+ mongo_data:

readme.md

@@ -22,7 +22,7 @@ A lightweight, news intelligence service for ingesting free news sources, taggin
  cd justinsight
 
 
-## Setting up Celery Beat:
+## Setting up Celery Beat
 - On MacOS
  - run: docker compose up --build
  - OR for running in the background: docker compose up -d
@@ -34,5 +34,15 @@ A lightweight, news intelligence service for ingesting free news sources, taggin
  - use EC2 section of entrypoint.sh
  - to shut down celery: docker compose down
 
-*When running in background you can use "docker logs <container_name_or_id>" to see the log and "docker ps"
-to find the ID.
+Note: When running in background you can use "docker logs <container_name_or_id>" to see the log and "docker ps" to find the ID.
+
+## How to Create and Schedule a Task
+1) Create a method for your task wherever you are working on it that can be called and complete the task. It should be in a package (ie there should be an \_\_init\_\_.py file in the same folder as the file you are working on).
+2) Go to src/justinsight/tasks.py and create a new task following the format of the sample task. Remember to add an import statement if one is needed.
+3) Go to src/justinsight/celery.py and schedule a new task after the comment showing where to schedule them. The new task can follow the same format as those above
+
+ "sample-task-name": { 
+ "task": "justinsight.tasks.your_task_name", 
+ "schedule": x, #where x is the number of seconds between when the task should happen 
+ "args": (), #potential arguments for your task 
+ },

requirements.txt

@@ -1 +1,6 @@
-celery[redis]
+beautifulsoup4
+celery[redis]
+feedparser
+playwright
+pymongo
+requests

src/ingest/base_ingestor.py

@@ -0,0 +1,52 @@
+import datetime
+import feedparser
+import hashlib
+import re
+from src.ingest.save_to_database import save_entry
+
+class BaseIngestor:
+ RSS_URL = None #will be set by the subclasses 
+
+ def slugify(self, text):
+ # Convert title to a filesystem-friendly slug
+ text = text.lower()
+ text = re.sub(r'[^a-z0-9]+', '-', text)
+ return text.strip('-')
+
+ def format_date(self, entry):
+ # Extract and format the published date
+ try:
+ dt = datetime.datetime(*entry.published_parsed[:6])
+ return dt.strftime("%Y-%m-%d")
+ except:
+ return "unknown-date"
+
+ def fetch_full_text(self, url):
+ raise NotImplementedError(
+ "Subclass must implement fetch_full_text()"
+ )
+
+ def generate_entry_hash(self, entry):
+ hash_input = f"{entry.title}{entry.link}{entry.get('published', '')}"
+ return hashlib.sha256(hash_input.encode('utf-8')).hexdigest()
+
+ def format_entry(self, entry):
+ title_slug = self.slugify(entry.title)
+ date_str = self.format_date(entry)
+ full_text = self.fetch_full_text(entry.link)
+
+ data = {
+ "title": entry.title,
+ "link": entry.link,
+ "published": entry.get("published", ""),
+ "summary": entry.get("summary", ""),
+ "full_text": full_text,
+ "id": self.generate_entry_hash(entry)
+ }
+ return data
+
+ def check_and_save_new_entries(self):
+ feed = feedparser.parse(self.RSS_URL)
+
+ for entry in feed.entries:
+ save_entry(self.format_entry(entry))

src/ingest/bbc_ingestor.py

@@ -0,0 +1,15 @@
+from bs4 import BeautifulSoup
+import requests
+from src.ingest.base_ingestor import BaseIngestor
+
+class BBCIngestor(BaseIngestor):
+ RSS_URL = "http://feeds.bbci.co.uk/news/world/rss.xml"
+
+ def fetch_full_text(self, article_url):
+ response = requests.get(self.RSS_URL)
+ soup = BeautifulSoup(response.content, 'html.parser')
+
+ article = soup.find('article')
+
+ if article:
+ return(article.get_text())

src/ingest/save_to_database.py

@@ -0,0 +1,18 @@
+from pymongo import MongoClient
+
+# Include username, password, and authentication database
+client = MongoClient("mongodb://myuser:mypassword@mongo:27017/justinsightdb?authSource=admin")
+
+# Get (or create) a database
+db = client["justinsightdb"]
+
+# Get (or create) a collection
+collection = db["articles"]
+
+def save_entry(entry):
+ #check if the entry has already been saved and if it has not then save it
+ entry_hash = entry["id"]
+ if collection.count_documents({"id": entry_hash}) == 0:
+ collection.insert_one(entry)
+ else:
+ print(f"{entry['title']} already in use!")

src/justinsight/celery.py

@@ -0,0 +1,32 @@
+# celery.py
+from celery import Celery
+
+app = Celery(
+ "justinsight",
+ broker="redis://redis:6379/0",
+ backend="redis://redis:6379/0",
+ include=["justinsight.tasks"], 
+)
+
+# Optional beat schedule
+app.conf.beat_schedule = {
+ # "sample-task-every-5-seconds": {
+ # "task": "justinsight.tasks.sample_task",
+ # "schedule": 5.0,
+ # "args": (),
+ # },
+
+ "check-BBCfeed-every-5-minutes": {
+ "task": "justinsight.tasks.bbcLogger_task",
+ "schedule": 5.0,
+ "args": (),
+ },
+
+ # "check-NYTfeed-every-5-minutes": {
+ # "task": "justinsight.tasks.nytLogger_task",
+ # "schedule": 300.0,
+ # "args": (),
+ # },
+
+ #schedule more tasks here
+}

src/justinsight/tasks.py

@@ -0,0 +1,23 @@
+# tasks.py
+from celery import shared_task
+from ingest.bbc_ingestor import BBCIngestor
+
+
+@shared_task
+def sample_task():
+ print("Sample task!")
+ return "done!"
+
+@shared_task
+def bbcLogger_task():
+ # Create an instance of the class
+ ingestor = BBCIngestor()
+ ingestor.check_and_save_new_entries() # this will invoke the inherited logic
+ return "BBC RSS Feed checked."
+
+@shared_task
+def nytLogger_task():
+ check_and_save_nyt()
+ return "NYT RSS Feed checked."
+
+#Add more tasks here in the format of the one above