#!/usr/bin/env python3 """批量抓取四川各市州组织部/党校2026预算 - 使用浏览器Fetch API下载PDF""" import json, time, re, os, subprocess, urllib.parse from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC PDF_DIR = "/data/www/files/budget_pdfs" RESULTS_FILE = "/tmp/all_cities_2026_budget.json" # 所有可达的市州（从curl测试得出200的） cities = { "leshan": {"name": "乐山市", "domain": "leshan.gov.cn", "prefix": "lsswszf"}, "dazhou": {"name": "达州市", "domain": "dazhou.gov.cn", "prefix": "dazhou"}, "yaan": {"name": "雅安市", "domain": "yaan.gov.cn", "prefix": "yaan"}, "ziyang": {"name": "资阳市", "domain": "ziyang.gov.cn", "prefix": "ziyang"}, "ganzi": {"name": "甘孜州", "domain": "ganzi.gov.cn", "prefix": "ganzi"}, "yibin": {"name": "宜宾市", "domain": "yibin.gov.cn", "prefix": "yibin"}, "mianyang": {"name": "绵阳市", "domain": "mianyang.gov.cn", "prefix": "mianyang"}, "deyang": {"name": "德阳市", "domain": "deyang.gov.cn", "prefix": "deyang"}, "nanchong": {"name": "南充市", "domain": "nanchong.gov.cn", "prefix": "nanchong"}, "luzhou": {"name": "泸州市", "domain": "luzhou.gov.cn", "prefix": "luzhou"}, "suining": {"name": "遂宁市", "domain": "suining.gov.cn", "prefix": "suining"}, "neijiang": {"name": "内江市", "domain": "neijiang.gov.cn", "prefix": "neijiang"}, "liangshan": {"name": "凉山州", "domain": "liangshan.gov.cn", "prefix": "liangshan"}, "guangan": {"name": "广安市", "domain": "guangan.gov.cn", "prefix": "guangan"}, "meishan": {"name": "眉山市", "domain": "meishan.gov.cn", "prefix": "meishan"}, "zigong": {"name": "自贡市", "domain": "zigong.gov.cn", "prefix": "zigong"}, "bazhong": {"name": "巴中市", "domain": "bazhong.gov.cn", "prefix": "bazhong"}, } # 以及成都区县 chengdu_districts = { "jinniu": {"name": "金牛区", "domain": "jinniu.gov.cn", "prefix": "jinniu"}, "jinjiang": {"name": "锦江区", "domain": "jinjiang.gov.cn", "prefix": "jinjiang"}, # 注意可能是晋江 "shuangliu": {"name": "双流区", "domain": "shuangliu.gov.cn", "prefix": "shuangliu"}, "qingyang": {"name": "青羊区", "domain": "cdzy.gov.cn", "prefix": "qingyang"}, "wuhou": {"name": "武侯区", "domain": "cdwh.gov.cn", "prefix": "wuhou"}, "chenghua": {"name": "成华区", "domain": "chenghua.gov.cn", "prefix": "chenghua"}, "longquanyi": {"name": "龙泉驿区", "domain": "longquanyi.gov.cn", "prefix": "longquanyi"}, "xindu": {"name": "新都区", "domain": "xindu.gov.cn", "prefix": "xindu"}, "wenjiang": {"name": "温江区", "domain": "wenjiang.gov.cn", "prefix": "wenjiang"}, "pidu": {"name": "郫都区", "domain": "pidu.gov.cn", "prefix": "pidu"}, "xinjin": {"name": "新津区", "domain": "xinjin.gov.cn", "prefix": "xinjin"}, "jintang": {"name": "金堂县", "domain": "jintang.gov.cn", "prefix": "jintang"}, "dayi": {"name": "大邑县", "domain": "dayi.gov.cn", "prefix": "dayi"}, "pujiang": {"name": "蒲江县", "domain": "pujiang.gov.cn", "prefix": "pujiang"}, "dujiangyan": {"name": "都江堰市", "domain": "dujiangyan.gov.cn", "prefix": "dujiangyan"}, "pengzhou": {"name": "彭州市", "domain": "pengzhou.gov.cn", "prefix": "pengzhou"}, "qionglai": {"name": "邛崃市", "domain": "qionglai.gov.cn", "prefix": "qionglai"}, "chongzhou": {"name": "崇州市", "domain": "chongzhou.gov.cn", "prefix": "chongzhou"}, "jianyang": {"name": "简阳市", "domain": "jianyang.gov.cn", "prefix": "jianyang"}, "qingbaijiang": {"name": "青白江区", "domain": "qbj.gov.cn", "prefix": "qingbaijiang"}, } all_results = {} def setup_driver(): opts = Options() opts.add_argument("--headless=new") opts.add_argument("--no-sandbox") opts.add_argument("--disable-dev-shm-usage") opts.add_argument("--disable-blink-features=AutomationControlled") opts.add_argument("--window-size=1920,1080") opts.add_experimental_option("excludeSwitches", ["enable-automation"]) opts.add_experimental_option('useAutomationExtension', False) d = webdriver.Chrome(options=opts) d.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument", { "source": "Object.defineProperty(navigator, 'webdriver', {get: () => undefined})" }) return d def find_budget_page(driver, domain, prefix, city_name): """导航到预算公开页面""" # 策略1: 直接访问 /ysjs/ 路径（乐山模式） urls_to_try = [ f"https://www.{domain}/{prefix}/ysjs/", f"https://www.{domain}/{prefix}/ysjs/index.html", f"https://www.{domain}/{prefix}/c110728/", f"https://www.{domain}/{prefix}/c110728/yusuan/", ] for url in urls_to_try: try: driver.get(url) time.sleep(2) html = driver.page_source if len(html) > 500: # 检查是否包含预算相关内容 if any(k in html for k in ["预算", "部门预算", "2026"]): print(f" ✓ 找到预算页面: {url}") return url except: continue # 策略2: 从首页导航 try: driver.get(f"https://www.{domain}/") time.sleep(3) # 尝试搜索"预算"或"财政" search_input = None try: search_input = driver.find_element(By.CSS_SELECTOR, "input[type='text'], input.search, input[name='search']") except: pass if search_input: search_input.send_keys("2026年部门预算") try: search_btn = driver.find_element(By.CSS_SELECTOR, "button.search, input[type='submit']") search_btn.click() time.sleep(2) return driver.current_url except: pass except: pass return None def extract_org_dx_links(driver, budget_url, city_name): """从预算公开页面提取组织部和党校的链接""" org_links = [] dx_links = [] try: all_a = driver.find_elements(By.TAG_NAME, "a") for a in all_a: try: text = a.text.strip() href = a.get_attribute("href") or "" if not text or not href: continue if any(k in text for k in ["组织部", "组织部门"]): org_links.append({"title": text, "url": href}) print(f" → 组织部: {text}") if any(k in text for k in ["党校", "行政学院"]): dx_links.append({"title": text, "url": href}) print(f" → 党校: {text}") except: continue except Exception as e: print(f" 提取链接出错: {e}") return org_links, dx_links def download_pdf_via_browser(driver, url, filename): """通过浏览器Fetch API下载PDF文件（绕过WAF）""" filepath = os.path.join(PDF_DIR, filename) if os.path.exists(filepath) and os.path.getsize(filepath) > 1000: print(f" 📄 已存在: {filename}") return filepath try: js_code = f""" return fetch('{url}') .then(r => r.blob()) .then(blob => {{ return new Promise(resolve => {{ const reader = new FileReader(); reader.onload = () => resolve(reader.result.split(',')[1]); reader.readAsDataURL(blob); }}); }}) .catch(e => 'ERROR:' + e.message); """ b64_data = driver.execute_async_script(js_code) if b64_data and not b64_data.startswith("ERROR"): import base64 with open(filepath, "wb") as f: f.write(base64.b64decode(b64_data)) size = os.path.getsize(filepath) print(f" ✓ 下载成功: {filename} ({size} bytes)") return filepath except Exception as e: print(f" ✗ 下载失败: {e}") return None # 主程序 print("="*60) print("批量抓取四川各市州组织部/党校2026预算") print("="*60) driver = setup_driver() # 合并市州和区县 all_targets = {} all_targets.update(cities) all_targets.update(chengdu_districts) for slug, info in all_targets.items(): name = info["name"] domain = info["domain"] prefix = info["prefix"] print(f"\n{'='*40}") print(f"🔍 {name} ({domain})") print(f"{'='*40}") result = {"org": {"links": [], "pdfs": []}, "dx": {"links": [], "pdfs": []}} # 1. 找预算公开页面 budget_url = find_budget_page(driver, domain, prefix, name) if budget_url: # 2. 提取组织部/党校链接 org_links, dx_links = extract_org_dx_links(driver, budget_url, name) result["org"]["links"] = org_links result["dx"]["links"] = dx_links # 3. 访问每个链接，提取PDF附件 for link in org_links + dx_links: dept = "org" if "组织" in link["title"] else "dx" try: driver.get(link["url"]) time.sleep(2) # 查找PDF附件 all_a = driver.find_elements(By.TAG_NAME, "a") for a in all_a: try: href = a.get_attribute("href") or "" text = a.text.strip() if any(ext in href.lower() for ext in [".pdf", ".xls", ".xlsx", ".doc"]): fname = href.split("/")[-1].split("?")[0] if not fname: fname = f"{slug}_{dept}_2026.pdf" local_name = f"{slug}_{dept}_{fname}" pdf_path = download_pdf_via_browser(driver, href, local_name) if pdf_path: result[dept]["pdfs"].append({ "title": text or fname, "url": href, "local": f"/budget_pdfs/{local_name}" }) except: continue except: continue else: print(f" ✗ 未找到预算公开页面") all_results[slug] = result driver.quit() # 保存结果 with open(RESULTS_FILE, "w") as f: json.dump(all_results, f, ensure_ascii=False, indent=2) print(f"\n结果已保存到 {RESULTS_FILE}") # 汇总 total_org = sum(len(v["org"]["links"]) for v in all_results.values()) total_dx = sum(len(v["dx"]["links"]) for v in all_results.values()) total_pdf = sum(len(v["org"]["pdfs"]) + len(v["dx"]["pdfs"]) for v in all_results.values()) print(f"\n汇总: 组织部{total_org}条, 党校{total_dx}条, PDF{total_pdf}个")