fix ths issue related to the success log url based on new airflow3 ui…

airflow/models/taskinstance.py

@@ -2044,38 +2044,40 @@ def generate_command(
 
     @property
     def log_url(self) -> str:
-        """Log URL for TaskInstance."""
+        """Log URL for TaskInstance in Airflow 3."""
         run_id = quote(self.run_id)
         base_url = conf.get_mandatory_value("webserver", "BASE_URL")
         map_index = f"&map_index={self.map_index}" if self.map_index >= 0 else ""
+
         _log_uri = (
             f"{base_url}"
-            f"/dags"
-            f"/{self.dag_id}"
-            f"/grid"
+            f"/dags/{self.dag_id}"
+            f"/taskInstances/{self.task_id}"
+            f"/logs"
             f"?dag_run_id={run_id}"
-            f"&task_id={self.task_id}"
             f"{map_index}"
-            "&tab=logs"
         )
+
         if self.logical_date:
             base_date = quote(self.logical_date.strftime("%Y-%m-%dT%H:%M:%S%z"))
             _log_uri = f"{_log_uri}&base_date={base_date}"
+
         return _log_uri
 
     @property
     def mark_success_url(self) -> str:
-        """URL to mark TI success."""
+        """URL to mark TaskInstance success in Airflow 3."""
         base_url = conf.get_mandatory_value("webserver", "BASE_URL")
+
         return (
             f"{base_url}"
-            "/confirm"
-            f"?task_id={self.task_id}"
-            f"&dag_id={self.dag_id}"
-            f"&dag_run_id={quote(self.run_id)}"
+            f"/dags/{self.dag_id}"
+            f"/dagRuns/{quote(self.run_id)}"
+            f"/taskInstances/{self.task_id}"
+            f"/setState"
+            "?state=success"
             "&upstream=false"
             "&downstream=false"
-            "&state=success"
         )
 
     @provide_session