routes/datasets.py

from __future__ import annotations

# P0 transitional router module.
# Route functions remain behavior-compatible while ui_server.py becomes
# an application assembly layer. Runtime dependencies are injected once
# during registration to avoid circular imports.

def _inject(context):
    globals().update(context)

async def sidebar_datasets(request: Request, refresh: bool = False):
    await resolve_request_user(request)

    hf_error = ""
    catalog_error = ""
    live_tree_items = []
    live_files = []
    marine_catalog = {}
    repo_errors = {}

    try:
        live_files, repo_errors = await hf_all_live_files(force=refresh)
        live_tree_items = live_files
        hf_error = "; ".join(f"{repo}: {msg}" for repo, msg in repo_errors.items())
    except Exception as exc:
        hf_error = str(exc)[:500]

    try:
        marine_catalog = await _marine_api_get("/catalog")
    except Exception as exc:
        catalog_error = str(exc)[:500]

    fisheries_sources = []
    for source, aliases in _HF_SOURCE_ALIASES.items():
        category, category_zh = _HF_SOURCE_CATEGORIES.get(
            source,
            ("general", "综合渔业数据"),
        )
        matched = [
            item for item in live_files
            if any(alias in item["path_lower"] for alias in aliases)
        ]
        matched_aliases = [
            alias for alias in aliases
            if any(alias in item["path_lower"] for item in matched)
        ]
        fisheries_sources.append({
            "key": source.lower().replace(" ", "_"),
            "name": source,
            "name_zh": _HF_SOURCE_NAMES_ZH.get(source, source),
            "category": category,
            "category_zh": category_zh,
            "status": "available" if matched else "not_found",
            "file_count": len(matched),
            "size_bytes": sum(item["size_bytes"] for item in matched),
            "examples": [item["path"] for item in matched[:4]],
            "matched_aliases": matched_aliases,
            "metadata": {
                "repository": "、".join(sorted({
                    item.get("repository", "") for item in matched
                    if item.get("repository")
                })) or "未命中",
                "branch": "main",
                "file_count": str(len(matched)),
                "total_size": _human_bytes(sum(item["size_bytes"] for item in matched)),
                "inventory_source": "Hugging Face 实时文件树",
                "matching_rule": (
                    "路径命中：" + "、".join(matched_aliases)
                    if matched_aliases
                    else "当前文件树未命中该来源别名"
                ),
            },
            "query_prompt": (
                f"查询 Hugging Face 正式数据集中 {source} 当前已经入库的数据，"
                "按数据类型说明可用于哪些研究；必须用 live inventory 核验"
            ),
        })

    tuna_files = [item for item in live_files if item.get("repository_domain") == "tuna"]
    squid_files = [item for item in live_files if item.get("repository_domain") == "squid"]

    ocean_sources = [
        {
            "key": key,
            "name": name,
            "name_zh": name_zh,
            "variables": list(variables),
            "variable_labels": {
                variable: _OCEAN_VARIABLE_NAMES_ZH.get(variable, variable)
                for variable in variables
            },
            "metadata": _catalog_metadata(
                _find_catalog_entry(marine_catalog, key)
            ),
            "status": "connected" if marine_catalog else "unverified",
            "query_prompt": f"查询 {name} 当前支持的数据变量、时间范围和空间分辨率",
        }
        for key, name, name_zh, variables in _OCEAN_CATALOG
    ]

    return {
        "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
        "ocean": {
            "status": "connected" if marine_catalog else "unavailable",
            "error": catalog_error,
            "sources": ocean_sources,
        },
        "fisheries": {
            "status": "connected" if live_files else "unavailable",
            "error": hf_error,
            "repository": HF_DATASET_REPO,
            "repositories": HF_DATASET_REPOS,
            "repository_errors": repo_errors,
            "tree_object_count": len(live_tree_items),
            "file_count": len(live_files),
            "size_bytes": sum(item["size_bytes"] for item in live_files),
            "tuna_file_count": len(tuna_files),
            "squid_file_count": len(squid_files),
            "available_source_count": sum(
                1 for item in fisheries_sources
                if item["status"] == "available"
            ),
            "missing_source_count": sum(
                1 for item in fisheries_sources
                if item["status"] == "not_found"
            ),
            "sources": fisheries_sources,
        },
    }


