server.py

"""
github-insight-mcp -- MCP server for GitHub repository & package analysis.
GitHub API + npm Registry + PyPI + deps.dev (Google Open Source Insights).
Works without any API key. Optional GitHub token for higher rate limits.
"""

import argparse
import os
import sys

from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP

import core.github as github
import core.npm as npm
import core.pypi as pypi
import core.deps as deps

# Load .env file if present (for GITHUB_TOKEN)
load_dotenv()

mcp = FastMCP("github-insight")


# ──────────────────────────────────────────────
# API Status
# ──────────────────────────────────────────────

@mcp.tool()
def api_status() -> str:
    """Check which APIs are available and current rate limit status."""
    lines = ["## API Status\n"]
    lines.append("  GitHub REST API:     [OK] Ready")
    if github.is_token_set():
        lines.append("    -> Token detected: 5,000 requests/hour")
    else:
        lines.append("    -> No token: 60 requests/hour (set GITHUB_TOKEN for more)")
    lines.append("  npm Registry:        [OK] Ready (no key needed)")
    lines.append("  PyPI + pypistats:    [OK] Ready (no key needed)")
    lines.append("  deps.dev (Google):   [OK] Ready (no key needed)")
    lines.append("")
    lines.append("All features available.")
    return "\n".join(lines)


# ──────────────────────────────────────────────
# Analyze Repository (main tool)
# ──────────────────────────────────────────────

@mcp.tool()
def analyze_repo(owner: str, repo: str) -> str:
    """Comprehensive GitHub repository analysis.
    Stars, forks, recent activity, contributors, releases, and topics.
    This is the main tool for understanding a repository at a glance.

    - owner: Repository owner (e.g. 'facebook')
    - repo: Repository name (e.g. 'react')
    """
    sections = [f"# Repository Analysis: {owner}/{repo}\n"]

    # 1. Basic Info
    try:
        info = github.get_repo(owner, repo)
    except Exception as e:
        return f"[Error] Could not fetch repository: {e}"

    sections.append("## Overview\n")
    sections.append(f"  **Name:** {info['full_name']}")
    sections.append(f"  **Description:** {info.get('description') or 'N/A'}")
    sections.append(f"  **Language:** {info.get('language') or 'N/A'}")
    sections.append(f"  **License:** {info['license']}")
    sections.append(f"  **Default Branch:** {info['default_branch']}")
    sections.append(f"  **Link:** {info['html_url']}")
    if info.get("topics"):
        sections.append(f"  **Topics:** {', '.join(info['topics'])}")
    sections.append("")

    # Stars / Forks / Issues
    sections.append("## Stats\n")
    sections.append(f"  Stars: {info['stargazers_count']:,}")
    sections.append(f"  Forks: {info['forks_count']:,}")
    sections.append(f"  Watchers: {info['watchers_count']:,}")
    sections.append(f"  Open Issues: {info['open_issues_count']:,}")
    sections.append(f"  Created: {info['created_at'][:10]}")
    sections.append(f"  Last Push: {info['pushed_at'][:10]}")
    sections.append("")

    # 2. Recent Activity (30 days)
    sections.append("## Recent Activity (30 days)\n")
    try:
        commits = github.get_recent_commits(owner, repo, days=30)
        sections.append(f"  Commits: {len(commits)}")
    except Exception:
        sections.append("  Commits: [Could not fetch]")

    try:
        issues = github.get_recent_issues(owner, repo, days=30)
        sections.append(f"  Issues: {issues['total']} (opened: {issues['opened']}, closed: {issues['closed']})")
    except Exception:
        sections.append("  Issues: [Could not fetch]")

    try:
        pulls = github.get_recent_pulls(owner, repo, days=30)
        sections.append(f"  Pull Requests: {pulls['total']} (merged: {pulls['merged']}, open: {pulls['open']})")
    except Exception:
        sections.append("  Pull Requests: [Could not fetch]")
    sections.append("")

    # 3. Top Contributors
    sections.append("## Top Contributors\n")
    try:
        contributors = github.get_contributors(owner, repo, count=5)
        for i, c in enumerate(contributors, 1):
            sections.append(f"  {i}. {c['login']} ({c['contributions']} commits)")
    except Exception:
        sections.append("  [Could not fetch contributors]")
    sections.append("")

    # 4. Recent Releases
    sections.append("## Recent Releases\n")
    try:
        releases = github.get_releases(owner, repo, count=3)
        if releases:
            for r in releases:
                name = r.get("name") or r["tag_name"]
                date = (r.get("published_at") or "")[:10]
                sections.append(f"  - {name} ({date})")
        else:
            sections.append("  No releases found.")
    except Exception:
        sections.append("  [Could not fetch releases]")
    sections.append("")

    return "\n".join(sections)


