实现插件配置持久化与任务队列持久化

插件配置持久化：
- plugin_settings 表新增 config_json 字段，支持存储每个插件的自定义配置
- BaseCrawlerPlugin 新增 default_config 属性和 update_config 方法
- PluginSettingsRepository 新增 get_config / set_config 方法
- PluginService 新增 get_plugin_config 和 update_plugin_config
- api/routes/plugins.py 新增 GET /{id}/config 和 POST /{id}/config 接口
- 前端 Plugins.vue 增加配置编辑对话框，支持动态渲染数字/布尔/字符串类型配置
- ip3366 插件示例化：增加 max_pages 配置项，验证配置生效后会动态更新爬取 URL

任务队列持久化：
- 新建 validation_tasks 表：id, ip, port, protocol, status, result, response_time_ms, created_at, updated_at
- 新建 ValidationTaskRepository，提供 insert_batch / acquire_pending / complete_task / reset_processing 等方法
- ValidationQueue 重构：
  - submit() 时把任务写入数据库并唤醒 Worker
  - Worker 通过 acquire_pending 原子取任务并验证
  - 验证完成后更新任务状态并入库有效代理
  - 启动时自动恢复之前中断的 processing 任务为 pending
  - 支持 drain() 等待所有 pending 完成
- 调度器验证流程同样自动持久化到任务表

其他适配：
- 更新 api/deps.py 和 api/lifespan.py，移除对已删除 settings_service 的残留引用
- 更新前端 pluginService.js 和 api/index.js 增加配置相关 API

api/deps.py

@@ -36,7 +36,6 @@ def create_scheduler_service() -> SchedulerService:
     queue = ValidationQueue(
         validator=validator,
         proxy_repo=proxy_repo,
-        db_ctx=get_db,
         worker_count=app_settings.validator_max_concurrency,
         score_valid=app_settings.score_valid,
         score_invalid=app_settings.score_invalid,
@@ -44,7 +43,3 @@ def create_scheduler_service() -> SchedulerService:
         score_max=app_settings.score_max,
     )
     return SchedulerService(validation_queue=queue, proxy_repo=proxy_repo)
-
-
-# 避免循环导入
-from core.db import get_db

api/routes/plugins.py

@@ -2,7 +2,6 @@
 from fastapi import APIRouter, Depends
 from services.plugin_service import PluginService
 from services.scheduler_service import SchedulerService
-from models.schemas import PluginToggleRequest
 from api.deps import get_plugin_service, get_scheduler_service
 from core.log import logger
 
