How to check sources coverage

import aiohttp
import asyncio
import csv
import logging
from typing import List, Dict, Optional
from tqdm.asyncio import tqdm
import os

# Constants
API_KEY: str = os.getenv("NEWSCATCHER_API_KEY")
if not API_KEY:
    raise EnvironmentError("API key not set in environment variables")
URL: str = "https://v3-api.newscatcherapi.com/api/sources"
HEADERS: Dict[str, str] = {"x-api-token": API_KEY}
INPUT_FILE: str = "source_urls.csv"
OUTPUT_FILE: str = "uncovered_sources.csv"
MAX_CONCURRENT_REQUESTS: int = 5  # Set the desired number of concurrent requests
MAX_RETRIES: int = 3  # Number of retries for failed requests
TIMEOUT: int = 30  # Timeout for each request in seconds

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
 logging.FileHandler("check_sources.log"),  # Log to a file
 logging.StreamHandler(),  # Also log to console
 ],
)


async def fetch_sources(
    session: aiohttp.ClientSession,
    batch: List[str],
    headers: Dict[str, str],
    retries: int = MAX_RETRIES,
) -> Optional[Dict]:
    """Fetch source coverage in a single batch with retry logic."""
 payload: Dict[str, object] = {"source_url": batch, "include_additional_info": True}
    for attempt in range(retries):
        try:
            async with session.post(
 URL, headers=headers, json=payload, timeout=TIMEOUT
 ) as response:
 response.raise_for_status()
 data: Dict = await response.json()
                return data
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
 logging.error(f"Attempt {attempt + 1}/{retries} failed: {e}")
            await asyncio.sleep(2**attempt)  # Exponential backoff
        except Exception as e:
 logging.error(f"Unexpected error: {e}")
            break
    return None


async def fetch_sources_with_semaphore(
    semaphore: asyncio.Semaphore,
    session: aiohttp.ClientSession,
    batch: List[str],
    headers: Dict[str, str],
) -> Optional[Dict]:
    """Fetch sources with a semaphore to limit the number of concurrent requests."""
    async with semaphore:
        return await fetch_sources(session, batch, headers)


async def check_sources_concurrently(
    source_urls: List[str], batch_size: int = 1000
) -> Dict[str, bool]:
    """Check if sources are covered by the API concurrently."""
 coverage: Dict[str, bool] = {}
 semaphore: asyncio.Semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)

    async with aiohttp.ClientSession(
        connector=aiohttp.TCPConnector(limit=MAX_CONCURRENT_REQUESTS)
 ) as session:
 tasks: List[asyncio.Task] = []
 num_batches: int = (
            len(source_urls) + batch_size - 1
 ) // batch_size  # Calculate the number of batches

        # Use tqdm.asyncio for async progress bar
        for i in range(0, len(source_urls), batch_size):
 batch: List[str] = source_urls[i : i + batch_size]
 task: asyncio.Task = fetch_sources_with_semaphore(
 semaphore, session, batch, HEADERS
 )
 tasks.append(task)

        # Use tqdm to visualize the progress of batch processing
 results: List[Optional[Dict]] = await tqdm.gather(
 *tasks, desc="Checking source coverage", unit="batch"
 )

        # Process results
        for i, result in enumerate(results):
            if isinstance(result, Exception):
 logging.error(f"Error processing batch {i+1}: {result}")
                continue
            if result:
                # Mark sources as covered
                for source in result.get("sources", []):
 coverage[source["domain_url"]] = True
                # Mark sources not in the response as uncovered
 batch = source_urls[i * batch_size : (i + 1) * batch_size]
                for url in batch:
                    if url not in coverage:
 coverage[url] = False

 logging.info("Finished checking sources.")
    return coverage


def read_sources_from_csv(file_path: str) -> List[str]:
    """Read source URLs from a CSV file."""
    with open(file_path, "r") as file:
 reader = csv.reader(file)
 sources: List[str] = [
 row[0] for row in reader if row
 ]  # Assuming URLs are in the first column
 logging.info(f"Loaded {len(sources)} sources from {file_path}.")
    return sources


def write_uncovered_sources(uncovered_sources: List[str], file_path: str) -> None:
    """Write uncovered sources to a CSV file."""
    with open(file_path, "w", newline="") as file:
 writer = csv.writer(file)
 writer.writerow(["Uncovered Source URL"])  # Header
        for source in uncovered_sources:
 writer.writerow([source])
 logging.info(f"Uncovered sources written to {file_path}.")


def main() -> None:
    # Read sources from CSV
 source_urls: List[str] = read_sources_from_csv(INPUT_FILE)

    # Check sources coverage concurrently
 coverage: Dict[str, bool] = asyncio.run(check_sources_concurrently(source_urls))

    # Identify uncovered sources
 uncovered_sources: List[str] = [
 url for url, is_covered in coverage.items() if not is_covered
 ]

    # Write uncovered sources to CSV
 write_uncovered_sources(uncovered_sources, OUTPUT_FILE)

 logging.info(f"Total sources: {len(source_urls)}")
 logging.info(f"Covered sources: {len(source_urls) - len(uncovered_sources)}")
 logging.info(f"Uncovered sources: {len(uncovered_sources)}")


if __name__ == "__main__":
 main()