#!/usr/bin/env python3 """ 本地提取数据 → SSH管道 → 服务器导入 search.db 不依赖SCP，通过SSH stdin传输 """ import os, sys, json, sqlite3, subprocess, re, tempfile from datetime import datetime BASE = os.path.expanduser("~/Nutstore Files/我的坚果云/Crawler/gov_crawler") PASSWORD = "" # 2026-09-10 改用 SSH 密钥认证(~/.ssh/config: hw-backup), 明文已移除 SERVER = "root@1.94.217.116" def normalize_date(d): d = str(d or "") m = re.search(r'(\d{4})-(\d{1,2})-(\d{1,2})', d) if m: return f"{m.group(1)}-{int(m.group(2)):02d}-{int(m.group(3)):02d}" m = re.search(r'(\d{4})年(\d{1,2})月(\d{1,2})日', d) if m: return f"{m.group(1)}-{int(m.group(2)):02d}-{int(m.group(3)):02d}" return d[:10] def extract_gov_raw(db_path): """从 gov_raw 表提取""" if not os.path.exists(db_path): return [] db = sqlite3.connect(db_path) items = [] try: cols = [c[1] for c in db.execute("PRAGMA table_info(gov_raw)").fetchall()] url_col = "page_url" if "page_url" in cols else "url" cur = db.execute(f"SELECT title, {url_col}, content, publish_date, site_name FROM gov_raw") for r in cur.fetchall(): url = str(r[1] or "").strip() if not url: continue items.append({"title": str(r[0] or ""), "page_url": url, "content": str(r[2] or ""), "publish_date": normalize_date(r[3]), "site_name": str(r[4] or "")}) except: pass db.close() return items def extract_crawl_results(db_path, site_name=None): """从 crawl_results 表提取""" if not os.path.exists(db_path): return [] db = sqlite3.connect(db_path) items = [] try: cols = [c[1] for c in db.execute("PRAGMA table_info(crawl_results)").fetchall()] if "url" not in cols and "page_url" not in cols: db.close() return [] url_col = "page_url" if "page_url" in cols else "url" sn_col = "'" + (site_name or "") + "'" if site_name else ("site_name" if "site_name" in cols else "''") pub_col = "publish_date" if "publish_date" in cols else "''" cur = db.execute(f"SELECT title, {url_col}, content, {pub_col}, {sn_col} FROM crawl_results") for r in cur.fetchall(): url = str(r[1] or "").strip() if not url: continue items.append({"title": str(r[0] or ""), "page_url": url, "content": str(r[2] or ""), "publish_date": normalize_date(r[3]), "site_name": str(r[4] or site_name or "")}) except: pass db.close() return items def pipe_to_server(items, label): """通过SSH管道推送JSONL""" if not items: return 0 # 去重 seen = set() unique = [] for it in items: u = it.get("page_url", "") if u and u not in seen: seen.add(u) unique.append(it) if not unique: print(f" ⏭️ {label}: 去重后无数据") return 0 jsonl = "\n".join(json.dumps(it, ensure_ascii=False) for it in unique) + "\n" # 通过管道传给服务器 cmd = f"ssh -o StrictHostKeyChecking=no {SERVER} 'python3 /tmp/import_jsonl.py /dev/stdin \"{label}\"'" proc = subprocess.run(cmd, shell=True, input=jsonl, capture_output=True, text=True, timeout=120) out = (proc.stdout or "").strip() if out: print(f" {out}") else: err = (proc.stderr or "").strip()[:200] print(f" {label}: {'错误: ' + err if err else '完成'}") return len(unique) def main(): print("="*60) print("📤 统一同步 → 服务器 search.db (SSH管道传输)") print("="*60) total = 0 # 1. 共享 search.db (gov_raw) print("\n--- 1. 本地 search.db (gov_raw) ---") items = extract_gov_raw(os.path.join(BASE, "search.db")) total += pipe_to_server(items, "shared search.db") # 2. quality_results.db (crawl_results) print("\n--- 2. quality_results.db (crawl_results) ---") items = extract_crawl_results(os.path.join(BASE, "quality_results.db")) total += pipe_to_server(items, "quality_results") # 3. xxx_results.db（按站点分组） sources = [ ("hipac_results.db", "淮安工业园区"), ("yanshougs_results.db", "工程建设验收公示网"), ("eiacloud_results.db", "环评云"), ("sthj_xz_results.db", "徐州市生态环境局"), ("ylnh_results.db", "延长中煤榆林能源化工"), ("taixing_results.db", "泰兴市生态环境局"), ("qingdao_results.db", "青岛市生态环境局"), ("xinhui_results.db", "新会区-征集调查"), ("debaoenv_results.db", "德宝环境"), ("cz_sthjj_results.db", "郴州市生态环境局"), ] for db_name, site in sources: 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} ---") items = extract_crawl_results(db_path, site_name=site) total += pipe_to_server(items, site) # 4. SQL文件 — 传过去再执行 sql_files = [ "jzssthjj.zwgk.jingzhou.gov.cn_import.sql", "www.xjhbcy.cn_import.sql", ] for sql_name in sql_files: sql_path = os.path.join(BASE, sql_name) if os.path.exists(sql_path): print(f"\n--- {sql_name} ---") size = os.path.getsize(sql_path) print(f" 传输 {size//1024//1024}MB...") with open(sql_path) as f: sql_content = f.read() # 管道传输 + 执行 cmd = f"ssh -o StrictHostKeyChecking=no {SERVER} 'cd /root && sqlite3 search.db'" proc = subprocess.run(cmd, shell=True, input=sql_content, capture_output=True, text=True, timeout=300) # 管道传SQL给sqlite3直接执行 # 通过临时文件更可靠 tag = f"sql_{int(datetime.now().timestamp())}" tmp_cmd = f"ssh -o StrictHostKeyChecking=no {SERVER} 'cat > /tmp/{tag}.sql'" proc = subprocess.run(tmp_cmd, shell=True, input=sql_content, capture_output=True, text=True, timeout=120) if proc.returncode == 0: exec_cmd = f"ssh -o StrictHostKeyChecking=no {SERVER} 'cd /root && sqlite3 search.db < /tmp/{tag}.sql && rm -f /tmp/{tag}.sql && echo SQL导入完成'" proc2 = subprocess.run(exec_cmd, shell=True, capture_output=True, text=True, timeout=120) print(f" {sql_name}: {proc2.stdout.strip()}") else: print(f" ✗ {sql_name}: 传输失败") # 5. 重建FTS print("\n--- 5. 重建FTS索引 ---") cmd = f"ssh -o StrictHostKeyChecking=no {SERVER} 'sqlite3 /root/search.db \"DELETE FROM gov_search\"; 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!='\"'\"'\"; echo FTS重建完成; sqlite3 /root/search.db \"SELECT COUNT(*) FROM gov_raw\"'" proc = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=60) print(f" {proc.stdout.strip()}") print("\n" + "="*60) print(f"✅ 同步完成！共推送 {total} 条") print("="*60) if __name__ == "__main__": main()