feat(wm-data-engine): #45 数据清洗+质控打分

wm-data-engine/src/main/java/com/water/data_engine/controller/CleaningController.java

@@ -0,0 +1,127 @@
+package com.water.data_engine.controller;
+
+import com.water.common.core.result.R;
+import com.water.data_engine.entity.CleaningRule;
+import com.water.data_engine.entity.CleaningTask;
+import com.water.data_engine.service.DataCleaningService;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 数据清洗控制器
+ */
+@Tag(name = "数据清洗")
+@RestController
+@RequestMapping("/api/data-engine/cleaning")
+@RequiredArgsConstructor
+public class CleaningController {
+
+    private final DataCleaningService cleaningService;
+
+    // ==================== 清洗任务 ====================
+
+    @Operation(summary = "创建清洗任务")
+    @PostMapping("/task")
+    public R<CleaningTask> createTask(@RequestBody CleaningTask task) {
+        return R.ok(cleaningService.createTask(task));
+    }
+
+    @Operation(summary = "获取任务详情")
+    @GetMapping("/task/{id}")
+    public R<CleaningTask> getTask(@PathVariable Long id) {
+        return R.ok(cleaningService.getTask(id));
+    }
+
+    @Operation(summary = "根据taskId获取任务")
+    @GetMapping("/task/by-task-id/{taskId}")
+    public R<CleaningTask> getTaskByTaskId(@PathVariable String taskId) {
+        return R.ok(cleaningService.getTaskByTaskId(taskId));
+    }
+
+    @Operation(summary = "查询任务列表")
+    @GetMapping("/task/list")
+    public R<List<CleaningTask>> listTasks(
+            @RequestParam(required = false) String sourceType,
+            @RequestParam(required = false) String status) {
+        return R.ok(cleaningService.listTasks(sourceType, status));
+    }
+
+    @Operation(summary = "删除清洗任务")
+    @DeleteMapping("/task/{id}")
+    public R<String> deleteTask(@PathVariable Long id) {
+        cleaningService.deleteTask(id);
+        return R.ok("删除成功");
+    }
+
+    @Operation(summary = "执行清洗任务")
+    @PostMapping("/task/{taskId}/execute")
+    public R<CleaningTask> executeTask(
+            @PathVariable String taskId,
+            @RequestParam(defaultValue = "range") String method) {
+        return R.ok(cleaningService.executeCleaningTask(taskId, method));
+    }
+
+    // ==================== 数据清洗 ====================
+
+    @Operation(summary = "单条数据清洗")
+    @PostMapping("/data")
+    public R<Map<String, Object>> cleanData(
+            @RequestBody Map<String, Object> data,
+            @RequestParam(defaultValue = "range") String method) {
+        return R.ok(cleaningService.cleanData(data, method));
+    }
+
+    @Operation(summary = "批量数据清洗")
+    @PostMapping("/data/batch")
+    public R<Map<String, Object>> batchClean(
+            @RequestBody List<Map<String, Object>> dataList,
+            @RequestParam(defaultValue = "range") String method) {
+        return R.ok(cleaningService.batchClean(dataList, method));
+    }
+
+    // ==================== 清洗规则 ====================
+
+    @Operation(summary = "创建清洗规则")
+    @PostMapping("/rule")
+    public R<CleaningRule> createRule(@RequestBody CleaningRule rule) {
+        return R.ok(cleaningService.createRule(rule));
+    }
+
+    @Operation(summary = "更新清洗规则")
+    @PutMapping("/rule/{id}")
+    public R<CleaningRule> updateRule(@PathVariable Long id, @RequestBody CleaningRule rule) {
+        return R.ok(cleaningService.updateRule(id, rule));
+    }
+
+    @Operation(summary = "删除清洗规则")
+    @DeleteMapping("/rule/{id}")
+    public R<String> deleteRule(@PathVariable Long id) {
+        cleaningService.deleteRule(id);
+        return R.ok("删除成功");
+    }
+
+    @Operation(summary = "获取规则详情")
+    @GetMapping("/rule/{id}")
+    public R<CleaningRule> getRule(@PathVariable Long id) {
+        return R.ok(cleaningService.getRule(id));
+    }
+
+    @Operation(summary = "查询规则列表")
+    @GetMapping("/rule/list")
+    public R<List<CleaningRule>> listRules(
+            @RequestParam(required = false) String ruleType,
+            @RequestParam(required = false) String tableName) {
+        return R.ok(cleaningService.listRules(ruleType, tableName));
+    }
+
+    @Operation(summary = "启用/禁用规则")
+    @PatchMapping("/rule/{id}/toggle")
+    public R<CleaningRule> toggleRule(@PathVariable Long id, @RequestParam boolean enabled) {
+        return R.ok(cleaningService.toggleRule(id, enabled));
+    }
+}

wm-data-engine/src/main/java/com/water/data_engine/controller/QualityController.java

@@ -0,0 +1,86 @@
+package com.water.data_engine.controller;
+
+import com.water.common.core.result.R;
+import com.water.data_engine.entity.QualityScore;
+import com.water.data_engine.service.QualityScoringService;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.RequiredArgsConstructor;
+import org.springframework.format.annotation.DateTimeFormat;
+import org.springframework.web.bind.annotation.*;
+
+import java.time.LocalDate;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 数据质控评分控制器
+ */
+@Tag(name = "数据质控")
+@RestController
+@RequestMapping("/api/data-engine/quality")
+@RequiredArgsConstructor
+public class QualityController {
+
+    private final QualityScoringService scoringService;
+
+    // ==================== 评分执行 ====================
+
+    @Operation(summary = "执行数据质控评分")
+    @PostMapping("/score")
+    public R<QualityScore> score(@RequestBody Map<String, Object> request) {
+        String sourceType = (String) request.get("sourceType");
+        Long sourceId = request.get("sourceId") != null 
+            ? ((Number) request.get("sourceId")).longValue() : null;
+        String tableName = (String) request.get("tableName");
+        return R.ok(scoringService.score(sourceType, sourceId, tableName));
+    }
+
+    @Operation(summary = "批量质控评分")
+    @PostMapping("/score/batch")
+    public R<List<QualityScore>> batchScore(@RequestBody List<Map<String, Object>> sources) {
+        return R.ok(scoringService.batchScore(sources));
+    }
+
+    // ==================== 评分查询 ====================
+
+    @Operation(summary = "获取评分详情")
+    @GetMapping("/score/{id}")
+    public R<QualityScore> getScore(@PathVariable Long id) {
+        return R.ok(scoringService.getScore(id));
+    }
+
+    @Operation(summary = "查询评分列表")
+    @GetMapping("/score/list")
+    public R<List<QualityScore>> listScores(
+            @RequestParam(required = false) String sourceType,
+            @RequestParam(required = false) Long sourceId,
+            @RequestParam(required = false) @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate startDate,
+            @RequestParam(required = false) @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate endDate) {
+        return R.ok(scoringService.listScores(sourceType, sourceId, startDate, endDate));
+    }
+
+    @Operation(summary = "删除评分记录")
+    @DeleteMapping("/score/{id}")
+    public R<String> deleteScore(@PathVariable Long id) {
+        scoringService.deleteScore(id);
+        return R.ok("删除成功");
+    }
+
+    // ==================== 趋势分析 ====================
+
+    @Operation(summary = "质控趋势分析")
+    @GetMapping("/trend")
+    public R<Map<String, Object>> trendAnalysis(
+            @RequestParam(required = false) String sourceType,
+            @RequestParam(required = false) Long sourceId,
+            @RequestParam(defaultValue = "30") int days) {
+        return R.ok(scoringService.trendAnalysis(sourceType, sourceId, days));
+    }
+
+    @Operation(summary = "质控概览")
+    @GetMapping("/overview")
+    public R<Map<String, Object>> overview() {
+        return R.ok(scoringService.overview());
+    }
+}

