Deploy v4.2.2 download resilience

README.md

@@ -9,7 +9,7 @@ app_port: 7860
 
 # Global Marine Foundation Data Agent
 
-Current UI release: **v4.2.1**.
+Current UI release: **v4.2.2**.
 
 
 ## v3.4.0 稳定性重构（第一阶段）

VERSION

@@ -1 +1 @@
-4.2.1
+4.2.2

routes/ocean_batch.py

@@ -45,6 +45,16 @@ async def retry_ocean_batch(job_id:str,request:Request):
     if raw.get("user_id") and uid and raw.get("user_id")!=uid:raise HTTPException(403,"forbidden")
     return await OCEAN_BATCH_MANAGER.retry_failed(job_id)
 
+async def refresh_ocean_batch_download(job_id:str,subtask_id:str,request:Request):
+    uid,_=await resolve_request_user(request)
+    raw=OCEAN_BATCH_MANAGER._read(job_id)
+    if not raw:raise HTTPException(404,"batch job not found")
+    if raw.get("user_id") and uid and raw.get("user_id")!=uid:raise HTTPException(403,"forbidden")
+    if not any(x.get("subtask_id")==subtask_id for x in raw.get("subtasks",[])):
+        raise HTTPException(404,"subtask not found")
+    try:return await OCEAN_BATCH_MANAGER.refresh_download_url(job_id,subtask_id)
+    except Exception as exc:raise HTTPException(502,f"刷新下载链接失败：{str(exc)[:300]}") from exc
+
 async def ocean_batch_manifest(job_id:str,request:Request):
     uid,_=await resolve_request_user(request)
     raw=OCEAN_BATCH_MANAGER._read(job_id)
@@ -59,4 +69,5 @@ def register_ocean_batch_routes(app,context):
     app.add_api_route("/api/ocean/batch/{job_id}",cancel_ocean_batch,methods=["DELETE"])
     app.add_api_route("/api/ocean/batch/{job_id}/resume",resume_ocean_batch,methods=["POST"])
     app.add_api_route("/api/ocean/batch/{job_id}/retry",retry_ocean_batch,methods=["POST"])
+    app.add_api_route("/api/ocean/batch/{job_id}/files/{subtask_id}/refresh",refresh_ocean_batch_download,methods=["POST"])
     app.add_api_route("/api/ocean/batch/{job_id}/manifest",ocean_batch_manifest,methods=["GET"])

services/ocean_batch.py

@@ -176,7 +176,7 @@ class OceanBatchManager:
             "progress_percent":(done/total*100 if total else 0),
             "message":job.get("message",""),"cancel_requested":bool(job.get("cancel_requested")),
             "results":[
-                {k:s.get(k) for k in ("subtask_id","date","label","source","variable","status","attempts","download_url","filename","size_bytes","detail")}
+                {k:s.get(k) for k in ("subtask_id","date","label","source","variable","status","attempts","download_url","download_url_issued_at","filename","size_bytes","detail")}
                 for s in subs if s.get("status") in {"completed","failed"}
             ],
             "known_size_bytes":sum(int(s.get("size_bytes") or 0) for s in subs if s.get("status")=="completed"),
