marine_mcp.py

"""Marine MCP bridge to the school Marine Data FastAPI."""
from __future__ import annotations
import os, re, time
from datetime import date as calendar_date, timedelta
from typing import Any
import httpx
from mcp.server.mcpserver import MCPServer
from fisheries_hf import (
    analyze_and_export,
    download_dataset_file,
    local_dataset_tree,
    merge_local_with_remote_tree,
)

API_URL = os.environ.get("MARINE_API_URL", "").strip().rstrip("/")
if not API_URL:
    raise RuntimeError("MARINE_API_URL is not configured")
# Browser-facing download URLs must not expose the private Marine API address.
# On the school deployment this is set to the fixed UI proxy prefix, e.g.
# https://subway-unbiased-barcode.ngrok-free.dev/marine.  If it is unset we
# retain the old local URL for private/operator-only deployments.
PUBLIC_BASE_URL = (
    os.environ.get("MARINE_PUBLIC_BASE_URL")
    or os.environ.get("PUBLIC_BASE_URL")
    or ""
).strip().rstrip("/")


def _download_url(path: str) -> str:
    path = str(path or "").strip()
    if not path.startswith("/download/"):
        return ""
    return f"{PUBLIC_BASE_URL or API_URL}{path}"

mcp = MCPServer(
    "Marine Data",
    instructions=(
        "Gateway to the user's school Marine Data Server. "
        "Use health/domains/status for live state. "
        "Use marine_query and marine_subset for real data retrieval. "
        "School-server Ocean data and Hugging Face fisheries data are separate data planes. "
        "Search both configured fisheries repositories and preserve repository provenance. "
        "Use fisheries_analyze_export with both repository and path for actual CSV/TSV/ZIP content, filtering, statistics and CSV export. "
        "Never invent files or values."
    ),
)

def _get(path: str) -> dict[str, Any]:
    with httpx.Client(timeout=60.0, follow_redirects=True) as client:
        r = client.get(f"{API_URL}{path}")
        r.raise_for_status()
        return r.json()

def _post(path: str, payload: dict[str, Any]) -> dict[str, Any]:
    with httpx.Client(timeout=30.0, follow_redirects=True) as client:
        response = client.post(
            f"{API_URL}{path}",
            json=payload,
        )

        if response.is_error:
            try:
                body = response.json()
                detail = (
                    body.get("detail")
                    if isinstance(body, dict)
                    else None
                )
            except Exception:
                detail = None

            if not detail:
                detail = (
                    response.text.strip()
                    or response.reason_phrase
                )

            return {
                "status": "error",
                "http_status": response.status_code,
                "detail": detail,
            }

        return response.json()

def _domain(value: str) -> str:
    value = value.strip().lower()
    if value not in {"ocean", "tuna", "squid"}:
        raise ValueError("domain must be one of: ocean, tuna, squid")
    return value


def _norm_domain(value: str) -> str:
    value = value.strip().lower()
    if value not in {"ocean", "tuna", "squid"}:
        raise ValueError("domain must be one of: ocean, tuna, squid")
    return value


@mcp.tool()
def marine_health() -> dict[str, Any]:
    """Check whether the school Marine Data Server is reachable."""
    return _get("/health")

@mcp.tool()
def marine_domains() -> dict[str, Any]:
    """Return live overview for ocean, tuna and squid."""
    return _get("/domains")

@mcp.tool()
def marine_status(domain: str = "ocean") -> dict[str, Any]:
    """Return detailed live status for one data center."""
    d = _domain(domain)
    return _get("/status" if d == "ocean" else f"/status/{d}")

@mcp.tool()
def marine_catalog() -> dict[str, Any]:
    'Return the live Ocean data catalog from the school server.'
    return _get("/catalog")


