from __future__ import annotations from services.project_history import ProjectHistoryStore # 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 analyze_project_package(body: ProjectDataPackageRequest, request: Request): await resolve_request_user(request) return _project_package_plan(body.project) async def estimate_project_package(body: ProjectDataPackageRequest, request: Request): """Estimate package contents from the live inventories without downloading files.""" await resolve_request_user(request) plan = _project_package_plan(body.project) if body.selected_ocean_keys is not None: selected_ocean = {str(x).strip() for x in body.selected_ocean_keys if str(x).strip()} plan["ocean"] = [x for x in plan.get("ocean", []) if x.get("key") in selected_ocean] if body.selected_fisheries_names is not None: selected_fish = {str(x).strip() for x in body.selected_fisheries_names if str(x).strip()} plan["fisheries"] = [x for x in plan.get("fisheries", []) if x.get("name") in selected_fish] cap = max(20, min(int(body.max_package_mb or 300), 1200)) * 1024 * 1024 per_db: dict[str, dict[str, Any]] = defaultdict(lambda: {"file_count": 0, "size_bytes": 0, "sources": []}) selected_files = [] repo_errors = {} if body.include_tuna or body.include_squid: live_files, repo_errors = await hf_all_live_files(force=False) for src in plan.get("fisheries", []): planned_db = src.get("database") if planned_db == "Tuna-Fisheries-Dataset" and not body.include_tuna: continue if planned_db == "squid_dataset" and not body.include_squid: continue aliases = _HF_SOURCE_ALIASES.get(src.get("name"), ()) candidates = [x for x in live_files if any(a in x.get("path_lower", "") for a in aliases)] candidates = [x for x in candidates if (x.get("repository_domain") == "tuna" and body.include_tuna) or (x.get("repository_domain") == "squid" and body.include_squid)] candidates = [x for x in candidates if Path(x.get("path", "")).suffix.lower() in {".csv", ".tsv", ".zip"} and int(x.get("size_bytes") or 0) > 0] candidates.sort(key=lambda x: (int(x.get("size_bytes") or 0), x.get("path", ""))) for item in candidates[:2]: db = "Tuna-Fisheries-Dataset" if item.get("repository_domain") == "tuna" else "squid_dataset" size = int(item.get("size_bytes") or 0) per_db[db]["file_count"] += 1 per_db[db]["size_bytes"] += size if src.get("name") not in per_db[db]["sources"]: per_db[db]["sources"].append(src.get("name")) selected_files.append({"database": db, "source": src.get("name"), "path": item.get("path"), "size_bytes": size}) ocean_items = [] if body.include_ocean: for src in plan.get("ocean", []): variable = (src.get("variables") or [""])[0] ocean_items.append({"database": "Ocean", "source": src.get("name"), "variable": variable, "export_ready": bool(plan.get("ocean_export_ready"))}) if ocean_items: per_db["Ocean"]["file_count"] = len(ocean_items) if plan.get("ocean_export_ready") else 0 per_db["Ocean"]["sources"] = [x.get("source") for x in ocean_items] fish_bytes = sum(int(x.get("size_bytes") or 0) for x in selected_files) return { "status": "ok", "package_limit_bytes": cap, "estimated_known_bytes": fish_bytes, "estimated_known_file_count": len(selected_files), "within_limit": fish_bytes <= cap, "per_database": dict(per_db), "selected_files": selected_files[:100], "ocean": { "request_count": len(ocean_items), "export_ready": bool(plan.get("ocean_export_ready")), "size_known": False, "items": ocean_items, }, "repository_errors": repo_errors, "note": "渔业文件大小来自 Hugging Face 实时文件树；Ocean NetCDF 大小需实际导出后才能确定，因此不计入已知大小。", } async def build_project_package(body: ProjectDataPackageRequest, request: Request): uid, _auth_user = await resolve_request_user(request) plan = _project_package_plan(body.project) # v2.9.0: allow the user to review the recommendation and package only # explicitly selected data sources. None means "use all recommendations". if body.selected_ocean_keys is not None: selected_ocean = {str(x).strip() for x in body.selected_ocean_keys if str(x).strip()} plan["ocean"] = [x for x in plan.get("ocean", []) if x.get("key") in selected_ocean] if body.selected_fisheries_names is not None: selected_fish = {str(x).strip() for x in body.selected_fisheries_names if str(x).strip()} plan["fisheries"] = [x for x in plan.get("fisheries", []) if x.get("name") in selected_fish] cap = max(20, min(int(body.max_package_mb or 300), 1200)) * 1024 * 1024 token = secrets.token_urlsafe(24) work = PROJECT_PACKAGE_ROOT / token work.mkdir(parents=True, exist_ok=False) included=[]; skipped=[]; used=0 # Always write a reproducible plan/manifest. profile = plan.get("project_profile") or {} priority = plan.get("priority_summary") or {} (work / "README.md").write_text( "# 项目数据包\n\n" + body.project + "\n\n" f"任务类型：{profile.get('task_type') or '综合分析'}\n\n" f"研究对象：{'、'.join(profile.get('species') or []) or '未识别'}\n\n" f"区域：{profile.get('region') or '未识别'}；时间：{profile.get('time_range') or '未识别'}\n\n" f"数据优先级：必需 {priority.get('required', 0)} / 推荐 {priority.get('recommended', 0)} / 可选 {priority.get('optional', 0)}\n\n" "目录按数据库分类：Ocean、Tuna-Fisheries-Dataset、squid_dataset、Fisheries。\n" "project_plan.json 保存 Agent 数据规划；manifest.json 记录真实文件来源与跳过原因。\n", encoding="utf-8", ) (work / "project_plan.json").write_text( json.dumps(plan, ensure_ascii=False, indent=2), encoding="utf-8" ) # Fisheries: include real repository files, picking smaller matching files first. if body.include_tuna or body.include_squid: live_files, repo_errors = await hf_all_live_files(force=False) for src in plan["fisheries"]: planned_db = src["database"] if planned_db == "Tuna-Fisheries-Dataset" and not body.include_tuna: continue if planned_db == "squid_dataset" and not body.include_squid: continue aliases = _HF_SOURCE_ALIASES.get(src["name"], ()) candidates = [x for x in live_files if any(a in x["path_lower"] for a in aliases)] candidates = [x for x in candidates if (x.get("repository_domain") == "tuna" and body.include_tuna) or (x.get("repository_domain") == "squid" and body.include_squid)] candidates = [x for x in candidates if Path(x["path"]).suffix.lower() in {".csv", ".tsv", ".zip"} and int(x.get("size_bytes") or 0) > 0] candidates.sort(key=lambda x: (int(x.get("size_bytes") or 0), x["path"])) picked=0 for item in candidates: size=int(item.get("size_bytes") or 0) if picked >= 2: break db = "Tuna-Fisheries-Dataset" if item.get("repository_domain") == "tuna" else "squid_dataset" if used + size > cap: skipped.append({"database":db,"source":src["name"],"path":item["path"],"reason":"超过数据包大小上限"}); continue try: local, revision = await asyncio.to_thread(download_dataset_file, item["path"], size, item.get("repository")) dest_dir=work/db/_safe_package_part(src["name"]); dest_dir.mkdir(parents=True, exist_ok=True) dest=dest_dir/_safe_package_part(Path(item["path"]).name) shutil.copy2(local,dest); used += dest.stat().st_size; picked += 1 included.append({"database":db,"source":src["name"],"repository":item.get("repository"),"revision":revision,"path":item["path"],"size_bytes":size,"zip_path":str(dest.relative_to(work))}) except Exception as exc: skipped.append({"database":db,"source":src["name"],"path":item["path"],"reason":str(exc)[:300]}) # Ocean: export real data only when an exact date + named-region bbox can be inferred. ocean_requests=[] if body.include_ocean: ocean_dir=work/"Ocean"; ocean_dir.mkdir(exist_ok=True) for src in plan["ocean"]: variable=(src.get("variables") or [""])[0] req={"source":src["key"],"variable":variable,"date":plan.get("date"),"bbox":plan.get("bbox"),"reason":src.get("reason")} ocean_requests.append(req) if not plan.get("ocean_export_ready") or not variable: continue if used > cap * 0.85: break bbox=plan["bbox"] payload={"domain":"ocean","source":src["key"],"date":plan["date"],"variable":variable,"lon_min":bbox[0],"lon_max":bbox[1],"lat_min":bbox[2],"lat_max":bbox[3],"format":"netcdf"} try: result=await _marine_api_post("/data/export",payload,timeout=90) path=str(result.get("download_path") or "") if not path.startswith("/download/"): skipped.append({"database":"Ocean","source":src["name"],"reason":"Marine API 未返回可下载文件","detail":result}); continue async with httpx.AsyncClient(timeout=httpx.Timeout(connect=10,read=180,write=20,pool=20),follow_redirects=True) as client: r=await client.get(f"{MARINE_API_URL}{path}") if r.status_code>=400: raise RuntimeError(f"Marine API download {r.status_code}") if used+len(r.content)>cap: skipped.append({"database":"Ocean","source":src["name"],"reason":"导出文件超过剩余数据包上限"}); continue dest_dir=ocean_dir/_safe_package_part(src["name"]);dest_dir.mkdir(parents=True,exist_ok=True) dest=dest_dir/f"{_safe_package_part(src['key'])}_{variable}_{plan['date']}.nc";dest.write_bytes(r.content);used+=len(r.content) included.append({"database":"Ocean","source":src["name"],"path":path,"size_bytes":len(r.content),"zip_path":str(dest.relative_to(work)),"request":payload}) except Exception as exc: skipped.append({"database":"Ocean","source":src["name"],"reason":str(exc)[:300]}) (ocean_dir/"data_requests.json").write_text(json.dumps(ocean_requests,ensure_ascii=False,indent=2),encoding="utf-8") manifest={"created_at":datetime.now().astimezone().isoformat(timespec="seconds"),"user_id":uid,"project":body.project,"plan":plan,"included_files":included,"skipped":skipped,"package_bytes":used,"package_limit_bytes":cap} (work/"manifest.json").write_text(json.dumps(manifest,ensure_ascii=False,indent=2),encoding="utf-8") zip_path=PROJECT_PACKAGE_ROOT/f"project-data-package-{token}.zip" with zipfile.ZipFile(zip_path,"w",compression=zipfile.ZIP_DEFLATED,allowZip64=True) as zf: for f in work.rglob("*"): if f.is_file(): zf.write(f,arcname=f.relative_to(work)) shutil.rmtree(work,ignore_errors=True) PROJECT_PACKAGE_TOKENS[token]={"path":str(zip_path),"created":time.time(),"user_id":uid} _save_project_package_tokens(PROJECT_PACKAGE_TOKENS) return {"status":"ok","token":token,"download_url":f"/api/sidebar/project-package/download/{token}","included_file_count":len(included),"skipped_count":len(skipped),"size_bytes":zip_path.stat().st_size,"plan":plan,"included_files":included[:200],"skipped":skipped[:200],"package_limit_bytes":cap} async def download_project_package(token: str, request: Request): uid, _auth_user=await resolve_request_user(request) _cleanup_project_package_tokens() meta=PROJECT_PACKAGE_TOKENS.get(token) if not meta: raise HTTPException(404,"数据包不存在或已过期") if time.time()-float(meta.get("created") or 0)>PROJECT_PACKAGE_TTL_SECONDS: Path(meta.get("path") or "").unlink(missing_ok=True);PROJECT_PACKAGE_TOKENS.pop(token,None);_save_project_package_tokens(PROJECT_PACKAGE_TOKENS);raise HTTPException(410,"数据包已过期") if meta.get("user_id") and uid and uid!=meta.get("user_id"): raise HTTPException(403,"无权下载该数据包") path=Path(meta["path"]) if not path.exists(): raise HTTPException(404,"数据包文件不存在") return FileResponse(path,media_type="application/zip",filename="project_data_package.zip") async def get_project_package_history(request: Request): supplied_uid = str(request.query_params.get("user_id") or "").strip() uid, auth_user = await resolve_request_user(request, supplied_uid) if AUTH_ENABLED and not auth_user: raise HTTPException(401, "Authentication required") return { "items": PROJECT_HISTORY_STORE.read(uid), "user_id": uid, "sync_enabled": bool(auth_user), "storage_mode": _favorites_storage_mode(), } async def put_project_package_history(request: Request): body = await request.json() supplied_uid = str(body.get("user_id") or "").strip() if isinstance(body, dict) else "" uid, auth_user = await resolve_request_user(request, supplied_uid) if AUTH_ENABLED and not auth_user: raise HTTPException(401, "Authentication required") entry = body.get("entry") if isinstance(body, dict) else None try: items = PROJECT_HISTORY_STORE.upsert(uid, entry) except ValueError as exc: raise HTTPException(400, str(exc)) from exc return { "items": items, "count": len(items), "user_id": uid, "sync_enabled": bool(auth_user), } async def delete_project_package_history(item_id: str, request: Request): supplied_uid = str(request.query_params.get("user_id") or "").strip() uid, auth_user = await resolve_request_user(request, supplied_uid) if AUTH_ENABLED and not auth_user: raise HTTPException(401, "Authentication required") items = PROJECT_HISTORY_STORE.delete(uid, item_id) return {"items": items, "count": len(items), "user_id": uid} def register_project_package_routes(app, context): _inject(context) global PROJECT_HISTORY_STORE PROJECT_HISTORY_STORE = ProjectHistoryStore(USER_STATE_ROOT, valid_user_id, max_items=40) app.add_api_route("/api/sidebar/project-package/history", get_project_package_history, methods=["GET"]) app.add_api_route("/api/sidebar/project-package/history", put_project_package_history, methods=["PUT"]) app.add_api_route("/api/sidebar/project-package/history/{item_id}", delete_project_package_history, methods=["DELETE"]) app.add_api_route("/api/sidebar/project-package/analyze", analyze_project_package, methods=["POST"]) app.add_api_route("/api/sidebar/project-package/estimate", estimate_project_package, methods=["POST"]) app.add_api_route("/api/sidebar/project-package/build", build_project_package, methods=["POST"]) app.add_api_route("/api/sidebar/project-package/download/{token}", download_project_package, methods=["GET"])