@@ -193,7 +193,7 @@ class OceanBatchManager:
                 subs.append({
                     "subtask_id":f"{n:04d}","date":d.isoformat(),"key":key,
                     "label":meta["label"],"source":meta["source"],"variable":meta["variable"],
-                    "status":"pending","attempts":0,"download_url":"","filename":"","size_bytes":0,"detail":"",
+                    "status":"pending","attempts":0,"download_url":"","filename":"","size_bytes":0,"detail":"","download_url_issued_at":"",
                 })
         job={
             "job_id":jid,"user_id":str(user_id or ""),"prompt":prompt,"spec":spec,
@@ -253,6 +253,7 @@ class OceanBatchManager:
             if isinstance(result,dict) and result.get("download_url"):
                 sub["status"]="completed"
                 sub["download_url"]=str(result["download_url"])
+                sub["download_url_issued_at"]=datetime.now().astimezone().isoformat(timespec="seconds")
                 guessed=Path(urlparse(sub["download_url"]).path).name
                 ext={"netcdf":".nc","csv":".csv","xlsx":".xlsx","json":".json"}.get(job["spec"].get("format") or "netcdf",".bin")
                 sub["filename"]=str(
@@ -316,6 +317,55 @@ class OceanBatchManager:
         self._write(job);await self.start(jid)
         return self._public(self._read(jid))
 
+    async def refresh_download_url(self,jid:str,subtask_id:str):
+        """Re-export one completed subtask to obtain a fresh signed URL."""
+        job=self._read(jid)
+        if not job:return None
+        sub=next((x for x in job.get("subtasks",[]) if x.get("subtask_id")==subtask_id),None)
+        if not sub:return None
+        payload={
+            "domain":"ocean","source":sub["source"],"date":sub["date"],"variable":sub["variable"],
+            **job["spec"]["bbox"],"format":job["spec"].get("format") or "netcdf",
+        }
+        result={};detail=""
+        for attempt in range(3):
+            try:
+                async with httpx.AsyncClient(
+                    timeout=httpx.Timeout(connect=8,read=120,write=20,pool=20),follow_redirects=True
+                ) as client:
+                    r=await client.post(self.api_url+"/data/export",json=payload)
+                    try:result=r.json()
+                    except Exception:result={"detail":r.text[:500]}
+                    if r.status_code<400 and isinstance(result,dict) and (result.get("download_url") or result.get("download_path")):
+                        break
+                    detail=str((result or {}).get("detail") or f"HTTP {r.status_code}")[:500]
+            except Exception as exc:
+                detail=str(exc)[:500]
+            if attempt<2:await asyncio.sleep(1.0*(2**attempt))
+        if isinstance(result,dict) and result.get("download_path") and not result.get("download_url"):
+            result["download_url"]=self.api_url+str(result["download_path"])
+        if not isinstance(result,dict) or not result.get("download_url"):
+            raise RuntimeError(detail or "无法刷新 Ocean 下载链接")
+        job=self._read(jid)
+        if not job:return None
+        sub=next((x for x in job.get("subtasks",[]) if x.get("subtask_id")==subtask_id),None)
+        if not sub:return None
+        sub["download_url"]=str(result["download_url"])
+        sub["download_url_issued_at"]=datetime.now().astimezone().isoformat(timespec="seconds")
+        guessed=Path(urlparse(sub["download_url"]).path).name
+        if result.get("filename") or result.get("file_name") or guessed:
+            sub["filename"]=str(result.get("filename") or result.get("file_name") or guessed)[:240]
+        raw_size=result.get("size_bytes") or result.get("file_size_bytes") or result.get("content_length") or result.get("bytes") or 0
+        try:sub["size_bytes"]=max(0,int(raw_size))
+        except Exception:pass
+        sub["detail"]=""
+        self._write(job)
+        return {
+            "job_id":jid,"subtask_id":subtask_id,"download_url":sub["download_url"],
+            "download_url_issued_at":sub["download_url_issued_at"],
+            "filename":sub.get("filename") or "download","size_bytes":int(sub.get("size_bytes") or 0),
+        }
+
     def get(self,jid): 
         job=self._read(jid)
         return self._public(job) if job else None
@@ -338,6 +388,7 @@ class OceanBatchManager:
                 "filename":filename,
                 "size_bytes":int(s.get("size_bytes") or 0),
                 "download_url":s.get("download_url"),
+                "download_url_issued_at":s.get("download_url_issued_at") or "",
             })
         return {
             "job_id":jid,"spec":job.get("spec"),"status":job.get("status"),

static/js/app.js

@@ -1197,6 +1197,40 @@ function waitManagedDownloadStatus(job,state,card){
   });
 }
 
