Compatible for bohrium OpenAPI sandbox (#553)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

## Release Notes

* **Bug Fixes**
* Improved reliability of job grouping by safely handling missing data
fields with sensible defaults.
  * Fixed conditional log retrieval to only fetch logs when appropriate.
* Enhanced exit code handling for better control over job completion
status.

* **Improvements**
* Ensured automatic creation of required directories at initialization
for smoother setup.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Signed-off-by: zjgemi <[email protected]>
Co-authored-by: dingzhaohan <[email protected]>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Jinzhe Zeng <[email protected]>
Co-authored-by: Jinzhe Zeng <[email protected]>

Author: 4 people

dpdispatcher/contexts/openapi_context.py

@@ -170,7 +170,7 @@ def upload_job(self, job, common_files=None):
         object_key = os.path.join(data["storePath"], zip_filename)  # type: ignore
         job.upload_path = object_key
         job.job_id = data["jobId"]  # type: ignore
-        job.jgid = data["jobGroupId"]  # type: ignore
+        job.jgid = data.get("jobGroupId", "")  # type: ignore
         self.storage.upload_From_file_multi_part(
             object_key=object_key, file_path=upload_zip, token=token
         )

dpdispatcher/machines/openapi.py

@@ -129,10 +129,13 @@ def do_submit(self, job):
             ),
             "out_files": self._gen_backward_files_list(job),
             "platform": self.remote_profile.get("platform", "ali"),
-            "image_address": self.remote_profile.get("image_address", ""),
+            "image_name": self.remote_profile.get("image_address", ""),
         }
-        if job.job_state == JobStatus.unsubmitted:
-            openapi_params["job_id"] = job.job_id
+        if "real_user_id" in self.remote_profile:
+            openapi_params["real_user_id"] = self.remote_profile["real_user_id"]
+        if "session_id" in self.remote_profile:
+            openapi_params["session_id"] = self.remote_profile["session_id"]
+        openapi_params["job_id"] = job.job_id
         data = self.job.insert(**openapi_params)
 
         job.job_id = data.get("jobId", 0)  # type: ignore
@@ -182,8 +185,8 @@ def check_status(self, job):
             self.ignore_exit_code,
         )
         if job_state == JobStatus.finished:
-            job_log = self.job.log(job_id)
             if self.remote_profile.get("output_log"):
+                job_log = self.job.log(job_id)
                 print(job_log, end="")
             self._download_job(job)
         elif self.remote_profile.get("output_log") and job_state == JobStatus.running:
@@ -193,7 +196,7 @@ def check_status(self, job):
 
     def _download_job(self, job):
         data = self.job.detail(job.job_id)
-        job_url = data["jobFiles"]["outFiles"][0]["url"]  # type: ignore
+        job_url = data["resultUrl"]  # type: ignore
         if not job_url:
             return
         job_hash = job.job_hash
@@ -243,7 +246,7 @@ def map_dp_job_state(status, exit_code, ignore_exit_code=True):
         if status not in map_dict:
             dlog.error(f"unknown job status {status}")
             return JobStatus.unknown
-        if status == -1 and exit_code != 0 and ignore_exit_code:
+        if status == -1 and ignore_exit_code:
             return JobStatus.finished
         return map_dict[status]