#!/usr/bin/env python3 """ 本地提取数据 → 临时文件 → SSH管道传文件 → 服务器导入 逐个处理，确保稳定 """ import os, sys, json, sqlite3, subprocess, re, tempfile, time 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 run(cmd, inp=None, timeout=60): proc = subprocess.run(cmd, shell=True, input=inp, capture_output=True, text=True, timeout=timeout) return proc.returncode, proc.stdout.strip(), proc.stderr.strip() 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): 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): 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" pub_col = "publish_date" if "publish_date" in cols else "''" sn_col = "'" + (site_name or "") + "'" if site_name else ("site_name" if "site_name" 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 push(label, items): if not items: return 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 jsonl = "\n".join(json.dumps(it, ensure_ascii=False) for it in unique) tag = f"sync_{int(time.time())}_{os.getpid()}" # SSH管道传文件到服务器临时文件 rc, out, err = run( f"ssh -o StrictHostKeyChecking=no {SERVER} 'cat > /tmp/{tag}.jsonl'", inp=jsonl, timeout=120 ) if rc != 0: print(f" ✗ {label}: 传输失败: {err[:100]}") return # 服务器导入 rc, out, err = run( f"ssh -o StrictHostKeyChecking=no {SERVER} 'python3 /tmp/import_jsonl.py /tmp/{tag}.jsonl'", timeout=120 ) if rc != 0: print(f" ✗ {label}: 导入失败: {(out or err)[:200]}") else: print(f" {out}") def main(): print("="*60) print("📤 统一同步 → 服务器 search.db (SSH管道+临时文件)") print("="*60) sources = [ ("共享 search.db (gov_raw)", lambda: extract_gov_raw(os.path.join(BASE, "search.db"))), ("quality_results.db", lambda: extract_crawl_results(os.path.join(BASE, "quality_results.db"))), ("淮安工业园区", lambda: extract_crawl_results(os.path.join(BASE, "hipac_results.db"), "淮安工业园区")), ("验收公示网", lambda: extract_crawl_results(os.path.join(BASE, "yanshougs_results.db"), "工程建设验收公示网")), ("环评云", lambda: extract_crawl_results(os.path.join(BASE, "eiacloud_results.db"), "环评云")), ("徐州市", lambda: extract_crawl_results(os.path.join(BASE, "sthj_xz_results.db"), "徐州市生态环境局")), ("延长中煤", lambda: extract_crawl_results(os.path.join(BASE, "ylnh_results.db"), "延长中煤榆林能源化工")), ("泰兴市", lambda: extract_crawl_results(os.path.join(BASE, "taixing_results.db"), "泰兴市生态环境局")), ("青岛市", lambda: extract_crawl_results(os.path.join(BASE, "qingdao_results.db"), "青岛市生态环境局")), ("新会区", lambda: extract_crawl_results(os.path.join(BASE, "xinhui_results.db"), "新会区-征集调查")), ("德宝环境", lambda: extract_crawl_results(os.path.join(BASE, "debaoenv_results.db"), "德宝环境")), ("郴州市", lambda: extract_crawl_results(os.path.join(BASE, "cz_sthjj_results.db"), "郴州市生态环境局")), ] for label, extractor in sources: print(f"\n--- {label} ---") try: items = extractor() push(label, items) except Exception as e: print(f" ✗ {label}: 错误 {e}") # SQL文件直接管道传给 sqlite3 print("\n--- 荆州(44MB SQL) ---") sql_path = os.path.join(BASE, "jzssthjj.zwgk.jingzhou.gov.cn_import.sql") if os.path.exists(sql_path): with open(sql_path) as f: sql = f.read() tag = f"sql_{int(time.time())}" rc, out, err = run( f"ssh -o StrictHostKeyChecking=no {SERVER} 'cat > /tmp/{tag}.sql'", inp=sql, timeout=120 ) if rc == 0: rc2, out2, err2 = run( f"ssh -o StrictHostKeyChecking=no {SERVER} 'cd /root && sqlite3 search.db < /tmp/{tag}.sql && rm -f /tmp/{tag}.sql && echo SQL导入完成'", timeout=120 ) print(f" 荆州: {out2 or err2[:100]}") print("\n--- 新疆环协(48MB SQL) ---") sql_path2 = os.path.join(BASE, "www.xjhbcy.cn_import.sql") if os.path.exists(sql_path2): with open(sql_path2) as f: sql = f.read() tag = f"sql_{int(time.time())}" rc, out, err = run( f"ssh -o StrictHostKeyChecking=no {SERVER} 'cat > /tmp/{tag}.sql'", inp=sql, timeout=180 ) if rc == 0: rc2, out2, err2 = run( f"ssh -o StrictHostKeyChecking=no {SERVER} 'cd /root && sqlite3 search.db < /tmp/{tag}.sql && rm -f /tmp/{tag}.sql && echo SQL导入完成'", timeout=180 ) print(f" 新疆环协: {out2 or err2[:100]}") # 重建FTS print("\n--- 重建FTS ---") rc, out, err = 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重建完成\"" , timeout=60 ) print(f" {out}") # 统计 rc, out, err = run( f"ssh -o StrictHostKeyChecking=no {SERVER} 'sqlite3 /root/search.db \"SELECT site_name, COUNT(*) FROM gov_raw GROUP BY site_name ORDER BY COUNT(*) DESC\"'", timeout=10 ) print(f"\n📊 服务器 search.db 统计:") for line in out.split('\n'): if '|' in line: sn, cnt = line.split('|') print(f" {sn.strip():30s} | {cnt.strip()}条") print("\n✅ 同步完成") if __name__ == "__main__": main()