@mcp.tool()
def marine_query(
    date: str,
    variable: str,
    source: str,
    domain: str = "ocean",
) -> dict[str, Any]:
    'Check whether a source/variable/date exists on the school Ocean server.'
    return _post(
        "/data/query",
        {
            "domain": _norm_domain(domain),
            "source": source.strip().lower(),
            "date": date.strip(),
            "variable": variable.strip().lower(),
        },
    )

@mcp.tool()
def marine_subset(
    date: str,
    lon_min: float,
    lon_max: float,
    lat_min: float,
    lat_max: float,
    variable: str,
    source: str,
    domain: str = "ocean",
    depth: float | None = None,
) -> dict[str, Any]:
    'Create a NetCDF subset from any supported Ocean source.'
    payload = {
        "domain": _norm_domain(domain),
        "source": source.strip().lower(),
        "date": date.strip(),
        "variable": variable.strip().lower(),
        "lon_min": float(lon_min),
        "lon_max": float(lon_max),
        "lat_min": float(lat_min),
        "lat_max": float(lat_max),
        "format": "netcdf",
    }
    if depth is not None:
        payload["depth"] = float(depth)
    result = _post("/data/export", payload)
    path = result.get("download_path")
    if isinstance(path, str) and path.startswith("/download/"):
        result["download_url"] = _download_url(path)
    return result

@mcp.tool()
def marine_download(token: str) -> dict[str, Any]:
    """Convert an export token into a browser HTTPS download URL."""
    token = token.strip()
    if not re.fullmatch(r"[A-Za-z0-9_-]{20,160}", token):
        raise ValueError("invalid download token")
    return {"download_url": _download_url(f"/download/{token}")}



@mcp.tool()
def marine_fisheries_catalog() -> dict:
    """Compatibility alias for the live Hugging Face squid catalog."""
    return fisheries_catalog("squid")


@mcp.tool()
def marine_export(
    date: str,
    lon_min: float,
    lon_max: float,
    lat_min: float,
    lat_max: float,
    variable: str,
    source: str,
    format: str = "netcdf",
    domain: str = "ocean",
    depth: float | None = None,
) -> dict[str, Any]:
    'Export Ocean data as netcdf/csv/xlsx/json/geotiff/png.'
    payload = {
        "domain": _norm_domain(domain),
        "source": source.strip().lower(),
        "date": date.strip(),
        "variable": variable.strip().lower(),
        "lon_min": float(lon_min),
        "lon_max": float(lon_max),
        "lat_min": float(lat_min),
        "lat_max": float(lat_max),
        "format": format.strip().lower(),
    }
    if depth is not None:
        payload["depth"] = float(depth)
    result = _post("/data/export", payload)
    path = result.get("download_path")
    if isinstance(path, str) and path.startswith("/download/"):
        result["download_url"] = _download_url(path)
    return result


def _range_export_limit() -> int:
    """Return a bounded, operator-configurable batch size.

    This deliberately never accepts a user/model supplied limit: a malformed
    request must not fan out into thousands of school-server export jobs.
    """
    try:
        configured = int(os.environ.get("MARINE_EXPORT_RANGE_MAX_DAYS", "31"))
    except ValueError:
        configured = 31
    return max(1, min(configured, 31))