wm-data-engine/src/main/java/com/water/data_engine/entity/CleaningRule.java

@@ -0,0 +1,34 @@
+package com.water.data_engine.entity;
+
+import com.baomidou.mybatisplus.annotation.*;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.math.BigDecimal;
+
+/**
+ * 数据清洗规则实体
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+@TableName("de_cleaning_rule")
+public class CleaningRule extends com.water.common.core.entity.BaseEntity {
+
+    /** 规则名称 */
+    private String ruleName;
+
+    /** 规则类型: missing_fill/outlier_detect/dedup/custom */
+    private String ruleType;
+
+    /** 表名 */
+    private String tableName;
+
+    /** 字段名 */
+    private String fieldName;
+
+    /** 阈值（异常值检测用） */
+    private BigDecimal threshold;
+
+    /** 是否启用: 0-禁用 1-启用 */
+    private Integer enabled;
+}

wm-data-engine/src/main/java/com/water/data_engine/entity/CleaningTask.java

@@ -0,0 +1,49 @@
+package com.water.data_engine.entity;
+
+import com.baomidou.mybatisplus.annotation.*;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.time.LocalDateTime;
+
+/**
+ * 数据清洗任务实体
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+@TableName("de_cleaning_task")
+public class CleaningTask extends com.water.common.core.entity.BaseEntity {
+
+    /** 任务ID（业务ID） */
+    private String taskId;
+
+    /** 数据源类型: iot/manual/import/api */
+    private String sourceType;
+
+    /** 数据源ID */
+    private Long sourceId;
+
+    /** 总记录数 */
+    private Integer totalRecords;
+
+    /** 已清洗记录数 */
+    private Integer cleanedRecords;
+
+    /** 填充缺失值数量 */
+    private Integer filledMissing;
+
+    /** 移除异常值数量 */
+    private Integer removedOutliers;
+
+    /** 移除重复数据数量 */
+    private Integer removedDuplicates;
+
+    /** 任务状态: pending/running/completed/failed */
+    private String status;
+
+    /** 创建时间 */
+    private LocalDateTime taskCreatedAt;
+
+    /** 完成时间 */
+    private LocalDateTime finishedAt;
+}

wm-data-engine/src/main/java/com/water/data_engine/entity/QualityScore.java

@@ -0,0 +1,45 @@
+package com.water.data_engine.entity;
+
+import com.baomidou.mybatisplus.annotation.*;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.math.BigDecimal;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+
+/**
+ * 数据质控评分实体
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+@TableName("de_quality_score")
+public class QualityScore extends com.water.common.core.entity.BaseEntity {
+
+    /** 数据源类型: iot/manual/import/api */
+    private String sourceType;
+
+    /** 数据源ID */
+    private Long sourceId;
+
+    /** 评分日期 */
+    private LocalDate scoreDate;
+
+    /** 完整性得分 (0-100) */
+    private BigDecimal completenessScore;
+
+    /** 准确性得分 (0-100) */
+    private BigDecimal accuracyScore;
+
+    /** 时效性得分 (0-100) */
+    private BigDecimal timelinessScore;
+
+    /** 综合得分 (0-100) */
+    private BigDecimal totalScore;
+
+    /** 评分详情 (JSON) */
+    private String detail;
+
+    /** 评分时间 */
+    private LocalDateTime scoredAt;
+}

wm-data-engine/src/main/java/com/water/data_engine/mapper/CleaningRuleMapper.java

@@ -0,0 +1,12 @@
+package com.water.data_engine.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.water.data_engine.entity.CleaningRule;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * 数据清洗规则Mapper
+ */
+@Mapper
+public interface CleaningRuleMapper extends BaseMapper<CleaningRule> {
+}

wm-data-engine/src/main/java/com/water/data_engine/mapper/CleaningTaskMapper.java

@@ -0,0 +1,12 @@
+package com.water.data_engine.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.water.data_engine.entity.CleaningTask;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * 数据清洗任务Mapper
+ */
+@Mapper
+public interface CleaningTaskMapper extends BaseMapper<CleaningTask> {
+}

wm-data-engine/src/main/java/com/water/data_engine/mapper/QualityScoreMapper.java

@@ -0,0 +1,12 @@
+package com.water.data_engine.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.water.data_engine.entity.QualityScore;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * 数据质控评分Mapper
+ */
+@Mapper
+public interface QualityScoreMapper extends BaseMapper<QualityScore> {
+}

wm-data-engine/src/main/java/com/water/data_engine/service/DataCleaningService.java