@@ -26,7 +25,7 @@ async def list_plugins(service: PluginService = Depends(get_plugin_service)):
             "plugins": [
                 {
                     "id": p.id,
-                    "name": p.display_name,  # 保持旧版本兼容：name 用于展示
+                    "name": p.display_name,
                     "display_name": p.display_name,
                     "description": p.description,
                     "enabled": p.enabled,
@@ -43,18 +42,47 @@ async def list_plugins(service: PluginService = Depends(get_plugin_service)):
 @router.put("/{plugin_id}/toggle")
 async def toggle_plugin(
     plugin_id: str,
-    request: PluginToggleRequest,
+    request: dict,
     service: PluginService = Depends(get_plugin_service),
 ):
-    success = await service.toggle_plugin(plugin_id, request.enabled)
+    enabled = request.get("enabled")
+    if enabled is None:
+        return error_response("缺少 enabled 参数", 400)
+    success = await service.toggle_plugin(plugin_id, enabled)
     if not success:
         return error_response("插件不存在", 404)
     return success_response(
-        f"插件 {plugin_id} 已{'启用' if request.enabled else '禁用'}",
+        f"插件 {plugin_id} 已{'启用' if enabled else '禁用'}",
-        {"plugin_id": plugin_id, "enabled": request.enabled},
+        {"plugin_id": plugin_id, "enabled": enabled},
     )
 
 
+@router.get("/{plugin_id}/config")
+async def get_plugin_config(
+    plugin_id: str,
+    service: PluginService = Depends(get_plugin_service),
+):
+    config = await service.get_plugin_config(plugin_id)
+    if config is None:
+        return error_response("插件不存在", 404)
+    return success_response("获取插件配置成功", {"plugin_id": plugin_id, "config": config})
+
+
+@router.post("/{plugin_id}/config")
+async def update_plugin_config(
+    plugin_id: str,
+    request: dict,
+    service: PluginService = Depends(get_plugin_service),
+):
+    config = request.get("config", {})
+    if not isinstance(config, dict):
+        return error_response("config 必须是对象", 400)
+    success = await service.update_plugin_config(plugin_id, config)
+    if not success:
+        return error_response("插件不存在或配置无效", 404)
+    return success_response("保存插件配置成功", {"plugin_id": plugin_id, "config": config})
+
+
 @router.post("/{plugin_id}/crawl")
 async def crawl_plugin(
     plugin_id: str,

core/db.py

@@ -64,11 +64,36 @@ async def init_db():
             CREATE TABLE IF NOT EXISTS plugin_settings (
                 plugin_id TEXT PRIMARY KEY,
                 enabled INTEGER DEFAULT 1,
+                config_json TEXT DEFAULT '{}',
                 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                 updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
             )
         """)
 
+        # 迁移：为旧版 plugin_settings 表增加 config_json 列
+        try:
+            await db.execute("SELECT config_json FROM plugin_settings LIMIT 1")
+        except Exception:
+            await db.execute("ALTER TABLE plugin_settings ADD COLUMN config_json TEXT DEFAULT '{}'")
+            logger.info("Migrated: added config_json column to plugin_settings")
+
+        # 验证任务队列表
+        await db.execute("""
+            CREATE TABLE IF NOT EXISTS validation_tasks (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                ip TEXT NOT NULL,
+                port INTEGER NOT NULL,
+                protocol TEXT DEFAULT 'http',
+                status TEXT DEFAULT 'pending',
+                result TEXT,
+                response_time_ms REAL,
+                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+            )
+        """)
+        await db.execute("CREATE INDEX IF NOT EXISTS idx_validation_status ON validation_tasks(status)")
+        await db.execute("CREATE INDEX IF NOT EXISTS idx_validation_created ON validation_tasks(created_at)")
+
         # 系统设置表
         await db.execute("""
             CREATE TABLE IF NOT EXISTS settings (

core/plugin_system/base.py

@@ -1,7 +1,7 @@
 """插件基类 - 所有爬虫插件必须继承此基类"""
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import List
+from typing import List, Dict, Any
 
 
 @dataclass
@@ -30,6 +30,20 @@ class BaseCrawlerPlugin(ABC):
     display_name: str = ""
     description: str = ""
     enabled: bool = True
+    default_config: Dict[str, Any] = {}
+
+    def __init__(self):
+        self._config: Dict[str, Any] = dict(self.default_config or {})
+
+    @property
+    def config(self) -> Dict[str, Any]:
+        return self._config
+
+    def update_config(self, updates: Dict[str, Any]) -> None:
+        """更新插件配置，只覆盖存在的键"""
+        for key, value in updates.items():
+            if key in self._config:
+                self._config[key] = value
 
     @abstractmethod
     async def crawl(self) -> List[ProxyRaw]:

core/tasks/queue.py

@@ -1,24 +1,26 @@
-"""验证任务队列 - 解耦爬取与验证，支持背压控制"""
+"""验证任务队列 - 解耦爬取与验证，支持背压控制和持久化"""
 import asyncio
 from typing import Optional
 from models.domain import ProxyRaw
+from repositories.task_repo import ValidationTaskRepository
+from core.db import get_db
 from core.log import logger
 
 
 class ValidationQueue:
-    """代理验证队列
+    """代理验证队列（支持持久化到 SQLite）
 
     工作流程：
-    1. 爬虫将原始代理 submit() 到队列
+    1. 爬虫将原始代理 submit() 到队列（写入数据库 + 内存信号）
-    2. Worker 池从队列消费并验证
+    2. Worker 池从数据库消费并验证
     3. 验证通过的代理写入数据库
+    4. 服务重启时自动恢复未完成的 pending 任务
     """
 
     def __init__(
         self,
         validator,
         proxy_repo,
-        db_ctx,
         worker_count: int = 50,
         score_valid: int = 10,
         score_invalid: int = -5,
@@ -27,16 +29,17 @@ class ValidationQueue:
     ):
         self.validator = validator
         self.proxy_repo = proxy_repo
-        self.db_ctx = db_ctx
+        self.task_repo = ValidationTaskRepository()
         self.worker_count = worker_count
         self.score_valid = score_valid
         self.score_invalid = score_invalid
         self.score_min = score_min
         self.score_max = score_max
 
-        self._queue: asyncio.Queue[Optional[ProxyRaw]] = asyncio.Queue()
+        self._signal: asyncio.Queue[None] = asyncio.Queue()
         self._workers: list[asyncio.Task] = []
         self._running = False
+        self._db_lock = asyncio.Lock()
 
         # 统计
         self.valid_count = 0
@@ -46,6 +49,16 @@ class ValidationQueue:
         if self._running:
             return
         self._running = True
+
+        # 恢复之前中断的 processing 任务
+        async with get_db() as db:
+            recovered = await self.task_repo.reset_processing(db)
+            pending = await self.task_repo.get_pending_count(db)
+        if recovered:
+            logger.info(f"ValidationQueue recovered {recovered} interrupted tasks")
+        if pending:
+            logger.info(f"ValidationQueue has {pending} pending tasks to process")
+
         for i in range(self.worker_count):
             self._workers.append(asyncio.create_task(self._worker_loop(i)))
         logger.info(f"ValidationQueue started with {self.worker_count} workers")
@@ -55,42 +68,60 @@ class ValidationQueue:
             return
         self._running = False
         for _ in self._workers:
-            self._queue.put_nowait(None)  # sentinel
+            self._signal.put_nowait(None)  # sentinel
         if self._workers:
             await asyncio.gather(*self._workers, return_exceptions=True)
         self._workers.clear()
         logger.info("ValidationQueue stopped")
 
     async def submit(self, proxies: list[ProxyRaw]):
-        """提交代理到验证队列"""
+        """提交代理到验证队列（持久化 + 唤醒 Worker）"""
-        for p in proxies:
+        async with self._db_lock:
-            await self._queue.put(p)
+            async with get_db() as db:
+                inserted = await self.task_repo.insert_batch(db, proxies)
+        if inserted:
+            for _ in range(min(inserted, self.worker_count)):
+                self._signal.put_nowait(None)
 
     async def submit_one(self, proxy: ProxyRaw):
-        await self._queue.put(proxy)
+        await self.submit([proxy])
 
     async def drain(self):
-        """等待队列中当前所有任务处理完毕"""
+        """等待队列中当前所有 pending 任务处理完毕"""
-        await self._queue.join()
+        while True:
+            async with get_db() as db:
+                count = await self.task_repo.get_pending_count(db)
+            if count == 0:
+                break
+            await asyncio.sleep(0.5)
 
     async def _worker_loop(self, worker_id: int):
         while True:
-            item = await self._queue.get()
+            await self._signal.get()
-            if item is None:
+            self._signal.task_done()
-                self._queue.task_done()
+            if not self._running:
                 break
-            try:
+            await self._process_one_task(worker_id)
-                await self._validate_and_save(item)
-            except Exception as e:
-                logger.error(f"Worker {worker_id} validation error: {e}")
-            finally:
-                self._queue.task_done()
 
-    async def _validate_and_save(self, proxy: ProxyRaw):
+    async def _process_one_task(self, worker_id: int):
+        """从数据库取一个任务并验证"""
+        async with self._db_lock:
+            async with get_db() as db:
+                task = await self.task_repo.acquire_pending(db)
+        if not task:
+            return
+
+        proxy = ProxyRaw(task["ip"], task["port"], task["protocol"])
+        try:
             is_valid, latency = await self.validator.validate(
                 proxy.ip, proxy.port, proxy.protocol
             )
-        async with self.db_ctx() as db:
+        except Exception as e:
+            logger.error(f"Worker {worker_id} validation error: {e}")
+            is_valid, latency = False, 0.0
+
+        async with self._db_lock:
+            async with get_db() as db:
                 if is_valid:
                     await self.proxy_repo.insert_or_update(
                         db, proxy.ip, proxy.port, proxy.protocol, score=self.score_valid
@@ -99,10 +130,11 @@ class ValidationQueue:
                         await self.proxy_repo.update_response_time(
                             db, proxy.ip, proxy.port, latency
                         )
+                    await self.task_repo.complete_task(db, task["id"], True, latency)
                     self.valid_count += 1
                     logger.debug(f"ValidationQueue: valid {proxy.ip}:{proxy.port}")
                 else:
-                # 对于新爬取的无效代理，不需要入库，直接丢弃
+                    await self.task_repo.complete_task(db, task["id"], False, 0.0)
                     self.invalid_count += 1
                     logger.debug(f"ValidationQueue: invalid {proxy.ip}:{proxy.port}")
 

frontend/src/api/index.js

@@ -79,6 +79,8 @@ export const proxiesAPI = {
 export const pluginsAPI = {
   getPlugins: () => api.get('/api/plugins'),
   togglePlugin: (pluginId, enabled) => api.put(`/api/plugins/${pluginId}/toggle`, { enabled }),
+  getPluginConfig: (pluginId) => api.get(`/api/plugins/${pluginId}/config`),
+  updatePluginConfig: (pluginId, config) => api.post(`/api/plugins/${pluginId}/config`, { config }),
   crawlPlugin: (pluginId) => api.post(`/api/plugins/${pluginId}/crawl`),
   crawlAll: () => api.post('/api/plugins/crawl-all')
 }

frontend/src/services/pluginService.js

@@ -9,6 +9,14 @@ export const pluginService = {
     return pluginsAPI.togglePlugin(pluginId, enabled)
   },
 
+  async getPluginConfig(pluginId) {
+    return pluginsAPI.getPluginConfig(pluginId)
+  },
+
+  async updatePluginConfig(pluginId, config) {
+    return pluginsAPI.updatePluginConfig(pluginId, config)
+  },
+
   async crawlPlugin(pluginId) {
     return pluginsAPI.crawlPlugin(pluginId)
   },

frontend/src/views/Plugins.vue

@@ -74,11 +74,19 @@
           </template>
         </el-table-column>
 
-        <el-table-column label="操作" width="150" fixed="right" align="center">
+        <el-table-column label="操作" width="200" fixed="right" align="center">
           <template #default="{ row }">
             <el-button
               type="primary"
               size="small"
+              @click="handleOpenConfig(row)"
+            >
+              <el-icon class="btn-icon"><Setting /></el-icon>
+              配置
+            </el-button>
+            <el-button
+              type="success"
+              size="small"
               @click="handleCrawl(row.id)"
               :loading="crawlingPlugin === row.id"
               :disabled="!row.enabled"
@@ -123,11 +131,50 @@
         </template>
       </el-alert>
     </el-card>
+
+    <!-- 配置编辑对话框 -->
+    <el-dialog
+      v-model="configDialogVisible"
+      title="插件配置"
+      width="400px"
+      :close-on-click-modal="false"
+    >
+      <div v-if="currentPlugin">
+        <div class="config-plugin-name">{{ currentPlugin.name }}</div>
+        <el-form label-width="120px">
+          <el-form-item
+            v-for="(value, key) in configForm"
+            :key="key"
+            :label="String(key)"
+          >
+            <el-input-number
+              v-if="typeof value === 'number'"
+              v-model="configForm[key]"
+              :min="0"
+              style="width: 180px"
+            />
+            <el-switch
+              v-else-if="typeof value === 'boolean'"
+              v-model="configForm[key]"
+            />
+            <el-input
+              v-else
+              v-model="configForm[key]"
+              style="width: 180px"
+            />
+          </el-form-item>
+        </el-form>
+      </div>
+      <template #footer>
+        <el-button @click="configDialogVisible = false">取消</el-button>
+        <el-button type="primary" @click="handleSaveConfig" :loading="savingConfig">保存</el-button>
+      </template>
+    </el-dialog>
   </div>
 </template>
 
 <script setup>
-import { ref, onMounted } from 'vue'
+import { ref, reactive, onMounted } from 'vue'
 import { ElMessage, ElMessageBox } from 'element-plus'
 import { 
   Connection, 
@@ -135,7 +182,8 @@ import {
   Promotion, 
   CircleCheck, 
   CircleClose,
-  Box
+  Box,
+  Setting
 } from '@element-plus/icons-vue'
 import { usePluginsStore } from '../stores/plugins'
 import { pluginService } from '../services/pluginService'
@@ -147,6 +195,12 @@ const crawlingPlugin = ref(null)
 const crawlingAll = ref(false)
 const lastCrawlResult = ref(null)
 
+// 配置对话框
+const configDialogVisible = ref(false)
+const currentPlugin = ref(null)
+const configForm = reactive({})
+const savingConfig = ref(false)
+
 // ==================== 事件处理 ====================
 async function handleRefresh() {
   await pluginsStore.fetchPlugins()
@@ -158,11 +212,40 @@ async function handleToggle(pluginId, enabled) {
   if (success) {
     ElMessage.success(enabled ? '插件已启用' : '插件已禁用')
   } else {
-    // 失败时刷新列表恢复状态
     await pluginsStore.fetchPlugins()
   }
 }
 
+async function handleOpenConfig(row) {
+  currentPlugin.value = row
+  const response = await pluginService.getPluginConfig(row.id)
+  if (response.code === 200) {
+    Object.keys(configForm).forEach(key => delete configForm[key])
+    Object.assign(configForm, response.data.config || {})
+    configDialogVisible.value = true
+  } else {
+    ElMessage.error('获取插件配置失败')
+  }
+}
+
+async function handleSaveConfig() {
+  if (!currentPlugin.value) return
+  savingConfig.value = true
+  try {
+    const response = await pluginService.updatePluginConfig(currentPlugin.value.id, { ...configForm })
+    if (response.code === 200) {
+      ElMessage.success('配置保存成功')
+      configDialogVisible.value = false
+    } else {
+      ElMessage.error(response.message || '保存失败')
+    }
+  } catch (error) {
+    ElMessage.error('保存配置出错')
+  } finally {
+    savingConfig.value = false
+  }
+}
+
 async function handleCrawl(pluginId) {
   try {
     crawlingPlugin.value = pluginId
@@ -176,7 +259,6 @@ async function handleCrawl(pluginId) {
         message: response.message,
         data: response.data
       }
-      // 刷新插件统计
       await pluginsStore.fetchPlugins()
     } else {
       lastCrawlResult.value = {
@@ -196,7 +278,6 @@ async function handleCrawl(pluginId) {
 
 async function handleCrawlAll() {
   try {
-    // 确认是否爬取所有插件
     const enabledPlugins = pluginsStore.plugins.filter(p => p.enabled)
     if (enabledPlugins.length === 0) {
       ElMessage.warning('没有启用的插件')
@@ -225,7 +306,6 @@ async function handleCrawlAll() {
         data: response.data
       }
       ElMessage.success('批量爬取完成')
-      // 刷新插件统计
       await pluginsStore.fetchPlugins()
     } else {
       lastCrawlResult.value = {
@@ -373,4 +453,11 @@ onMounted(async () => {
 .invalid-count {
   color: var(--danger);
 }
+
+.config-plugin-name {
+  font-size: 16px;
+  font-weight: 600;
+  margin-bottom: 16px;
+  color: var(--text-primary);
+}
 </style>

plugins/ip3366.py

@@ -12,13 +12,18 @@ class Ip3366Plugin(BaseHTTPPlugin):
     name = "ip3366"
     display_name = "IP3366"
     description = "从 IP3366 网站爬取免费代理"
+    default_config = {"max_pages": 5}
 
     def __init__(self):
         super().__init__()
+        self._update_urls()
+
+    def _update_urls(self):
+        max_pages = self.config.get("max_pages", 5)
         self.urls = [
-            f"http://www.ip3366.net/free/?stype=1&page={i}" for i in range(1, 6)
+            f"http://www.ip3366.net/free/?stype=1&page={i}" for i in range(1, max_pages + 1)
         ] + [
-            f"http://www.ip3366.net/free/?stype=2&page={i}" for i in range(1, 6)
+            f"http://www.ip3366.net/free/?stype=2&page={i}" for i in range(1, max_pages + 1)
         ]
 
     async def crawl(self) -> List[ProxyRaw]:

repositories/__init__.py

@@ -1,3 +1,5 @@
 from .proxy_repo import ProxyRepository
+from .settings_repo import SettingsRepository, PluginSettingsRepository
+from .task_repo import ValidationTaskRepository
 
-__all__ = ["ProxyRepository"]
+__all__ = ["ProxyRepository", "SettingsRepository", "PluginSettingsRepository", "ValidationTaskRepository"]

repositories/settings_repo.py

@@ -93,10 +93,48 @@ class PluginSettingsRepository:
             return False
 
     @staticmethod
-    async def list_all(db: aiosqlite.Connection) -> Dict[str, bool]:
+    async def get_config(db: aiosqlite.Connection, plugin_id: str) -> Optional[Dict[str, Any]]:
+        async with db.execute(
+            "SELECT config_json FROM plugin_settings WHERE plugin_id = ?", (plugin_id,)
+        ) as cursor:
+            row = await cursor.fetchone()
+            if row and row[0]:
+                try:
+                    return json.loads(row[0])
+                except json.JSONDecodeError:
+                    return None
+            return None
+
+    @staticmethod
+    async def set_config(db: aiosqlite.Connection, plugin_id: str, config: Dict[str, Any]) -> bool:
+        try:
+            await db.execute(
+                """
+                INSERT INTO plugin_settings (plugin_id, config_json, updated_at)
+                VALUES (?, ?, CURRENT_TIMESTAMP)
+                ON CONFLICT(plugin_id) DO UPDATE SET
+                    config_json = excluded.config_json,
+                    updated_at = CURRENT_TIMESTAMP
+                """,
+                (plugin_id, json.dumps(config, ensure_ascii=False)),
+            )
+            await db.commit()
+            return True
+        except Exception as e:
+            logger.error(f"set_config failed for {plugin_id}: {e}")
+            return False
+
+    @staticmethod
+    async def list_all(db: aiosqlite.Connection) -> Dict[str, Dict[str, Any]]:
         result = {}
-        async with db.execute("SELECT plugin_id, enabled FROM plugin_settings") as cursor:
+        async with db.execute("SELECT plugin_id, enabled, config_json FROM plugin_settings") as cursor:
             rows = await cursor.fetchall()
-            for plugin_id, enabled in rows:
+            for plugin_id, enabled, config_json in rows:
-                result[plugin_id] = bool(enabled)
+                config = {}
+                if config_json:
+                    try:
+                        config = json.loads(config_json)
+                    except json.JSONDecodeError:
+                        pass
+                result[plugin_id] = {"enabled": bool(enabled), "config": config}
         return result

repositories/task_repo.py

@@ -0,0 +1,135 @@
+"""验证任务队列持久化层"""
+import aiosqlite
+from typing import List, Optional
+from models.domain import ProxyRaw
+from core.log import logger
+
+
+class ValidationTaskRepository:
+    """验证任务 Repository —— 支持队列持久化"""
+
+    @staticmethod
+    async def insert_batch(db: aiosqlite.Connection, proxies: List[ProxyRaw]) -> int:
+        if not proxies:
+            return 0
+        try:
+            rows = [(p.ip, p.port, p.protocol) for p in proxies]
+            await db.executemany(
+                """
+                INSERT INTO validation_tasks (ip, port, protocol, status, created_at)
+                VALUES (?, ?, ?, 'pending', CURRENT_TIMESTAMP)
+                """,
+                rows,
+            )
+            await db.commit()
+            return len(rows)
+        except Exception as e:
+            logger.error(f"insert_batch validation tasks failed: {e}")
+            return 0
+
+    @staticmethod
+    async def acquire_pending(db: aiosqlite.Connection) -> Optional[dict]:
+        """原子性地获取一个 pending 任务并将其标记为 processing"""
+        try:
+            async with db.execute(
+                """
+                SELECT id, ip, port, protocol FROM validation_tasks
+                WHERE status = 'pending'
+                ORDER BY id ASC
+                LIMIT 1
+                """
+            ) as cursor:
+                row = await cursor.fetchone()
+                if not row:
+                    return None
+                task_id = row[0]
+                await db.execute(
+                    "UPDATE validation_tasks SET status = 'processing', updated_at = CURRENT_TIMESTAMP WHERE id = ?",
+                    (task_id,),
+                )
+                await db.commit()
+                return {"id": task_id, "ip": row[1], "port": row[2], "protocol": row[3]}
+        except Exception as e:
+            logger.error(f"acquire_pending failed: {e}")
+            return None
+
+    @staticmethod
+    async def complete_task(
+        db: aiosqlite.Connection,
+        task_id: int,
+        is_valid: bool,
+        response_time_ms: Optional[float] = None,
+    ) -> bool:
+        try:
+            await db.execute(
+                """
+                UPDATE validation_tasks
+                SET status = 'completed',
+                    result = ?,
+                    response_time_ms = ?,
+                    updated_at = CURRENT_TIMESTAMP
+                WHERE id = ?
+                """,
+                ("valid" if is_valid else "invalid", response_time_ms, task_id),
+            )
+            await db.commit()
+            return True
+        except Exception as e:
+            logger.error(f"complete_task failed: {e}")
+            return False
+
+    @staticmethod
+    async def fail_task(db: aiosqlite.Connection, task_id: int) -> bool:
+        try:
+            await db.execute(
+                """
+                UPDATE validation_tasks
+                SET status = 'failed',
+                    result = 'invalid',
+                    updated_at = CURRENT_TIMESTAMP
+                WHERE id = ?
+                """,
+                (task_id,),
+            )
+            await db.commit()
+            return True
+        except Exception as e:
+            logger.error(f"fail_task failed: {e}")
+            return False
+
+    @staticmethod
+    async def get_pending_count(db: aiosqlite.Connection) -> int:
+        async with db.execute(
+            "SELECT COUNT(*) FROM validation_tasks WHERE status = 'pending'"
+        ) as cursor:
+            row = await cursor.fetchone()
+            return row[0] if row else 0
+
+    @staticmethod
+    async def reset_processing(db: aiosqlite.Connection) -> int:
+        """将异常中断的 processing 任务重置为 pending，用于启动恢复"""
+        try:
+            await db.execute(
+                """
+                UPDATE validation_tasks
+                SET status = 'pending', updated_at = CURRENT_TIMESTAMP
+                WHERE status = 'processing'
+                """
+            )
+            await db.commit()
+            return db.total_changes
+        except Exception as e:
+            logger.error(f"reset_processing failed: {e}")
+            return 0
+
+    @staticmethod
+    async def cleanup_old(db: aiosqlite.Connection, days: int = 7) -> int:
+        try:
+            await db.execute(
+                "DELETE FROM validation_tasks WHERE updated_at < datetime('now', '-{} days')".format(days)
+            )
+            await db.commit()
+            return db.total_changes
+        except Exception as e:
+            logger.error(f"cleanup_old tasks failed: {e}")
+            return 0

services/plugin_service.py

@@ -10,22 +10,25 @@ from core.log import logger
 
 
 class PluginService:
-    """插件业务服务：管理插件生命周期、执行爬取"""
+    """插件业务服务：管理插件生命周期、执行爬取、配置管理"""
 
     def __init__(self):
         self.plugin_settings_repo = PluginSettingsRepository()
         self._stats: dict[str, dict] = {}
 
     async def list_plugins(self) -> List[PluginInfo]:
-        """获取所有插件信息（合并持久化状态）"""
+        """获取所有插件信息（合并持久化状态和配置）"""
         async with get_db() as db:
             db_states = await self.plugin_settings_repo.list_all(db)
 
         result = []
         for plugin in registry.list_plugins():
-            # 如果有持久化状态，覆盖内存状态
+            # 合并持久化状态
-            if plugin.name in db_states:
+            state = db_states.get(plugin.name, {})
-                plugin.enabled = db_states[plugin.name]
+            if "enabled" in state:
+                plugin.enabled = state["enabled"]
+            if "config" in state and isinstance(state["config"], dict):
+                plugin.update_config(state["config"])
 
             stat = self._stats.get(plugin.name, {
                 "success_count": 0,
@@ -55,6 +58,31 @@ class PluginService:
             logger.info(f"Plugin {plugin_id} toggled to {enabled}")
         return success
 
+    async def get_plugin_config(self, plugin_id: str) -> Optional[dict]:
+        """获取插件当前配置（合并默认值和持久化值）"""
+        plugin = registry.get(plugin_id)
+        if not plugin:
+            return None
+        async with get_db() as db:
+            saved = await self.plugin_settings_repo.get_config(db, plugin_id)
+        config = dict(plugin.default_config)
+        if saved:
+            config.update(saved)
+        return config
+
+    async def update_plugin_config(self, plugin_id: str, config: dict) -> bool:
+        """更新插件配置（只保存已存在于 default_config 中的键）"""
+        plugin = registry.get(plugin_id)
+        if not plugin:
+            return False
+        # 过滤非法键
+        safe_config = {k: v for k, v in config.items() if k in plugin.default_config}
+        if not safe_config:
+            return False
+        plugin.update_config(safe_config)
+        async with get_db() as db:
+            return await self.plugin_settings_repo.set_config(db, plugin_id, plugin.config)
+
     def get_plugin(self, plugin_id: str) -> Optional[BaseCrawlerPlugin]:
         return registry.get(plugin_id)