async def sidebar_dataset_detail(
    group: str,
    source_key: str,
    request: Request,
    refresh: bool = False,
):
    """Return one dataset's current evidence, not just its card summary."""
    await resolve_request_user(request)
    group_key = group.strip().lower()
    source_key = source_key.strip().lower()
    checked_at = datetime.now().astimezone().isoformat(timespec="seconds")

    if group_key == "ocean":
        match = next(
            (item for item in _OCEAN_CATALOG if item[0] == source_key),
            None,
        )
        if not match:
            raise HTTPException(404, "unknown Ocean source")
        key, name, name_zh, variables = match
        paths = ("/catalog", "/status/ocean", "/domains")
        responses = await asyncio.gather(
            *(_marine_api_get(path) for path in paths),
            return_exceptions=True,
        )
        metadata: dict[str, str] = {}
        provenance = []
        errors = []
        source_entry_found = False
        for path, response in zip(paths, responses):
            if isinstance(response, Exception):
                errors.append(f"{path}: {str(response)[:240]}")
                continue
            provenance.append(f"学校 Marine API {path}")
            entry = _find_catalog_entry(response, key)
            if entry:
                source_entry_found = True
                for field, value in _catalog_metadata(entry).items():
                    metadata.setdefault(field, value)

        reference = _OCEAN_SOURCE_DETAILS.get(key, {})
        metadata.update({
            "data_plane": "学校 Ocean Marine Server",
            "source_key": key,
            "variable_count": str(len(variables)),
            "supported_formats": "NetCDF、CSV、XLSX、JSON、GeoTIFF、PNG",
            "availability_check": "按日期、变量调用 /data/query 实时核验",
            "detail_checked_at": checked_at,
        })
        completeness = _metadata_completeness(metadata)
        return {
            "group": "Ocean",
            "key": key,
            "name": name,
            "name_zh": name_zh,
            "status": "connected" if provenance else "unverified",
            "variables": list(variables),
            "variable_labels": {
                variable: _OCEAN_VARIABLE_NAMES_ZH.get(variable, variable)
                for variable in variables
            },
            "metadata": metadata,
            "reference": reference,
            "provenance": provenance,
            "metadata_completeness": completeness,
            "missing_fields": completeness["missing_fields"],
            "source_entry_found": source_entry_found,
            "errors": errors,
            "query_prompt": f"查询 {name} 当前支持的数据变量、时间范围和空间分辨率",
        }

    if group_key == "fisheries":
        match = next(
            (
                (name, aliases)
                for name, aliases in _HF_SOURCE_ALIASES.items()
                if name.lower().replace(" ", "_") == source_key
            ),
            None,
        )
        if not match:
            raise HTTPException(404, "unknown Fisheries source")
        name, aliases = match
        category, category_zh = _HF_SOURCE_CATEGORIES.get(
            name,
            ("general", "综合渔业数据"),
        )
        try:
            files, repo_errors = await hf_all_live_files(force=refresh)
            matched = [
                item for item in files
                if any(alias in item["path_lower"] for alias in aliases)
            ]
            error = "; ".join(f"{repo}: {msg}" for repo, msg in repo_errors.items())
        except Exception as exc:
            matched = []
            error = str(exc)[:500]
        matched_aliases = [
            alias for alias in aliases
            if any(alias in item["path_lower"] for item in matched)
        ]
        extension_counts: dict[str, int] = {}
        directories = set()
        for item in matched:
            suffix = Path(item["path"]).suffix.lower() or "无扩展名"
            extension_counts[suffix] = extension_counts.get(suffix, 0) + 1
            parts = Path(item["path"]).parts
            if len(parts) > 1:
                directories.add("/".join(parts[:2]))
        total_size = sum(item["size_bytes"] for item in matched)
        completeness = _metadata_completeness({})
        return {
            "group": "Fisheries",
            "key": source_key,
            "name": name,
            "name_zh": _HF_SOURCE_NAMES_ZH.get(name, name),
            "category": category,
            "category_zh": category_zh,
            "status": "available" if matched else "not_found",
            "variables": [],
            "variable_labels": {},
            "metadata": {
                "data_plane": "Hugging Face Dataset",
                "repository": "、".join(sorted({
                    item.get("repository", "") for item in matched
                    if item.get("repository")
                })) or "未命中",
                "branch": "main",
                "file_count": str(len(matched)),
                "total_size": _human_bytes(total_size),
                "file_types": "、".join(
                    f"{suffix} × {count}"
                    for suffix, count in sorted(extension_counts.items())
                ) or "实时目录未发现匹配文件",
                "directory_count": str(len(directories)),
                "inventory_source": "Hugging Face main 分支完整实时文件树",
                "matching_rule": (
                    "路径命中：" + "、".join(matched_aliases)
                    if matched_aliases
                    else "当前文件树未命中该来源别名"
                ),
                "inventory_interpretation": (
                    "当前仓库已收录"
                    if matched
                    else "当前 main 分支未收录；不是接口读取失败"
                ),
                "source_category": category_zh,
                "classification_basis": (
                    "按数据来源组织职责分类；具体文件中的物种仍以文件字段核验"
                ),
                "detail_checked_at": checked_at,
            },
            "reference": {
                "description": _FISHERIES_SOURCE_DETAILS.get(name, "渔业数据来源"),
                "data_shape": "时间、空间、物种和渔业指标以具体文件字段为准",
            },
            "provenance": [
                f"Hugging Face Dataset {repo}@main"
                for repo in sorted({
                    item.get("repository", "") for item in matched
                    if item.get("repository")
                })
            ],
            "metadata_completeness": completeness,
            "missing_fields": completeness["missing_fields"],
            "examples": [item["path"] for item in matched[:20]],
            "directories": sorted(directories)[:20],
            "error": error,
            "query_prompt": (
                f"查询 Hugging Face 正式数据集中 {name} 当前已经入库的数据，"
                "按数据类型说明可用于哪些研究；必须用 live inventory 核验"
            ),
        }

    raise HTTPException(404, "dataset group must be Ocean or Fisheries")