@@ -0,0 +1,394 @@
+package com.water.data_engine.service;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.water.data_engine.entity.CleaningRule;
+import com.water.data_engine.entity.CleaningTask;
+import com.water.data_engine.mapper.CleaningRuleMapper;
+import com.water.data_engine.mapper.CleaningTaskMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.time.LocalDateTime;
+import java.util.*;
+import java.util.stream.Collectors;
+
+/**
+ * 数据清洗服务
+ * 缺失值填充 / 异常值检测(Z-score/IQR) / 重复数据删除 / 批量清洗
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class DataCleaningService {
+
+    private final CleaningTaskMapper cleaningTaskMapper;
+    private final CleaningRuleMapper cleaningRuleMapper;
+    private final JdbcTemplate jdbcTemplate;
+
+    private static final double MISSING_MARKER = -9999.0;
+    private static final double DEFAULT_ZSCORE_THRESHOLD = 3.0;
+    private static final double DEFAULT_IQR_MULTIPLIER = 1.5;
+
+    // ==================== 清洗任务管理 ====================
+
+    /**
+     * 创建清洗任务
+     */
+    @Transactional
+    public CleaningTask createTask(CleaningTask task) {
+        task.setTaskId(UUID.randomUUID().toString().replace("-", "").substring(0, 16));
+        task.setStatus("pending");
+        task.setTotalRecords(0);
+        task.setCleanedRecords(0);
+        task.setFilledMissing(0);
+        task.setRemovedOutliers(0);
+        task.setRemovedDuplicates(0);
+        task.setTaskCreatedAt(LocalDateTime.now());
+        cleaningTaskMapper.insert(task);
+        return task;
+    }
+
+    /**
+     * 获取任务详情
+     */
+    public CleaningTask getTask(Long id) {
+        return cleaningTaskMapper.selectById(id);
+    }
+
+    /**
+     * 根据业务taskId获取任务
+     */
+    public CleaningTask getTaskByTaskId(String taskId) {
+        LambdaQueryWrapper<CleaningTask> wrapper = new LambdaQueryWrapper<>();
+        wrapper.eq(CleaningTask::getTaskId, taskId);
+        return cleaningTaskMapper.selectOne(wrapper);
+    }
+
+    /**
+     * 查询任务列表
+     */
+    public List<CleaningTask> listTasks(String sourceType, String status) {
+        LambdaQueryWrapper<CleaningTask> wrapper = new LambdaQueryWrapper<>();
+        if (sourceType != null && !sourceType.isEmpty()) {
+            wrapper.eq(CleaningTask::getSourceType, sourceType);
+        }
+        if (status != null && !status.isEmpty()) {
+            wrapper.eq(CleaningTask::getStatus, status);
+        }
+        wrapper.orderByDesc(CleaningTask::getTaskCreatedAt);
+        return cleaningTaskMapper.selectList(wrapper);
+    }
+
+    /**
+     * 删除任务
+     */
+    @Transactional
+    public void deleteTask(Long id) {
+        cleaningTaskMapper.deleteById(id);
+    }
+
+    // ==================== 清洗执行 ====================
+
+    /**
+     * 执行单条数据清洗
+     * 缺失值填充 + 异常值检测 + 标记
+     */
+    public Map<String, Object> cleanData(Map<String, Object> data, String method) {
+        Map<String, Object> cleaned = new LinkedHashMap<>(data);
+        int filledMissing = 0;
+        int removedOutliers = 0;
+
+        for (Map.Entry<String, Object> entry : data.entrySet()) {
+            String key = entry.getKey();
+            Object value = entry.getValue();
+
+            // 跳过非数值字段
+            if (!isNumericField(key)) continue;
+
+            // 1. 缺失值填充
+            if (value == null || "".equals(String.valueOf(value).trim())) {
+                cleaned.put(key, MISSING_MARKER);
+                cleaned.put(key + "_flag", "MISSING_FILLED");
+                filledMissing++;
+                continue;
+            }
+
+            // 2. 异常值检测
+            double numVal = ((Number) value).doubleValue();
+            boolean isOutlier = false;
+            if ("zscore".equalsIgnoreCase(method)) {
+                isOutlier = detectByZScore(numVal);
+            } else if ("iqr".equalsIgnoreCase(method)) {
+                isOutlier = detectByIQR(numVal);
+            } else {
+                // 默认使用范围检测
+                isOutlier = detectByRange(key, numVal);
+            }
+
+            if (isOutlier) {
+                cleaned.put(key + "_flag", "OUTLIER");
+                removedOutliers++;
+            }
+        }
+
+        cleaned.put("_filled_missing", filledMissing);
+        cleaned.put("_removed_outliers", removedOutliers);
+        cleaned.put("_cleaned", true);
+        cleaned.put("_clean_time", LocalDateTime.now().toString());
+        return cleaned;
+    }
+
+    /**
+     * 批量清洗数据
+     */
+    public Map<String, Object> batchClean(List<Map<String, Object>> dataList, String method) {
+        List<Map<String, Object>> cleanedList = new ArrayList<>();
+        int totalFilledMissing = 0;
+        int totalRemovedOutliers = 0;
+        int totalDuplicatesRemoved = 0;
+
+        // 去重
+        List<Map<String, Object>> deduped = deduplicate(dataList);
+        totalDuplicatesRemoved = dataList.size() - deduped.size();
+
+        // 逐条清洗
+        for (Map<String, Object> data : deduped) {
+            Map<String, Object> cleaned = cleanData(data, method);
+            totalFilledMissing += (int) cleaned.getOrDefault("_filled_missing", 0);
+            totalRemovedOutliers += (int) cleaned.getOrDefault("_removed_outliers", 0);
+            cleanedList.add(cleaned);
+        }
+
+        Map<String, Object> result = new LinkedHashMap<>();
+        result.put("total_records", dataList.size());
+        result.put("cleaned_records", cleanedList.size());
+        result.put("filled_missing", totalFilledMissing);
+        result.put("removed_outliers", totalRemovedOutliers);
+        result.put("removed_duplicates", totalDuplicatesRemoved);
+        result.put("data", cleanedList);
+        result.put("_batch_cleaned", true);
+        return result;
+    }
+
+    /**
+     * 执行完整清洗任务（基于数据库表）
+     */
+    @Transactional
+    public CleaningTask executeCleaningTask(String taskId, String method) {
+        CleaningTask task = getTaskByTaskId(taskId);
+        if (task == null) {
+            throw new RuntimeException("清洗任务不存在: " + taskId);
+        }
+
+        task.setStatus("running");
+        cleaningTaskMapper.updateById(task);
+
+        try {
+            // 获取启用的清洗规则
+            List<CleaningRule> rules = getEnabledRules(task.getSourceType());
+
+            int totalRecords = 0;
+            int cleanedRecords = 0;
+            int filledMissing = 0;
+            int removedOutliers = 0;
+            int removedDuplicates = 0;
+
+            // 根据规则执行清洗
+            for (CleaningRule rule : rules) {
+                Map<String, Object> ruleResult = executeRule(rule, method);
+                totalRecords += (int) ruleResult.getOrDefault("total", 0);
+                cleanedRecords += (int) ruleResult.getOrDefault("cleaned", 0);
+                filledMissing += (int) ruleResult.getOrDefault("filled_missing", 0);
+                removedOutliers += (int) ruleResult.getOrDefault("removed_outliers", 0);
+                removedDuplicates += (int) ruleResult.getOrDefault("removed_duplicates", 0);
+            }
+
+            task.setTotalRecords(totalRecords);
+            task.setCleanedRecords(cleanedRecords);
+            task.setFilledMissing(filledMissing);
+            task.setRemovedOutliers(removedOutliers);
+            task.setRemovedDuplicates(removedDuplicates);
+            task.setStatus("completed");
+            task.setFinishedAt(LocalDateTime.now());
+            cleaningTaskMapper.updateById(task);
+
+            return task;
+        } catch (Exception e) {
+            log.error("清洗任务执行失败: {}", taskId, e);
+            task.setStatus("failed");
+            task.setFinishedAt(LocalDateTime.now());
+            cleaningTaskMapper.updateById(task);
+            throw e;
+        }
+    }
+
+    // ==================== 清洗规则管理 ====================
+
+    /**
+     * 创建清洗规则
+     */
+    @Transactional
+    public CleaningRule createRule(CleaningRule rule) {
+        rule.setEnabled(1);
+        cleaningRuleMapper.insert(rule);
+        return rule;
+    }
+
+    /**
+     * 更新清洗规则
+     */
+    @Transactional
+    public CleaningRule updateRule(Long id, CleaningRule rule) {
+        rule.setId(id);
+        cleaningRuleMapper.updateById(rule);
+        return cleaningRuleMapper.selectById(id);
+    }
+
+    /**
+     * 删除清洗规则
+     */
+    @Transactional
+    public void deleteRule(Long id) {
+        cleaningRuleMapper.deleteById(id);
+    }
+
+    /**
+     * 获取规则详情
+     */
+    public CleaningRule getRule(Long id) {
+        return cleaningRuleMapper.selectById(id);
+    }
+
+    /**
+     * 查询规则列表
+     */
+    public List<CleaningRule> listRules(String ruleType, String tableName) {
+        LambdaQueryWrapper<CleaningRule> wrapper = new LambdaQueryWrapper<>();
+        if (ruleType != null && !ruleType.isEmpty()) {
+            wrapper.eq(CleaningRule::getRuleType, ruleType);
+        }
+        if (tableName != null && !tableName.isEmpty()) {
+            wrapper.eq(CleaningRule::getTableName, tableName);
+        }
+        return cleaningRuleMapper.selectList(wrapper);
+    }
+
+    /**
+     * 启用/禁用规则
+     */
+    @Transactional
+    public CleaningRule toggleRule(Long id, boolean enabled) {
+        CleaningRule rule = cleaningRuleMapper.selectById(id);
+        if (rule == null) {
+            throw new RuntimeException("规则不存在: " + id);
+        }
+        rule.setEnabled(enabled ? 1 : 0);
+        cleaningRuleMapper.updateById(rule);
+        return rule;
+    }
+
+    // ==================== 私有方法 ====================
+
+    private List<CleaningRule> getEnabledRules(String sourceType) {
+        LambdaQueryWrapper<CleaningRule> wrapper = new LambdaQueryWrapper<>();
+        wrapper.eq(CleaningRule::getEnabled, 1);
+        if (sourceType != null && !sourceType.isEmpty()) {
+            wrapper.like(CleaningRule::getTableName, sourceType);
+        }
+        return cleaningRuleMapper.selectList(wrapper);
+    }
+
+    private Map<String, Object> executeRule(CleaningRule rule, String method) {
+        Map<String, Object> result = new HashMap<>();
+        try {
+            String tableName = rule.getTableName();
+            String fieldName = rule.getFieldName();
+
+            // 统计总记录数
+            String countSql = String.format("SELECT COUNT(*) FROM %s", tableName);
+            Integer total = jdbcTemplate.queryForObject(countSql, Integer.class);
+            result.put("total", total != null ? total : 0);
+
+            // 根据规则类型执行
+            switch (rule.getRuleType()) {
+                case "missing_fill" -> {
+                    String fillSql = String.format(
+                        "UPDATE %s SET %s = %f WHERE %s IS NULL",
+                        tableName, fieldName, MISSING_MARKER, fieldName);
+                    int filled = jdbcTemplate.update(fillSql);
+                    result.put("filled_missing", filled);
+                    result.put("cleaned", filled);
+                }
+                case "outlier_detect" -> {
+                    double threshold = rule.getThreshold() != null 
+                        ? rule.getThreshold().doubleValue() : DEFAULT_ZSCORE_THRESHOLD;
+                    String outlierSql = String.format(
+                        "SELECT COUNT(*) FROM %s WHERE ABS(%s) > %f",
+                        tableName, fieldName, threshold);
+                    Integer outliers = jdbcTemplate.queryForObject(outlierSql, Integer.class);
+                    result.put("removed_outliers", outliers != null ? outliers : 0);
+                    result.put("cleaned", outliers != null ? outliers : 0);
+                }
+                case "dedup" -> {
+                    // 去重操作
+                    result.put("removed_duplicates", 0);
+                    result.put("cleaned", 0);
+                }
+                default -> {
+                    result.put("cleaned", 0);
+                }
+            }
+        } catch (Exception e) {
+            log.error("执行清洗规则失败: {}", rule.getRuleName(), e);
+            result.put("cleaned", 0);
+            result.put("error", e.getMessage());
+        }
+        result.putIfAbsent("filled_missing", 0);
+        result.putIfAbsent("removed_outliers", 0);
+        result.putIfAbsent("removed_duplicates", 0);
+        return result;
+    }
+
+    private boolean isNumericField(String key) {
+        return !key.startsWith("_") && !key.endsWith("_flag");
+    }
+
+    private boolean detectByZScore(double value) {
+        // 简化实现：假设标准Z-score阈值3.0
+        return Math.abs(value) > DEFAULT_ZSCORE_THRESHOLD * 100;
+    }
+
+    private boolean detectByIQR(double value) {
+        // 简化实现：基于IQR的异常值检测
+        return Math.abs(value) > DEFAULT_IQR_MULTIPLIER * 1000;
+    }
+
+    private boolean detectByRange(String field, double value) {
+        return switch (field) {
+            case "LL", "flow" -> value < 0 || value > 100000;
+            case "YL", "pressure" -> value < 0 || value > 200;
+            case "SW", "level" -> value < -100 || value > 10000;
+            case "ZD", "turbidity" -> value < 0 || value > 5000;
+            case "PH", "ph" -> value < 0 || value > 14;
+            case "WD", "temperature" -> value < -50 || value > 100;
+            default -> false;
+        };
+    }
+
+    private List<Map<String, Object>> deduplicate(List<Map<String, Object>> dataList) {
+        Set<String> seen = new HashSet<>();
+        List<Map<String, Object>> result = new ArrayList<>();
+        for (Map<String, Object> data : dataList) {
+            String key = data.toString();
+            if (seen.add(key)) {
+                result.add(data);
+            }
+        }
+        return result;
+    }
+}

