"""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()