"""Gitea MCP server — git-style tools for a self-hosted Gitea instance."""

from __future__ import annotations

import base64
import os

import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    "Gitea MCP",
    instructions="MCP server for interacting with a Gitea instance. "
    "Provides tools for repositories, issues, pull requests, files, branches, "
    "releases, organizations, and users.",
)

GITEA_URL = os.environ.get("GITEA_URL", "http://localhost:3000")
GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "")

_TIMEOUT = 15.0


def _client() -> httpx.Client:
    return httpx.Client(
        base_url=f"{GITEA_URL}/api/v1",
        headers={
            "Authorization": f"token {GITEA_TOKEN}",
            "Accept": "application/json",
            "Content-Type": "application/json",
        },
        timeout=_TIMEOUT,
    )


def _request(method: str, path: str, **kwargs) -> dict | list | str:
    with _client() as c:
        try:
            resp = c.request(method, path, **kwargs)
            resp.raise_for_status()
            if resp.status_code == 204:
                return {"ok": True}
            return resp.json()
        except httpx.HTTPStatusError as e:
            body = e.response.text[:500]
            return {"error": f"HTTP {e.response.status_code}", "detail": body}
        except httpx.RequestError as e:
            return {"error": str(e)}


def _get(path: str, **params) -> dict | list | str:
    return _request("GET", path, params=params)


def _post(path: str, body: dict | None = None) -> dict | list | str:
    return _request("POST", path, json=body or {})


def _patch(path: str, body: dict) -> dict | list | str:
    return _request("PATCH", path, json=body)


def _delete(path: str) -> dict | list | str:
    return _request("DELETE", path)


# ───────────────────────────────────────────────────────────────────
# User / Auth
# ───────────────────────────────────────────────────────────────────


@mcp.tool()
def get_authenticated_user() -> dict | list | str:
    """Get the currently authenticated Gitea user."""
    return _get("/user")


# ───────────────────────────────────────────────────────────────────
# Repositories
# ───────────────────────────────────────────────────────────────────


@mcp.tool()
def list_repos(limit: int = 20, page: int = 1) -> dict | list | str:
    """List repositories accessible to the authenticated user.

    Args:
        limit: Number of results per page. Default 20.
        page: Page number. Default 1.
    """
    return _get("/user/repos", limit=limit, page=page)


@mcp.tool()
def get_repo(owner: str, repo: str) -> dict | list | str:
    """Get details of a repository.

    Args:
        owner: Repository owner username.
        repo: Repository name.
    """
    return _get(f"/repos/{owner}/{repo}")


@mcp.tool()
def create_repo(
    name: str,
    description: str = "",
    private: bool = False,
    auto_init: bool = True,
    default_branch: str = "main",
) -> dict | list | str:
    """Create a new repository for the authenticated user.

    Args:
        name: Repository name.
        description: Repository description.
        private: Whether the repo is private. Default False.
        auto_init: Initialize with a README. Default True.
        default_branch: Default branch name. Default "main".
    """
    return _post(
        "/user/repos",
        {
            "name": name,
            "description": description,
            "private": private,
            "auto_init": auto_init,
            "default_branch": default_branch,
        },
    )


@mcp.tool()
def delete_repo(owner: str, repo: str) -> dict | list | str:
    """Delete a repository. This is irreversible.

    Args:
        owner: Repository owner username.
        repo: Repository name.
    """
    return _delete(f"/repos/{owner}/{repo}")


@mcp.tool()
def search_repos(query: str, limit: int = 10) -> dict | list | str:
    """Search for repositories.

    Args:
        query: Search query string.
        limit: Maximum results to return. Default 10.
    """
    return _get("/repos/search", q=query, limit=limit)


@mcp.tool()
def fork_repo(owner: str, repo: str, new_name: str | None = None) -> dict | list | str:
    """Fork a repository.

    Args:
        owner: Owner of the repository to fork.
        repo: Repository name to fork.
        new_name: Optional new name for the fork.
    """
    body: dict = {}
    if new_name:
        body["name"] = new_name
    return _post(f"/repos/{owner}/{repo}/forks", body)


# ───────────────────────────────────────────────────────────────────
# Branches
# ───────────────────────────────────────────────────────────────────