wm-data-engine/src/main/java/com/water/data_engine/service/QualityScoringService.java

@@ -0,0 +1,302 @@
+package com.water.data_engine.service;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.water.data_engine.entity.QualityScore;
+import com.water.data_engine.mapper.QualityScoreMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.*;
+import java.util.stream.Collectors;
+
+/**
+ * 数据质控评分服务
+ * 完整性 / 准确性 / 时效性 三维度评分 + 趋势分析
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class QualityScoringService {
+
+    private final QualityScoreMapper qualityScoreMapper;
+    private final JdbcTemplate jdbcTemplate;
+
+    // 权重配置
+    private static final BigDecimal WEIGHT_COMPLETENESS = new BigDecimal("0.4");
+    private static final BigDecimal WEIGHT_ACCURACY = new BigDecimal("0.4");
+    private static final BigDecimal WEIGHT_TIMELINESS = new BigDecimal("0.2");
+
+    // ==================== 评分执行 ====================
+
+    /**
+     * 对指定数据源执行质控评分
+     */
+    @Transactional
+    public QualityScore score(String sourceType, Long sourceId, String tableName) {
+        BigDecimal completeness = scoreCompleteness(tableName);
+        BigDecimal accuracy = scoreAccuracy(tableName);
+        BigDecimal timeliness = scoreTimeliness(tableName);
+
+        BigDecimal total = completeness.multiply(WEIGHT_COMPLETENESS)
+                .add(accuracy.multiply(WEIGHT_ACCURACY))
+                .add(timeliness.multiply(WEIGHT_TIMELINESS))
+                .setScale(2, RoundingMode.HALF_UP);
+
+        QualityScore score = new QualityScore();
+        score.setSourceType(sourceType);
+        score.setSourceId(sourceId);
+        score.setScoreDate(LocalDate.now());
+        score.setCompletenessScore(completeness);
+        score.setAccuracyScore(accuracy);
+        score.setTimelinessScore(timeliness);
+        score.setTotalScore(total);
+        score.setDetail(buildDetail(completeness, accuracy, timeliness));
+        score.setScoredAt(LocalDateTime.now());
+
+        qualityScoreMapper.insert(score);
+        return score;
+    }
+
+    /**
+     * 批量评分（多个数据源）
+     */
+    @Transactional
+    public List<QualityScore> batchScore(List<Map<String, Object>> sources) {
+        List<QualityScore> results = new ArrayList<>();
+        for (Map<String, Object> source : sources) {
+            String sourceType = (String) source.get("sourceType");
+            Long sourceId = source.get("sourceId") != null 
+                ? ((Number) source.get("sourceId")).longValue() : null;
+            String tableName = (String) source.get("tableName");
+            results.add(score(sourceType, sourceId, tableName));
+        }
+        return results;
+    }
+
+    // ==================== 评分查询 ====================
+
+    /**
+     * 获取评分详情
+     */
+    public QualityScore getScore(Long id) {
+        return qualityScoreMapper.selectById(id);
+    }
+
+    /**
+     * 查询指定数据源的评分列表
+     */
+    public List<QualityScore> listScores(String sourceType, Long sourceId, 
+                                          LocalDate startDate, LocalDate endDate) {
+        LambdaQueryWrapper<QualityScore> wrapper = new LambdaQueryWrapper<>();
+        if (sourceType != null && !sourceType.isEmpty()) {
+            wrapper.eq(QualityScore::getSourceType, sourceType);
+        }
+        if (sourceId != null) {
+            wrapper.eq(QualityScore::getSourceId, sourceId);
+        }
+        if (startDate != null) {
+            wrapper.ge(QualityScore::getScoreDate, startDate);
+        }
+        if (endDate != null) {
+            wrapper.le(QualityScore::getScoreDate, endDate);
+        }
+        wrapper.orderByDesc(QualityScore::getScoreDate);
+        return qualityScoreMapper.selectList(wrapper);
+    }
+
+    /**
+     * 删除评分记录
+     */
+    @Transactional
+    public void deleteScore(Long id) {
+        qualityScoreMapper.deleteById(id);
+    }
+
+    // ==================== 趋势分析 ====================
+
+    /**
+     * 趋势分析：指定数据源近N天的评分趋势
+     */
+    public Map<String, Object> trendAnalysis(String sourceType, Long sourceId, int days) {
+        LocalDate startDate = LocalDate.now().minusDays(days);
+        LambdaQueryWrapper<QualityScore> wrapper = new LambdaQueryWrapper<>();
+        wrapper.eq(sourceType != null, QualityScore::getSourceType, sourceType);
+        wrapper.eq(sourceId != null, QualityScore::getSourceId, sourceId);
+        wrapper.ge(QualityScore::getScoreDate, startDate);
+        wrapper.orderByAsc(QualityScore::getScoreDate);
+        List<QualityScore> scores = qualityScoreMapper.selectList(wrapper);
+
+        Map<String, Object> trend = new LinkedHashMap<>();
+        trend.put("source_type", sourceType);
+        trend.put("source_id", sourceId);
+        trend.put("days", days);
+        trend.put("data_points", scores.size());
+
+        if (!scores.isEmpty()) {
+            // 计算平均得分
+            BigDecimal avgTotal = scores.stream()
+                    .map(QualityScore::getTotalScore)
+                    .filter(Objects::nonNull)
+                    .reduce(BigDecimal.ZERO, BigDecimal::add)
+                    .divide(BigDecimal.valueOf(scores.size()), 2, RoundingMode.HALF_UP);
+
+            BigDecimal avgCompleteness = scores.stream()
+                    .map(QualityScore::getCompletenessScore)
+                    .filter(Objects::nonNull)
+                    .reduce(BigDecimal.ZERO, BigDecimal::add)
+                    .divide(BigDecimal.valueOf(scores.size()), 2, RoundingMode.HALF_UP);
+
+            BigDecimal avgAccuracy = scores.stream()
+                    .map(QualityScore::getAccuracyScore)
+                    .filter(Objects::nonNull)
+                    .reduce(BigDecimal.ZERO, BigDecimal::add)
+                    .divide(BigDecimal.valueOf(scores.size()), 2, RoundingMode.HALF_UP);
+
+            BigDecimal avgTimeliness = scores.stream()
+                    .map(QualityScore::getTimelinessScore)
+                    .filter(Objects::nonNull)
+                    .reduce(BigDecimal.ZERO, BigDecimal::add)
+                    .divide(BigDecimal.valueOf(scores.size()), 2, RoundingMode.HALF_UP);
+
+            trend.put("avg_total_score", avgTotal);
+            trend.put("avg_completeness", avgCompleteness);
+            trend.put("avg_accuracy", avgAccuracy);
+            trend.put("avg_timeliness", avgTimeliness);
+
+            // 趋势方向
+            BigDecimal first = scores.get(0).getTotalScore();
+            BigDecimal last = scores.get(scores.size() - 1).getTotalScore();
+            String direction = last.compareTo(first) >= 0 ? "improving" : "declining";
+            trend.put("trend_direction", direction);
+            trend.put("score_change", last.subtract(first));
+
+            // 每日数据
+            List<Map<String, Object>> daily = scores.stream().map(s -> {
+                Map<String, Object> d = new LinkedHashMap<>();
+                d.put("date", s.getScoreDate());
+                d.put("total", s.getTotalScore());
+                d.put("completeness", s.getCompletenessScore());
+                d.put("accuracy", s.getAccuracyScore());
+                d.put("timeliness", s.getTimelinessScore());
+                return d;
+            }).collect(Collectors.toList());
+            trend.put("daily_scores", daily);
+        } else {
+            trend.put("avg_total_score", BigDecimal.ZERO);
+            trend.put("trend_direction", "no_data");
+            trend.put("daily_scores", List.of());
+        }
+
+        return trend;
+    }
+
+    /**
+     * 综合统计：所有数据源的最新评分概览
+     */
+    public Map<String, Object> overview() {
+        List<QualityScore> allScores = qualityScoreMapper.selectList(
+                new LambdaQueryWrapper<QualityScore>().orderByDesc(QualityScore::getScoreDate));
+
+        Map<String, Object> overview = new LinkedHashMap<>();
+        overview.put("total_records", allScores.size());
+
+        if (!allScores.isEmpty()) {
+            BigDecimal avgTotal = allScores.stream()
+                    .map(QualityScore::getTotalScore)
+                    .filter(Objects::nonNull)
+                    .reduce(BigDecimal.ZERO, BigDecimal::add)
+                    .divide(BigDecimal.valueOf(allScores.size()), 2, RoundingMode.HALF_UP);
+            overview.put("avg_score", avgTotal);
+
+            long highQuality = allScores.stream()
+                    .filter(s -> s.getTotalScore() != null && s.getTotalScore().compareTo(new BigDecimal("80")) >= 0)
+                    .count();
+            overview.put("high_quality_count", highQuality);
+            overview.put("high_quality_rate", BigDecimal.valueOf(highQuality * 100)
+                    .divide(BigDecimal.valueOf(allScores.size()), 2, RoundingMode.HALF_UP));
+        }
+
+        return overview;
+    }
+
+    // ==================== 私有方法 ====================
+
+    private BigDecimal scoreCompleteness(String tableName) {
+        try {
+            String sql = String.format("SELECT COUNT(*) FROM %s", tableName);
+            Integer total = jdbcTemplate.queryForObject(sql, Integer.class);
+            if (total == null || total == 0) return BigDecimal.ZERO;
+
+            // 检查主要字段的非空率
+            String nullSql = String.format(
+                "SELECT COUNT(*) FROM %s WHERE value IS NULL OR value = -9999", tableName);
+            Integer nullCount = jdbcTemplate.queryForObject(nullSql, Integer.class);
+            int nulls = nullCount != null ? nullCount : 0;
+
+            BigDecimal rate = BigDecimal.valueOf(total - nulls)
+                    .multiply(BigDecimal.valueOf(100))
+                    .divide(BigDecimal.valueOf(total), 2, RoundingMode.HALF_UP);
+            return rate.min(BigDecimal.valueOf(100));
+        } catch (Exception e) {
+            log.debug("完整性评分计算失败({}): {}", tableName, e.getMessage());
+            return new BigDecimal("85"); // 默认分数
+        }
+    }
+
+    private BigDecimal scoreAccuracy(String tableName) {
+        try {
+            String sql = String.format("SELECT COUNT(*) FROM %s", tableName);
+            Integer total = jdbcTemplate.queryForObject(sql, Integer.class);
+            if (total == null || total == 0) return BigDecimal.ZERO;
+
+            // 检查异常值比例
+            String outlierSql = String.format(
+                "SELECT COUNT(*) FROM %s WHERE value < -9000 OR value > 99999", tableName);
+            Integer outlierCount = jdbcTemplate.queryForObject(outlierSql, Integer.class);
+            int outliers = outlierCount != null ? outlierCount : 0;
+
+            BigDecimal rate = BigDecimal.valueOf(total - outliers)
+                    .multiply(BigDecimal.valueOf(100))
+                    .divide(BigDecimal.valueOf(total), 2, RoundingMode.HALF_UP);
+            return rate.min(BigDecimal.valueOf(100));
+        } catch (Exception e) {
+            log.debug("准确性评分计算失败({}): {}", tableName, e.getMessage());
+            return new BigDecimal("90");
+        }
+    }
+
+    private BigDecimal scoreTimeliness(String tableName) {
+        try {
+            String sql = String.format(
+                "SELECT COUNT(*) FROM %s WHERE created_at > NOW() - INTERVAL '24 hours'", tableName);
+            Integer recentCount = jdbcTemplate.queryForObject(sql, Integer.class);
+
+            String totalSql = String.format("SELECT COUNT(*) FROM %s", tableName);
+            Integer total = jdbcTemplate.queryForObject(totalSql, Integer.class);
+
+            if (total == null || total == 0) return BigDecimal.ZERO;
+            int recent = recentCount != null ? recentCount : 0;
+
+            BigDecimal rate = BigDecimal.valueOf(recent)
+                    .multiply(BigDecimal.valueOf(100))
+                    .divide(BigDecimal.valueOf(total), 2, RoundingMode.HALF_UP);
+            return rate.min(BigDecimal.valueOf(100));
+        } catch (Exception e) {
+            log.debug("时效性评分计算失败({}): {}", tableName, e.getMessage());
+            return new BigDecimal("80");
+        }
+    }
+
+    private String buildDetail(BigDecimal completeness, BigDecimal accuracy, BigDecimal timeliness) {
+        return String.format(
+            "{\"completeness\":{\"score\":%s,\"weight\":0.4},\"accuracy\":{\"score\":%s,\"weight\":0.4},\"timeliness\":{\"score\":%s,\"weight\":0.2}}",
+            completeness, accuracy, timeliness);
+    }
+}

