Update bs4_quality_scanner.py

bs4_quality_scanner.py

@@ -8,24 +8,58 @@ url = os.environ.get("SUPABASE_URL")
 key = os.environ.get("SUPABASE_SERVICE_ROLE_KEY")
 supabase = create_client(url, key)
 
-def analyze_value(val):
-    s = str(val or "").strip()
+# --- 定义每个字段的“正常范围”和“异常规则” ---
+# min_len: 低于此长度疑似抓取不全; max_len: 高于此长度疑似抓错位置或吞了后面内容
+FIELD_RULES = {
+    'latin_name': {'min': 5, 'max': 80},
+    'common_name': {'min': 2, 'max': 100},
+    'family': {'min': 3, 'max': 50},
+    'title': {'min': 10, 'max': 200},
+    'usda_hardiness': {'min': 1, 'max': 15},
+    'known_hazards': {'min': 4, 'max': 2000},
+    'habitats': {'min': 5, 'max': 1000},
+    'range': {'min': 5, 'max': 1000},
+    'physical_characteristics': {'min': 20, 'max': 5000},
+    'edible_uses': {'min': 10, 'max': 5000},
+    'medicinal_uses': {'min': 10, 'max': 5000},
+    'other_uses': {'min': 10, 'max': 5000},
+    'cultivation_details': {'min': 20, 'max': 8000},
+    'propagation': {'min': 20, 'max': 4000}
+}
+
+def analyze_value_pro(field_name, val):
+    """增强版检测：判定空值、乱码及离群异常"""
+    raw_s = str(val or "")
+    s = raw_s.strip()
+    
     is_empty = not s or s.startswith("[Empty") or s.lower() == "none"
+    if is_empty:
+        return True, False, 0
+    
+    # 1. 乱码检测
     has_garbage = bool(re.search(r'[ÂÃÅÐÑÒÓÔÕÖ×Ø]', s))
-    return is_empty, has_garbage, len(s)
+    
+    # 2. 离群值/异常检测 (Outlier)
+    is_outlier = False
+    length = len(s)
+    
+    if field_name in FIELD_RULES:
+        rule = FIELD_RULES[field_name]
+        # 长度异常判定
+        if length < rule['min'] or length > rule['max']:
+            is_outlier = True
+        # 结构异常判定：如果包含超过 5 个连续空格，或者包含 HTML 标签残余
+        if "     " in raw_s or "<div>" in raw_s.lower() or "<span>" in raw_s.lower():
+            is_outlier = True
+
+    return False, (has_garbage or is_outlier), length
 
 def run_bs4_audit():
-    # --- 补全后的完整字段列表 ---
-    target_columns = [
-        'latin_name', 'common_name', 'family', 'title', 
-        'usda_hardiness', 'known_hazards', 'habitats', 'range',
-        'physical_characteristics', 'edible_uses', 'medicinal_uses', 
-        'other_uses', 'cultivation_details', 'propagation'
-    ]
-    
-    stats = {col: {'filled': 0, 'empty': 0, 'dirty': 0, 'len_sum': 0, 'score_sum': 0} for col in target_columns}
+    target_columns = list(FIELD_RULES.keys())
+    # 增加 outlier 统计项
+    stats = {col: {'filled': 0, 'empty': 0, 'outlier': 0, 'len_sum': 0, 'score_sum': 0} for col in target_columns}
     
-    log(f"\n[{time.strftime('%H:%M:%S')}] 📋 开始全量巡检...")
+    log(f"\n[{time.strftime('%H:%M:%S')}] 🛡️ 启动离群值扫描 (深度体检模式)...")
     
     offset = 0
     limit = 200
@@ -35,50 +69,59 @@ def run_bs4_audit():
         
         for row in res.data:
             for col in target_columns:
-                is_emp, is_dirty, length = analyze_value(row.get(col))
-                stats[col]['filled'] += 0 if is_emp else 1
-                stats[col]['empty'] += 1 if is_emp else 0
-                stats[col]['dirty'] += 1 if is_dirty else 0
-                stats[col]['len_sum'] += length
-                stats[col]['score_sum'] += 0 if is_emp else (70 if is_dirty else 100)
+                is_emp, is_bad, length = analyze_value_pro(col, row.get(col))
+                
+                if is_emp:
+                    stats[col]['empty'] += 1
+                else:
+                    stats[col]['filled'] += 1
+                    stats[col]['len_sum'] += length
+                    if is_bad:
+                        stats[col]['outlier'] += 1
+                
+                # 评分逻辑优化：离群异常和乱码按 50 分计（警告级），空值 0 分
+                if is_emp: score = 0
+                elif is_bad: score = 50 
+                else: score = 100
+                stats[col]['score_sum'] += score
         
         offset += len(res.data)
 
-    # --- 输出控制台报告 ---
-    log("\n" + "="*80)
-    log(f"{'字段名称':<25} | {'填充数':<8} | {'空值':<8} | {'均长':<8} | {'评分'}")
-    log("-" * 80)
+    # --- 控制台可视化报告 ---
+    log("\n" + "="*95)
+    log(f"{'字段名称':<25} | {'填充':<6} | {'异常/离群':<10} | {'空缺':<6} | {'均长':<6} | {'健康分'}")
+    log("-" * 95)
     
     report_batch = []
-    # 排序：按评分从低到高，让你一眼看到最惨的字段
+    # 按健康分从低到高排序
     sorted_cols = sorted(stats.items(), key=lambda x: (x[1]['score_sum']/offset if offset>0 else 0))
     
     for col, data in sorted_cols:
         avg_len = int(data['len_sum'] / data['filled']) if data['filled'] > 0 else 0
-        score = round(data['score_sum'] / offset, 2)
+        total_score = round(data['score_sum'] / offset, 2)
         
-        log(f"{col:<25} | {data['filled']:<8} | {data['empty']:<8} | {avg_len:<8} | {score}")
+        log(f"{col:<25} | {data['filled']:<6} | {data['outlier']:<10} | {data['empty']:<6} | {avg_len:<6} | {total_score}")
         
+        # 为了兼容你之前的数据库表，我们将 outlier 存入原本的 dirty_count 字段，或者你可以扩充字段
         report_batch.append({
             "field_name": col,
             "total_records": offset,
             "filled_count": data['filled'],
             "empty_count": data['empty'],
-            "dirty_count": data['dirty'],
+            "dirty_count": data['outlier'], # 现在这个字段代表广义的“坏数据”
             "avg_char_count": avg_len,
-            "quality_score": score,
+            "quality_score": total_score,
             "updated_at": "now()"
         })
 
-    # --- 回写数据库 ---
     supabase.table("bs4_field_quality_report").upsert(report_batch).execute()
-    log("="*80)
-    log(f"✅ 报告已同步至 SQL 大盘。下次巡检将在 10 分钟后...")
+    log("="*95)
+    log(f"✅ 深度巡检完成。异常记录已标记，下次巡检 10 分钟后...")
 
 if __name__ == "__main__":
     while True:
         try:
             run_bs4_audit()
         except Exception as e:
-            log(f"❌ 运行出错: {e}")
-        time.sleep(600) # 10分钟间隔
+            log(f"❌ 运行报错: {e}")
+        time.sleep(600)