Machine Learning

In order to determine if a news article is negative or not, we use a sentiment analysis model on a the news title. In particular, I use the Flair Neural Network. Since this model is trained on comments and not news articles, the sentiment score is not always accurate. However, it is a good starting point for determining sentiment and I am building a dataset of news articles to train a custom model in the future. Additionally, I want to expand this to the content of the article as well.

Code Highlights

Backend Code Snippets

---

Django Models:

class Source(models.Model):
    # NOTE: it is not called 'id' since that is not allowed unless it is a primary key.
    #       Since this is not a primary key, we need to rename it to something else to 
    #       avoid migration errors.
    source_id = models.CharField(max_length=100, unique=True, null=True)
    name      = models.CharField(max_length=100)

    def __str__(self):
        return self.name
    
    def as_dict(self):
        return {
            'source_id': self.source_id,
            'name': self.name
        }

class Article(models.Model):
    source       = models.ForeignKey(Source, on_delete=models.CASCADE) # many articles to one Source, but not one Article to many sources
    title        = models.CharField(max_length=300)
    url          = models.URLField(max_length=600)
    published_at = models.DateTimeField()
    sentiment_score = models.FloatField() # -1.0 to 1.0
    author       = models.CharField(max_length=600, null=True, blank=True)
    description  = models.TextField(null=True, blank=True)
    url_to_image = models.URLField(max_length=600, null=True, blank=True)
    content      = models.TextField(null=True, blank=True)
    # categories   = models.ManyToManyField("Category", related_name="articles")

    def __str__(self):
        return self.title
    
    def as_dict(self):
        return {
            'source': self.source.as_dict(),  # Or any other representation of Source you want
            'title': self.title,
            'url': self.url,
            'published_at': self.published_at.isoformat(),  # Convert datetime to string
            'sentiment_score': self.sentiment_score,
            'author': self.author,
            'description': self.description,
            'url_to_image': self.url_to_image,
            'content': self.content
        }

Django View Example:

def fetch_latest_top_news(request) -> JsonResponse:
    """Fetch the latest top news from Google News API and store it in our posgresql database. If it already exists,
    simply get the data from the database. 

    Args:
        request (_type_): request object from Django. This will hold the query parameters for the request.
        country (str, optional): Country code to fetch the news from. Defaults to 'us'.
        category (str, optional): One of the 7 categories from google news api. Defaults to "general".
        keyword (str, optional): Optional keywords to search for in news. Defaults to "".
        page_size (int, optional): number of articles to return. Defaults to 20.
        postive_news_only (bool, optional): whether to filter out all negative news. Defaults to False.

    Returns:
        JsonResponse: Returns the response in JSON format. If success, it will be the articles. Otherwise, return the error.
    """
    country = request.GET.get('country', 'us')
    category = request.GET.get('category', '')
    keyword = request.GET.get('keyword')
    page_size = int(request.GET.get('page_size', 20))
    postive_news_only = request.GET.get('positive_news_only', 'false').lower() == 'true'
    
    if country not in NEWS_API_VALID_COUNTRY_CODES:
        return JsonResponse({'status': 'failure',
                             'message': f'"{country}" is not a valid country code.'})

    if category not in NEWS_API_VALID_CATEGORIES:
        return JsonResponse({'status': 'failure',
                             'message': f'"{category}" is not a valid category.'})

    # Setup Params for NewsAPI fetch request
    params = {
        'country': country,
        'pageSize': int(page_size),
        'apiKey': settings.NEWS_API_KEY,
    }
    if category != "":
        params['category'] = category
    if keyword != "":
        params['q'] = keyword

    response = requests.get(NEWS_API_TOP_HEADLINES_URL, params=params)
    data = response.json()

    if data['status'] == 'error':
        # ERROR
        return JsonResponse({'status': 'failure',
                             'code': data['code'],
                             'message': data['message']})

    return_articles = []

    for article_data in data['articles']:
        source_data = article_data['source']
        source, _ = Source.objects.get_or_create(
            source_id = source_data['id'],
            name      = source_data['name']
        )

        # TODO: add category data. Should allow for multiple and will update with any new categories of already stored entries
        article, _ = Article.objects.get_or_create(
            source       = source,
            url          = article_data['url'],
            title        = article_data['title'],
            published_at = article_data['publishedAt'],
            sentiment_score = sentiment_classifer.get_sentiment_score_from_string(article_data['title']),
            defaults = {
                'author': article_data.get('author', ''),
                'description': article_data.get('description', ''),
                'url_to_image': article_data.get('urlToImage', ''),
                'content': article_data.get('content', ''),
            }
        )
        print(article_data['publishedAt'])

        # TODO: This method, while it works, will not guarentee we have `page_size` number of articles.
        if postive_news_only:
            if article.sentiment_score >= 0.0:
                return_articles.append(article)
        else:
            return_articles.append(article)

    # sort based on published time
    return_articles = sorted(return_articles, key=lambda article: article.published_at, reverse=True)[:page_size]
    
        
    return JsonResponse({'status': 'success',
                         'articles': [article.as_dict() for article in return_articles]})

Frontend Code Snippets

---

Coming Soon...