#!/usr/bin/env python3 """ 统一同步：提取本地各爬虫数据库 → JSONL → SCP到服务器 → 导入 search.db """ import os, sys, json, subprocess, tempfile, sqlite3, time, re BASE = os.path.dirname(os.path.abspath(__file__)) SERVER = "root@1.94.217.116" PASSWORD = "" # 2026-09-10 改用 SSH 密钥认证(~/.ssh/config: hw-backup), 明文已移除 def safe_str(v): return str(v).strip() if v else "" def extract_items(db_path, table, col_map, site_name=None): """从数据库提取条目，col_map: {输出列: 源列名}""" if not os.path.exists(db_path): return [] items = [] try: db = sqlite3.connect(db_path) src_cols = [c[1] for c in db.execute(f"PRAGMA table_info({table})").fetchall()] # 验证列是否存在 valid_cols = {out: src for out, src in col_map.items() if src in src_cols} if not valid_cols: db.close() return [] select_cols = ", ".join(valid_cols.values()) cur = db.execute(f"SELECT {select_cols} FROM {table}") for row in cur.fetchall(): item = {} for i, (out_col, _) in enumerate(valid_cols.items()): item[out_col] = safe_str(row[i]) if site_name: item["site_name"] = site_name if not item.get("site_name"): item["site_name"] = site_name or "unknown" if item.get("page_url"): items.append(item) db.close() except Exception as e: print(f" ⚠️ 读取错误: {e}") return items def push_jsonl(items, label): """推JSONL到服务器并导入""" if not items: print(f" ⏭️ {label}: 无数据") return 0 # 去重 seen = set() unique = [] for it in items: url = it.get("page_url", "") if url and url not in seen: seen.add(url) unique.append(it) if not unique: print(f" ⏭️ {label}: 去重后无数据") return 0 # 写入JSONL tag = f"sync_{int(time.time())}_{os.urandom(4).hex()}" jsonl_path = f"/tmp/{tag}.jsonl" with open(jsonl_path, 'w') as f: for it in unique: # 标准化日期 d = it.get("publish_date", "") m = re.search(r'(\d{4})-(\d{1,2})-(\d{1,2})', d) if m: it["publish_date"] = f"{m.group(1)}-{int(m.group(2)):02d}-{int(m.group(3)):02d}" m2 = re.search(r'(\d{4})年(\d{1,2})月(\d{1,2})日', d) if m2: it["publish_date"] = f"{m2.group(1)}-{int(m2.group(2)):02d}-{int(m2.group(3)):02d}" f.write(json.dumps(it, ensure_ascii=False) + '\n') # SCP remote_path = f"/tmp/{tag}.jsonl" ret = os.system(f"scp -o StrictHostKeyChecking=no -q {jsonl_path} {SERVER}:{remote_path}") os.unlink(jsonl_path) if ret != 0: print(f" ✗ {label}: SCP失败") return 0 # 服务器导入 import_script = ( f"ssh -o StrictHostKeyChecking=no {SERVER} " f'"python3 -c \'' f"import json, sqlite3; " f"db = sqlite3.connect(os.getenv("SEARCH_DB", "/root/search.db")); " f"a = 0; e = 0; " f"with open('{remote_path}') as f:" f" for line in f:" f" try:" f" item = json.loads(line);" f" db.execute('INSERT OR IGNORE INTO gov_raw (title,page_url,content,publish_date,site_name,source_url,status) VALUES (?,?,?,?,?,?,\\'synced\\')'," f" (item.get('title',''),item.get('page_url',''),item.get('content',''),item.get('publish_date',''),item.get('site_name',''),item.get('page_url','')));" f" a += 1;" f" except: e += 1;" f"db.commit();" f"db.execute('DELETE FROM gov_search');" f"db.execute('INSERT INTO gov_search(rowid,title,site_name,summary) SELECT rowid,title,site_name,substr(content,1,500) FROM gov_raw WHERE content IS NOT NULL AND content!=\\'\\'');" f"db.commit();" f"t = db.execute('SELECT COUNT(*) FROM gov_raw').fetchone()[0];" f"print(f\\\"OK {label}: +{{a}} err{{e}} total{{t}}\\\");" f"import os; os.remove('{remote_path}');" f"'\"" ) result = subprocess.run(import_script, shell=True, capture_output=True, text=True, timeout=120) print(f" {result.stdout.strip()}") return len(unique) def main(): print("=" * 60) print("📤 统一同步所有本地数据 → 服务器 search.db") print(f" 服务器: {SERVER}") print("=" * 60) total_push = 0 # 1. 共享 search.db（gov_raw表） print("\n--- 1. 本地 search.db (gov_raw表) ---") items = extract_items(os.path.join(BASE, "search.db"), "gov_raw", {"title": "title", "page_url": "page_url", "content": "content", "publish_date": "publish_date", "site_name": "site_name"}) if not items: items = extract_items(os.path.join(BASE, "search.db"), "gov_raw", {"title": "title", "page_url": "url", "content": "content", "publish_date": "publish_date", "site_name": "site_name"}) total_push += push_jsonl(items, "search.db") # 2. quality_results.db (crawl_results表) print("\n--- 2. quality_results.db ---") items = extract_items(os.path.join(BASE, "quality_results.db"), "crawl_results", {"title": "title", "page_url": "url", "content": "content", "publish_date": "publish_date", "site_name": "site_name"}) total_push += push_jsonl(items, "crawl_results") # 3. 各个 xxx_results.db result_configs = [ ("hipac_results.db", "crawl_results", "淮安工业园区", {"title": "title", "page_url": "url", "content": "content", "publish_date": "publish_date"}), ("yanshougs_results.db", "crawl_results", "工程建设验收公示网", {"title": "title", "page_url": "url", "content": "content", "publish_date": "publish_date"}), ("eiacloud_results.db", "crawl_results", "环评云", {"title": "title", "page_url": "url", "content": "content", "publish_date": "publish_date"}), ("sthj_xz_results.db", "crawl_results", "徐州市生态环境局", {"title": "title", "page_url": "url", "content": "content", "publish_date": "publish_date"}), ("ylnh_results.db", "crawl_results", "延长中煤榆林能源化工", {"title": "title", "page_url": "url", "content": "content", "publish_date": "publish_date"}), ("taixing_results.db", "crawl_results", "泰兴市生态环境局", {"title": "title", "page_url": "url", "content": "content", "publish_date": "publish_date"}), ("qingdao_results.db", "crawl_results", "青岛市生态环境局", {"title": "title", "page_url": "url", "content": "content", "publish_date": "publish_date"}), ("xinhui_results.db", "crawl_results", "新会区-征集调查", {"title": "title", "page_url": "url", "content": "content", "publish_date": "publish_date"}), ("debaoenv_results.db", "crawl_results", "德宝环境", {"title": "title", "page_url": "url", "content": "content", "publish_date": "publish_date"}), ("cz_sthjj_results.db", "crawl_results", "郴州市生态环境局", {"title": "title", "page_url": "url", "content": "content", "publish_date": "publish_date"}), ("baiyinqu_results.db", "crawl_results", "白银区人民政府", {"title": "title", "page_url": "url", "content": "content", "publish_date": "publish_date"}), ("boxing_results.db", "crawl_results", "博兴县人民政府", {"title": "title", "page_url": "url", "content": "content", "publish_date": "publish_date"}), ] for db_name, table, site, col_map in result_configs: db_path = os.path.join(BASE, db_name) if os.path.exists(db_path) and os.path.getsize(db_path) > 1000: print(f"\n--- {site} ({db_name}) ---") items = extract_items(db_path, table, col_map, site_name=site) total_push += push_jsonl(items, site) # 4. SQL文件直接导入（荆州、新疆环协） sql_files = [ ("jzssthjj.zwgk.jingzhou.gov.cn_import.sql", "荆州市生态环境局"), ("www.xjhbcy.cn_import.sql", "新疆生态环境保护产业协会"), ] for sql_name, label in sql_files: sql_path = os.path.join(BASE, sql_name) if os.path.exists(sql_path): print(f"\n--- {label} ({sql_name}) ---") tag = f"sync_{int(time.time())}" remote_sql = f"/tmp/{tag}.sql" ret = os.system(f"scp -o StrictHostKeyChecking=no -q {sql_path} {SERVER}:{remote_sql}") if ret == 0: result = subprocess.run( f"ssh -o StrictHostKeyChecking=no {SERVER} " f"\"cd /root && sqlite3 search.db < {remote_sql} 2>&1; " f"total=$(sqlite3 search.db 'SELECT COUNT(*) FROM gov_raw'); " f"echo \\\"总计: $total 条\\\"; rm -f {remote_sql}\"", shell=True, capture_output=True, text=True, timeout=300 ) print(f" {label}: {result.stdout.strip()}") # 5. 重建FTS + 重启服务 print("\n--- 5. 重建FTS索引 ---") subprocess.run( f"ssh -o StrictHostKeyChecking=no {SERVER} " f"\"sqlite3 /root/search.db 'DELETE FROM gov_search'; " f"sqlite3 /root/search.db 'INSERT INTO gov_search(rowid,title,site_name,summary) SELECT rowid,title,site_name,substr(content,1,500) FROM gov_raw WHERE content IS NOT NULL AND content!='\"'\"'; " f"echo FTS重建完成; " f"total=\\\$(sqlite3 /root/search.db 'SELECT COUNT(*) FROM gov_raw'); " f"echo 累计: \\\$total 条\"", shell=True, capture_output=True, text=True, timeout=60 ) print(" ✅ FTS索引重建完成") print("\n" + "=" * 60) print("✅ 全部同步完成！") print("=" * 60) if __name__ == "__main__": main()