@mcp.tool()
def list_branches(owner: str, repo: str) -> dict | list | str:
    """List branches of a repository.

    Args:
        owner: Repository owner.
        repo: Repository name.
    """
    return _get(f"/repos/{owner}/{repo}/branches")


@mcp.tool()
def get_branch(owner: str, repo: str, branch: str) -> dict | list | str:
    """Get details of a specific branch.

    Args:
        owner: Repository owner.
        repo: Repository name.
        branch: Branch name.
    """
    return _get(f"/repos/{owner}/{repo}/branches/{branch}")


@mcp.tool()
def create_branch(
    owner: str, repo: str, branch_name: str, old_branch: str = "main"
) -> dict | list | str:
    """Create a new branch.

    Args:
        owner: Repository owner.
        repo: Repository name.
        branch_name: Name of the new branch.
        old_branch: Branch to create from. Default "main".
    """
    return _post(
        f"/repos/{owner}/{repo}/branches",
        {
            "new_branch_name": branch_name,
            "old_branch_name": old_branch,
        },
    )


@mcp.tool()
def delete_branch(owner: str, repo: str, branch: str) -> dict | list | str:
    """Delete a branch.

    Args:
        owner: Repository owner.
        repo: Repository name.
        branch: Branch name to delete.
    """
    return _delete(f"/repos/{owner}/{repo}/branches/{branch}")


# ───────────────────────────────────────────────────────────────────
# File contents
# ───────────────────────────────────────────────────────────────────


@mcp.tool()
def get_file(
    owner: str, repo: str, filepath: str, ref: str | None = None
) -> dict | list | str:
    """Get the contents of a file from a repository. Returns decoded text content.

    Args:
        owner: Repository owner.
        repo: Repository name.
        filepath: Path to the file in the repo.
        ref: Optional branch/tag/commit to read from.
    """
    params = {}
    if ref:
        params["ref"] = ref
    result = _get(f"/repos/{owner}/{repo}/contents/{filepath}", **params)
    # Decode base64 content for convenience
    if (
        isinstance(result, dict)
        and "content" in result
        and result.get("encoding") == "base64"
    ):
        try:
            result["content"] = base64.b64decode(result["content"]).decode(
                "utf-8", errors="replace"
            )
            result["encoding"] = "utf-8"
        except Exception:
            pass
    return result


@mcp.tool()
def create_or_update_file(
    owner: str,
    repo: str,
    filepath: str,
    content: str,
    message: str,
    branch: str | None = None,
    sha: str | None = None,
) -> dict | list | str:
    """Create or update a file in a repository.

    To update an existing file you must provide the current sha (from get_file).

    Args:
        owner: Repository owner.
        repo: Repository name.
        filepath: Path where the file will be created/updated.
        content: File content (plain text, will be base64-encoded).
        message: Commit message.
        branch: Branch to commit to. Uses repo default if omitted.
        sha: Current SHA of the file (required for updates, omit for creation).
    """
    body: dict = {
        "content": base64.b64encode(content.encode()).decode(),
        "message": message,
    }
    if branch:
        body["branch"] = branch
    if sha:
        body["sha"] = sha
    method = "PUT"
    with _client() as c:
        try:
            resp = c.request(
                method, f"/repos/{owner}/{repo}/contents/{filepath}", json=body
            )
            resp.raise_for_status()
            return resp.json()
        except httpx.HTTPStatusError as e:
            return {
                "error": f"HTTP {e.response.status_code}",
                "detail": e.response.text[:500],
            }
        except httpx.RequestError as e:
            return {"error": str(e)}


@mcp.tool()
def delete_file(
    owner: str,
    repo: str,
    filepath: str,
    message: str,
    sha: str,
    branch: str | None = None,
) -> dict | list | str:
    """Delete a file from a repository.

    Args:
        owner: Repository owner.
        repo: Repository name.
        filepath: Path of the file to delete.
        message: Commit message.
        sha: Current SHA of the file (from get_file).
        branch: Branch to delete from. Uses repo default if omitted.
    """
    body: dict = {"message": message, "sha": sha}
    if branch:
        body["branch"] = branch
    with _client() as c:
        try:
            resp = c.request(
                "DELETE", f"/repos/{owner}/{repo}/contents/{filepath}", json=body
            )
            resp.raise_for_status()
            return resp.json()
        except httpx.HTTPStatusError as e:
            return {
                "error": f"HTTP {e.response.status_code}",
                "detail": e.response.text[:500],
            }
        except httpx.RequestError as e:
            return {"error": str(e)}


