diff --git a/backend/app/services/collectors/base.py b/backend/app/services/collectors/base.py index fa2faaac..dcdbe36a 100644 --- a/backend/app/services/collectors/base.py +++ b/backend/app/services/collectors/base.py @@ -282,6 +282,7 @@ class BaseCollector(ABC): "execution_time_seconds": (datetime.now(UTC) - start_time).total_seconds(), } except asyncio.CancelledError: + await db.rollback() task.status = "cancelled" task.phase = "cancelled" task.error_message = "Collection cancelled by operator and rolled back" @@ -297,6 +298,7 @@ class BaseCollector(ABC): await self._publish_task_update(force=True) raise except Exception as e: + await db.rollback() task.status = "failed" task.phase = "failed" task.error_message = str(e) diff --git a/frontend/src/pages/DataSources/DataSources.tsx b/frontend/src/pages/DataSources/DataSources.tsx index b560b44d..9a667978 100644 --- a/frontend/src/pages/DataSources/DataSources.tsx +++ b/frontend/src/pages/DataSources/DataSources.tsx @@ -9,7 +9,7 @@ import { CheckCircleOutlined, CloseCircleOutlined, ExperimentOutlined, SyncOutlined, ClearOutlined, CopyOutlined } from '@ant-design/icons' -import axios from 'axios' +import axios, { type AxiosResponse } from 'axios' import AppLayout from '../../components/AppLayout/AppLayout' import { formatDateTimeZhCN } from '../../utils/datetime' import { useWebSocket } from '../../hooks/useWebSocket' @@ -60,6 +60,57 @@ interface BulkProgressBatch { items: Record } +type TriggerDatasourceConflict = { + reason?: string + message?: string + progress?: number | null + phase?: string | null + records_processed?: number | null + total_records?: number | null +} + +type TriggerDatasourceResult = + | { ok: true; response: AxiosResponse } + | { ok: false; status: number; detail?: string | TriggerDatasourceConflict } + +type DatasourceTaskStatus = { + is_running: boolean + task_id?: number | null + progress?: number | null + phase?: string | null + records_processed?: number | null + total_records?: number | null + status?: string | null +} + +const phaseProgressRanges: Record = { + queued: [0, 0], + fetching: [0, 30], + transforming: [30, 50], + saving: [50, 100], + completed: [100, 100], + success: [100, 100], + failed: [0, 0], + cancelled: [0, 0], +} + +function mapPhaseProgressToOverall( + phase?: string | null, + progress?: number | null, + status?: string | null, +): number { + if (status && status !== 'running') { + return status === 'success' ? 100 : 0 + } + + const normalizedPhase = phase || 'queued' + const [start, end] = phaseProgressRanges[normalizedPhase] ?? [0, 100] + if (start === end) return start + + const boundedProgress = Math.max(0, Math.min(100, progress ?? 0)) + return start + ((end - start) * boundedProgress) / 100 +} + function finalizeBulkProgressBatch(batch: BulkProgressBatch | null): BulkProgressBatch | null { if (!batch || batch.sourceIds.length === 0) { return null @@ -122,6 +173,7 @@ interface ViewDataSource { function DataSources() { const [messageApi, contextHolder] = message.useMessage() + const [modal, modalContextHolder] = Modal.useModal() const [activeTab, setActiveTab] = useState('builtin') const [builtInSources, setBuiltInSources] = useState([]) const [customSources, setCustomSources] = useState([]) @@ -173,17 +225,18 @@ function DataSources() { bulkProgressBatch.sourceIds.reduce((sum, sourceId) => { const item = bulkProgressBatch.items[sourceId] if (!item) return sum - if (item.status && item.status !== 'running') { - return sum + 100 - } - return sum + (item.progress || 0) + return sum + mapPhaseProgressToOverall(item.phase, item.progress, item.status) }, 0) / bulkProgressBatch.sourceIds.length ) : runningBuiltInSources.length > 0 ? Math.round( runningBuiltInSources.reduce((sum, source) => { const trackedTask = taskProgress[source.id] - return sum + (trackedTask?.progress ?? source.progress ?? 0) + return sum + mapPhaseProgressToOverall( + trackedTask?.phase ?? source.phase ?? 'queued', + trackedTask?.progress ?? source.progress ?? 0, + trackedTask?.status ?? (trackedTask?.is_running || source.is_running ? 'running' : source.last_status ?? null), + ) }, 0) / runningBuiltInSources.length ) : 0 @@ -249,7 +302,7 @@ function DataSources() { ...prev.items, [sourceId]: { task_id: payload.task_id ?? prev.items[sourceId]?.task_id ?? null, - progress: payload.status && payload.status !== 'running' ? 100 : (payload.progress ?? prev.items[sourceId]?.progress ?? 0), + progress: payload.progress ?? prev.items[sourceId]?.progress ?? 0, is_running: payload.status === 'running', phase: payload.phase ?? prev.items[sourceId]?.phase ?? null, status: payload.status ?? prev.items[sourceId]?.status ?? null, @@ -361,7 +414,7 @@ function DataSources() { if (!prev.sourceIds.includes(numericSourceId)) continue nextItems[numericSourceId] = { task_id: state.task_id, - progress: state.status && state.status !== 'running' ? 100 : state.progress, + progress: state.progress, is_running: state.is_running, phase: state.phase ?? null, status: state.status ?? null, @@ -386,8 +439,17 @@ function DataSources() { const force = options?.force ?? false const res = await axios.post(`/api/v1/datasources/${id}/trigger`, null, { params: { force }, + validateStatus: (status) => status < 500, }) + if (res.status >= 400) { + return { + ok: false, + status: res.status, + detail: res.data?.detail, + } satisfies TriggerDatasourceResult + } + if (res.data.task_id) { setTaskProgress(prev => ({ ...prev, @@ -406,7 +468,15 @@ function DataSources() { } fetchData() - return res + return { + ok: true, + response: res, + } satisfies TriggerDatasourceResult + } + + const fetchDatasourceTaskStatus = async (id: number) => { + const res = await axios.get(`/api/v1/datasources/${id}/task-status`) + return res.data } const confirmForceTrigger = (id: number, taskInfo?: { @@ -422,7 +492,7 @@ function DataSources() { ? `${taskInfo.records_processed}${typeof taskInfo?.total_records === 'number' && taskInfo.total_records > 0 ? ` / ${taskInfo.total_records}` : ''}` : '未知' - Modal.confirm({ + modal.confirm({ title: '当前任务未完成', content: (
@@ -437,38 +507,58 @@ function DataSources() { cancelText: '取消', okButtonProps: { danger: true }, onOk: async () => { - try { - await triggerDatasource(id, { force: true }) + const result = await triggerDatasource(id, { force: true }) + if (result.ok) { messageApi.success('已强制重新触发,未完成采集将回滚') - } catch (error: unknown) { - const err = error as { response?: { data?: { detail?: string | { message?: string } } } } - const detail = err.response?.data?.detail - messageApi.error(typeof detail === 'string' ? detail : detail?.message || '强制重新采集失败') + return } + + const detail = result.detail + messageApi.error(typeof detail === 'string' ? detail : detail?.message || '强制重新采集失败') }, }) } - const handleTrigger = async (id: number) => { - try { - await triggerDatasource(id) - messageApi.success('任务已触发') - } catch (error: unknown) { - const err = error as { response?: { status?: number; data?: { detail?: string | { - reason?: string - message?: string - progress?: number | null - phase?: string | null - records_processed?: number | null - total_records?: number | null - } } } } - const detail = err.response?.data?.detail - if (err.response?.status === 409 && typeof detail === 'object' && detail?.reason === 'running_task_in_progress') { - confirmForceTrigger(id, detail) - return - } - messageApi.error(typeof detail === 'string' ? detail : detail?.message || '触发失败') + const triggerDatasourceWithPrecheck = async ( + id: number, + options?: { + successMessage: string + onSuccess?: () => void + errorMessage?: string + }, + ) => { + const taskStatus = await fetchDatasourceTaskStatus(id) + if (taskStatus.is_running) { + confirmForceTrigger(id, { + message: '当前采集任务尚未完成,重新触发会丢失本次未完成进度。是否强制重新采集?', + progress: taskStatus.progress ?? null, + phase: taskStatus.phase ?? null, + records_processed: taskStatus.records_processed ?? null, + total_records: taskStatus.total_records ?? null, + }) + return } + + const result = await triggerDatasource(id) + if (result.ok) { + messageApi.success(options?.successMessage || '任务已触发') + options?.onSuccess?.() + return + } + + const detail = result.detail + if (result.status === 409 && typeof detail === 'object' && detail?.reason === 'running_task_in_progress') { + confirmForceTrigger(id, detail) + return + } + messageApi.error(typeof detail === 'string' ? detail : detail?.message || options?.errorMessage || '触发失败') + } + + const handleTrigger = async (id: number) => { + await triggerDatasourceWithPrecheck(id, { + successMessage: '任务已触发', + errorMessage: '触发失败', + }) } const handleTriggerAll = async () => { @@ -588,26 +678,13 @@ function DataSources() { const handleUpdateSource = async () => { if (!viewingSource) return - try { - await triggerDatasource(viewingSource.id) - messageApi.success('已触发更新') - setViewDrawerVisible(false) - } catch (error: unknown) { - const err = error as { response?: { status?: number; data?: { detail?: string | { - reason?: string - message?: string - progress?: number | null - phase?: string | null - records_processed?: number | null - total_records?: number | null - } } } } - const detail = err.response?.data?.detail - if (err.response?.status === 409 && typeof detail === 'object' && detail?.reason === 'running_task_in_progress') { - confirmForceTrigger(viewingSource.id, detail) - return - } - messageApi.error(typeof detail === 'string' ? detail : detail?.message || '更新失败') - } + await triggerDatasourceWithPrecheck(viewingSource.id, { + successMessage: '已触发更新', + errorMessage: '更新失败', + onSuccess: () => { + setViewDrawerVisible(false) + }, + }) } const handleTest = async () => { @@ -1017,6 +1094,7 @@ function DataSources() { return ( {contextHolder} + {modalContextHolder}

数据源管理

diff --git a/planet.sh b/planet.sh index 4f109889..d2badfe9 100755 --- a/planet.sh +++ b/planet.sh @@ -165,7 +165,6 @@ run_with_retry() { fi if [ "$attempt" -eq "$max_retries" ]; then - clear_wait_spinner log_error "${failure_message}" return 1 fi @@ -337,7 +336,6 @@ start_ai_provider_service() { fi if [ "$retry" -eq "$AI_PROVIDER_START_MAX_RETRIES" ]; then - clear_wait_spinner log_error "AI Provider 启动失败,已重试 ${AI_PROVIDER_START_MAX_RETRIES} 次" docker logs --tail 20 planet_aiprovider 2>/dev/null || true exit 1 @@ -348,7 +346,6 @@ start_ai_provider_service() { retry=$((retry + 1)) done - clear_wait_spinner log_error "AI Provider 启动失败" docker logs --tail 20 planet_aiprovider 2>/dev/null || true exit 1 @@ -387,7 +384,6 @@ ensure_database_services_healthy() { fi if [ "$retry" -eq "$DATABASE_START_MAX_RETRIES" ]; then - clear_wait_spinner log_error "数据库启动失败,已重试 ${DATABASE_START_MAX_RETRIES} 次" docker logs --tail 20 planet_postgres 2>/dev/null || true docker logs --tail 20 planet_redis 2>/dev/null || true @@ -409,7 +405,6 @@ ensure_postgres_service_healthy() { fi if [ "$retry" -eq "$DATABASE_START_MAX_RETRIES" ]; then - clear_wait_spinner log_error "PostgreSQL 启动失败,已重试 ${DATABASE_START_MAX_RETRIES} 次" docker logs --tail 20 planet_postgres 2>/dev/null || true exit 1 @@ -433,7 +428,6 @@ restart_database_service() { fi if [ "$retry" -eq "$DATABASE_START_MAX_RETRIES" ]; then - clear_wait_spinner log_error "数据库重启失败,已重试 ${DATABASE_START_MAX_RETRIES} 次" docker logs --tail 20 planet_postgres 2>/dev/null || true docker logs --tail 20 planet_redis 2>/dev/null || true @@ -809,25 +803,28 @@ health() { log() { case "$1" in -f|--frontend) - echo "📝 前端日志 (Ctrl+C 退出):" + log_step "查看前端日志" + log_note "按 Ctrl+C 退出" tail -f /tmp/planet_frontend.log ;; -b|--backend) - echo "📝 后端日志 (Ctrl+C 退出):" + log_step "查看后端日志" + log_note "按 Ctrl+C 退出" tail -f /tmp/planet_backend.log ;; -a|--ai-provider) - echo "📝 AI Provider 日志 (Ctrl+C 退出):" + log_step "查看 AI Provider 日志" + log_note "按 Ctrl+C 退出" docker logs -f planet_aiprovider ;; *) - echo "📝 最近日志:" - echo "--- 后端 ---" - tail -20 /tmp/planet_backend.log 2>/dev/null || echo "无日志" - echo "--- AI Provider ---" - docker logs --tail 20 planet_aiprovider 2>/dev/null || echo "无日志" - echo "--- 前端 ---" - tail -20 /tmp/planet_frontend.log 2>/dev/null || echo "无日志" + log_step "查看最近日志" + log_note "后端" + tail -20 /tmp/planet_backend.log 2>/dev/null || log_note "无日志" + log_note "AI Provider" + docker logs --tail 20 planet_aiprovider 2>/dev/null || log_note "无日志" + log_note "前端" + tail -20 /tmp/planet_frontend.log 2>/dev/null || log_note "无日志" ;; esac } @@ -854,17 +851,15 @@ case "$1" in log "$2" ;; *) - echo "用法: ./planet.sh {start|stop|restart|createuser|health|log}" - echo "" - echo "命令:" - echo " start 启动服务,可选: -b <后端端口> -f <前端端口> -a " - echo " stop 停止服务" - echo " restart 重启服务,可选: -b [后端端口] -f [前端端口] -a [AI Provider 端口] -d" - echo " createuser 交互创建用户" - echo " health 检查健康状态" - echo " log 查看日志" - echo " log -f 查看前端日志" - echo " log -b 查看后端日志" - echo " log -a 查看 AI Provider 日志" + log_error "用法: ./planet.sh {start|stop|restart|createuser|health|log}" + log_note "start 启动服务,可选: -b <后端端口> -f <前端端口> -a " + log_note "stop 停止服务" + log_note "restart 重启服务,可选: -b [后端端口] -f [前端端口] -a [AI Provider 端口] -d" + log_note "createuser 交互创建用户" + log_note "health 检查健康状态" + log_note "log 查看日志" + log_note "log -f 查看前端日志" + log_note "log -b 查看后端日志" + log_note "log -a 查看 AI Provider 日志" ;; esac