diff --git a/src/familiar/charutil.py b/src/familiar/charutil.py index 0872a19..ff9b5a3 100644 --- a/src/familiar/charutil.py +++ b/src/familiar/charutil.py @@ -37,6 +37,16 @@ def _names() -> dict[int, str | None]: return {c["character_id"]: c["character_name"] for c in auth.list_characters()} +def character_targets(character: str | int | None) -> list[tuple[int, str | None]]: + """``(character_id, name)`` pairs for one character or all — no token needed. + + For public lookups keyed by character id (e.g. zKillboard) that still want + to default to the logged-in characters. + """ + names = _names() + return [(cid, names.get(cid)) for cid in _targets(character)] + + async def run_per_character(character: str | int | None, fetch: CharFetcher) -> dict[str, Any]: """Run ``fetch`` for one character or all, returning a uniform list.""" names = _names() diff --git a/src/familiar/killmails.py b/src/familiar/killmails.py new file mode 100644 index 0000000..97521d7 --- /dev/null +++ b/src/familiar/killmails.py @@ -0,0 +1,81 @@ +"""Read-only killmail tools: recent kills (character + corp) and zKill stats.""" + +from __future__ import annotations + +from typing import Any + +from fastmcp import FastMCP + +from . import zkill +from .charutil import character_targets, run_per_character, run_per_corp +from .esi import client + + +async def _enrich(refs: list[dict[str, Any]], limit: int) -> list[dict[str, Any]]: + """Turn killmail (id, hash) refs into trimmed detail for the most recent few.""" + out = [] + for ref in refs[: max(1, limit)]: + km = await client().get(f"/killmails/{ref['killmail_id']}/{ref['killmail_hash']}/") + victim = km.get("victim", {}) + out.append( + { + "killmail_id": km.get("killmail_id"), + "time": km.get("killmail_time"), + "solar_system_id": km.get("solar_system_id"), + "victim_ship_type_id": victim.get("ship_type_id"), + "victim_character_id": victim.get("character_id"), + "attacker_count": len(km.get("attackers") or []), + } + ) + return out + + +async def get_killmails(character: str | int | None = None, limit: int = 10) -> dict[str, Any]: + """Recent killmails for one character or all (most recent enriched with detail).""" + + async def fetch(token: str, cid: int) -> dict[str, Any]: + refs = await client().get_all_pages(f"/characters/{cid}/killmails/recent/", token=token) + return {"count": len(refs), "recent": await _enrich(refs, limit)} + + return await run_per_character(character, fetch) + + +async def get_corp_killmails(character: str | int | None = None, limit: int = 10) -> dict[str, Any]: + """Recent corp killmails (needs a corp role) for one character's corp or all.""" + + async def fetch(token: str, corp_id: int, cid: int) -> dict[str, Any]: + refs = await client().get_all_pages( + f"/corporations/{corp_id}/killmails/recent/", token=token + ) + return {"count": len(refs), "recent": await _enrich(refs, limit)} + + return await run_per_corp(character, fetch) + + +async def get_zkill_stats(character: str | int | None = None) -> dict[str, Any]: + """PvP stats from zKillboard for one character or all (public, no auth).""" + results = [] + for cid, name in character_targets(character): + stats = await zkill.client().get(f"/stats/characterID/{cid}/") or {} + results.append( + { + "character_id": cid, + "character_name": name, + "ships_destroyed": stats.get("shipsDestroyed"), + "ships_lost": stats.get("shipsLost"), + "isk_destroyed": stats.get("iskDestroyed"), + "isk_lost": stats.get("iskLost"), + "danger_ratio": stats.get("dangerRatio"), + "gang_ratio": stats.get("gangRatio"), + } + ) + return {"characters": results} + + +TOOLS = [get_killmails, get_corp_killmails, get_zkill_stats] + + +def register(mcp: FastMCP) -> None: + """Register the killmail tools onto the given server.""" + for fn in TOOLS: + mcp.tool(fn) diff --git a/src/familiar/server.py b/src/familiar/server.py index de3a464..d74cf2b 100644 --- a/src/familiar/server.py +++ b/src/familiar/server.py @@ -11,7 +11,17 @@ import datetime as _dt from fastmcp import FastMCP -from . import __version__, account, industry, market, structures, trade, universe, writes +from . import ( + __version__, + account, + industry, + killmails, + market, + structures, + trade, + universe, + writes, +) mcp: FastMCP = FastMCP( name="familiar", @@ -54,3 +64,6 @@ structures.register(mcp) # Phase 4: write-action safety (the gate every write tool goes through). writes.register(mcp) + +# Phase 4: killmails + zKillboard (read-only). +killmails.register(mcp) diff --git a/src/familiar/zkill.py b/src/familiar/zkill.py new file mode 100644 index 0000000..08db3c8 --- /dev/null +++ b/src/familiar/zkill.py @@ -0,0 +1,46 @@ +"""Tiny client for the public zKillboard API (kill/PvP stats). + +Same shape as the ESI client so tests can inject an ``httpx.MockTransport``. +""" + +from __future__ import annotations + +import os +from typing import Any + +import httpx + +from . import __version__ + +ZKILL_BASE = os.environ.get("FAMILIAR_ZKILL_BASE", "https://zkillboard.com/api") +USER_AGENT = f"familiar/{__version__} (+https://git.hiddenden.cafe/Hiddenden/familiar)" + + +class ZkillClient: + def __init__(self, base_url: str = ZKILL_BASE, *, transport=None) -> None: + self._client = httpx.AsyncClient( + base_url=base_url, + transport=transport, + timeout=30.0, + headers={"User-Agent": USER_AGENT, "Accept": "application/json"}, + ) + + async def get(self, path: str) -> Any: + resp = await self._client.get(path) + resp.raise_for_status() + return resp.json() if resp.content else None + + +_default: ZkillClient | None = None + + +def client() -> ZkillClient: + global _default + if _default is None: + _default = ZkillClient() + return _default + + +def set_client(new: ZkillClient | None) -> None: + global _default + _default = new diff --git a/tests/test_killmails.py b/tests/test_killmails.py new file mode 100644 index 0000000..5115ab6 --- /dev/null +++ b/tests/test_killmails.py @@ -0,0 +1,96 @@ +"""Killmail tools (ESI char + corp) and zKillboard stats.""" + +import asyncio + +import httpx +import pytest + +from conftest import route +from familiar import auth, killmails, zkill +from familiar.server import mcp + +CHARS = [ + {"character_id": 99, "character_name": "Alpha", "scopes": []}, + {"character_id": 100, "character_name": "Beta", "scopes": []}, +] + + +@pytest.fixture +def fake_auth(monkeypatch): + monkeypatch.setattr(auth, "list_characters", lambda: list(CHARS)) + + async def get_auth(cid, **kw): + return f"token-{cid}", cid + + monkeypatch.setattr(auth, "get_auth", get_auth) + + def resolve(identifier): + for c in CHARS: + if ( + str(c["character_id"]) == str(identifier) + or (c["character_name"] or "").lower() == str(identifier).lower() + ): + return c["character_id"] + raise auth.NotAuthenticated("no such character") + + monkeypatch.setattr(auth, "resolve_character", resolve) + return CHARS + + +def test_get_killmails_enriches(mock_esi, fake_auth): + detail = { + "killmail_id": 1, + "killmail_time": "2026-06-01T00:00:00Z", + "solar_system_id": 30000142, + "victim": {"ship_type_id": 670, "character_id": 99}, + "attackers": [{}, {}, {}], + } + fixtures = { + "/killmails/recent/": [{"killmail_id": 1, "killmail_hash": "abc"}], + "/killmails/": detail, + } + mock_esi(lambda r: route(r, fixtures)) + res = asyncio.run(killmails.get_killmails(character="Alpha", limit=5)) + entry = res["characters"][0] + assert entry["count"] == 1 + assert entry["recent"][0]["attacker_count"] == 3 + assert entry["recent"][0]["solar_system_id"] == 30000142 + + +def test_get_corp_killmails(mock_esi, fake_auth): + detail = {"killmail_id": 9, "victim": {"ship_type_id": 1}, "attackers": [{}]} + fixtures = { + "/killmails/recent/": [{"killmail_id": 9, "killmail_hash": "h"}], + "/killmails/": detail, + "/characters/": {"corporation_id": 2000}, + } + mock_esi(lambda r: route(r, fixtures)) + res = asyncio.run(killmails.get_corp_killmails(character=99)) + assert res["characters"][0]["corporation_id"] == 2000 + assert res["characters"][0]["count"] == 1 + + +@pytest.fixture +def mock_zkill(): + def _make(payload): + zkill.set_client( + zkill.ZkillClient( + transport=httpx.MockTransport(lambda r: httpx.Response(200, json=payload)) + ) + ) + + yield _make + zkill.set_client(None) + + +def test_zkill_stats(fake_auth, mock_zkill): + mock_zkill({"shipsDestroyed": 100, "shipsLost": 5, "dangerRatio": 95}) + res = asyncio.run(killmails.get_zkill_stats()) + assert len(res["characters"]) == 2 + assert res["characters"][0]["ships_destroyed"] == 100 + assert res["characters"][0]["danger_ratio"] == 95 + + +def test_killmail_tools_registered(): + tools = {t.name for t in asyncio.run(mcp.list_tools())} + assert {"get_killmails", "get_corp_killmails", "get_zkill_stats"} <= tools