@mcp.tool()
def marine_export_range(
    start_date: str,
    end_date: str,
    lon_min: float,
    lon_max: float,
    lat_min: float,
    lat_max: float,
    variable: str,
    source: str,
    format: str = "netcdf",
    domain: str = "ocean",
    depth: float | None = None,
) -> dict[str, Any]:
    """Export one Ocean request for every day in an inclusive date range.

    The school Ocean API is single-day by design.  This safe batch endpoint
    makes multi-day downloads explicit and returns one real download URL per
    successful day.  Ranges are capped at 31 days to protect the server; a
    longer period must be requested as calendar-month batches.
    """
    try:
        start = calendar_date.fromisoformat(str(start_date).strip())
        end = calendar_date.fromisoformat(str(end_date).strip())
    except ValueError as exc:
        raise ValueError("start_date and end_date must be YYYY-MM-DD") from exc
    if end < start:
        raise ValueError("end_date must not be earlier than start_date")
    total_days = (end - start).days + 1
    max_days = _range_export_limit()
    if total_days > max_days:
        return {
            "status": "error",
            "detail": (
                f"日期范围共 {total_days} 天，超过单次安全上限 {max_days} 天。"
                "请按不超过 31 天的自然月批次提交。"
            ),
            "start_date": start.isoformat(),
            "end_date": end.isoformat(),
            "max_days": max_days,
        }

    files: list[dict[str, Any]] = []
    failures: list[dict[str, Any]] = []
    current = start
    while current <= end:
        day = current.isoformat()
        result = marine_export(
            date=day,
            lon_min=lon_min,
            lon_max=lon_max,
            lat_min=lat_min,
            lat_max=lat_max,
            variable=variable,
            source=source,
            format=format,
            domain=domain,
            depth=depth,
        )
        if str(result.get("status") or "").lower() == "error":
            failures.append({"date": day, "detail": str(result.get("detail") or "export failed")})
        else:
            files.append({"date": day, **result})
        current += timedelta(days=1)

    status = "ok" if not failures else ("partial" if files else "error")
    return {
        "status": status,
        "start_date": start.isoformat(),
        "end_date": end.isoformat(),
        "requested_days": total_days,
        "files": files,
        "failures": failures,
    }

# ============================================================================
# Hugging Face fisheries data bridge
# ============================================================================
HF_SQUID_DATASET_REPO = (
    os.environ.get("HF_SQUID_DATASET_REPO")
    or os.environ.get("HF_DATASET_REPO")
    or "globalsquiddatabase/squid_dataset"
).strip()
HF_TUNA_DATASET_REPO = (
    os.environ.get("HF_TUNA_DATASET_REPO")
    or "globalsquiddatabase/Tuna-Fisheries-Dataset"
).strip()
HF_DATASET_REPOS = {
    "squid": HF_SQUID_DATASET_REPO,
    "tuna": HF_TUNA_DATASET_REPO,
}
HF_DATASET_REPO = HF_SQUID_DATASET_REPO

_HF_TREE_CACHE: dict[str, dict[str, Any]] = {}

