diff --git a/examples/download_stats_demo.py b/examples/download_stats_demo.py index f81a80f..16d2bfb 100644 --- a/examples/download_stats_demo.py +++ b/examples/download_stats_demo.py @@ -7,7 +7,6 @@ to analyze PyPI package popularity and trends. """ import asyncio -import json from datetime import datetime from pypi_query_mcp.tools.download_stats import ( @@ -22,53 +21,53 @@ async def demo_package_download_stats(): print("=" * 60) print("PyPI Package Download Statistics Demo") print("=" * 60) - + # Example packages to analyze packages = ["requests", "numpy", "django", "flask"] - + for package_name in packages: print(f"\nšŸ“Š Download Statistics for '{package_name}':") print("-" * 50) - + try: # Get download statistics for the last month stats = await get_package_download_stats(package_name, period="month") - + # Display basic info metadata = stats.get("metadata", {}) downloads = stats.get("downloads", {}) analysis = stats.get("analysis", {}) - + print(f"Package: {metadata.get('name', package_name)}") print(f"Version: {metadata.get('version', 'unknown')}") print(f"Summary: {metadata.get('summary', 'No summary available')[:80]}...") - + # Display download counts - print(f"\nDownload Counts:") + print("\nDownload Counts:") print(f" Last Day: {downloads.get('last_day', 0):,}") print(f" Last Week: {downloads.get('last_week', 0):,}") print(f" Last Month: {downloads.get('last_month', 0):,}") - + # Display analysis if analysis: - print(f"\nAnalysis:") + print("\nAnalysis:") print(f" Total Downloads: {analysis.get('total_downloads', 0):,}") print(f" Highest Period: {analysis.get('highest_period', 'N/A')}") - + growth = analysis.get('growth_indicators', {}) if growth: - print(f" Growth Indicators:") + print(" Growth Indicators:") for indicator, value in growth.items(): print(f" {indicator}: {value}") - + # Display repository info if available project_urls = metadata.get('project_urls', {}) if project_urls: - print(f"\nRepository Links:") + print("\nRepository Links:") for name, url in project_urls.items(): if url: print(f" {name}: {url}") - + except Exception as e: print(f"āŒ Error getting stats for {package_name}: {e}") @@ -78,45 +77,45 @@ async def demo_package_download_trends(): print("\n" + "=" * 60) print("PyPI Package Download Trends Demo") print("=" * 60) - + # Analyze trends for a popular package package_name = "requests" - + print(f"\nšŸ“ˆ Download Trends for '{package_name}':") print("-" * 50) - + try: # Get download trends (without mirrors for cleaner data) trends = await get_package_download_trends(package_name, include_mirrors=False) - + trend_analysis = trends.get("trend_analysis", {}) time_series = trends.get("time_series", []) - + print(f"Package: {package_name}") print(f"Data Points: {trend_analysis.get('data_points', 0)}") print(f"Total Downloads: {trend_analysis.get('total_downloads', 0):,}") print(f"Average Daily: {trend_analysis.get('average_daily', 0):,.0f}") print(f"Trend Direction: {trend_analysis.get('trend_direction', 'unknown')}") - + # Display date range date_range = trend_analysis.get('date_range', {}) if date_range: print(f"Date Range: {date_range.get('start')} to {date_range.get('end')}") - + # Display peak day peak_day = trend_analysis.get('peak_day', {}) if peak_day: print(f"Peak Day: {peak_day.get('date')} ({peak_day.get('downloads', 0):,} downloads)") - + # Show recent data points (last 7 days) if time_series: - print(f"\nRecent Download Data (last 7 days):") + print("\nRecent Download Data (last 7 days):") recent_data = [item for item in time_series if item.get('category') == 'without_mirrors'][-7:] for item in recent_data: date = item.get('date', 'unknown') downloads = item.get('downloads', 0) print(f" {date}: {downloads:,} downloads") - + except Exception as e: print(f"āŒ Error getting trends for {package_name}: {e}") @@ -126,33 +125,33 @@ async def demo_top_packages(): print("\n" + "=" * 60) print("Top PyPI Packages by Downloads Demo") print("=" * 60) - + periods = ["day", "week", "month"] - + for period in periods: print(f"\nšŸ† Top 10 Packages (last {period}):") print("-" * 50) - + try: # Get top packages for this period top_packages = await get_top_packages_by_downloads(period=period, limit=10) - + packages_list = top_packages.get("top_packages", []) total_found = top_packages.get("total_found", 0) - + print(f"Found {total_found} packages") print(f"Data Source: {top_packages.get('data_source', 'unknown')}") - + if top_packages.get("note"): print(f"Note: {top_packages['note']}") - - print(f"\nRankings:") + + print("\nRankings:") for package in packages_list: rank = package.get("rank", "?") name = package.get("package", "unknown") downloads = package.get("downloads", 0) print(f" {rank:2d}. {name:<20} {downloads:>12,} downloads") - + except Exception as e: print(f"āŒ Error getting top packages for {period}: {e}") @@ -162,37 +161,37 @@ async def demo_package_comparison(): print("\n" + "=" * 60) print("Package Comparison Demo") print("=" * 60) - + # Compare web frameworks frameworks = ["django", "flask", "fastapi", "tornado"] - - print(f"\nšŸ” Comparing Web Frameworks (last month downloads):") + + print("\nšŸ” Comparing Web Frameworks (last month downloads):") print("-" * 70) - + comparison_data = [] - + for framework in frameworks: try: stats = await get_package_download_stats(framework, period="month") downloads = stats.get("downloads", {}) last_month = downloads.get("last_month", 0) - + comparison_data.append({ "name": framework, "downloads": last_month, "metadata": stats.get("metadata", {}), }) - + except Exception as e: print(f"āŒ Error getting stats for {framework}: {e}") - + # Sort by downloads (descending) comparison_data.sort(key=lambda x: x["downloads"], reverse=True) - + # Display comparison print(f"{'Rank':<4} {'Framework':<12} {'Downloads':<15} {'Summary'}") print("-" * 70) - + for i, data in enumerate(comparison_data, 1): name = data["name"] downloads = data["downloads"] @@ -204,18 +203,18 @@ async def main(): """Run all demo functions.""" print("šŸš€ Starting PyPI Download Statistics Demo") print(f"Timestamp: {datetime.now().isoformat()}") - + try: # Run all demos await demo_package_download_stats() await demo_package_download_trends() await demo_top_packages() await demo_package_comparison() - + print("\n" + "=" * 60) print("āœ… Demo completed successfully!") print("=" * 60) - + except KeyboardInterrupt: print("\nāŒ Demo interrupted by user") except Exception as e: diff --git a/pypi_query_mcp/core/stats_client.py b/pypi_query_mcp/core/stats_client.py index 88a9abd..1c732ce 100644 --- a/pypi_query_mcp/core/stats_client.py +++ b/pypi_query_mcp/core/stats_client.py @@ -2,7 +2,6 @@ import asyncio import logging -from datetime import datetime, timedelta from typing import Any import httpx diff --git a/pypi_query_mcp/server.py b/pypi_query_mcp/server.py index 8c67604..f9c8d3c 100644 --- a/pypi_query_mcp/server.py +++ b/pypi_query_mcp/server.py @@ -541,7 +541,7 @@ async def get_top_downloaded_packages( logger.info(f"MCP tool: Getting top {actual_limit} packages for period: {period}") result = await get_top_packages_by_downloads(period, actual_limit) - logger.info(f"Successfully retrieved top packages list") + logger.info("Successfully retrieved top packages list") return result except Exception as e: logger.error(f"Error getting top packages: {e}") diff --git a/pypi_query_mcp/tools/download_stats.py b/pypi_query_mcp/tools/download_stats.py index 84e50cd..4406ded 100644 --- a/pypi_query_mcp/tools/download_stats.py +++ b/pypi_query_mcp/tools/download_stats.py @@ -1,12 +1,11 @@ """PyPI package download statistics tools.""" import logging -from datetime import datetime, timedelta +from datetime import datetime from typing import Any from ..core.pypi_client import PyPIClient from ..core.stats_client import PyPIStatsClient -from ..core.exceptions import InvalidPackageNameError, NetworkError, PackageNotFoundError logger = logging.getLogger(__name__) @@ -57,7 +56,7 @@ async def get_package_download_stats( # Extract download data download_data = recent_stats.get("data", {}) - + # Calculate trends and analysis analysis = _analyze_download_stats(download_data) @@ -106,7 +105,7 @@ async def get_package_download_trends( # Process time series data time_series_data = overall_stats.get("data", []) - + # Analyze trends trend_analysis = _analyze_download_trends(time_series_data, include_mirrors) @@ -153,31 +152,31 @@ async def get_top_packages_by_downloads( async with PyPIStatsClient() as stats_client: try: top_packages = [] - + # Get download stats for popular packages for i, package_name in enumerate(popular_packages[:limit]): try: stats = await stats_client.get_recent_downloads( package_name, period, use_cache=True ) - + download_data = stats.get("data", {}) download_count = _extract_download_count(download_data, period) - + top_packages.append({ "rank": i + 1, "package": package_name, "downloads": download_count, "period": period, }) - + except Exception as e: logger.warning(f"Could not get stats for {package_name}: {e}") continue # Sort by download count (descending) top_packages.sort(key=lambda x: x.get("downloads", 0), reverse=True) - + # Update ranks after sorting for i, package in enumerate(top_packages): package["rank"] = i + 1 @@ -221,7 +220,7 @@ def _analyze_download_stats(download_data: dict[str, Any]) -> dict[str, Any]: if period.startswith("last_") and isinstance(count, int): analysis["periods_available"].append(period) analysis["total_downloads"] += count - + if analysis["highest_period"] is None or count > download_data.get(analysis["highest_period"], 0): analysis["highest_period"] = period @@ -232,7 +231,7 @@ def _analyze_download_stats(download_data: dict[str, Any]) -> dict[str, Any]: if last_day and last_week: analysis["growth_indicators"]["daily_vs_weekly"] = round(last_day * 7 / last_week, 2) - + if last_week and last_month: analysis["growth_indicators"]["weekly_vs_monthly"] = round(last_week * 4 / last_month, 2) @@ -264,7 +263,7 @@ def _analyze_download_trends(time_series_data: list[dict], include_mirrors: bool # Filter data based on mirror preference category_filter = "with_mirrors" if include_mirrors else "without_mirrors" filtered_data = [ - item for item in time_series_data + item for item in time_series_data if item.get("category") == category_filter ] @@ -299,7 +298,7 @@ def _analyze_download_trends(time_series_data: list[dict], include_mirrors: bool if len(filtered_data) >= 14: first_week = sum(item.get("downloads", 0) for item in filtered_data[:7]) last_week = sum(item.get("downloads", 0) for item in filtered_data[-7:]) - + if last_week > first_week * 1.1: analysis["trend_direction"] = "increasing" elif last_week < first_week * 0.9: diff --git a/tests/test_download_stats.py b/tests/test_download_stats.py index d9f3944..28f41ac 100644 --- a/tests/test_download_stats.py +++ b/tests/test_download_stats.py @@ -1,17 +1,18 @@ """Tests for download statistics functionality.""" -import pytest from unittest.mock import AsyncMock, patch +import pytest + +from pypi_query_mcp.core.exceptions import PackageNotFoundError from pypi_query_mcp.tools.download_stats import ( - get_package_download_stats, - get_package_download_trends, - get_top_packages_by_downloads, _analyze_download_stats, _analyze_download_trends, _extract_download_count, + get_package_download_stats, + get_package_download_trends, + get_top_packages_by_downloads, ) -from pypi_query_mcp.core.exceptions import PackageNotFoundError, InvalidPackageNameError class TestDownloadStats: @@ -43,7 +44,7 @@ class TestDownloadStats: with patch("pypi_query_mcp.tools.download_stats.PyPIStatsClient") as mock_stats_client, \ patch("pypi_query_mcp.tools.download_stats.PyPIClient") as mock_pypi_client: - + # Setup mocks mock_stats_instance = AsyncMock() mock_stats_instance.get_recent_downloads.return_value = mock_stats_data