# ──────────────────────────────────────────────
# Compare Repositories
# ──────────────────────────────────────────────

@mcp.tool()
def compare_repos(owner1: str, repo1: str, owner2: str, repo2: str) -> str:
    """Compare two GitHub repositories side by side.
    Stars, forks, activity, language, and license comparison.

    - owner1: First repo owner (e.g. 'expressjs')
    - repo1: First repo name (e.g. 'express')
    - owner2: Second repo owner (e.g. 'fastify')
    - repo2: Second repo name (e.g. 'fastify')
    """
    try:
        info1 = github.get_repo(owner1, repo1)
    except Exception as e:
        return f"[Error] Could not fetch {owner1}/{repo1}: {e}"

    try:
        info2 = github.get_repo(owner2, repo2)
    except Exception as e:
        return f"[Error] Could not fetch {owner2}/{repo2}: {e}"

    name1 = info1["full_name"]
    name2 = info2["full_name"]

    lines = [f"# Repository Comparison\n"]
    lines.append(f"## {name1} vs {name2}\n")

    # Table format
    lines.append(f"  | Metric | {name1} | {name2} |")
    lines.append(f"  |--------|---|---|")
    lines.append(f"  | Stars | {info1['stargazers_count']:,} | {info2['stargazers_count']:,} |")
    lines.append(f"  | Forks | {info1['forks_count']:,} | {info2['forks_count']:,} |")
    lines.append(f"  | Open Issues | {info1['open_issues_count']:,} | {info2['open_issues_count']:,} |")
    lines.append(f"  | Language | {info1.get('language') or 'N/A'} | {info2.get('language') or 'N/A'} |")
    lines.append(f"  | License | {info1['license']} | {info2['license']} |")
    lines.append(f"  | Created | {info1['created_at'][:10]} | {info2['created_at'][:10]} |")
    lines.append(f"  | Last Push | {info1['pushed_at'][:10]} | {info2['pushed_at'][:10]} |")
    lines.append("")

    # Activity comparison
    lines.append("## Recent Activity (30 days)\n")
    for label, owner, repo in [(name1, owner1, repo1), (name2, owner2, repo2)]:
        lines.append(f"  **{label}:**")
        try:
            commits = github.get_recent_commits(owner, repo, days=30)
            lines.append(f"    Commits: {len(commits)}")
        except Exception:
            lines.append("    Commits: [Error]")
        try:
            issues = github.get_recent_issues(owner, repo, days=30)
            lines.append(f"    Issues: {issues['total']} (opened: {issues['opened']}, closed: {issues['closed']})")
        except Exception:
            lines.append("    Issues: [Error]")
        lines.append("")

    lines.append(f"  Links: {info1['html_url']} | {info2['html_url']}")
    return "\n".join(lines)


# ──────────────────────────────────────────────
# Repository Activity Timeline
# ──────────────────────────────────────────────

@mcp.tool()
def repo_activity(owner: str, repo: str, days: int = 30) -> str:
    """Get recent repository activity: commits, issues, and pull requests.

    - owner: Repository owner (e.g. 'pallets')
    - repo: Repository name (e.g. 'flask')
    - days: Number of days to look back (default: 30)
    """
    lines = [f"# Recent Activity: {owner}/{repo} (last {days} days)\n"]

    # Commits
    lines.append("## Commits\n")
    try:
        commits = github.get_recent_commits(owner, repo, days=days)
        if commits:
            lines.append(f"  Total: {len(commits)} commits\n")
            for c in commits[:10]:
                date = (c.get("date") or "")[:10]
                lines.append(f"  - [{c['sha']}] {c['message'][:80]} ({c['author']}, {date})")
            if len(commits) > 10:
                lines.append(f"  ... and {len(commits) - 10} more")
        else:
            lines.append("  No commits in this period.")
    except Exception as e:
        lines.append(f"  [Error] {e}")
    lines.append("")

    # Issues
    lines.append("## Issues\n")
    try:
        issues = github.get_recent_issues(owner, repo, days=days)
        lines.append(f"  Total: {issues['total']} (opened: {issues['opened']}, closed: {issues['closed']})")
    except Exception as e:
        lines.append(f"  [Error] {e}")
    lines.append("")

    # Pull Requests
    lines.append("## Pull Requests\n")
    try:
        pulls = github.get_recent_pulls(owner, repo, days=days)
        lines.append(f"  Total: {pulls['total']} (merged: {pulls['merged']}, open: {pulls['open']})")
    except Exception as e:
        lines.append(f"  [Error] {e}")
    lines.append("")

    return "\n".join(lines)


