Update sync_job.py

sync_job.py

@@ -1,5 +1,10 @@
 import os, threading, time, subprocess, sys, logging, importlib, re
-import json
+from contextlib import asynccontextmanager
+from fastapi import FastAPI
+import uvicorn
+from datasets import load_dataset
+from supabase import create_client
+from bs4 import BeautifulSoup
 
 # --- 1. 环境自修复 ---
 try:
@@ -8,18 +13,12 @@ except ImportError:
     subprocess.check_call([sys.executable, "-m", "pip", "install", "bleach"])
     import bleach
 
-from fastapi import FastAPI
-import uvicorn
-from datasets import load_dataset
-from supabase import create_client
-from bs4 import BeautifulSoup
-
 # 屏蔽无用日志
 logging.getLogger("uvicorn.access").setLevel(logging.ERROR)
 
 # --- 2. 动态加载配置 ---
 CONFIG_MODULE_NAME = os.getenv("CONFIG_FILE", "fields_config_plants")
-TASK_KEY = os.getenv("SYNC_TASK", "edible")  # 可传入 'all' 或具体任务名
+TASK_KEY = os.getenv("SYNC_TASK", "all")  # 默认跑全量，收割 title
 
 try:
     cfg_module = importlib.import_module(CONFIG_MODULE_NAME)
@@ -33,34 +32,50 @@ KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY")
 
 # 状态追踪
 status = {"total_todo": 0, "scanned": 0, "hits": 0, "speed": 0, "task": TASK_KEY}
-app = FastAPI()
 
-# --- 3. 增强版清洗函数 (针对药用字段深度优化) ---
+# --- 3. 增强版清洗函数 ---
 def extract_and_clean_v2(html, rule):
     if not html: 
         return "[Empty:NoHtml]"
     try:
-        # 使用 lxml 提升解析速度和容错
         soup = BeautifulSoup(html, 'lxml')
+        col_name = rule["col"]
+        
+        # A. 正常寻找配置中的 ID
         el = soup.find('span', id=rule["html_id"])
         
+        # B. 针对 Title 的专项保底逻辑 (处理 id="Head1" 问题)
+        if not el and col_name == "title":
+            title_tag = soup.find('title')
+            if title_tag:
+                raw_t = title_tag.get_text()
+                # 剔除 PFAF 常见后缀
+                return raw_t.replace("PFAF Plant Database", "").strip()
+        
+        # C. 针对 Common Name 的拼写纠错保底
+        if not el and col_name == "common_name":
+            # 尝试 PFAF 常见的拼写错误 ID
+            el = soup.find('span', id="ContentPlaceHolder1_lblCommanName")
+            
         if not el: 
             return "[Empty:NoTag]"
         
-        # 核心逻辑：如果字段是 medicinal_uses 或明确关闭了 bleach
-        if rule["col"] == "medicinal_uses" or not rule.get("use_bleach"):
-            # 穿透所有 <a>, <i>, <br> 标签，获取完整纯文本
+        # D. 深度内容提取逻辑
+        # 药用、食用、俗名等字段强制穿透所有内嵌标签
+        force_text_fields = ['medicinal_uses', 'edible_uses', 'common_name', 'title']
+        
+        if col_name in force_text_fields or not rule.get("use_bleach"):
             content = el.get_text(separator=" ", strip=True)
             
-            # 剔除 PFAF 特有的长免责声明（避免每个条目都带这段废话）
+            # 剔除药用免责声明
             disclaimer = "Plants For A Future can not take any responsibility for any adverse effects from the use of plants. Always seek advice from a professional before using a plant medicinally."
             if disclaimer in content:
                 content = content.replace(disclaimer, "").strip()
             
-            # 清理因 separator 产生的多余空格
+            # 清理多余空格
             content = re.sub(r'\s+', ' ', content)
         else:
-            # 原有的 HTML 片段保留逻辑
+            # 需要保留基础格式的字段 (如栽培详情)
             content = bleach.clean(str(el), tags=['p', 'br', 'b', 'i'], strip=True)
             
         return content if content.strip() else "[Empty:Blank]"
@@ -76,25 +91,19 @@ def run_sync_logic():
         print(f"❌ Supabase 客户端初始化失败: {e}")
         return
 
-    # A. 确定任务范围
+    # 确定任务
     if TASK_KEY.lower() == "all":
         active_tasks = CFG["FIELD_RULES"]
-        print(f"🌟 模式: 全字段模式 (All-in-one)")
     else:
-        if TASK_KEY not in CFG["FIELD_RULES"]:
-            print(f"❌ 任务 Key '{TASK_KEY}' 不在配置中")
-            os._exit(1)
         active_tasks = {TASK_KEY: CFG["FIELD_RULES"][TASK_KEY]}