_SQUID_CATALOG = [
    {
        "source": "FAO FishStatJ",
        "resource": "全球柔鱼科捕捞量",
        "variables": ["catch", "species", "country_or_area", "year"],
        "time_range": "1998-2024（正式资源清单口径；实际标准层以live inventory为准）",
        "spatial_scale": "全球；无经纬度网格",
        "temporal_scale": "年",
        "science_uses": ["长期捕捞量变化", "国家/地区贡献结构", "物种捕捞组成变化"],
        "caveats": ["不能用于精细渔场位置分析", "没有努力量时不能直接得到CPUE"],
    },
    {
        "source": "Sea Around Us",
        "resource": "全球柔鱼科重建捕捞量",
        "variables": ["reconstructed_catch", "species", "area", "year"],
        "time_range": "1950-2019",
        "spatial_scale": "0.5°×0.5°",
        "temporal_scale": "年",
        "science_uses": ["历史空间捕捞格局", "渔场重心变化", "区域热点年代际变化"],
        "caveats": ["属于重建数据", "使用时必须说明重建口径"],
    },
    {
        "source": "SPRFMO",
        "resource": "南太平洋捕捞量与努力量",
        "variables": ["catch", "effort", "year", "grid"],
        "time_range": "2007-2021（后续补充以live inventory为准）",
        "spatial_scale": "5°×5°",
        "temporal_scale": "年/仓库后续标准层可能含月",
        "science_uses": ["区域作业格局", "捕捞强度变化", "重算CPUE后做相对丰度分析"],
        "caveats": ["CPUE必须用总catch÷总effort重算", "不同努力量单位不可直接相加"],
    },
    {
        "source": "WCPFC",
        "resource": "中西太平洋月度捕捞数据",
        "variables": ["catch", "year", "month", "grid", "coverage"],
        "time_range": "1967-2024",
        "spatial_scale": "1°×1°",
        "temporal_scale": "月",
        "science_uses": ["月尺度捕捞热点", "渔场季节迁移", "与SST/锋面/ENSO做时空匹配"],
        "caveats": ["需结合coverage解释缺测", "缺测不能直接解释为零捕捞"],
    },
    {
        "source": "RAM Legacy",
        "resource": "茎柔鱼资源评估数据",
        "variables": ["catch", "biomass", "recruitment", "CPUE"],
        "time_range": "1950-2024（不同种群覆盖不同）",
        "spatial_scale": "评估种群/stock",
        "temporal_scale": "年",
        "science_uses": ["资源量长期变化", "补充量变化", "资源状态与捕捞压力分析"],
        "caveats": ["不同评估模型单位/标准化口径不同", "跨种群比较前需统一数据字典"],
    },
    {
        "source": "Global Fishing Watch",
        "resource": "全球AIS表观渔船作业努力量",
        "variables": ["apparent_fishing_hours", "vessel_presence", "flag", "gear_type"],
        "time_range": "2012-2024",
        "spatial_scale": "0.1°×0.1°",
        "temporal_scale": "月",
        "science_uses": ["渔船活动强度", "作业努力热点迁移", "与渔获/CPUE联合分析捕捞压力"],
        "caveats": ["AIS+模型推断的表观努力量", "不能等同于捕捞量、日志努力量或资源丰度"],
    },
    {
        "source": "VIIRS VBD",
        "resource": "夜光船探测三变量",
        "variables": ["n_detect", "avg_rade9", "pct_detect"],
        "time_range": "2017-2024",
        "spatial_scale": "15 arcsec 原始；仓库可能含1°标准层",
        "temporal_scale": "月",
        "science_uses": ["夜光作业船热点", "灯光强度与探测稳定性", "补充AIS不足区的活动证据"],
        "caveats": ["夜光探测不是捕捞量", "必须结合cvg评估观测机会"],
    },
    {
        "source": "VIIRS CVG",
        "resource": "卫星覆盖次数/观测机会",
        "variables": ["cvg"],
        "time_range": "2017-2024",
        "spatial_scale": "15 arcsec",
        "temporal_scale": "月",
        "science_uses": ["夜光质量控制", "覆盖偏差校正", "区域/月际可比性评估"],
        "caveats": ["cvg不是渔船活动量", "不能当作捕捞努力量"],
    },
]

_TUNA_SOURCE_TERMS = {
    "WCPFC": ["wcpfc"],
    "IATTC": ["iattc"],
    "ICCAT": ["iccat"],
    "IOTC": ["iotc"],
    "CCSBT": ["ccsbt"],
    "FAO": ["fao"],
    "GFW": ["global fishing watch", "gfw"],
}

_DOMAIN_TERMS = {
    "squid": [
        "柔鱼", "鱿鱼", "squid", "ommastre", "dosidicus", "illex", "todarodes",
        "sprfmo", "npfc", "ram legacy", "viirs", "vbd", "sea around", "sea_around", "gfw",
    ],
    "tuna": [
        "金枪鱼", "tuna", "wcpfc", "iattc", "iccat", "iotc", "ccsbt",
        "yellowfin", "bigeye", "skipjack", "albacore", "bluefin", "yft", "bet", "skj",
    ],
}

def _hf_headers() -> dict[str, str]:
    token = os.environ.get("HF_TOKEN", "").strip()
    return {"Authorization": f"Bearer {token}"} if token else {}