async def sidebar_ocean_availability(
    source_key: str,
    body: DatasetAvailabilityCheck,
    request: Request,
):
    await resolve_request_user(request)
    source_key = source_key.strip().lower()
    match = next(
        (item for item in _OCEAN_CATALOG if item[0] == source_key),
        None,
    )
    if not match:
        raise HTTPException(404, "unknown Ocean source")
    _key, name, name_zh, variables = match
    date = body.date.strip()
    variable = body.variable.strip().lower()
    try:
        datetime.strptime(date, "%Y-%m-%d")
    except ValueError as exc:
        raise HTTPException(400, "date must use YYYY-MM-DD") from exc
    if variable not in variables:
        raise HTTPException(
            400,
            f"variable must be one of: {', '.join(variables)}",
        )
    result = await _marine_api_post(
        "/data/query",
        {
            "domain": "ocean",
            "source": source_key,
            "date": date,
            "variable": variable,
        },
    )
    return {
        "source": source_key,
        "source_name": name,
        "source_name_zh": name_zh,
        "date": date,
        "variable": variable,
        "variable_zh": _OCEAN_VARIABLE_NAMES_ZH.get(variable, variable),
        "checked_at": datetime.now().astimezone().isoformat(timespec="seconds"),
        "result": result,
    }


async def sidebar_fisheries_files(
    source_key: str,
    request: Request,
    q: str = "",
    offset: int = 0,
    limit: int = 30,
    refresh: bool = False,
):
    await resolve_request_user(request)
    source_key = source_key.strip().lower()
    match = next(
        (
            (name, aliases)
            for name, aliases in _HF_SOURCE_ALIASES.items()
            if name.lower().replace(" ", "_") == source_key
        ),
        None,
    )
    if not match:
        raise HTTPException(404, "unknown Fisheries source")
    name, aliases = match
    offset = max(0, offset)
    limit = min(100, max(1, limit))
    query = q.strip().lower()[:160]
    files, repo_errors = await hf_all_live_files(force=refresh)
    source_files = [
        item for item in files
        if any(alias in item["path_lower"] for alias in aliases)
    ]
    filtered = [
        item for item in source_files
        if not query or query in item["path_lower"]
    ]
    page = filtered[offset:offset + limit]
    return {
        "source": name,
        "source_key": source_key,
        "query": q.strip()[:160],
        "source_total": len(source_files),
        "total": len(filtered),
        "offset": offset,
        "limit": limit,
        "has_more": offset + limit < len(filtered),
        "files": [
            {
                "path": item["path"],
                "repository": item.get("repository", ""),
                "size_bytes": item["size_bytes"],
                "size": _human_bytes(item["size_bytes"]),
                "extension": Path(item["path"]).suffix.lower() or "无扩展名",
            }
            for item in page
        ],
        "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
        "repository_errors": repo_errors,
    }