+function oceanDownloadUrlAgeMs(file){
+  const raw=file?.download_url_issued_at;
+  if(!raw)return Infinity; // legacy jobs: refresh safely before downloading.
+  const t=Date.parse(raw);return Number.isFinite(t)?Math.max(0,Date.now()-t):Infinity;
+}
+async function refreshOceanBatchFileUrl(state,file){
+  if(!state.batch_job_id||!file?.subtask_id)throw Error("缺少批量任务或子任务编号");
+  state.status_text="正在刷新下载链接…";
+  const fresh=await j("/api/ocean/batch/"+encodeURIComponent(state.batch_job_id)+"/files/"+encodeURIComponent(file.subtask_id)+"/refresh",{method:"POST"});
+  Object.assign(file,fresh||{});
+  persistOceanBatchQueue(state);
+  return file;
+}
+async function createManagedOceanDownload(state,file){
+  // Ocean signed links are documented by the upstream response as short-lived.
+  // Refresh links older than 45 minutes, and all legacy links without issue time.
+  if(oceanDownloadUrlAgeMs(file)>=45*60*1000)await refreshOceanBatchFileUrl(state,file);
+  let lastErr=null;
+  for(let attempt=0;attempt<3;attempt++){
+    try{
+      return await j("/api/download/jobs",{method:"POST",body:JSON.stringify({url:file.download_url,filename:state.current_filename})});
+    }catch(err){
+      lastErr=err;
+      if(attempt===0){
+        // A stale/expired signed URL commonly surfaces while creating the
+        // managed download. Refresh it once before transport retries.
+        await refreshOceanBatchFileUrl(state,file);
+      }
+      if(attempt<2)await new Promise(r=>setTimeout(r,800*Math.pow(2,attempt)));
+    }
+  }
+  throw lastErr||Error("无法创建下载任务");
+}
+
 async function runOceanBatchDownloadQueue(card,state){
   if(state._looping)return;
   state._looping=true;state.running=true;state.paused=false;
@@ -1207,14 +1241,26 @@ async function runOceanBatchDownloadQueue(card,state){
       state.current_bytes=0;state.status_text="正在创建下载任务…";
       persistOceanBatchQueue(state);renderOceanBatchDownloadQueue(card,state);
       try{
-        const child=await j("/api/download/jobs",{method:"POST",body:JSON.stringify({url:file.download_url,filename:state.current_filename})});
+        const child=await createManagedOceanDownload(state,file);
         if(child.total_bytes){
           state.file_totals=state.file_totals||{};
           state.file_totals[state.index]=Number(child.total_bytes);
           state.total_known_bytes=Object.values(state.file_totals).reduce((a,b)=>a+Number(b||0),0);
         }
         triggerBrowserSave(child.stream_url,state.current_filename);
-        const result=await waitManagedDownloadStatus(child,state,card);
+        let result=await waitManagedDownloadStatus(child,state,card);
+        if(result.status==="paused")break;
+        for(let retryNo=1;result.status!=="completed" && retryNo<=2 && !state.paused;retryNo++){
+          state.status_text="下载失败，自动重试 "+retryNo+"/2";
+          state.note="正在刷新临时下载链接并重新建立传输；无需手动操作。";
+          persistOceanBatchQueue(state);renderOceanBatchDownloadQueue(card,state);
+          await new Promise(r=>setTimeout(r,1000*Math.pow(2,retryNo-1)));
+          await refreshOceanBatchFileUrl(state,file);
+          const retryJob=await createManagedOceanDownload(state,file);
+          triggerBrowserSave(retryJob.stream_url,state.current_filename);
+          result=await waitManagedDownloadStatus(retryJob,state,card);
+          if(result.status==="paused")break;
+        }
         if(result.status==="paused")break;
         if(result.status==="completed"){
           state.completed_files++;
@@ -1265,7 +1311,7 @@ async function startOceanBatchDownloadAll(card,batchJob){
       completed_files:0,failed_files:0,completed_bytes:0,current_bytes:0,
       file_totals: fileTotals,
       total_known_bytes:Object.values(fileTotals).reduce((a,b)=>a+Number(b||0),0),
-      running:true,paused:false,status_text:"准备批量下载",current_filename:"",note:"文件会逐个流式下载，不会在页面内存中拼接大文件。"
+      running:true,paused:false,status_text:"准备批量下载",current_filename:"",note:"文件会逐个流式下载；过期链接会自动刷新，失败传输最多自动重试 2 次。"
     };
     persistOceanBatchQueue(state);renderOceanBatchDownloadQueue(card,state);
     runOceanBatchDownloadQueue(card,state);

tests/test_chat_lifecycle_progress_v393.py

@@ -3,7 +3,7 @@ import pathlib
 ROOT=pathlib.Path(__file__).resolve().parents[1]
 
 def test_version_393():
-    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.1"
+    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.2"
 
 def test_frontend_lifecycle_progress():
     js=(ROOT/"static"/"js"/"app.js").read_text(encoding="utf-8")

tests/test_download_progress_v391.py

@@ -3,7 +3,7 @@ import pathlib
 ROOT=pathlib.Path(__file__).resolve().parents[1]
 
 def test_version_391():
-    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.1"
+    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.2"
 
 def test_p1_label_removed():
     js=(ROOT/"static"/"js"/"app.js").read_text(encoding="utf-8")

tests/test_download_resilience_v422.py

@@ -0,0 +1,24 @@
+
+import pathlib
+ROOT=pathlib.Path(__file__).resolve().parents[1]
+
+def test_version():
+    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.2"
+
+def test_refresh_route_registered():
+    t=(ROOT/"routes"/"ocean_batch.py").read_text(encoding="utf-8")
+    assert '/files/{subtask_id}/refresh' in t
+    assert 'refresh_download_url(job_id,subtask_id)' in t
+
+def test_ocean_batch_records_link_issue_time():
+    t=(ROOT/"services"/"ocean_batch.py").read_text(encoding="utf-8")
+    assert 'download_url_issued_at' in t
+    assert 'async def refresh_download_url' in t
+    assert 'for attempt in range(3)' in t
+
+def test_frontend_proactive_refresh_and_retries():
+    t=(ROOT/"static"/"js"/"app.js").read_text(encoding="utf-8")
+    assert '45*60*1000' in t
+    assert 'createManagedOceanDownload' in t
+    assert 'retryNo<=2' in t
+    assert '正在刷新下载链接' in t

tests/test_history_sidebar_v373.py

@@ -3,7 +3,7 @@ import pathlib
 ROOT=pathlib.Path(__file__).resolve().parents[1]
 
 def test_version_373():
-    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.1"
+    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.2"
 
 def test_project_history_backend():
     route=(ROOT/"routes"/"project_package.py").read_text(encoding="utf-8")

tests/test_large_download_jobs_v400.py

@@ -3,7 +3,7 @@ import pathlib
 ROOT=pathlib.Path(__file__).resolve().parents[1]
 
 def test_version_400():
-    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.1"
+    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.2"
 
 def test_download_job_backend():
     route=(ROOT/"routes"/"downloads.py").read_text(encoding="utf-8")

tests/test_ocean_batch_downloads_v420.py

@@ -3,7 +3,7 @@ import pathlib
 ROOT=pathlib.Path(__file__).resolve().parents[1]
 
 def test_version_420():
-    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.1"
+    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.2"
 
 def test_manifest_has_filename_and_sizes():
     svc=(ROOT/"services"/"ocean_batch.py").read_text(encoding="utf-8")

tests/test_ocean_batch_routing_v421.py

@@ -4,7 +4,7 @@ from services.ocean_batch import parse_ocean_batch_request
 ROOT=pathlib.Path(__file__).resolve().parents[1]
 
 def test_version_421():
-    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.1"
+    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.2"
 
 def test_exact_user_month_query_routes_to_batch():
     p=parse_ocean_batch_request("查询 2002 年 8 月西北太平洋的 SST 数据，并导出为 CSV 文件供我下载")

tests/test_ocean_batch_v410.py

@@ -4,7 +4,7 @@ from services.ocean_batch import parse_ocean_batch_request
 ROOT=pathlib.Path(__file__).resolve().parents[1]
 
 def test_version_410():
-    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.1"
+    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.2"
 
 def test_month_three_variable_batch_plan():
     p=parse_ocean_batch_request("查询并导出 2002 年 8 月西北太平洋 120°E–140°E、20°N–40°N 的 SST、SSH、CHL 数据，并打包下载")

tests/test_p0_finish_v380.py

@@ -3,7 +3,7 @@ import pathlib, re, subprocess, shutil
 ROOT=pathlib.Path(__file__).resolve().parents[1]
 
 def test_version_380():
-    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.1"
+    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.2"
 
 def test_chat_runtime_extracted():
     ui=(ROOT/"ui_server.py").read_text(encoding="utf-8")

tests/test_p0_phase2_v371.py

@@ -3,7 +3,7 @@ import pathlib
 ROOT=pathlib.Path(__file__).resolve().parents[1]
 
 def test_version():
-    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.1"
+    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.2"
 
 def test_frontend_feature_modules_exist():
     html=(ROOT/"app.html").read_text(encoding="utf-8")

tests/test_p0_phase3_v372.py

@@ -3,7 +3,7 @@ import pathlib
 ROOT=pathlib.Path(__file__).resolve().parents[1]
 
 def test_version_372():
-    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.1"
+    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.2"
 
 def test_route_modules_exist():
     for name in ("datasets.py","user_state.py","project_package.py","data_health.py"):

tests/test_quality_center_v362.py

@@ -3,7 +3,7 @@ import pathlib
 ROOT = pathlib.Path(__file__).resolve().parents[1]
 
 def test_version_371():
-    assert (ROOT / "VERSION").read_text(encoding="utf-8").strip() == "4.2.1"
+    assert (ROOT / "VERSION").read_text(encoding="utf-8").strip() == "4.2.2"
 
 def test_native_health_backend():
     text = (ROOT / "routes" / "data_health.py").read_text(encoding="utf-8")

tests/test_release_stability.py

@@ -7,7 +7,7 @@ def test_version_is_single_source():
     version = (ROOT / "VERSION").read_text(encoding="utf-8").strip()
     ui = (ROOT / "ui_server.py").read_text(encoding="utf-8")
     html = (ROOT / "app.html").read_text(encoding="utf-8")
-    assert version == "4.2.1"
+    assert version == "4.2.2"
     assert 'with_name("VERSION")' in ui
     assert "__APP_VERSION__" in html
 

tests/test_runtime_hotfix_v376.py

@@ -3,7 +3,7 @@ import pathlib
 ROOT=pathlib.Path(__file__).resolve().parents[1]
 
 def test_version_376():
-    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.1"
+    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.2"
 
 def test_no_stray_async_before_sidebar_toggle():
     js=(ROOT/"static"/"js"/"app.js").read_text(encoding="utf-8")

tests/test_sidebar_collapse_v374.py

@@ -3,7 +3,7 @@ import pathlib
 ROOT=pathlib.Path(__file__).resolve().parents[1]
 
 def test_version_374():
-    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.1"
+    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.2"
 
 def test_sidebar_toggle_markup():
     html=(ROOT/"app.html").read_text(encoding="utf-8")

tests/test_sidebar_partial_collapse_v375.py

@@ -3,7 +3,7 @@ import pathlib
 ROOT=pathlib.Path(__file__).resolve().parents[1]
 
 def test_version_375():
-    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.1"
+    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.2"
 
 def test_sidebar_order_and_groups():
     html=(ROOT/"app.html").read_text(encoding="utf-8")

tests/test_sse_heartbeat_v401.py

@@ -3,7 +3,7 @@ import pathlib
 ROOT=pathlib.Path(__file__).resolve().parents[1]
 
 def test_version_401():
-    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.1"
+    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.2"
 
 def test_chat_has_independent_heartbeat_wrapper():
     route=(ROOT/"routes"/"chat.py").read_text(encoding="utf-8")

tests/test_unified_download_manager_v392.py

@@ -3,7 +3,7 @@ import pathlib
 ROOT=pathlib.Path(__file__).resolve().parents[1]
 
 def test_version_392():
-    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.1"
+    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.2"
 
 def test_download_proxy_route_registered():
     route=(ROOT/"routes"/"downloads.py").read_text(encoding="utf-8")

tests/test_unified_query_v390.py

@@ -4,7 +4,7 @@ from services.unified_query import plan_unified_query
 ROOT=pathlib.Path(__file__).resolve().parents[1]
 
 def test_version_390():
-    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.1"
+    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.2"
 
 def test_joint_squid_environment_plan():
     p=plan_unified_query("查询 2002 年 8 月西北太平洋柔鱼记录，并匹配 SST、SSH、CHL")

tests/test_v381_navigation_ocean_intent.py

@@ -3,7 +3,7 @@ import pathlib, ast
 ROOT=pathlib.Path(__file__).resolve().parents[1]
 
 def test_version_381():
-    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.1"
+    assert (ROOT/"VERSION").read_text(encoding="utf-8").strip()=="4.2.2"
 
 def test_recent_conversation_switches_to_chat():
     js=(ROOT/"static"/"js"/"app.js").read_text(encoding="utf-8")