style: fix code formatting and linting issues

- Remove unused imports in stats_client.py and download_stats.py
- Fix import sorting in test files
- Remove unnecessary f-strings in server.py and demo script
- Clean up whitespace and formatting issues
- Ensure all files pass ruff and isort checks

Signed-off-by: longhao <hal.long@outlook.com>
This commit is contained in:
longhao 2025-05-27 21:13:09 +08:00 committed by Hal
parent 99c603ed37
commit 3d9d7b4208
5 changed files with 66 additions and 68 deletions

View File

@ -7,7 +7,6 @@ to analyze PyPI package popularity and trends.
""" """
import asyncio import asyncio
import json
from datetime import datetime from datetime import datetime
from pypi_query_mcp.tools.download_stats import ( from pypi_query_mcp.tools.download_stats import (
@ -22,53 +21,53 @@ async def demo_package_download_stats():
print("=" * 60) print("=" * 60)
print("PyPI Package Download Statistics Demo") print("PyPI Package Download Statistics Demo")
print("=" * 60) print("=" * 60)
# Example packages to analyze # Example packages to analyze
packages = ["requests", "numpy", "django", "flask"] packages = ["requests", "numpy", "django", "flask"]
for package_name in packages: for package_name in packages:
print(f"\n📊 Download Statistics for '{package_name}':") print(f"\n📊 Download Statistics for '{package_name}':")
print("-" * 50) print("-" * 50)
try: try:
# Get download statistics for the last month # Get download statistics for the last month
stats = await get_package_download_stats(package_name, period="month") stats = await get_package_download_stats(package_name, period="month")
# Display basic info # Display basic info
metadata = stats.get("metadata", {}) metadata = stats.get("metadata", {})
downloads = stats.get("downloads", {}) downloads = stats.get("downloads", {})
analysis = stats.get("analysis", {}) analysis = stats.get("analysis", {})
print(f"Package: {metadata.get('name', package_name)}") print(f"Package: {metadata.get('name', package_name)}")
print(f"Version: {metadata.get('version', 'unknown')}") print(f"Version: {metadata.get('version', 'unknown')}")
print(f"Summary: {metadata.get('summary', 'No summary available')[:80]}...") print(f"Summary: {metadata.get('summary', 'No summary available')[:80]}...")
# Display download counts # Display download counts
print(f"\nDownload Counts:") print("\nDownload Counts:")
print(f" Last Day: {downloads.get('last_day', 0):,}") print(f" Last Day: {downloads.get('last_day', 0):,}")
print(f" Last Week: {downloads.get('last_week', 0):,}") print(f" Last Week: {downloads.get('last_week', 0):,}")
print(f" Last Month: {downloads.get('last_month', 0):,}") print(f" Last Month: {downloads.get('last_month', 0):,}")
# Display analysis # Display analysis
if analysis: if analysis:
print(f"\nAnalysis:") print("\nAnalysis:")
print(f" Total Downloads: {analysis.get('total_downloads', 0):,}") print(f" Total Downloads: {analysis.get('total_downloads', 0):,}")
print(f" Highest Period: {analysis.get('highest_period', 'N/A')}") print(f" Highest Period: {analysis.get('highest_period', 'N/A')}")
growth = analysis.get('growth_indicators', {}) growth = analysis.get('growth_indicators', {})
if growth: if growth:
print(f" Growth Indicators:") print(" Growth Indicators:")
for indicator, value in growth.items(): for indicator, value in growth.items():
print(f" {indicator}: {value}") print(f" {indicator}: {value}")
# Display repository info if available # Display repository info if available
project_urls = metadata.get('project_urls', {}) project_urls = metadata.get('project_urls', {})
if project_urls: if project_urls:
print(f"\nRepository Links:") print("\nRepository Links:")
for name, url in project_urls.items(): for name, url in project_urls.items():
if url: if url:
print(f" {name}: {url}") print(f" {name}: {url}")
except Exception as e: except Exception as e:
print(f"❌ Error getting stats for {package_name}: {e}") print(f"❌ Error getting stats for {package_name}: {e}")
@ -78,45 +77,45 @@ async def demo_package_download_trends():
print("\n" + "=" * 60) print("\n" + "=" * 60)
print("PyPI Package Download Trends Demo") print("PyPI Package Download Trends Demo")
print("=" * 60) print("=" * 60)
# Analyze trends for a popular package # Analyze trends for a popular package
package_name = "requests" package_name = "requests"
print(f"\n📈 Download Trends for '{package_name}':") print(f"\n📈 Download Trends for '{package_name}':")
print("-" * 50) print("-" * 50)
try: try:
# Get download trends (without mirrors for cleaner data) # Get download trends (without mirrors for cleaner data)
trends = await get_package_download_trends(package_name, include_mirrors=False) trends = await get_package_download_trends(package_name, include_mirrors=False)
trend_analysis = trends.get("trend_analysis", {}) trend_analysis = trends.get("trend_analysis", {})
time_series = trends.get("time_series", []) time_series = trends.get("time_series", [])
print(f"Package: {package_name}") print(f"Package: {package_name}")
print(f"Data Points: {trend_analysis.get('data_points', 0)}") print(f"Data Points: {trend_analysis.get('data_points', 0)}")
print(f"Total Downloads: {trend_analysis.get('total_downloads', 0):,}") print(f"Total Downloads: {trend_analysis.get('total_downloads', 0):,}")
print(f"Average Daily: {trend_analysis.get('average_daily', 0):,.0f}") print(f"Average Daily: {trend_analysis.get('average_daily', 0):,.0f}")
print(f"Trend Direction: {trend_analysis.get('trend_direction', 'unknown')}") print(f"Trend Direction: {trend_analysis.get('trend_direction', 'unknown')}")
# Display date range # Display date range
date_range = trend_analysis.get('date_range', {}) date_range = trend_analysis.get('date_range', {})
if date_range: if date_range:
print(f"Date Range: {date_range.get('start')} to {date_range.get('end')}") print(f"Date Range: {date_range.get('start')} to {date_range.get('end')}")
# Display peak day # Display peak day
peak_day = trend_analysis.get('peak_day', {}) peak_day = trend_analysis.get('peak_day', {})
if peak_day: if peak_day:
print(f"Peak Day: {peak_day.get('date')} ({peak_day.get('downloads', 0):,} downloads)") print(f"Peak Day: {peak_day.get('date')} ({peak_day.get('downloads', 0):,} downloads)")
# Show recent data points (last 7 days) # Show recent data points (last 7 days)
if time_series: 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:] recent_data = [item for item in time_series if item.get('category') == 'without_mirrors'][-7:]
for item in recent_data: for item in recent_data:
date = item.get('date', 'unknown') date = item.get('date', 'unknown')
downloads = item.get('downloads', 0) downloads = item.get('downloads', 0)
print(f" {date}: {downloads:,} downloads") print(f" {date}: {downloads:,} downloads")
except Exception as e: except Exception as e:
print(f"❌ Error getting trends for {package_name}: {e}") print(f"❌ Error getting trends for {package_name}: {e}")
@ -126,33 +125,33 @@ async def demo_top_packages():
print("\n" + "=" * 60) print("\n" + "=" * 60)
print("Top PyPI Packages by Downloads Demo") print("Top PyPI Packages by Downloads Demo")
print("=" * 60) print("=" * 60)
periods = ["day", "week", "month"] periods = ["day", "week", "month"]
for period in periods: for period in periods:
print(f"\n🏆 Top 10 Packages (last {period}):") print(f"\n🏆 Top 10 Packages (last {period}):")
print("-" * 50) print("-" * 50)
try: try:
# Get top packages for this period # Get top packages for this period
top_packages = await get_top_packages_by_downloads(period=period, limit=10) top_packages = await get_top_packages_by_downloads(period=period, limit=10)
packages_list = top_packages.get("top_packages", []) packages_list = top_packages.get("top_packages", [])
total_found = top_packages.get("total_found", 0) total_found = top_packages.get("total_found", 0)
print(f"Found {total_found} packages") print(f"Found {total_found} packages")
print(f"Data Source: {top_packages.get('data_source', 'unknown')}") print(f"Data Source: {top_packages.get('data_source', 'unknown')}")
if top_packages.get("note"): if top_packages.get("note"):
print(f"Note: {top_packages['note']}") print(f"Note: {top_packages['note']}")
print(f"\nRankings:") print("\nRankings:")
for package in packages_list: for package in packages_list:
rank = package.get("rank", "?") rank = package.get("rank", "?")
name = package.get("package", "unknown") name = package.get("package", "unknown")
downloads = package.get("downloads", 0) downloads = package.get("downloads", 0)
print(f" {rank:2d}. {name:<20} {downloads:>12,} downloads") print(f" {rank:2d}. {name:<20} {downloads:>12,} downloads")
except Exception as e: except Exception as e:
print(f"❌ Error getting top packages for {period}: {e}") print(f"❌ Error getting top packages for {period}: {e}")
@ -162,37 +161,37 @@ async def demo_package_comparison():
print("\n" + "=" * 60) print("\n" + "=" * 60)
print("Package Comparison Demo") print("Package Comparison Demo")
print("=" * 60) print("=" * 60)
# Compare web frameworks # Compare web frameworks
frameworks = ["django", "flask", "fastapi", "tornado"] frameworks = ["django", "flask", "fastapi", "tornado"]
print(f"\n🔍 Comparing Web Frameworks (last month downloads):") print("\n🔍 Comparing Web Frameworks (last month downloads):")
print("-" * 70) print("-" * 70)
comparison_data = [] comparison_data = []
for framework in frameworks: for framework in frameworks:
try: try:
stats = await get_package_download_stats(framework, period="month") stats = await get_package_download_stats(framework, period="month")
downloads = stats.get("downloads", {}) downloads = stats.get("downloads", {})
last_month = downloads.get("last_month", 0) last_month = downloads.get("last_month", 0)
comparison_data.append({ comparison_data.append({
"name": framework, "name": framework,
"downloads": last_month, "downloads": last_month,
"metadata": stats.get("metadata", {}), "metadata": stats.get("metadata", {}),
}) })
except Exception as e: except Exception as e:
print(f"❌ Error getting stats for {framework}: {e}") print(f"❌ Error getting stats for {framework}: {e}")
# Sort by downloads (descending) # Sort by downloads (descending)
comparison_data.sort(key=lambda x: x["downloads"], reverse=True) comparison_data.sort(key=lambda x: x["downloads"], reverse=True)
# Display comparison # Display comparison
print(f"{'Rank':<4} {'Framework':<12} {'Downloads':<15} {'Summary'}") print(f"{'Rank':<4} {'Framework':<12} {'Downloads':<15} {'Summary'}")
print("-" * 70) print("-" * 70)
for i, data in enumerate(comparison_data, 1): for i, data in enumerate(comparison_data, 1):
name = data["name"] name = data["name"]
downloads = data["downloads"] downloads = data["downloads"]
@ -204,18 +203,18 @@ async def main():
"""Run all demo functions.""" """Run all demo functions."""
print("🚀 Starting PyPI Download Statistics Demo") print("🚀 Starting PyPI Download Statistics Demo")
print(f"Timestamp: {datetime.now().isoformat()}") print(f"Timestamp: {datetime.now().isoformat()}")
try: try:
# Run all demos # Run all demos
await demo_package_download_stats() await demo_package_download_stats()
await demo_package_download_trends() await demo_package_download_trends()
await demo_top_packages() await demo_top_packages()
await demo_package_comparison() await demo_package_comparison()
print("\n" + "=" * 60) print("\n" + "=" * 60)
print("✅ Demo completed successfully!") print("✅ Demo completed successfully!")
print("=" * 60) print("=" * 60)
except KeyboardInterrupt: except KeyboardInterrupt:
print("\n❌ Demo interrupted by user") print("\n❌ Demo interrupted by user")
except Exception as e: except Exception as e:

View File

@ -2,7 +2,6 @@
import asyncio import asyncio
import logging import logging
from datetime import datetime, timedelta
from typing import Any from typing import Any
import httpx import httpx

View File

@ -541,7 +541,7 @@ async def get_top_downloaded_packages(
logger.info(f"MCP tool: Getting top {actual_limit} packages for period: {period}") logger.info(f"MCP tool: Getting top {actual_limit} packages for period: {period}")
result = await get_top_packages_by_downloads(period, actual_limit) 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 return result
except Exception as e: except Exception as e:
logger.error(f"Error getting top packages: {e}") logger.error(f"Error getting top packages: {e}")

View File

@ -1,12 +1,11 @@
"""PyPI package download statistics tools.""" """PyPI package download statistics tools."""
import logging import logging
from datetime import datetime, timedelta from datetime import datetime
from typing import Any from typing import Any
from ..core.pypi_client import PyPIClient from ..core.pypi_client import PyPIClient
from ..core.stats_client import PyPIStatsClient from ..core.stats_client import PyPIStatsClient
from ..core.exceptions import InvalidPackageNameError, NetworkError, PackageNotFoundError
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -57,7 +56,7 @@ async def get_package_download_stats(
# Extract download data # Extract download data
download_data = recent_stats.get("data", {}) download_data = recent_stats.get("data", {})
# Calculate trends and analysis # Calculate trends and analysis
analysis = _analyze_download_stats(download_data) analysis = _analyze_download_stats(download_data)
@ -106,7 +105,7 @@ async def get_package_download_trends(
# Process time series data # Process time series data
time_series_data = overall_stats.get("data", []) time_series_data = overall_stats.get("data", [])
# Analyze trends # Analyze trends
trend_analysis = _analyze_download_trends(time_series_data, include_mirrors) 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: async with PyPIStatsClient() as stats_client:
try: try:
top_packages = [] top_packages = []
# Get download stats for popular packages # Get download stats for popular packages
for i, package_name in enumerate(popular_packages[:limit]): for i, package_name in enumerate(popular_packages[:limit]):
try: try:
stats = await stats_client.get_recent_downloads( stats = await stats_client.get_recent_downloads(
package_name, period, use_cache=True package_name, period, use_cache=True
) )
download_data = stats.get("data", {}) download_data = stats.get("data", {})
download_count = _extract_download_count(download_data, period) download_count = _extract_download_count(download_data, period)
top_packages.append({ top_packages.append({
"rank": i + 1, "rank": i + 1,
"package": package_name, "package": package_name,
"downloads": download_count, "downloads": download_count,
"period": period, "period": period,
}) })
except Exception as e: except Exception as e:
logger.warning(f"Could not get stats for {package_name}: {e}") logger.warning(f"Could not get stats for {package_name}: {e}")
continue continue
# Sort by download count (descending) # Sort by download count (descending)
top_packages.sort(key=lambda x: x.get("downloads", 0), reverse=True) top_packages.sort(key=lambda x: x.get("downloads", 0), reverse=True)
# Update ranks after sorting # Update ranks after sorting
for i, package in enumerate(top_packages): for i, package in enumerate(top_packages):
package["rank"] = i + 1 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): if period.startswith("last_") and isinstance(count, int):
analysis["periods_available"].append(period) analysis["periods_available"].append(period)
analysis["total_downloads"] += count analysis["total_downloads"] += count
if analysis["highest_period"] is None or count > download_data.get(analysis["highest_period"], 0): if analysis["highest_period"] is None or count > download_data.get(analysis["highest_period"], 0):
analysis["highest_period"] = period 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: if last_day and last_week:
analysis["growth_indicators"]["daily_vs_weekly"] = round(last_day * 7 / last_week, 2) analysis["growth_indicators"]["daily_vs_weekly"] = round(last_day * 7 / last_week, 2)
if last_week and last_month: if last_week and last_month:
analysis["growth_indicators"]["weekly_vs_monthly"] = round(last_week * 4 / last_month, 2) 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 # Filter data based on mirror preference
category_filter = "with_mirrors" if include_mirrors else "without_mirrors" category_filter = "with_mirrors" if include_mirrors else "without_mirrors"
filtered_data = [ filtered_data = [
item for item in time_series_data item for item in time_series_data
if item.get("category") == category_filter 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: if len(filtered_data) >= 14:
first_week = sum(item.get("downloads", 0) for item in filtered_data[:7]) 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:]) last_week = sum(item.get("downloads", 0) for item in filtered_data[-7:])
if last_week > first_week * 1.1: if last_week > first_week * 1.1:
analysis["trend_direction"] = "increasing" analysis["trend_direction"] = "increasing"
elif last_week < first_week * 0.9: elif last_week < first_week * 0.9:

View File

@ -1,17 +1,18 @@
"""Tests for download statistics functionality.""" """Tests for download statistics functionality."""
import pytest
from unittest.mock import AsyncMock, patch from unittest.mock import AsyncMock, patch
import pytest
from pypi_query_mcp.core.exceptions import PackageNotFoundError
from pypi_query_mcp.tools.download_stats import ( 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_stats,
_analyze_download_trends, _analyze_download_trends,
_extract_download_count, _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: class TestDownloadStats:
@ -43,7 +44,7 @@ class TestDownloadStats:
with patch("pypi_query_mcp.tools.download_stats.PyPIStatsClient") as mock_stats_client, \ 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: patch("pypi_query_mcp.tools.download_stats.PyPIClient") as mock_pypi_client:
# Setup mocks # Setup mocks
mock_stats_instance = AsyncMock() mock_stats_instance = AsyncMock()
mock_stats_instance.get_recent_downloads.return_value = mock_stats_data mock_stats_instance.get_recent_downloads.return_value = mock_stats_data