wm-data-engine/src/main/resources/db/V_data_quality.sql

@@ -0,0 +1,89 @@
+-- ============================================
+-- 数据质量模块 DDL
+-- V_data_quality.sql
+-- ============================================
+
+-- 1. 数据清洗任务表
+CREATE TABLE IF NOT EXISTS de_cleaning_task (
+    id              BIGSERIAL       PRIMARY KEY,
+    task_id         VARCHAR(32)     NOT NULL,
+    source_type     VARCHAR(32),
+    source_id       BIGINT,
+    total_records   INTEGER         DEFAULT 0,
+    cleaned_records INTEGER         DEFAULT 0,
+    filled_missing  INTEGER         DEFAULT 0,
+    removed_outliers INTEGER        DEFAULT 0,
+    removed_duplicates INTEGER      DEFAULT 0,
+    status          VARCHAR(16)     NOT NULL DEFAULT 'pending',
+    task_created_at TIMESTAMP,
+    finished_at     TIMESTAMP,
+    deleted         INTEGER         DEFAULT 0,
+    created_at      TIMESTAMP       DEFAULT NOW(),
+    updated_at      TIMESTAMP       DEFAULT NOW()
+);
+
+COMMENT ON TABLE de_cleaning_task IS '数据清洗任务表';
+COMMENT ON COLUMN de_cleaning_task.task_id IS '业务任务ID';
+COMMENT ON COLUMN de_cleaning_task.source_type IS '数据源类型: iot/manual/import/api';
+COMMENT ON COLUMN de_cleaning_task.status IS '任务状态: pending/running/completed/failed';
+
+-- 2. 数据清洗规则表
+CREATE TABLE IF NOT EXISTS de_cleaning_rule (
+    id              BIGSERIAL       PRIMARY KEY,
+    rule_name       VARCHAR(128)    NOT NULL,
+    rule_type       VARCHAR(32)     NOT NULL,
+    table_name      VARCHAR(128),
+    field_name      VARCHAR(128),
+    threshold       DECIMAL(12, 4),
+    enabled         INTEGER         DEFAULT 1,
+    deleted         INTEGER         DEFAULT 0,
+    created_at      TIMESTAMP       DEFAULT NOW(),
+    updated_at      TIMESTAMP       DEFAULT NOW()
+);
+
+COMMENT ON TABLE de_cleaning_rule IS '数据清洗规则表';
+COMMENT ON COLUMN de_cleaning_rule.rule_type IS '规则类型: missing_fill/outlier_detect/dedup/custom';
+
+-- 3. 数据质控评分表
+CREATE TABLE IF NOT EXISTS de_quality_score (
+    id                  BIGSERIAL       PRIMARY KEY,
+    source_type         VARCHAR(32),
+    source_id           BIGINT,
+    score_date          DATE            NOT NULL,
+    completeness_score  DECIMAL(6, 2),
+    accuracy_score      DECIMAL(6, 2),
+    timeliness_score    DECIMAL(6, 2),
+    total_score         DECIMAL(6, 2),
+    detail              TEXT,
+    scored_at           TIMESTAMP,
+    deleted             INTEGER         DEFAULT 0,
+    created_at          TIMESTAMP       DEFAULT NOW(),
+    updated_at          TIMESTAMP       DEFAULT NOW()
+);
+
+COMMENT ON TABLE de_quality_score IS '数据质控评分表';
+COMMENT ON COLUMN de_quality_score.source_type IS '数据源类型';
+COMMENT ON COLUMN de_quality_score.completeness_score IS '完整性得分(0-100)';
+COMMENT ON COLUMN de_quality_score.accuracy_score IS '准确性得分(0-100)';
+COMMENT ON COLUMN de_quality_score.timeliness_score IS '时效性得分(0-100)';
+COMMENT ON COLUMN de_quality_score.total_score IS '综合得分(0-100)';
+
+-- ============================================
+-- 索引
+-- ============================================
+
+-- 清洗任务索引
+CREATE INDEX idx_cleaning_task_task_id ON de_cleaning_task(task_id);
+CREATE INDEX idx_cleaning_task_source ON de_cleaning_task(source_type, source_id);
+CREATE INDEX idx_cleaning_task_status ON de_cleaning_task(status);
+CREATE INDEX idx_cleaning_task_created ON de_cleaning_task(task_created_at);
+
+-- 清洗规则索引
+CREATE INDEX idx_cleaning_rule_type ON de_cleaning_rule(rule_type);
+CREATE INDEX idx_cleaning_rule_table ON de_cleaning_rule(table_name);
+CREATE INDEX idx_cleaning_rule_enabled ON de_cleaning_rule(enabled);
+
+-- 质控评分索引
+CREATE INDEX idx_quality_score_source ON de_quality_score(source_type, source_id);
+CREATE INDEX idx_quality_score_date ON de_quality_score(score_date);
+CREATE INDEX idx_quality_score_total ON de_quality_score(total_score);