@mcp.tool()
def list_directory(
    owner: str, repo: str, path: str = "", ref: str | None = None
) -> dict | list | str:
    """List files and directories at a path in a repository.

    Args:
        owner: Repository owner.
        repo: Repository name.
        path: Directory path. Empty string for root.
        ref: Optional branch/tag/commit.
    """
    params = {}
    if ref:
        params["ref"] = ref
    return _get(f"/repos/{owner}/{repo}/contents/{path}", **params)


# ───────────────────────────────────────────────────────────────────
# Commits
# ───────────────────────────────────────────────────────────────────


@mcp.tool()
def list_commits(
    owner: str, repo: str, sha: str | None = None, limit: int = 10, page: int = 1
) -> dict | list | str:
    """List commits in a repository.

    Args:
        owner: Repository owner.
        repo: Repository name.
        sha: Optional branch/tag/commit SHA to list from.
        limit: Number of commits per page. Default 10.
        page: Page number. Default 1.
    """
    params: dict = {"limit": limit, "page": page}
    if sha:
        params["sha"] = sha
    return _get(f"/repos/{owner}/{repo}/git/commits", **params)


@mcp.tool()
def get_commit(owner: str, repo: str, sha: str) -> dict | list | str:
    """Get details of a specific commit.

    Args:
        owner: Repository owner.
        repo: Repository name.
        sha: Commit SHA.
    """
    return _get(f"/repos/{owner}/{repo}/git/commits/{sha}")


# ───────────────────────────────────────────────────────────────────
# Issues
# ───────────────────────────────────────────────────────────────────


@mcp.tool()
def list_issues(
    owner: str,
    repo: str,
    state: str = "open",
    labels: str | None = None,
    limit: int = 20,
    page: int = 1,
) -> dict | list | str:
    """List issues in a repository.

    Args:
        owner: Repository owner.
        repo: Repository name.
        state: Filter by state: "open", "closed", or "all". Default "open".
        labels: Comma-separated label names to filter by.
        limit: Results per page. Default 20.
        page: Page number. Default 1.
    """
    params: dict = {"state": state, "limit": limit, "page": page, "type": "issues"}
    if labels:
        params["labels"] = labels
    return _get(f"/repos/{owner}/{repo}/issues", **params)


@mcp.tool()
def get_issue(owner: str, repo: str, index: int) -> dict | list | str:
    """Get details of an issue.

    Args:
        owner: Repository owner.
        repo: Repository name.
        index: Issue number.
    """
    return _get(f"/repos/{owner}/{repo}/issues/{index}")


@mcp.tool()
def create_issue(
    owner: str,
    repo: str,
    title: str,
    body: str = "",
    labels: list[int] | None = None,
    assignees: list[str] | None = None,
) -> dict | list | str:
    """Create a new issue.

    Args:
        owner: Repository owner.
        repo: Repository name.
        title: Issue title.
        body: Issue body/description.
        labels: List of label IDs to add.
        assignees: List of usernames to assign.
    """
    payload: dict = {"title": title, "body": body}
    if labels:
        payload["labels"] = labels
    if assignees:
        payload["assignees"] = assignees
    return _post(f"/repos/{owner}/{repo}/issues", payload)


@mcp.tool()
def edit_issue(
    owner: str,
    repo: str,
    index: int,
    title: str | None = None,
    body: str | None = None,
    state: str | None = None,
) -> dict | list | str:
    """Edit an existing issue.

    Args:
        owner: Repository owner.
        repo: Repository name.
        index: Issue number.
        title: New title (omit to keep current).
        body: New body (omit to keep current).
        state: Set to "open" or "closed".
    """
    payload: dict = {}
    if title is not None:
        payload["title"] = title
    if body is not None:
        payload["body"] = body
    if state is not None:
        payload["state"] = state
    return _patch(f"/repos/{owner}/{repo}/issues/{index}", payload)


