Update quality_audit_job.py

quality_audit_job.py

@@ -4,22 +4,23 @@ import time
 import sys
 import logging
 from supabase import create_client
-# from dotenv import load_dotenv
 
-# 强制不缓存输出，确保进度条能实时显示
-logging.basicConfig(
-    level=logging.INFO,
-    format='%(message)s', # 简化格式让进度条更美观
-    stream=sys.stdout
-)
+# 强制不缓存输出
+logging.basicConfig(level=logging.INFO, format='%(message)s', stream=sys.stdout)
 logger = logging.getLogger(__name__)
 
-# load_dotenv()
+# --- 核心修正点：匹配你截图里的变量名 ---
+url = os.environ.get("SUPABASE_URL")
+# 尝试读取截图中的 SERVICE_ROLE_KEY
+key = os.environ.get("SUPABASE_SERVICE_ROLE_KEY") 
 
-supabase = create_client(os.environ.get("SUPABASE_URL"), os.environ.get("SUPABASE_SERVICE_KEY"))
+if not url or not key:
+    print(f"❌ 环境变量缺失! URL: {'OK' if url else 'MISSING'}, KEY: {'OK' if key else 'MISSING'}", flush=True)
+    sys.exit(1)
 
-def get_progress_bar(current, total, bar_length=40):
-    """生成字符进度条"""
+supabase = create_client(url, key)
+
+def get_progress_bar(current, total, bar_length=30):
     fraction = current / total if total > 0 else 0
     arrow = int(fraction * bar_length) * '█'
     padding = (bar_length - len(arrow)) * '░'
@@ -30,81 +31,52 @@ def analyze_text(text):
     is_empty = text_str.startswith("[Empty") or len(text_str) == 0
     char_count = len(text_str)
     has_garbage = bool(re.search(r'[ÂÃÅÐÑÒÓÔÕÖ×Ø]', text_str))
-    
     score = 100
-    tags = []
-    if is_empty:
-        score = 0; tags.append("no_data")
+    if is_empty: score = 0
     else:
-        if char_count < 50: score -= 30; tags.append("short")
-        if has_garbage: score -= 20; tags.append("dirty_html")
-    
-    return {
-        "is_filled": not is_empty,
-        "is_empty_placeholder": is_empty,
-        "is_null": text is None,
-        "char_count": char_count,
-        "word_count": len(text_str.split()),
-        "has_garbage_chars": has_garbage,
-        "quality_score": score,
-        "issue_tags": tags
-    }
+        if char_count < 50: score -= 30
+        if has_garbage: score -= 20
+    return {"is_filled": not is_empty, "char_count": char_count, "has_garbage_chars": has_garbage, "quality_score": score}
 
 def run_audit():
-    # 1. 获取主表总记录数 (用于计算进度)
-    count_res = supabase.table("bs4_plants").select("id", count="exact").execute()
-    total_count = count_res.count if count_res.count else 7038
-    
-    logger.info(f"🚀 启动审计引擎 | 目标数据量: {total_count} 条")
-    fields = ['edible_uses', 'medicinal_uses', 'cultivation_details']
+    # 你的数据湖大小
+    total_count = 7038 
+    logger.info(f"🚀 审计引擎启动 | 目标量: {total_count}", flush=True)
 
     while True:
         try:
-            # 2. 获取当前审计表已处理的植物数量 (去重计数)
-            # 注意：因为一个植物有3个字段，我们按 plant_id 分组统计
-            current_audit_res = supabase.rpc("get_audited_count").execute()
-            processed_count = current_audit_res.data if current_audit_res.data else 0
+            # --- 修正点：不使用 RPC，改用普通 COUNT ---
+            # 每个植物审计3个字段，所以 count / 3
+            count_res = supabase.table("data_quality_audit").select("id", count="exact").execute()
+            processed_plants = (count_res.count or 0) // 3
             
             # 打印进度条
-            bar = get_progress_bar(processed_count, total_count)
-            logger.info(f"📊 进度: {bar} ({processed_count}/{total_count})")
+            print(f"📊 进度: {get_progress_bar(processed_plants, total_count)} ({processed_plants}/{total_count})", flush=True)
 
-            # 3. 获取水位线
-            last_audit_res = supabase.table("data_quality_audit") \
-                .select("updated_at") \
-                .order("updated_at", desc=True) \
-                .limit(1).execute()
-            
-            last_time = last_audit_res.data[0]['updated_at'] if last_audit_res.data else "1970-01-01T00:00:00Z"
+            # 水位线逻辑
+            last_res = supabase.table("data_quality_audit").select("updated_at").order("updated_at", desc=True).limit(1).execute()
+            last_time = last_res.data[0]['updated_at'] if last_res.data else "1970-01-01T00:00:00Z"
             
-            # 4. 抓取新变动 (每次50条)
-            res = supabase.table("bs4_plants") \
-                .select("plant_id, edible_uses, medicinal_uses, cultivation_details, updated_at") \
-                .gt("updated_at", last_time) \
-                .order("updated_at", desc=False) \
-                .limit(50).execute()
+            # 抓取主表变动
+            res = supabase.table("bs4_plants").select("plant_id, edible_uses, medicinal_uses, cultivation_details, updated_at")\
+                .gt("updated_at", last_time).order("updated_at").limit(20).execute()
 
             if not res.data:
-                if processed_count >= total_count:
-                    logger.info("✨ 100% 完成！正在监听增量更新...")
-                    time.sleep(300)
-                else:
-                    logger.info("⌛ 正在同步中，请稍候...")
-                    time.sleep(30)
+                time.sleep(30)
                 continue
 
             for row in res.data:
-                for field in fields:
-                    analysis = analyze_text(row.get(field))
+                for f in ['edible_uses', 'medicinal_uses', 'cultivation_details']:
+                    analysis = analyze_text(row.get(f))
                     supabase.table("data_quality_audit").upsert({
-                        "plant_id": row['plant_id'],
-                        "field_name": field,
-                        **analysis,
+                        "plant_id": row['plant_id'], 
+                        "field_name": f, 
+                        **analysis, 
                         "updated_at": "now()"
                     }).execute()
             
         except Exception as e:
-            logger.error(f"❌ 运行异常: {str(e)}")
+            print(f"❌ 运行异常: {e}", flush=True)
             time.sleep(10)
 
 if __name__ == "__main__":