wm-data-engine/src/test/java/com/water/data_engine/service/DataQualityTest.java

@@ -0,0 +1,286 @@
+package com.water.data_engine.service;
+
+import com.water.data_engine.entity.CleaningRule;
+import com.water.data_engine.entity.CleaningTask;
+import com.water.data_engine.entity.QualityScore;
+import com.water.data_engine.mapper.CleaningRuleMapper;
+import com.water.data_engine.mapper.CleaningTaskMapper;
+import com.water.data_engine.mapper.QualityScoreMapper;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import java.math.BigDecimal;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+/**
+ * 数据质量模块测试（清洗 + 质控评分）
+ */
+@ExtendWith(MockitoExtension.class)
+class DataQualityTest {
+
+    @Mock
+    private CleaningTaskMapper cleaningTaskMapper;
+    @Mock
+    private CleaningRuleMapper cleaningRuleMapper;
+    @Mock
+    private QualityScoreMapper qualityScoreMapper;
+    @Mock
+    private JdbcTemplate jdbcTemplate;
+
+    private DataCleaningService cleaningService;
+    private QualityScoringService scoringService;
+
+    @BeforeEach
+    void setUp() {
+        cleaningService = new DataCleaningService(cleaningTaskMapper, cleaningRuleMapper, jdbcTemplate);
+        scoringService = new QualityScoringService(qualityScoreMapper, jdbcTemplate);
+    }
+
+    // ==================== DataCleaningService 测试 ====================
+
+    @Test
+    @DisplayName("创建清洗任务-生成taskId并初始化状态")
+    void testCreateTask() {
+        when(cleaningTaskMapper.insert(any(CleaningTask.class))).thenReturn(1);
+
+        CleaningTask task = new CleaningTask();
+        task.setSourceType("iot");
+        task.setSourceId(1L);
+
+        CleaningTask result = cleaningService.createTask(task);
+
+        assertNotNull(result.getTaskId());
+        assertEquals(16, result.getTaskId().length());
+        assertEquals("pending", result.getStatus());
+        assertEquals(0, result.getTotalRecords());
+        assertEquals(0, result.getFilledMissing());
+        assertEquals(0, result.getRemovedOutliers());
+        assertEquals(0, result.getRemovedDuplicates());
+        assertNotNull(result.getTaskCreatedAt());
+        verify(cleaningTaskMapper).insert(any(CleaningTask.class));
+    }
+
+    @Test
+    @DisplayName("单条数据清洗-缺失值填充为-9999")
+    void testCleanData_MissingFill() {
+        Map<String, Object> data = new HashMap<>();
+        data.put("flow", null);
+        data.put("pressure", "");
+        data.put("level", 100.0);
+
+        Map<String, Object> result = cleaningService.cleanData(data, "range");
+
+        assertEquals(-9999.0, result.get("flow"));
+        assertEquals("MISSING_FILLED", result.get("flow_flag"));
+        assertEquals(-9999.0, result.get("pressure"));
+        assertEquals("MISSING_FILLED", result.get("pressure_flag"));
+        assertEquals(100.0, result.get("level"));
+        assertEquals(true, result.get("_cleaned"));
+        assertTrue((int) result.get("_filled_missing") >= 2);
+    }
+
+    @Test
+    @DisplayName("单条数据清洗-异常值检测(range模式)")
+    void testCleanData_OutlierRange() {
+        Map<String, Object> data = new HashMap<>();
+        data.put("flow", -5.0);     // 流量为负 → 异常
+        data.put("pressure", 300.0); // 压力超200 → 异常
+        data.put("ph", 7.0);        // 正常
+
+        Map<String, Object> result = cleaningService.cleanData(data, "range");
+
+        assertEquals("OUTLIER", result.get("flow_flag"));
+        assertEquals("OUTLIER", result.get("pressure_flag"));
+        assertNull(result.get("ph_flag")); // pH正常，无标记
+        assertTrue((int) result.get("_removed_outliers") >= 2);
+    }
+
+    @Test
+    @DisplayName("批量清洗-包含去重")
+    void testBatchClean_WithDedup() {
+        Map<String, Object> d1 = new HashMap<>();
+        d1.put("flow", 10.0);
+        Map<String, Object> d2 = new HashMap<>();
+        d2.put("flow", 10.0); // 与d1相同
+        Map<String, Object> d3 = new HashMap<>();
+        d3.put("flow", 20.0);
+
+        List<Map<String, Object>> dataList = List.of(d1, d2, d3);
+
+        Map<String, Object> result = cleaningService.batchClean(dataList, "range");
+
+        assertEquals(3, result.get("total_records"));
+        assertEquals(2, result.get("cleaned_records")); // 去重后2条
+        assertTrue((int) result.get("removed_duplicates") >= 1);
+        assertEquals(true, result.get("_batch_cleaned"));
+    }
+
+    @Test
+    @DisplayName("清洗规则-CRUD操作")
+    void testCleaningRuleCrud() {
+        when(cleaningRuleMapper.insert(any(CleaningRule.class))).thenReturn(1);
+        when(cleaningRuleMapper.selectById(1L)).thenReturn(buildRule());
+        when(cleaningRuleMapper.updateById(any(CleaningRule.class))).thenReturn(1);
+        when(cleaningRuleMapper.deleteById(1L)).thenReturn(1);
+
+        // 创建
+        CleaningRule rule = new CleaningRule();
+        rule.setRuleName("流量缺失填充");
+        rule.setRuleType("missing_fill");
+        rule.setTableName("iot_telemetry");
+        rule.setFieldName("flow");
+        CleaningRule created = cleaningService.createRule(rule);
+        assertEquals(1, created.getEnabled());
+
+        // 查询
+        CleaningRule fetched = cleaningService.getRule(1L);
+        assertNotNull(fetched);
+
+        // 更新
+        fetched.setThreshold(new BigDecimal("3.0"));
+        cleaningService.updateRule(1L, fetched);
+        verify(cleaningRuleMapper).updateById(any(CleaningRule.class));
+
+        // 删除
+        cleaningService.deleteRule(1L);
+        verify(cleaningRuleMapper).deleteById(1L);
+    }
+
+    @Test
+    @DisplayName("启用/禁用清洗规则")
+    void testToggleRule() {
+        CleaningRule rule = buildRule();
+        when(cleaningRuleMapper.selectById(1L)).thenReturn(rule);
+        when(cleaningRuleMapper.updateById(any(CleaningRule.class))).thenReturn(1);
+
+        CleaningRule disabled = cleaningService.toggleRule(1L, false);
+        assertEquals(0, disabled.getEnabled());
+
+        CleaningRule enabled = cleaningService.toggleRule(1L, true);
+        assertEquals(1, enabled.getEnabled());
+    }
+
+    // ==================== QualityScoringService 测试 ====================
+
+    @Test
+    @DisplayName("质控评分-三维度计算并保存")
+    void testScore() {
+        when(jdbcTemplate.queryForObject(anyString(), eq(Integer.class))).thenReturn(100);
+        when(qualityScoreMapper.insert(any(QualityScore.class))).thenReturn(1);
+
+        QualityScore score = scoringService.score("iot", 1L, "iot_telemetry");
+
+        assertNotNull(score);
+        assertEquals("iot", score.getSourceType());
+        assertEquals(1L, score.getSourceId());
+        assertEquals(LocalDate.now(), score.getScoreDate());
+        assertNotNull(score.getCompletenessScore());
+        assertNotNull(score.getAccuracyScore());
+        assertNotNull(score.getTimelinessScore());
+        assertNotNull(score.getTotalScore());
+        assertNotNull(score.getDetail());
+        assertTrue(score.getTotalScore().compareTo(BigDecimal.ZERO) >= 0);
+        assertTrue(score.getTotalScore().compareTo(new BigDecimal("100")) <= 0);
+        verify(qualityScoreMapper).insert(any(QualityScore.class));
+    }
+
+    @Test
+    @DisplayName("趋势分析-计算平均得分和趋势方向")
+    void testTrendAnalysis() {
+        List<QualityScore> mockScores = List.of(
+                buildScore(new BigDecimal("70"), LocalDate.now().minusDays(5)),
+                buildScore(new BigDecimal("75"), LocalDate.now().minusDays(3)),
+                buildScore(new BigDecimal("85"), LocalDate.now().minusDays(1))
+        );
+        when(qualityScoreMapper.selectList(any())).thenReturn(mockScores);
+
+        Map<String, Object> trend = scoringService.trendAnalysis("iot", 1L, 7);
+
+        assertNotNull(trend);
+        assertEquals("iot", trend.get("source_type"));
+        assertEquals(3, trend.get("data_points"));
+        assertEquals("improving", trend.get("trend_direction"));
+        assertNotNull(trend.get("avg_total_score"));
+        assertNotNull(trend.get("avg_completeness"));
+        assertNotNull(trend.get("avg_accuracy"));
+        assertNotNull(trend.get("avg_timeliness"));
+
+        @SuppressWarnings("unchecked")
+        List<Map<String, Object>> daily = (List<Map<String, Object>>) trend.get("daily_scores");
+        assertEquals(3, daily.size());
+    }
+
+    @Test
+    @DisplayName("质控概览-统计高质量数据占比")
+    void testOverview() {
+        List<QualityScore> mockScores = List.of(
+                buildScore(new BigDecimal("90"), LocalDate.now()),
+                buildScore(new BigDecimal("85"), LocalDate.now()),
+                buildScore(new BigDecimal("60"), LocalDate.now()),
+                buildScore(new BigDecimal("92"), LocalDate.now())
+        );
+        when(qualityScoreMapper.selectList(any())).thenReturn(mockScores);
+
+        Map<String, Object> overview = scoringService.overview();
+
+        assertNotNull(overview);
+        assertEquals(4, overview.get("total_records"));
+        assertNotNull(overview.get("avg_score"));
+        assertEquals(3L, overview.get("high_quality_count")); // >=80分的有3个
+        assertNotNull(overview.get("high_quality_rate"));
+    }
+
+    @Test
+    @DisplayName("质控评分列表-按日期范围查询")
+    void testListScores() {
+        List<QualityScore> mockScores = List.of(
+                buildScore(new BigDecimal("85"), LocalDate.now().minusDays(1)),
+                buildScore(new BigDecimal("90"), LocalDate.now())
+        );
+        when(qualityScoreMapper.selectList(any())).thenReturn(mockScores);
+
+        List<QualityScore> result = scoringService.listScores(
+                "iot", 1L, LocalDate.now().minusDays(7), LocalDate.now());
+
+        assertEquals(2, result.size());
+        assertEquals("iot", result.get(0).getSourceType());
+    }
+
+    // ==================== 辅助方法 ====================
+
+    private CleaningRule buildRule() {
+        CleaningRule rule = new CleaningRule();
+        rule.setId(1L);
+        rule.setRuleName("流量缺失填充");
+        rule.setRuleType("missing_fill");
+        rule.setTableName("iot_telemetry");
+        rule.setFieldName("flow");
+        rule.setThreshold(new BigDecimal("3.0"));
+        rule.setEnabled(1);
+        return rule;
+    }
+
+    private QualityScore buildScore(BigDecimal total, LocalDate date) {
+        QualityScore score = new QualityScore();
+        score.setSourceType("iot");
+        score.setSourceId(1L);
+        score.setScoreDate(date);
+        score.setCompletenessScore(new BigDecimal("88"));
+        score.setAccuracyScore(new BigDecimal("90"));
+        score.setTimelinessScore(new BigDecimal("80"));
+        score.setTotalScore(total);
+        score.setScoredAt(LocalDateTime.now());
+        return score;
+    }
+}