def _hf_tree(repo: str, force: bool = False) -> list[dict[str, Any]]:
    repo = repo.strip()
    local_items = local_dataset_tree(repo)
    now = time.time()
    cache = _HF_TREE_CACHE.get(repo) or {}
    if (
        not force
        and now - float(cache.get("ts") or 0) < 300
        and cache.get("items")
    ):
        return merge_local_with_remote_tree(cache["items"], repo)

    next_url = f"https://huggingface.co/api/datasets/{repo}/tree/main"
    params: dict[str, Any] | None = {
        "recursive": "true",
        "expand": "false",
        "limit": 1000,
    }
    items: list[dict[str, Any]] = []
    pages = 0
    try:
        with httpx.Client(timeout=30.0, follow_redirects=True) as client:
            while next_url and pages < 50:
                r = client.get(next_url, params=params, headers=_hf_headers())
                params = None
                pages += 1
                if r.status_code in {401, 403}:
                    raise RuntimeError(
                        f"无法读取 Hugging Face Dataset {repo}。请确认 Space Secret 中存在具有 Dataset 读取权限的 HF_TOKEN，"
                        "且运行时配置已将 HF_TOKEN 传给 marine MCP 子进程。"
                    )
                if r.status_code >= 400:
                    raise RuntimeError(
                        f"Hugging Face Dataset tree request failed: HTTP {r.status_code} ({repo}): {r.text[:300]}"
                    )
                data = r.json()
                if not isinstance(data, list):
                    raise RuntimeError(f"Hugging Face Dataset tree returned an unexpected response: {repo}")
                items.extend(x for x in data if isinstance(x, dict))
                next_url = (r.links.get("next") or {}).get("url")
    except Exception:
        # A network outage must not hide the school mirror, but only the
        # successfully fetched HF tree is authoritative when it is reachable.
        if local_items:
            return merge_local_with_remote_tree([], repo)
        raise

    if next_url:
        raise RuntimeError("Hugging Face Dataset 文件树超过在线分页安全上限。")
    _HF_TREE_CACHE[repo] = {"ts": now, "items": items}
    return merge_local_with_remote_tree(items, repo)


def _repos_for_domain(domain: str) -> list[tuple[str, str]]:
    d = (domain or "all").strip().lower()
    if d == "squid":
        return [("squid", HF_SQUID_DATASET_REPO)]
    if d == "tuna":
        return [("tuna", HF_TUNA_DATASET_REPO)]
    if d in {"all", "fisheries", "fishery"}:
        return list(HF_DATASET_REPOS.items())
    raise ValueError("domain must be one of: squid, tuna, all")


def _hf_files(
    domain: str = "all",
    force: bool = False,
) -> tuple[list[dict[str, Any]], dict[str, str]]:
    files: list[dict[str, Any]] = []
    errors: dict[str, str] = {}
    for repo_domain, repo in _repos_for_domain(domain):
        try:
            items = _files_only(_hf_tree(repo, force=force))
        except Exception as exc:
            errors[repo] = str(exc)[:500]
            continue
        for item in items:
            row = dict(item)
            row["repository"] = repo
            row["repository_domain"] = repo_domain
            files.append(row)
    return files, errors

