ft APIft.com ↗

"""
Financial Times Commodities News API Client
Get your API key from: https://parse.bot/settings
"""

import os
import requests
from datetime import datetime, timedelta


class ParseClient:
    """Client for the Financial Times Commodities News API via Parse Bot."""

    def __init__(self, api_key: str = None):
        """Initialize the Parse API client.
        
        Args:
            api_key: API key for Parse Bot. If not provided, reads from PARSE_API_KEY env var.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "0e37e2f1-5dc3-4915-956f-b8f461f1ae07"
        self.api_key = api_key or os.getenv("PARSE_API_KEY")
        
        if not self.api_key:
            raise ValueError("API key is required. Set PARSE_API_KEY environment variable or pass api_key parameter.")

    def _call(self, endpoint: str, method: str = "POST", **params) -> dict:
        """Make an API call to the Parse Bot scraper.
        
        Args:
            endpoint: The endpoint name (e.g., 'search_news', 'get_topic_articles')
            method: HTTP method - 'GET' or 'POST'
            **params: Query or body parameters depending on method
            
        Returns:
            dict: JSON response from the API
        """
        url = f"{self.base_url}/scraper/{self.scraper_id}/{endpoint}"
        headers = {
            "X-API-Key": self.api_key,
            "Content-Type": "application/json"
        }
        
        if method.upper() == "GET":
            response = requests.get(url, headers=headers, params=params)
        elif method.upper() == "POST":
            response = requests.post(url, headers=headers, json=params)
        else:
            raise ValueError(f"Unsupported HTTP method: {method}")
        
        response.raise_for_status()
        return response.json()

    def search_news(
        self,
        query: str = "commodities",
        sort: str = "date",
        page: int = 1,
        date_from: str = None,
        date_to: str = None
    ) -> dict:
        """Search Financial Times articles by keyword.
        
        Args:
            query: Search query string (e.g., 'commodities', 'oil prices')
            sort: Sort order - 'date' or 'relevance'
            page: Page number for pagination, starting from 1
            date_from: Start date in YYYY-MM-DD format
            date_to: End date in YYYY-MM-DD format
            
        Returns:
            dict: Search results with articles, total count, and pagination info
        """
        params = {
            "query": query,
            "sort": sort,
            "page": str(page)
        }
        
        if date_from:
            params["date_from"] = date_from
        if date_to:
            params["date_to"] = date_to
        
        return self._call("search_news", method="GET", **params)

    def get_topic_articles(self, topic: str = "commodities") -> dict:
        """Get the latest curated articles from an FT topic page.
        
        Args:
            topic: FT topic page slug (e.g., 'commodities', 'oil', 'gold', 'copper')
            
        Returns:
            dict: Topic articles with article count and details
        """
        return self._call("get_topic_articles", method="GET", topic=topic)


def main():
    """Practical workflow example: monitor commodity markets and track trends."""
    
    # Initialize the API client
    client = ParseClient()
    
    print("=" * 80)
    print("Financial Times Commodities Market Intelligence")
    print("=" * 80)
    print()
    
    # Workflow: Monitor multiple commodity topics and analyze recent coverage
    commodity_topics = ["commodities", "oil", "gold"]
    all_articles = []
    
    print("Step 1: Fetching latest curated articles from key commodity topics...")
    print("-" * 80)
    
    for topic in commodity_topics:
        print(f"\n📊 Checking {topic.upper()} topic page...")
        
        topic_data = client.get_topic_articles(topic=topic)
        
        print(f"   Found {topic_data['article_count']} articles")
        
        # Display first 3 articles from each topic
        for i, article in enumerate(topic_data['articles'][:3], 1):
            print(f"   {i}. {article['title']}")
            print(f"      Topic: {article['topic']}")
            if article['summary']:
                print(f"      Summary: {article['summary'][:60]}...")
            all_articles.append(article)
    
    print("\n" + "=" * 80)
    print("Step 2: Searching for recent oil price articles (last 30 days)...")
    print("-" * 80)
    
    # Calculate date range for last 30 days
    end_date = datetime.now()
    start_date = end_date - timedelta(days=30)
    
    date_from = start_date.strftime("%Y-%m-%d")
    date_to = end_date.strftime("%Y-%m-%d")
    
    search_results = client.search_news(
        query="oil prices",
        sort="date",
        page=1,
        date_from=date_from,
        date_to=date_to
    )
    
    print(f"\nSearch Query: 'oil prices'")
    print(f"Date Range: {date_from} to {date_to}")
    print(f"Total Results Found: {search_results['total_results']}")
    print(f"Results on Current Page: {len(search_results['articles'])}")
    print()
    
    # Display search results
    for i, article in enumerate(search_results['articles'][:5], 1):
        pub_date = article['published_date']
        if pub_date:
            pub_date = datetime.fromisoformat(pub_date.replace('Z', '+00:00')).strftime("%Y-%m-%d %H:%M")
        
        print(f"{i}. {article['title']}")
        print(f"   Published: {pub_date}")
        print(f"   Topic: {article['topic']}")
        print(f"   URL: {article['url']}")
        print()
    
    print("=" * 80)
    print("Step 3: Searching for metals market trends...")
    print("-" * 80)
    
    # Search for metals-related articles sorted by relevance
    metals_search = client.search_news(
        query="copper gold metals market",
        sort="relevance",
        page=1
    )
    
    print(f"\nSearch Query: 'copper gold metals market'")
    print(f"Sort Order: relevance")
    print(f"Total Results Found: {metals_search['total_results']}")
    print()
    
    # Analyze the results
    topic_distribution = {}
    for article in metals_search['articles'][:10]:
        topic = article['topic']
        topic_distribution[topic] = topic_distribution.get(topic, 0) + 1
    
    print("Topic Distribution in Results:")
    for topic, count in sorted(topic_distribution.items(), key=lambda x: x[1], reverse=True):
        print(f"  • {topic}: {count} articles")
    
    print("\n" + "=" * 80)
    print("Summary:")
    print(f"  • Total articles reviewed: {len(all_articles) + len(search_results['articles']) + len(metals_search['articles'])}")
    print(f"  • Topics monitored: {', '.join(commodity_topics)}")
    print(f"  • Search queries executed: 2 (oil prices, metals)")
    print("=" * 80)


if __name__ == "__main__":
    main()