# ──────────────────────────────────────────────
# Check Package (npm / PyPI)
# ──────────────────────────────────────────────

@mcp.tool()
def check_package(name: str, ecosystem: str = "npm") -> str:
    """Get package info and download stats from npm or PyPI.

    - name: Package name (e.g. 'express', 'requests', '@types/node')
    - ecosystem: 'npm' or 'pypi' (default: npm)
    """
    ecosystem = ecosystem.lower().strip()

    if ecosystem == "npm":
        return _check_npm(name)
    elif ecosystem == "pypi":
        return _check_pypi(name)
    else:
        return f"[Error] Unknown ecosystem: '{ecosystem}'. Use 'npm' or 'pypi'."


def _check_npm(name: str) -> str:
    lines = [f"# npm Package: {name}\n"]

    try:
        pkg = npm.get_package(name)
    except Exception as e:
        return f"[Error] npm lookup failed: {e}"

    lines.append("## Info\n")
    lines.append(f"  **Name:** {pkg['name']}")
    lines.append(f"  **Version:** {pkg['version']}")
    lines.append(f"  **Description:** {pkg.get('description') or 'N/A'}")
    lines.append(f"  **License:** {pkg.get('license') or 'N/A'}")
    if pkg.get("homepage"):
        lines.append(f"  **Homepage:** {pkg['homepage']}")
    lines.append(f"  **Dependencies:** {pkg['dependencies_count']}")
    if pkg.get("keywords"):
        lines.append(f"  **Keywords:** {', '.join(pkg['keywords'][:10])}")
    lines.append("")

    # Download stats
    lines.append("## Downloads\n")
    try:
        weekly = npm.get_downloads(name, "last-week")
        monthly = npm.get_downloads(name, "last-month")
        lines.append(f"  Last week: {weekly['downloads']:,}")
        lines.append(f"  Last month: {monthly['downloads']:,}")
    except Exception:
        lines.append("  [Could not fetch download stats]")
    lines.append("")

    # deps.dev info
    try:
        dep_info = deps.get_package_info("npm", name)
        lines.append(f"## deps.dev\n")
        lines.append(f"  Total versions: {dep_info['versions_count']}")
        lines.append(f"  Latest (deps.dev): {dep_info['latest_version']}")
        lines.append("")
    except Exception:
        pass

    return "\n".join(lines)


def _check_pypi(name: str) -> str:
    lines = [f"# PyPI Package: {name}\n"]

    try:
        pkg = pypi.get_package(name)
    except Exception as e:
        return f"[Error] PyPI lookup failed: {e}"

    lines.append("## Info\n")
    lines.append(f"  **Name:** {pkg['name']}")
    lines.append(f"  **Version:** {pkg['version']}")
    lines.append(f"  **Summary:** {pkg.get('summary') or 'N/A'}")
    lines.append(f"  **License:** {pkg.get('license') or 'N/A'}")
    lines.append(f"  **Requires Python:** {pkg.get('requires_python') or 'N/A'}")
    lines.append(f"  **Project URL:** {pkg['project_url']}")
    if pkg.get("dependencies"):
        lines.append(f"  **Dependencies ({len(pkg['dependencies'])}):** {', '.join(pkg['dependencies'][:10])}")
        if len(pkg["dependencies"]) > 10:
            lines.append(f"    ... and {len(pkg['dependencies']) - 10} more")
    if pkg.get("keywords"):
        lines.append(f"  **Keywords:** {', '.join(pkg['keywords'][:10])}")
    lines.append("")

    # Download stats
    lines.append("## Downloads\n")
    try:
        stats = pypi.get_download_stats(name)
        lines.append(f"  Last day: {stats['last_day']:,}")
        lines.append(f"  Last week: {stats['last_week']:,}")
        lines.append(f"  Last month: {stats['last_month']:,}")
    except Exception:
        lines.append("  [Could not fetch download stats]")
    lines.append("")

    # deps.dev info
    try:
        dep_info = deps.get_package_info("pypi", name)
        lines.append(f"## deps.dev\n")
        lines.append(f"  Total versions: {dep_info['versions_count']}")
        lines.append(f"  Latest (deps.dev): {dep_info['latest_version']}")
        lines.append("")
    except Exception:
        pass

    return "\n".join(lines)