@mcp.tool()
def list_issue_comments(owner: str, repo: str, index: int) -> dict | list | str:
    """List comments on an issue.

    Args:
        owner: Repository owner.
        repo: Repository name.
        index: Issue number.
    """
    return _get(f"/repos/{owner}/{repo}/issues/{index}/comments")


@mcp.tool()
def create_issue_comment(
    owner: str, repo: str, index: int, body: str
) -> dict | list | str:
    """Add a comment to an issue.

    Args:
        owner: Repository owner.
        repo: Repository name.
        index: Issue number.
        body: Comment text.
    """
    return _post(f"/repos/{owner}/{repo}/issues/{index}/comments", {"body": body})


# ───────────────────────────────────────────────────────────────────
# Labels
# ───────────────────────────────────────────────────────────────────


@mcp.tool()
def list_labels(owner: str, repo: str) -> dict | list | str:
    """List labels in a repository.

    Args:
        owner: Repository owner.
        repo: Repository name.
    """
    return _get(f"/repos/{owner}/{repo}/labels")


@mcp.tool()
def create_label(
    owner: str, repo: str, name: str, color: str = "#0075ca", description: str = ""
) -> dict | list | str:
    """Create a label in a repository.

    Args:
        owner: Repository owner.
        repo: Repository name.
        name: Label name.
        color: Hex color code. Default "#0075ca".
        description: Label description.
    """
    return _post(
        f"/repos/{owner}/{repo}/labels",
        {
            "name": name,
            "color": color,
            "description": description,
        },
    )


# ───────────────────────────────────────────────────────────────────
# Pull Requests
# ───────────────────────────────────────────────────────────────────


@mcp.tool()
def list_pull_requests(
    owner: str, repo: str, state: str = "open", limit: int = 20, page: int = 1
) -> dict | list | str:
    """List pull requests in a repository.

    Args:
        owner: Repository owner.
        repo: Repository name.
        state: Filter by state: "open", "closed", or "all". Default "open".
        limit: Results per page. Default 20.
        page: Page number. Default 1.
    """
    return _get(f"/repos/{owner}/{repo}/pulls", state=state, limit=limit, page=page)


@mcp.tool()
def get_pull_request(owner: str, repo: str, index: int) -> dict | list | str:
    """Get details of a pull request.

    Args:
        owner: Repository owner.
        repo: Repository name.
        index: Pull request number.
    """
    return _get(f"/repos/{owner}/{repo}/pulls/{index}")


@mcp.tool()
def create_pull_request(
    owner: str,
    repo: str,
    title: str,
    head: str,
    base: str = "main",
    body: str = "",
) -> dict | list | str:
    """Create a new pull request.

    Args:
        owner: Repository owner.
        repo: Repository name.
        title: PR title.
        head: Source branch.
        base: Target branch. Default "main".
        body: PR description.
    """
    return _post(
        f"/repos/{owner}/{repo}/pulls",
        {
            "title": title,
            "head": head,
            "base": base,
            "body": body,
        },
    )


@mcp.tool()
def merge_pull_request(
    owner: str, repo: str, index: int, merge_style: str = "merge", message: str = ""
) -> dict | list | str:
    """Merge a pull request.

    Args:
        owner: Repository owner.
        repo: Repository name.
        index: Pull request number.
        merge_style: One of "merge", "rebase", "rebase-merge", "squash". Default "merge".
        message: Optional merge commit message.
    """
    payload: dict = {"Do": merge_style}
    if message:
        payload["merge_message_field"] = message
    return _post(f"/repos/{owner}/{repo}/pulls/{index}/merge", payload)


@mcp.tool()
def list_pr_comments(owner: str, repo: str, index: int) -> dict | list | str:
    """List review comments on a pull request.

    Args:
        owner: Repository owner.
        repo: Repository name.
        index: Pull request number.
    """
    return _get(f"/repos/{owner}/{repo}/pulls/{index}/reviews")


@mcp.tool()
def get_pr_diff(owner: str, repo: str, index: int) -> dict | list | str:
    """Get the diff of a pull request.

    Args:
        owner: Repository owner.
        repo: Repository name.
        index: Pull request number.
    """
    with _client() as c:
        try:
            resp = c.get(f"/repos/{owner}/{repo}/pulls/{index}.diff")
            resp.raise_for_status()
            return {"diff": resp.text[:50000]}
        except httpx.HTTPStatusError as e:
            return {
                "error": f"HTTP {e.response.status_code}",
                "detail": e.response.text[:500],
            }
        except httpx.RequestError as e:
            return {"error": str(e)}