async def sidebar_dataset_quality(
    request: Request,
    refresh: bool = False,
):
    """Lightweight repository hygiene checks using the live HF file tree."""
    await resolve_request_user(request)
    files, repo_errors = await hf_all_live_files(force=refresh)
    extension_counts = Counter(
        Path(item["path"]).suffix.lower() or "无扩展名"
        for item in files
    )
    basename_groups: dict[str, list[dict]] = defaultdict(list)
    mapped_paths = set()
    all_aliases = tuple(
        alias
        for aliases in _HF_SOURCE_ALIASES.values()
        for alias in aliases
    )
    for item in files:
        basename_groups[Path(item["path"]).name.casefold()].append(item)
        if any(alias in item["path_lower"] for alias in all_aliases):
            mapped_paths.add(item["path"])

    duplicate_groups = [
        {
            "basename": Path(group[0]["path"]).name,
            "count": len(group),
            "paths": [item["path"] for item in group[:12]],
        }
        for group in basename_groups.values()
        if len(group) > 1
    ]
    duplicate_groups.sort(key=lambda item: (-item["count"], item["basename"]))
    zero_files = [item for item in files if item["size_bytes"] == 0]
    large_files = sorted(
        (item for item in files if item["size_bytes"] >= 1024 ** 3),
        key=lambda item: item["size_bytes"],
        reverse=True,
    )
    unmapped_files = [
        item for item in files
        if item["path"] not in mapped_paths
    ]
    hygiene_suffixes = {".log", ".pid", ".tmp", ".bak", ".pyc"}
    hygiene_files = [
        item for item in files
        if Path(item["path"]).suffix.lower() in hygiene_suffixes
    ]
    compressed_count = sum(
        extension_counts.get(suffix, 0)
        for suffix in (".zip", ".gz", ".7z", ".rar")
    )
    findings = []
    if zero_files:
        findings.append({
            "severity": "high",
            "title": "发现零字节文件",
            "detail": f"{len(zero_files)} 个文件大小为 0，需要核验上传完整性。",
        })
    if duplicate_groups:
        findings.append({
            "severity": "medium",
            "title": "存在同名文件",
            "detail": (
                f"{len(duplicate_groups)} 组文件 basename 相同；"
                "同名不等于内容重复，需结合路径或哈希复核。"
            ),
        })
    if unmapped_files:
        findings.append({
            "severity": "medium",
            "title": "存在未归类文件",
            "detail": (
                f"{len(unmapped_files)} 个文件未命中当前来源别名，"
                "建议补充目录命名或来源映射。"
            ),
        })
    if hygiene_files:
        findings.append({
            "severity": "low",
            "title": "存在运行残留文件",
            "detail": f"发现 {len(hygiene_files)} 个 log/pid/tmp/bak 文件。",
        })
    if compressed_count:
        findings.append({
            "severity": "info",
            "title": "压缩文件需要展开后质检",
            "detail": f"当前有 {compressed_count} 个压缩文件，文件树无法检查内部字段。",
        })

    return {
        "repository": HF_DATASET_REPO,
        "repositories": HF_DATASET_REPOS,
        "repository_errors": repo_errors,
        "branch": "main",
        "checked_at": datetime.now().astimezone().isoformat(timespec="seconds"),
        "tree_object_count": len(files),
        "file_count": len(files),
        "total_size_bytes": sum(item["size_bytes"] for item in files),
        "zero_byte_count": len(zero_files),
        "duplicate_basename_group_count": len(duplicate_groups),
        "duplicate_basename_file_count": sum(
            item["count"] for item in duplicate_groups
        ),
        "unmapped_file_count": len(unmapped_files),
        "mapped_file_count": len(mapped_paths),
        "large_file_count": len(large_files),
        "compressed_file_count": compressed_count,
        "hygiene_file_count": len(hygiene_files),
        "extension_counts": dict(extension_counts.most_common()),
        "findings": findings,
        "zero_byte_files": [item["path"] for item in zero_files[:50]],
        "duplicate_groups": duplicate_groups[:50],
        "unmapped_files": [item["path"] for item in unmapped_files[:80]],
        "large_files": [
            {
                "path": item["path"],
                "size_bytes": item["size_bytes"],
                "size": _human_bytes(item["size_bytes"]),
            }
            for item in large_files[:50]
        ],
        "hygiene_files": [item["path"] for item in hygiene_files[:50]],
        "notes": [
            "同名文件只表示 basename 重复，不代表文件内容重复。",
            "未归类表示未命中当前来源别名，不代表数据无效。",
            "该体检只分析仓库清单；CSV/NetCDF 内部缺失值和字段质量需另行质检。",
        ],
    }


