#!/usr/bin/env python3 """ 每日增量爬虫统一执行器 从 daily_crawl_config.json 读取配置，依次执行启用的脚本 支持两种同步模式： - direct_server: 脚本直写远程search.db（通过SSH） - scrapy_to_searchdb: 本地Scrapy→quality_results.db→SQL同步到服务器 """ import json import os import subprocess import sys import time import tempfile BASE_DIR = os.path.dirname(os.path.abspath(__file__)) CONFIG_PATH = os.path.join(BASE_DIR, 'daily_crawl_config.json') REMOTE = 'root@1.94.217.116' SERVER_DB = '/root/search.db' VENV_ACTIVATE = os.path.join(BASE_DIR, 'venv', 'bin', 'activate') def load_config(): with open(CONFIG_PATH, encoding='utf-8') as f: return json.load(f)['crawlers'] def run_cmd(cmd, timeout=300): """Run a shell command and return (output, exit_code, elapsed)""" t0 = time.time() try: r = subprocess.run( cmd, shell=True, capture_output=True, text=True, timeout=timeout ) elapsed = time.time() - t0 out = (r.stdout or '') + (r.stderr or '') return out.strip(), r.returncode, elapsed except subprocess.TimeoutExpired: elapsed = time.time() - t0 return f"[TIMEOUT {timeout}s]", -1, elapsed except Exception as e: elapsed = time.time() - t0 return f"[ERROR] {e}", -1, elapsed def _save_progress(result, ts_epoch, status_map): """Write one result to run_status.json (progressive - called after each script)""" run_status_file = "/root/gov_crawler/run_status.json" key = result.get("script") or result.get("name", "") if result["success"]: status_str = f"成功 +{result['new_count']}条" else: error_type = result.get("error_type", "") if result.get("timeout") or error_type == "timeout": status_str = "超时" elif error_type in ("connection_refused", "dns_resolve", "unreachable", "read_timeout"): status_str = "站点不可达" elif error_type == "script_error": status_str = "脚本异常" elif error_type == "exit_error": status_str = f"失败 rc={result.get('error','?')}" else: status_str = "失败" if key: status_map[key] = {"status": status_str, "ts": ts_epoch, "elapsed": result.get("elapsed", 0), "error_type": result.get("error_type", "")} name_key = result.get("name", "") if name_key and name_key != key: status_map[name_key] = {"status": status_str, "ts": ts_epoch, "elapsed": result.get("elapsed", 0), "error_type": result.get("error_type", "")} try: with open(run_status_file, "w") as f: json.dump(status_map, f, ensure_ascii=False, indent=2) except Exception: pass def run_local_script(script, args='', workdir=None): """Run local script (Python/JS/Shell based on extension)""" wd = workdir or BASE_DIR if script.endswith('.js'): cmd = f'cd "{wd}" && node "{script}" {args}' elif script.endswith('.sh'): cmd = f'cd "{wd}" && bash "{script}" {args}' else: venv_python = f'. "{VENV_ACTIVATE}" && python3' cmd = f'cd "{wd}" && {venv_python} "{script}" {args}' print(f" ▶ 执行: {cmd[:120]}") out, code, elapsed = run_cmd(cmd, timeout=600) if code != 0: print(f" ❌ 错误 (exit={code}) [{elapsed:.0f}s]") else: print(f" ✅ 完成 [{elapsed:.0f}s]") lines = out.split('\n') for l in lines[-15:]: if '新增' in l or '完成' in l or 'Error' in l or 'ERROR' in l or 'error' in l or '✅' in l or '❌' in l: print(f" {l}") return out, code, elapsed def sync_scrapy_to_server(site_name, script_name): """从 quality_results.db 查出该 domain 的新数据，同步到服务器""" domain_map = { '佛冈县人民政府': ('fogang', '佛冈'), '新疆生态环境保护产业协会': ('xjhbcy', '新疆环评协会'), '新疆准东经济技术开发区': ('xjzdkfq', '准东'), '环评爱好者': ('eiafans', '环评爱好者'), '新和成环境信息': ('cnhu', '新和成'), '荆州市生态环境局': ('jz', '荆州'), '石家庄炼化': ('sjlh', '石家庄炼化'), } domain_info = domain_map.get(site_name, (script_name.replace('crawl_', '').replace('.py', ''), site_name)) domain_key = domain_info[0] # Query quality_results.db qry = f""" SELECT title, page_url, content, publish_date, summary, category FROM quality_results.db.items WHERE domain LIKE '%{domain_key}%' AND created_at > datetime('now', '-1 day') """ # This is placeholder logic - the actual implementation depends on the specific Scrapy setup print(f" ↻ 同步 {site_name}: 检查 quality_results.db 新数据...") # TODO: 实际实现 - 查询quality_results.db → 生成SQL → SSH管道到服务器 return 0 def run_all(enabled_only=True, groups=None): crawlers = load_config() total = 0 failed = 0 results = [] ts_epoch = time.time() status_map = {} for c in crawlers: if enabled_only and not c.get('enabled', True): print(f" ⏭️ 跳过 {c['name']} (未启用)") continue if groups and c.get('group') not in groups: continue print(f"\n{'='*60}") print(f" 📍 {c['name']}") print(f" 脚本: {c['script']} | 模式: {c.get('sync_mode','?')}") print(f"{'='*60}") if not c.get('incremental', False): print(f" ⏭️ 无增量模式，跳过") continue script_path = os.path.join(BASE_DIR, c['script']) if not os.path.exists(script_path): print(f" ❌ 脚本不存在: {script_path}") failed += 1 continue try: out, code, elapsed = run_local_script(c['script'], c.get('args', '')) success_flag = code == 0 timeout_flag = code == -1 and "TIMEOUT" in out # 分类错误类型 error_type = "" if not success_flag: err_lower = out.lower() if timeout_flag: error_type = "timeout" elif any(x in err_lower for x in ["connection refused", "connection reset", "连接被拒绝", "connect failed"]): error_type = "connection_refused" elif any(x in err_lower for x in ["name or service not known", "temporary failure in name resolution", "无法解析", "resolve", "dns"]): error_type = "dns_resolve" elif any(x in err_lower for x in ["no route to host", "network is unreachable", "无法访问", "网络不可达", "连接超时"]): error_type = "unreachable" elif "timeout" in err_lower and "read timed out" in err_lower: error_type = "read_timeout" elif "traceback" in err_lower or "traceback" in err_lower: error_type = "script_error" elif code != 0: error_type = "exit_error" if success_flag: # 检查是否要同步到服务器 if c.get('sync_mode') == 'scrapy_to_searchdb': sync_scrapy_to_server(c['name'], c['script']) total += 1 else: failed += 1 results.append({ "name": c["name"], "script": c.get("script", ""), "success": success_flag, "new_count": 0, "timeout": timeout_flag, "elapsed": round(elapsed, 1), "error_type": error_type, "error": "" if success_flag else (out[:200] if timeout_flag else f"exit code {code}") }) # 逐步写入 run_status.json _save_progress(results[-1], ts_epoch, status_map) except Exception as e: print(f" ❌ 异常: {e}") failed += 1 results.append({ "name": c["name"], "script": c.get("script", ""), "success": False, "new_count": 0, "timeout": False, "elapsed": 0, "error_type": "exception", "error": str(e)[:300] }) # 逐步写入 run_status.json _save_progress(results[-1], ts_epoch, status_map) time.sleep(1) print(f"\n{'='*60}") print(f" 📊 完成: 成功 {total} / 失败 {failed}") print(f"{'='*60}") # 更新固化 run_status.json ts_epoch = time.time() status_map = {} try: with open("/root/gov_crawler/run_status.json") as _f: status_map = json.load(_f) except: pass for r in results: key = r.get("script") or r.get("name", "") if r["success"]: status_str = f"成功 +{r['new_count']}条" else: error_type = r.get("error_type", "") if r.get("timeout") or error_type == "timeout" or "timeout" in r.get("error", "").lower(): status_str = "超时" elif error_type in ("connection_refused", "dns_resolve", "unreachable", "read_timeout"): status_str = "站点不可达" elif error_type == "script_error": status_str = "脚本异常" elif error_type == "exit_error": status_str = f"失败 rc={r.get('error','?')}" else: status_str = "失败" if key: status_map[key] = { "status": status_str, "ts": ts_epoch, "elapsed": r.get("elapsed", 0), "error_type": r.get("error_type", "") } name_key = r.get("name", "") if name_key and name_key != key and name_key not in status_map: status_map[name_key] = { "status": status_str, "ts": ts_epoch, "elapsed": r.get("elapsed", 0), "error_type": r.get("error_type", "") } with open("/root/gov_crawler/run_status.json", "w") as f: json.dump(status_map, f, ensure_ascii=False, indent=2) print(f"run_status.json updated: {len(results)} entries") return total, failed if __name__ == '__main__': import argparse parser = argparse.ArgumentParser(description='每日增量爬虫执行器') parser.add_argument('--list', action='store_true', help='列出所有爬虫配置') parser.add_argument('--group', help='只运行指定组') parser.add_argument('--run-all', action='store_true', help='运行所有启用的爬虫') args = parser.parse_args() if args.list: crawlers = load_config() print(f"{'站点名称':<30} {'脚本':<25} {'组':<15} {'增量':<6} {'启用':<6}") print('-'*82) for c in crawlers: inc = '✅' if c.get('incremental') else '❌' en = '✅' if c.get('enabled', True) else '❌' grp = c.get('group', '') print(f"{c['name']:<30} {c['script']:<25} {grp:<15} {inc:<6} {en:<6}") elif args.run_all or args.group: groups = [args.group] if args.group else None run_all(enabled_only=True, groups=groups) else: parser.print_help()