# ──────────────────────────────────────────────
# Security Check (deps.dev advisories)
# ──────────────────────────────────────────────

@mcp.tool()
def check_security(name: str, version: str, ecosystem: str = "npm") -> str:
    """Check security advisories and dependencies for a package version.
    Uses Google's deps.dev (Open Source Insights).

    - name: Package name (e.g. 'lodash', 'requests')
    - version: Exact version (e.g. '4.17.20', '2.31.0')
    - ecosystem: 'npm' or 'pypi' (default: npm)
    """
    ecosystem = ecosystem.lower().strip()
    if ecosystem not in ("npm", "pypi"):
        return f"[Error] Unknown ecosystem: '{ecosystem}'. Use 'npm' or 'pypi'."

    lines = [f"# Security Report: {name}@{version} ({ecosystem})\n"]

    # Advisories
    lines.append("## Security Advisories\n")
    try:
        advisories = deps.get_advisories(ecosystem, name, version)
        if advisories:
            lines.append(f"  [!] Found {len(advisories)} advisory(ies):\n")
            for a in advisories:
                lines.append(f"  - **{a['title']}**")
                lines.append(f"    ID: {a['advisory_id']} | Severity: {a['severity']}")
                if a.get("url"):
                    lines.append(f"    Link: {a['url']}")
                lines.append("")
        else:
            lines.append("  No known security advisories. [OK]")
    except Exception as e:
        lines.append(f"  [Error] Could not check advisories: {e}")
    lines.append("")

    # Version info
    lines.append("## Version Info\n")
    try:
        ver_info = deps.get_version_info(ecosystem, name, version)
        if ver_info.get("licenses"):
            lines.append(f"  Licenses: {', '.join(ver_info['licenses'])}")
        if ver_info.get("published_at"):
            lines.append(f"  Published: {ver_info['published_at'][:10]}")
    except Exception:
        lines.append("  [Could not fetch version info]")
    lines.append("")

    # Direct dependencies
    lines.append("## Direct Dependencies\n")
    try:
        dep_list = deps.get_dependencies(ecosystem, name, version)
        if dep_list:
            lines.append(f"  Total: {len(dep_list)}\n")
            for d in dep_list[:20]:
                lines.append(f"  - {d['name']} {d['version_requirement']}")
            if len(dep_list) > 20:
                lines.append(f"  ... and {len(dep_list) - 20} more")
        else:
            lines.append("  No direct dependencies.")
    except Exception:
        lines.append("  [Could not fetch dependencies]")
    lines.append("")

    return "\n".join(lines)


# ──────────────────────────────────────────────
# Search Repos
# ──────────────────────────────────────────────

@mcp.tool()
def search_repos(query: str, sort: str = "stars", count: int = 10) -> str:
    """Search GitHub repositories.

    - query: Search keywords (e.g. 'machine learning python', 'mcp server')
    - sort: Sort by 'stars', 'forks', or 'updated' (default: stars)
    - count: Number of results (default: 10)
    """
    try:
        repos = github.search_repos(query, sort=sort, count=count)
    except Exception as e:
        return f"[Error] GitHub search failed: {e}"

    if not repos:
        return f"No repositories found for '{query}'."

    lines = [f"# GitHub Search: '{query}' (sorted by {sort})\n"]
    for i, r in enumerate(repos, 1):
        lines.append(f"  {i}. **{r['full_name']}**")
        lines.append(f"     {r.get('description') or 'No description'}")
        lines.append(f"     Stars: {r['stargazers_count']:,} | Forks: {r['forks_count']:,} | Language: {r.get('language') or 'N/A'}")
        lines.append(f"     Updated: {r['updated_at'][:10]}")
        lines.append(f"     Link: {r['html_url']}")
        lines.append("")

    return "\n".join(lines)


# ──────────────────────────────────────────────
# Entry point
# ──────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser(description="GitHub Insight MCP Server")
    args = parser.parse_args()

    print("github-insight-mcp: server starting", file=sys.stderr)
    mcp.run()


if __name__ == "__main__":
    main()