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