bs4_quality_scanner.py

import os, re, sys, time
from supabase import create_client

def log(msg):
    print(msg, flush=True)

url = os.environ.get("SUPABASE_URL")
key = os.environ.get("SUPABASE_SERVICE_ROLE_KEY")
supabase = create_client(url, key)

# --- 扩展后的字段规则字典 (同步 JSON 补齐字段) ---
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},
    # --- 新增补齐字段 ---
    'weed_potential': {'min': 2, 'max': 50},
    'found_in': {'min': 5, 'max': 5000},
    'conservation_status': {'min': 5, 'max': 500},
    'special_uses': {'min': 5, 'max': 1000},
    'author': {'min': 1, 'max': 50}
}

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))
    
    # 2. 离群值检测
    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
        # 结构异常判定：HTML残留或异常空白
        if "     " in raw_s or "<div>" in raw_s.lower():
            is_outlier = True

    return False, (has_garbage or is_outlier), length

def run_bs4_audit():
    target_columns = list(FIELD_RULES.keys())
    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')}] 🛡️ 启动全字段深度扫描 (含新增字段)...")
    
    offset = 0
    limit = 200
    while True:
        # 动态获取字段
        res = supabase.table("bs4_plants").select(",".join(target_columns)).range(offset, offset + limit - 1).execute()
        if not res.data: break
        
        for row in res.data:
            for col in target_columns:
                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
                
                # 质量评分
                score = 0 if is_emp else (50 if is_bad else 100)
                stats[col]['score_sum'] += score
        
        offset += len(res.data)

    # --- 输出可视化表格 ---
    log("\n" + "="*100)
    log(f"{'字段名称':<25} | {'填充':<6} | {'异常/离群':<10} | {'空缺':<6} | {'均长':<6} | {'健康分'}")
    log("-" * 100)
    
    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
        total_score = round(data['score_sum'] / offset, 2)
        
        log(f"{col:<25} | {data['filled']:<6} | {data['outlier']:<10} | {data['empty']:<6} | {avg_len:<6} | {total_score}")
        
        report_batch.append({
            "field_name": col,
            "total_records": offset,
            "filled_count": data['filled'],
            "empty_count": data['empty'],
            "dirty_count": data['outlier'],
            "avg_char_count": avg_len,
            "quality_score": total_score,
            "updated_at": "now()"
        })

    # 回写到统计表
    supabase.table("bs4_field_quality_report").upsert(report_batch).execute()
    log("="*100)
    log(f"✅ 巡检完成。共处理 {offset} 条记录。")

if __name__ == "__main__":
    # 保持之前商定的逻辑：启动时运行一遍。
    # 如果想手动触发，重启 Space 即可。
    try:
        run_bs4_audit()
    except Exception as e:
        log(f"❌ 运行报错: {e}")