async def sidebar_dataset_metadata_audit(
    request: Request,
    refresh: bool = False,
):
    """Audit metadata with Ocean/Fisheries-specific, evidence-based rules.

    Ocean fields are gathered from all three Marine API catalog/status endpoints
    plus the configured variable/data-shape registry. Fisheries repository-level
    metadata is scored from the live Hugging Face tree; content fields that require
    opening CSV/NetCDF files are marked as pending instead of being counted missing.
    """
    await resolve_request_user(request)
    checked_at = datetime.now().astimezone().isoformat(timespec="seconds")

    marine_paths = ("/catalog", "/status/ocean", "/domains")
    marine_responses = await asyncio.gather(
        *(_marine_api_get(path) for path in marine_paths),
        return_exceptions=True,
    )
    marine_payloads = {}
    ocean_errors = []
    for path, response in zip(marine_paths, marine_responses):
        if isinstance(response, Exception):
            ocean_errors.append(f"{path}: {str(response)[:240]}")
        else:
            marine_payloads[path] = response

    try:
        live_files, repo_errors = await hf_all_live_files(force=refresh)
        hf_error = "; ".join(f"{repo}: {msg}" for repo, msg in repo_errors.items())
    except Exception as exc:
        live_files = []
        hf_error = str(exc)[:500]

    ocean_expected = (
        "variables", "data_shape", "time_range", "temporal_resolution",
        "spatial_resolution", "spatial_coverage", "depth_range", "units",
        "updated_at",
    )
    fisheries_expected = (
        "repository", "file_count", "total_size", "file_types", "source_category",
    )
    fisheries_pending = (
        "species", "gear", "catch_effort_cpue", "time_range",
        "temporal_resolution", "spatial_coverage", "spatial_resolution", "units",
    )

    records = []
    for key, name, name_zh, variables in _OCEAN_CATALOG:
        metadata = {}
        provenance = []
        for path, payload in marine_payloads.items():
            entry = _find_catalog_entry(payload, key)
            if not entry:
                continue
            provenance.append(path)
            for field, value in _catalog_metadata(entry).items():
                metadata.setdefault(field, value)
        reference = _OCEAN_SOURCE_DETAILS.get(key, {})
        metadata["variables"] = "、".join(variables) if variables else ""
        metadata["data_shape"] = reference.get("data_shape", "")

        # Depth is not applicable to clearly 2-D products; do not penalize them.
        expected = list(ocean_expected)
        shape_text = str(metadata.get("data_shape") or "")
        if "二维" in shape_text and "三维" not in shape_text and "深度" not in shape_text:
            expected.remove("depth_range")

        completeness = _audit_completeness(metadata, expected)
        records.append({
            "group": "Ocean",
            "key": key,
            "name": name,
            "name_zh": name_zh,
            "status": "connected" if marine_payloads else "unverified",
            "file_count": None,
            "variable_count": len(variables),
            "completeness": completeness,
            "evidence": {"provenance": provenance, "metadata": metadata},
            "action": (
                "补充实时 Marine API 中仍未返回的元数据字段"
                if completeness["missing_fields"] else "当前可审计核心元数据已完整"
            ),
        })

    for name, aliases in _HF_SOURCE_ALIASES.items():
        matched = [
            item for item in live_files
            if any(alias in item["path_lower"] for alias in aliases)
        ]
        repos = sorted({
            item.get("repository", "") for item in matched if item.get("repository")
        })
        total_size = sum(int(item.get("size_bytes") or 0) for item in matched)
        extension_counts = {}
        for item in matched:
            suffix = Path(item["path"]).suffix.lower() or "无扩展名"
            extension_counts[suffix] = extension_counts.get(suffix, 0) + 1
        category, category_zh = _HF_SOURCE_CATEGORIES.get(name, ("general", "综合渔业数据"))
        metadata = {
            "repository": "、".join(repos) if repos else "",
            "file_count": str(len(matched)) if matched else "",
            "total_size": _human_bytes(total_size) if matched else "",
            "file_types": "、".join(
                f"{suffix} × {count}" for suffix, count in sorted(extension_counts.items())
            ) if matched else "",
            "source_category": category_zh if matched else "",
        }
        completeness = _audit_completeness(
            metadata, fisheries_expected, pending_fields=fisheries_pending
        )
        records.append({
            "group": "Fisheries",
            "key": name.lower().replace(" ", "_"),
            "name": name,
            "name_zh": _HF_SOURCE_NAMES_ZH.get(name, name),
            "status": "available" if matched else "not_found",
            "file_count": len(matched),
            "variable_count": None,
            "completeness": completeness,
            "evidence": {"metadata": metadata},
            "action": (
                "仓库级元数据已核验；物种/渔具/catch/effort/CPUE及时空字段需读取实际文件继续核验"
                if matched else "先将该来源文件收录到 main 分支"
            ),
        })

    audited = len(records)
    complete = sum(1 for item in records if item["completeness"]["score"] == 100)
    average = round(
        sum(item["completeness"]["score"] for item in records) / audited
    ) if audited else 0
    return {
        "checked_at": checked_at,
        "expected_fields": {
            "Ocean": list(ocean_expected),
            "Fisheries": list(fisheries_expected),
            "Fisheries_pending_file_content": list(fisheries_pending),
        },
        "summary": {
            "dataset_count": audited,
            "complete_count": complete,
            "incomplete_count": audited - complete,
            "average_score": average,
        },
        "records": records,
        "errors": {
            "ocean_api": "; ".join(ocean_errors),
            "hf_tree": hf_error,
        },
        "notes": [
            "完整度按 Ocean 与 Fisheries 两套规则分别计算，不再用同一组字段硬套全部数据源。",
            "Fisheries 的物种、渔具、catch、effort、CPUE、时空范围与单位必须读取实际文件后核验，当前显示为“待文件级核验”，不计作仓库元数据缺失。",
            "Ocean 会合并 /catalog、/status/ocean、/domains 三个实时接口证据，并计入已配置的变量和二维/三维数据形态。",
        ],
    }


def register_datasets_routes(app, context):
    _inject(context)
    app.add_api_route("/api/sidebar/datasets", sidebar_datasets, methods=["GET"])
    app.add_api_route("/api/sidebar/datasets/{group}/{source_key}", sidebar_dataset_detail, methods=["GET"])
    app.add_api_route("/api/sidebar/datasets/ocean/{source_key}/availability", sidebar_ocean_availability, methods=["POST"])
    app.add_api_route("/api/sidebar/datasets/fisheries/{source_key}/files", sidebar_fisheries_files, methods=["GET"])
    app.add_api_route("/api/sidebar/datasets/quality", sidebar_dataset_quality, methods=["GET"])
    app.add_api_route("/api/sidebar/datasets/metadata-audit", sidebar_dataset_metadata_audit, methods=["GET"])
    return sidebar_dataset_quality, sidebar_dataset_metadata_audit