-        print(f"📡 模式: 单任务模式 ({TASK_KEY})")
 
-    # B. 智能化获取待处理名单
-    print("🔍 正在检索数据库缺失项...")
+    print(f"🔍 模式: {TASK_KEY} | 正在检索缺失项...")
+    
     try:
-        # 注意：如果数据量极大，建议这里加 limit 或分批 select
         res = client.table(CFG["TARGET_TABLE"]).select("*").execute()
         db_rows = res.data if res.data else []
     except Exception as e:
-        print(f"❌ 无法读取数据库: {e}")
+        print(f"❌ 读取数据库失败: {e}")
         return
     
     todo_ids = set()
@@ -104,8 +113,8 @@ def run_sync_logic():
         p_id = r[CFG["TARGET_ID_COL"]]
         needs_update = False
         for t_key, rule in active_tasks.items():
-            # 检查字段是否为空 (None 或 空字符串)
             val = r.get(rule["col"])
+            # 识别 Null, 空值, 或之前的 [Empty:...] 标记
             if val is None or val == "" or str(val).startswith("[Empty:"): 
                 needs_update = True
                 break
@@ -114,17 +123,13 @@ def run_sync_logic():
             db_lookup[p_id] = r
 
     status["total_todo"] = len(todo_ids)
-    
-    print(f"\n" + "="*50)
-    print(f"📊 待处理总数: {status['total_todo']:,} 条")
-    print("="*50 + "\n")
+    print(f"📊 待修复/待补齐: {status['total_todo']:,} 条")
 
     if not todo_ids:
-        print("✅ 检查完毕：所有字段已填满，无需运行。")
+        print("✅ 状态完美，无需更新。")
         return
 
-    # C. 流式连接数据湖
-    print(f"📡 正在连接数据湖 {CFG['REPO_ID']}...")
+    # 连接湖泊
     ds = load_dataset(CFG["REPO_ID"], data_dir=CFG["DATA_DIR"], split="train", streaming=True)
     start_time = time.time()
 
@@ -139,49 +144,38 @@ def run_sync_logic():
 
             for t_key, rule in active_tasks.items():
                 col_name = rule["col"]
-                # 只有在 DB 中该字段确实为空时才重新清洗
                 db_val = current_db_row.get(col_name)
                 if db_val is None or db_val == "" or str(db_val).startswith("[Empty:"):
                     update_payload[col_name] = extract_and_clean_v2(html, rule)
             
-            # --- 原子更新 ---
             if update_payload:
                 try:
                     update_payload["updated_at"] = "now()"
                     client.table(CFG["TARGET_TABLE"]).update(update_payload).eq(CFG["TARGET_ID_COL"], p_id).execute()
                     status["hits"] += 1
-                except Exception as e:
-                    print(f"\n⚠️ 提交失败 (ID: {p_id}) | {str(e)[:100]}")
-                    time.sleep(2)
+                except Exception:
                     continue
 
-            # 进度打印
-            if status["hits"] > 0 and (status["hits"] % 20 == 0 or status["hits"] == status["total_todo"]):
+            if status["hits"] > 0 and status["hits"] % 50 == 0:
                 elapsed = time.time() - start_time
-                status["speed"] = status["hits"] / elapsed if elapsed > 0 else 0
-                print(f"🎯 进度: {status['hits']}/{status['total_todo']} | 速度: {status['speed']:.1f}条/秒")
+                status["speed"] = status["hits"] / elapsed
+                print(f"🎯 修复进度: {status['hits']}/{status['total_todo']} | 速度: {status['speed']:.1f}条/秒")
 
-            # 完成自毁逻辑
             if status["hits"] >= status["total_todo"]:
-                print(f"\n🎉 任务圆满完成！已修复/补齐 {status['hits']} 条记录。")
-                time.sleep(5)
+                print(f"🎉 修复完成！共处理 {status['hits']} 条。")
                 os._exit(0)
 
-        elif status["scanned"] % 1000 == 0:
-            print(f"📡 已扫描湖泊数据: {status['scanned']:,} 条...")
-
-@app.on_event("startup")
-def startup():
+# --- 5. FastAPI 入口 ---
+@asynccontextmanager
+async def lifespan(app: FastAPI):
     threading.Timer(1.0, run_sync_logic).start()
+    yield
+
+app = FastAPI(lifespan=lifespan)
 
 @app.get("/")
 def health():
-    return {
-        "status": "running",
-        "progress": f"{status['hits']}/{status['total_todo']}",
-        "scanned": status["scanned"],
-        "speed": f"{status['speed']:.2f} rows/sec"
-    }
+    return {"status": "running", "progress": f"{status['hits']}/{status['total_todo']}", "task": status["task"]}
 
 if __name__ == "__main__":
     uvicorn.run(app, host="0.0.0.0", port=7860, access_log=False)