from __future__ import annotations import re from collections import defaultdict from typing import Any from fastapi import HTTPException from sidebar_catalog import ( HF_SOURCE_CATEGORIES as _HF_SOURCE_CATEGORIES, HF_SOURCE_NAMES_ZH as _HF_SOURCE_NAMES_ZH, OCEAN_CATALOG as _OCEAN_CATALOG, ) def _project_package_plan(project: str) -> dict[str, Any]: """Build a reproducible project data plan from the user's project description. v3.5.1 keeps the planner deterministic/offline so a recommendation can always be reproduced, but upgrades it from a flat keyword list to a structured plan: project intent -> requirements -> priority -> source mapping -> coverage hints -> alternatives. Live file availability/size is still verified by the estimate/build endpoints before downloading anything. """ text = str(project or "").strip() q = text.lower() if len(text) < 4: raise HTTPException(400, "请至少用一句话描述项目目标。") ocean_scores: dict[str, int] = defaultdict(int) fish_scores: dict[str, int] = defaultdict(int) reasons: dict[str, list[str]] = defaultdict(list) requirement_tags: dict[str, set[str]] = defaultdict(set) def add_ocean(key: str, score: int, reason: str, requirement: str = ""): ocean_scores[key] += score reasons["ocean:" + key].append(reason) if requirement: requirement_tags["ocean:" + key].add(requirement) def add_fish(name: str, score: int, reason: str, requirement: str = ""): fish_scores[name] += score reasons["fish:" + name].append(reason) if requirement: requirement_tags["fish:" + name].add(requirement) # ---- 1) Interpret the project itself ------------------------------------------------- species = [] species_rules = [ (("柔鱼", "茎柔鱼", "鱿鱼", "squid"), "柔鱼/鱿鱼"), (("金枪鱼", "tuna"), "金枪鱼"), (("黄鳍", "yellowfin"), "黄鳍金枪鱼"), (("大眼", "bigeye"), "大眼金枪鱼"), (("长鳍", "albacore"), "长鳍金枪鱼"), (("蓝鳍", "bluefin"), "蓝鳍金枪鱼"), (("鲣", "skipjack"), "鲣"), ] for terms, label in species_rules: if any(t in q for t in terms) and label not in species: species.append(label) task_rules = [ (("中尺度涡旋", "涡旋", "eddy", "mesoscale eddy"), "物理海洋过程研究"), (("锋面", "海洋锋", "front"), "物理海洋过程研究"), (("上升流", "upwelling"), "物理海洋过程研究"), (("环流", "海流", "洋流", "circulation", "current"), "物理海洋过程研究"), (("生境", "适生区", "habitat", "species distribution", "sdm", "maxent", "分布预测"), "生境/物种分布预测"), (("资源评估", "种群评估", "stock assessment", "biomass", "资源量", "补充量"), "资源/种群评估"), (("cpue", "努力量", "effort", "catch rate"), "CPUE/捕捞努力分析"), (("渔场", "fishing ground", "捕捞热点", "hotspot"), "渔场/热点分析"), (("气候", "enso", "气候变化", "climate"), "气候影响分析"), (("趋势", "变化趋势", "时序", "time series"), "时间序列/趋势分析"), ] task_type = "综合海洋分析" for terms, label in task_rules: if any(t in q for t in terms): task_type = label break # Exact day, month, year and rough ranges. Keep exact date separately because Ocean # export currently needs a single date. date = "" time_range = "" m = re.search(r"(20\d{2})[-/年](1[0-2]|0?[1-9])[-/月](3[01]|[12]\d|0?[1-9])", text) if m: date = f"{int(m.group(1)):04d}-{int(m.group(2)):02d}-{int(m.group(3)):02d}" time_range = date else: compact = re.search(r"20\d{6}", text) if compact: raw = compact.group(0) date = f"{raw[:4]}-{raw[4:6]}-{raw[6:8]}" time_range = date else: yr = re.search(r"(20\d{2})\s*(?:年)?\s*(?:[-–—至到~～]\s*(20\d{2})\s*年?)?", text) mo = re.search(r"(20\d{2})[-/年](1[0-2]|0?[1-9])(?:月)?", text) if mo: time_range = f"{int(mo.group(1)):04d}-{int(mo.group(2)):02d}" elif yr: time_range = yr.group(1) + ((" 至 " + yr.group(2)) if yr.group(2) else "") regions = [ (("南海",), [99.0, 124.0, 0.0, 25.0], "南海"), (("东海",), [118.0, 132.0, 23.0, 34.0], "东海"), (("黄海",), [117.0, 126.0, 30.0, 41.0], "黄海"), (("西北太平洋",), [120.0, 180.0, 10.0, 50.0], "西北太平洋"), (("北太平洋",), [120.0, -100.0, 0.0, 60.0], "北太平洋"), # “太平洋”跨越日期变更线，不能安全地用一个普通 west str: if score >= 8: return "必需" if score >= 5: return "推荐" return "可选" alternatives = { "oisst": ["CMEMS Physics（若需要三维温度/更多物理量）"], "cmems_bgc": ["OC-CCI（若主要关注遥感叶绿素/海色）"], "occci": ["CMEMS BGC（若还需要营养盐、溶解氧等）"], "era5": ["ERA5 Accum（若重点是降水、辐射、热通量累计量）"], "cmems_surface": ["CMEMS Physics（若需要更完整三维物理场）"], } fish_alternatives = { "WCPFC": ["FAO", "Sea Around Us"], "IATTC": ["FAO", "Sea Around Us"], "ICCAT": ["FAO", "Sea Around Us"], "IOTC": ["FAO", "Sea Around Us"], "CCSBT": ["FAO", "Sea Around Us"], "SPRFMO": ["NPFC", "FAO"], "NPFC": ["SPRFMO", "FAO"], "GFW": ["VIIRS"], "VIIRS": ["GFW"], "RAM Legacy": ["FAO", "Sea Around Us"], } ocean_lookup = {x[0]: x for x in _OCEAN_CATALOG} ocean = [] for key, score in sorted(ocean_scores.items(), key=lambda x: (-x[1], x[0]))[:7]: entry = ocean_lookup.get(key) if not entry: continue _, name, name_zh, variables = entry ocean.append({ "database": "Ocean", "key": key, "name": name, "name_zh": name_zh, "variables": list(variables), "score": score, "priority": priority(score), "reason": "；".join(dict.fromkeys(reasons["ocean:" + key])), "requirements": sorted(requirement_tags["ocean:" + key]), "coverage": {"time": "打包前实时核验", "space": "打包前按所选区域核验", "resolution": "需结合模型尺度确认"}, "alternatives": alternatives.get(key, []), }) fisheries = [] for name, score in sorted(fish_scores.items(), key=lambda x: (-x[1], x[0]))[:10]: category = _HF_SOURCE_CATEGORIES.get(name, ("general", "综合渔业数据"))[0] db = "Tuna-Fisheries-Dataset" if category == "tuna" else "squid_dataset" if category == "squid" else "按实际 Hugging Face 仓库分类" fisheries.append({ "database": db, "name": name, "name_zh": _HF_SOURCE_NAMES_ZH.get(name, name), "category": category, "score": score, "priority": priority(score), "reason": "；".join(dict.fromkeys(reasons["fish:" + name])), "requirements": sorted(requirement_tags["fish:" + name]), "coverage": {"time": "打包前核验文件时间覆盖", "space": "依据来源组织/文件范围核验", "resolution": "以实际文件粒度为准"}, "alternatives": fish_alternatives.get(name, []), }) missing_conditions = [] if not region_name: missing_conditions.append("海域/空间范围") elif region_name == "太平洋" and not bbox: missing_conditions.append("更精确的太平洋子区域/经纬度范围") if not time_range: missing_conditions.append("时间范围") if not species and fisheries_intent: missing_conditions.append("目标物种") # Explicit variables plus inferred requirements give the user a compact project brief. requirements = sorted(reqs, key=lambda x: ({"必需": 0, "推荐": 1, "可选": 2}.get(x["priority"], 3), x["name"])) required_count = sum(1 for x in ocean + fisheries if x["priority"] == "必需") recommended_count = sum(1 for x in ocean + fisheries if x["priority"] == "推荐") optional_count = sum(1 for x in ocean + fisheries if x["priority"] == "可选") return { "planner_version": "3.5.1", "project": text, "project_profile": { "task_type": task_type, "research_domain": "物理海洋" if physical_ocean_intent and not fisheries_intent else ("渔业/生态" if fisheries_intent else "海洋综合"), "research_object": "中尺度涡旋" if eddy_intent else ("、".join(species) if species else ""), "species": species, "region": region_name, "bbox": bbox, "time_range": time_range, "exact_date": date, }, "requirements": requirements, "ocean": ocean, "fisheries": fisheries, "date": date, "time_range": time_range, "bbox": bbox, "region_name": region_name, "ocean_export_ready": bool(date and bbox), "missing_conditions": missing_conditions, "priority_summary": {"required": required_count, "recommended": recommended_count, "optional": optional_count}, "checks": [ "生成前通过实时 inventory 核验 Hugging Face 文件是否存在及大小。", "Ocean 原始格点导出会按日期与区域请求 Marine API；缺少精确日期时只生成可复现请求清单。", "时间覆盖、空间覆盖和分辨率属于数据源级条件，最终以实时文件/接口返回为准，不用规则结果冒充真实覆盖。", ], "note": "推荐按“必需 / 推荐 / 可选”分级；物理海洋任务默认只推荐 Ocean，只有出现物种、捕捞、CPUE、渔场等明确渔业意图时才加入 Fisheries。若主来源不可用，可参考替代来源。", }