feat: unify import flows and harden sqlite fallback

tests/test_main_agent_governance.py

@@ -1163,7 +1163,82 @@ class MainAgentGovernanceTests(unittest.TestCase):
         self.assertEqual(captured_request["video_provider"], "seedance2")
         self.assertEqual(captured_request["video_model"], "seedance-2.0-pro")
         self.assertEqual(payload["payload"]["job"]["artifacts"]["video_provider"], "seedance2")
-        self.assertEqual(payload["payload"]["job"]["artifacts"]["video_model"], "seedance-2.0-pro")
+
+    def test_import_video_link_and_text_actions_execute_through_oneliner(self) -> None:
+        self._insert_assistant()
+
+        captured_video_request: dict[str, Any] = {}
+        captured_text_request: dict[str, Any] = {}
+
+        async def fake_video_link_job(request: Any, account: dict[str, Any]) -> dict[str, Any]:
+            captured_video_request.update(
+                {
+                    "project_id": request.project_id,
+                    "knowledge_base_id": request.knowledge_base_id,
+                    "assistant_id": request.assistant_id,
+                    "title": request.title,
+                    "video_url": request.video_url,
+                    "language": request.language,
+                }
+            )
+            return {"id": "job_video_link_direct", "title": request.title or "视频链接分析任务"}
+
+        async def fake_text_job(request: Any, account: dict[str, Any]) -> dict[str, Any]:
+            captured_text_request.update(
+                {
+                    "project_id": request.project_id,
+                    "knowledge_base_id": request.knowledge_base_id,
+                    "assistant_id": request.assistant_id,
+                    "title": request.title,
+                    "content": request.content,
+                }
+            )
+            return {"id": "job_text_direct", "title": request.title}
+
+        with patch.object(self.core, "create_video_link_job", new=AsyncMock(side_effect=fake_video_link_job)):
+            video_response = self.client.post(
+                "/v2/oneliner/actions/execute",
+                headers=self.ctx["member_headers"],
+                json={
+                    "action_key": "import-video-link",
+                    "project_id": self.ctx["project_id"],
+                    "platform": "douyin",
+                    "payload": {
+                        "title": "短视频链接导入",
+                        "video_url": "https://example.com/video/123",
+                        "assistant_id": "asst_member_default",
+                        "language": "zh-CN",
+                    },
+                },
+            )
+        self.assertEqual(video_response.status_code, 200, video_response.text)
+        video_payload = video_response.json()
+        self.assertEqual(captured_video_request["video_url"], "https://example.com/video/123")
+        self.assertEqual(captured_video_request["assistant_id"], "asst_member_default")
+        self.assertEqual(video_payload["recommended_action"]["action"], "open-job-detail")
+        self.assertEqual(video_payload["recommended_action"]["job_id"], "job_video_link_direct")
+
+        with patch.object(self.core, "create_text_job", new=AsyncMock(side_effect=fake_text_job)):
+            text_response = self.client.post(
+                "/v2/oneliner/actions/execute",
+                headers=self.ctx["member_headers"],
+                json={
+                    "action_key": "import-text",
+                    "project_id": self.ctx["project_id"],
+                    "platform": "douyin",
+                    "payload": {
+                        "title": "文本导入",
+                        "content": "这是一段需要分析的文本。",
+                        "assistant_id": "asst_member_default",
+                    },
+                },
+            )
+        self.assertEqual(text_response.status_code, 200, text_response.text)
+        text_payload = text_response.json()
+        self.assertEqual(captured_text_request["title"], "文本导入")
+        self.assertEqual(captured_text_request["assistant_id"], "asst_member_default")
+        self.assertEqual(text_payload["recommended_action"]["action"], "open-job-detail")
+        self.assertEqual(text_payload["recommended_action"]["job_id"], "job_text_direct")
 
     def test_direct_oneliner_actions_execute_real_account_and_agent_flows(self) -> None:
         source_id = self._insert_content_source_account(