def _files_only(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
    return [
        x for x in items
        if str(x.get("type") or "").lower() in {"file", "blob"}
        or (
            not str(x.get("type") or "").strip()
            and "path" in x
            and "size" in x
        )
    ]

def _human_bytes(value: Any) -> str:
    try:
        n = float(value or 0)
    except Exception:
        n = 0.0
    units = ["B", "KB", "MB", "GB", "TB"]
    i = 0
    while n >= 1024 and i < len(units) - 1:
        n /= 1024.0
        i += 1
    return f"{n:.2f} {units[i]}"

def _domain_match(path: str, domain: str) -> bool:
    d = (domain or "all").strip().lower()
    if d in {"all", "fisheries", "fishery"}:
        return True
    terms = _DOMAIN_TERMS.get(d)
    if not terms:
        raise ValueError("domain must be one of: squid, tuna, all")
    p = path.lower()
    return any(term in p for term in terms)

def _query_terms(query: str) -> list[str]:
    q = (query or "").strip().lower()
    aliases = {
        "柔鱼": ["柔鱼", "鱿鱼", "squid"],
        "鱿鱼": ["柔鱼", "鱿鱼", "squid"],
        "金枪鱼": ["金枪鱼", "tuna"],
        "捕捞量": ["捕捞", "catch"],
        "努力量": ["努力", "effort", "fishing_hours", "fishing hours"],
        "cpue": ["cpue"],
        "渔船": ["gfw", "vessel", "ais", "viirs", "vbd"],
        "夜光": ["viirs", "vbd", "cvg", "n_detect", "rade"],
        "资源评估": ["ram", "assessment", "biomass", "recruitment"],
    }
    terms = [q] if q else []
    for key, vals in aliases.items():
        if key in q:
            terms.extend(vals)
    for token in re.split(r"[\s,，/、;；]+", q):
        if len(token) >= 2:
            terms.append(token)
    out = []
    for t in terms:
        if t and t not in out:
            out.append(t)
    return out

@mcp.tool()
def fisheries_catalog(domain: str = "squid") -> dict[str, Any]:
    """Return fisheries resources and the scientific questions they can support."""
    d = (domain or "squid").strip().lower()
    if d not in {"squid", "tuna", "all"}:
        raise ValueError("domain must be one of: squid, tuna, all")

    result: dict[str, Any] = {
        "status": "ok",
        "repositories": [repo for _, repo in _repos_for_domain(d)],
        "data_plane": "Hugging Face Dataset",
        "important_distinction": (
            "HF fisheries Dataset is separate from the school-server tuna_data/squid_data task databases. "
            "Empty school-server task databases do not mean the HF fisheries Dataset is empty."
        ),
        "aggregation_rules": {
            "catch": "SUM over time/space; preserve units",
            "effort": "SUM only within compatible units",
            "CPUE": "recompute aggregated total catch / aggregated total effort; never average monthly CPUE",
            "GFW": "AIS/model-derived apparent fishing effort; not catch or stock abundance",
            "VIIRS": "night-light vessel activity evidence; use CVG for observation-opportunity QC",
        },
    }

    if d in {"squid", "all"}:
        result["squid_semantic_catalog"] = _SQUID_CATALOG

    try:
        items, repo_errors = _hf_files(d)
        live = []
        for x in items:
            path = str(x.get("path") or "")
            live.append({
                "path": path,
                "repository": x.get("repository", ""),
                "repository_domain": x.get("repository_domain", ""),
                "size_bytes": int(x.get("size") or 0),
                "size": _human_bytes(x.get("size") or 0),
            })
        result["repository_errors"] = repo_errors

        total_bytes = sum(x["size_bytes"] for x in live)
        result["live_inventory"] = {
            "matched_file_count": len(live),
            "matched_size_bytes": total_bytes,
            "matched_size": _human_bytes(total_bytes),
            "path_preview": live[:40],
            "preview_truncated": len(live) > 40,
        }

        if d in {"tuna", "all"}:
            groups = {}
            for source, terms in _TUNA_SOURCE_TERMS.items():
                matched = [x for x in live if any(t in x["path"].lower() for t in terms)]
                if matched:
                    groups[source] = {
                        "file_count": len(matched),
                        "size": _human_bytes(sum(x["size_bytes"] for x in matched)),
                        "examples": [x["path"] for x in matched[:6]],
                    }
            result["tuna_live_groups"] = groups
            result["tuna_note"] = (
                "Tuna availability is derived from the live HF repository tree. "
                "Do not use a planned download list as proof that a tuna dataset is already present."
            )
    except Exception as exc:
        result["live_inventory"] = {"status": "error", "detail": str(exc)}

    return result

@mcp.tool()
def fisheries_inventory(
    domain: str = "all",
    keyword: str | None = None,
    max_results: int = 80,
    refresh: bool = False,
    query: str | None = None,
    source: str | None = None,
) -> dict[str, Any]:
    """Inspect both live fisheries trees.

    Preferred arguments are ``domain`` (squid/tuna/all) and ``keyword``.
    ``query`` and ``source`` are accepted as compatibility aliases because
    some chat runtimes emit those names for inventory searches.
    """
    if query and not keyword:
        keyword = str(query).strip()
    if source:
        source_text = str(source).strip()
        source_lower = source_text.lower()
        if source_lower in {"squid", "tuna", "all", "fisheries", "fishery"}:
            domain = source_lower
        elif source_text == HF_SQUID_DATASET_REPO:
            domain = "squid"
        elif source_text == HF_TUNA_DATASET_REPO:
            domain = "tuna"
        elif not keyword:
            keyword = source_text
    try:
        items, repo_errors = _hf_files(domain, force=bool(refresh))
    except Exception as exc:
        return {"status": "error", "repositories": HF_DATASET_REPOS, "detail": str(exc)}

    d = (domain or "all").strip().lower()
    limit = max(1, min(int(max_results or 80), 200))
    qterms = _query_terms(keyword or "")

    matches = []
    for x in items:
        path = str(x.get("path") or "")
        if d not in {"all", "fisheries", "fishery"} and x.get("repository_domain") != d:
            continue
        plow = path.lower()
        if qterms and not any(t in plow for t in qterms):
            continue
        matches.append({
            "path": path,
            "repository": x.get("repository", ""),
            "repository_domain": x.get("repository_domain", ""),
            "size_bytes": int(x.get("size") or 0),
            "size": _human_bytes(x.get("size") or 0),
        })

    total_bytes = sum(x["size_bytes"] for x in matches)
    return {
        "status": "ok",
        "repositories": [repo for _, repo in _repos_for_domain(d)],
        "repository_errors": repo_errors,
        "branch": "main",
        "domain": d,
        "keyword": keyword,
        "matched_file_count": len(matches),
        "matched_size_bytes": total_bytes,
        "matched_size": _human_bytes(total_bytes),
        "results": matches[:limit],
        "results_truncated": len(matches) > limit,
        "cache_seconds": 300,
    }

@mcp.tool()
def fisheries_search(query: str, max_results: int = 40) -> dict[str, Any]:
    """Search real HF fisheries files by source/species/metric/path keywords."""
    q = (query or "").strip()
    if not q:
        raise ValueError("query is required")

    try:
        items, repo_errors = _hf_files("all")
    except Exception as exc:
        return {"status": "error", "repositories": HF_DATASET_REPOS, "detail": str(exc)}

    terms = _query_terms(q)
    scored = []
    for x in items:
        path = str(x.get("path") or "")
        plow = path.lower()
        score = sum(1 for t in terms if t in plow)
        if score:
            scored.append((
                score,
                {
                    "path": path,
                    "repository": x.get("repository", ""),
                    "repository_domain": x.get("repository_domain", ""),
                    "size_bytes": int(x.get("size") or 0),
                    "size": _human_bytes(x.get("size") or 0),
                },
            ))

    scored.sort(key=lambda z: (-z[0], z[1]["path"]))
    limit = max(1, min(int(max_results or 40), 100))
    return {
        "status": "ok",
        "repositories": list(HF_DATASET_REPOS.values()),
        "repository_errors": repo_errors,
        "query": q,
        "matched_file_count": len(scored),
        "results": [x for _, x in scored[:limit]],
        "results_truncated": len(scored) > limit,
    }


def _live_file(path: str, repository: str | None = None) -> dict[str, Any]:
    clean = str(path or "").strip().lstrip("/")
    if not clean:
        raise ValueError("path is required")
    selector = str(repository or "all").strip()
    lowered = selector.lower()
    if lowered in HF_DATASET_REPOS:
        domain = lowered
    elif selector in HF_DATASET_REPOS.values():
        domain = next(k for k, v in HF_DATASET_REPOS.items() if v == selector)
    elif lowered in {"", "all"}:
        domain = "all"
    else:
        raise ValueError("repository must be squid, tuna, all, or an exact configured repository id")
    items, repo_errors = _hf_files(domain)
    exact = [item for item in items if str(item.get("path") or "") == clean]
    if not exact:
        raise ValueError(
            "请求的文件不在所选 Hugging Face main 实时文件树中；"
            "请先使用 fisheries_search 或 fisheries_inventory 确认精确路径。"
        )
    if len(exact) > 1:
        repos = ", ".join(str(item.get("repository") or "") for item in exact)
        raise ValueError(f"同一路径存在于多个仓库（{repos}），请显式指定 repository。")
    item = exact[0]
    return {
        "path": clean,
        "size_bytes": int(item.get("size") or 0),
        "repository": str(item.get("repository") or ""),
        "repository_domain": str(item.get("repository_domain") or ""),
        "repository_errors": repo_errors,
    }


@mcp.tool()
def fisheries_analyze_export(
    path: str,
    repository: str | None = None,
    year: int | None = None,
    lon_min: float | None = None,
    lon_max: float | None = None,
    lat_min: float | None = None,
    lat_max: float | None = None,
    metric_columns: str | None = None,
    max_rows: int = 2_000_000,
) -> dict[str, Any]:
    """Read a validated HF fisheries CSV/TSV/ZIP, analyze/filter it, and export a real CSV.

    The path must exactly match one configured Dataset live tree. Repository
    may be squid, tuna, or an exact configured repository id. The tool
    accepts optional year and bounding-box filters, reports actual columns,
    scanned/matched rows, missing values, exact duplicates, monthly counts and
    annual metric sums, then returns a tokenized HTTPS download URL.  It never
    accepts arbitrary URLs, repositories, shell commands, or local paths.
    """
    try:
        item = _live_file(path, repository=repository)
        local_path, revision = download_dataset_file(
            item["path"],
            item["size_bytes"],
            repository=item["repository"],
        )
        return analyze_and_export(
            local_path,
            dataset_path=item["path"],
            revision=revision,
            repository=item["repository"],
            year=year,
            lon_min=lon_min,
            lon_max=lon_max,
            lat_min=lat_min,
            lat_max=lat_max,
            metric_columns=metric_columns,
            max_rows=max_rows,
        )
    except Exception as exc:
        return {
            "status": "error",
            "repository": str(repository or "all"),
            "path": str(path or ""),
            "detail": str(exc),
        }

@mcp.tool()
def fisheries_data_rules() -> dict[str, Any]:
    """Return fisheries aggregation and interpretation rules."""
    return {
        "catch": {
            "aggregation": "sum",
            "rule": "时间/空间聚合采用求和，并保留原始单位。",
        },
        "effort": {
            "aggregation": "sum",
            "rule": "时间/空间聚合采用求和；fishing hours 与 vessel-days 等不同单位不可直接相加。",
        },
        "CPUE": {
            "aggregation": "recompute",
            "rule": "CPUE = 聚合后的总catch / 聚合后的总effort；禁止直接平均月度或格点CPUE。",
        },
        "GFW": {
            "rule": "apparent fishing hours 是AIS+模型推断的表观作业努力量，不等同于真实捕捞量或资源丰度。",
        },
        "VIIRS": {
            "rule": "n_detect/avg_rade9/pct_detect是夜光船活动指标；cvg是观测机会/覆盖质量控制变量。",
        },
        "missing_time": {
            "rule": "不得把月度/年度数据伪装成逐日数据；缺失月份必须显式报告。",
        },
    }


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