# ───────────────────────────────────────────────────────────────────
# Releases / Tags
# ───────────────────────────────────────────────────────────────────


@mcp.tool()
def list_releases(
    owner: str, repo: str, limit: int = 10, page: int = 1
) -> dict | list | str:
    """List releases in a repository.

    Args:
        owner: Repository owner.
        repo: Repository name.
        limit: Results per page. Default 10.
        page: Page number. Default 1.
    """
    return _get(f"/repos/{owner}/{repo}/releases", limit=limit, page=page)


@mcp.tool()
def create_release(
    owner: str,
    repo: str,
    tag_name: str,
    name: str = "",
    body: str = "",
    draft: bool = False,
    prerelease: bool = False,
    target: str | None = None,
) -> dict | list | str:
    """Create a new release.

    Args:
        owner: Repository owner.
        repo: Repository name.
        tag_name: Tag for the release (e.g. "v1.0.0").
        name: Release title.
        body: Release notes.
        draft: Whether this is a draft release.
        prerelease: Whether this is a pre-release.
        target: Branch or commit SHA to tag. Uses default branch if omitted.
    """
    payload: dict = {
        "tag_name": tag_name,
        "name": name or tag_name,
        "body": body,
        "draft": draft,
        "prerelease": prerelease,
    }
    if target:
        payload["target_commitish"] = target
    return _post(f"/repos/{owner}/{repo}/releases", payload)


@mcp.tool()
def list_tags(owner: str, repo: str) -> dict | list | str:
    """List tags in a repository.

    Args:
        owner: Repository owner.
        repo: Repository name.
    """
    return _get(f"/repos/{owner}/{repo}/tags")


# ───────────────────────────────────────────────────────────────────
# Organizations
# ───────────────────────────────────────────────────────────────────


@mcp.tool()
def list_orgs() -> dict | list | str:
    """List organizations the authenticated user belongs to."""
    return _get("/user/orgs")


@mcp.tool()
def get_org(org: str) -> dict | list | str:
    """Get details of an organization.

    Args:
        org: Organization name.
    """
    return _get(f"/orgs/{org}")


@mcp.tool()
def list_org_repos(org: str, limit: int = 20, page: int = 1) -> dict | list | str:
    """List repositories in an organization.

    Args:
        org: Organization name.
        limit: Results per page. Default 20.
        page: Page number. Default 1.
    """
    return _get(f"/orgs/{org}/repos", limit=limit, page=page)


# ───────────────────────────────────────────────────────────────────
# Users
# ───────────────────────────────────────────────────────────────────


@mcp.tool()
def get_user(username: str) -> dict | list | str:
    """Get a user's profile.

    Args:
        username: The username to look up.
    """
    return _get(f"/users/{username}")


@mcp.tool()
def list_user_repos(username: str, limit: int = 20, page: int = 1) -> dict | list | str:
    """List a user's repositories.

    Args:
        username: The username.
        limit: Results per page. Default 20.
        page: Page number. Default 1.
    """
    return _get(f"/users/{username}/repos", limit=limit, page=page)


# ───────────────────────────────────────────────────────────────────
# Milestones
# ───────────────────────────────────────────────────────────────────


@mcp.tool()
def list_milestones(owner: str, repo: str, state: str = "open") -> dict | list | str:
    """List milestones in a repository.

    Args:
        owner: Repository owner.
        repo: Repository name.
        state: Filter by state: "open", "closed", or "all". Default "open".
    """
    return _get(f"/repos/{owner}/{repo}/milestones", state=state)


@mcp.tool()
def create_milestone(
    owner: str, repo: str, title: str, description: str = ""
) -> dict | list | str:
    """Create a milestone in a repository.

    Args:
        owner: Repository owner.
        repo: Repository name.
        title: Milestone title.
        description: Milestone description.
    """
    return _post(
        f"/repos/{owner}/{repo}/milestones",
        {
            "title": title,
            "description": description,
        },
    )


if __name__ == "__main__":
    mcp.run()