#!/usr/bin/env python3 """用Selenium深度抓取5个可达市州的组织部/党校预算""" import json, time, re, os, urllib.request, 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 RESULTS_FILE = "/tmp/cities_budget_results.json" PDF_DIR = "/data/www/files/budget_pdfs" cities = { "leshan": "乐山市", "dazhou": "达州市", "yaan": "雅安市", "ziyang": "资阳市", "ganzi": "甘孜州" } 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 get_page(driver, url, wait=3): driver.get(url) time.sleep(wait) return driver.page_source def find_budget_links(driver, city_slug, city_name): """在财政信息页面查找组织部和党校的预算公开链接""" links = {"组织部": [], "党校": []} # 搜索所有链接 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 text_lower = text.lower() href_lower = href.lower() # 匹配组织部 if any(k in text for k in ["组织部", "组织部门", "党委组织部", "州委组织部", "市委组织部"]): if "预算" in text or "决算" in text or "2026" in text: links["组织部"].append({"title": text, "url": href}) elif "c110" in href_lower or "c1107" in href_lower or "cazhb" in href_lower: links["组织部"].append({"title": text, "url": href}) # 匹配党校 if any(k in text for k in ["党校", "行政学院"]): if "预算" in text or "决算" in text or "2026" in text: links["党校"].append({"title": text, "url": href}) elif "c110" in href_lower or "c1107" in href_lower or "cazhb" in href_lower: links["党校"].append({"title": text, "url": href}) except: continue return links def search_budget_section(driver, domain, city_slug, city_name): """搜索财政预决算栏目""" found = {"组织部": {"links": [], "pdfs": []}, "党校": {"links": [], "pdfs": []}} # 策略1: 尝试常见财政栏目URL模式 budget_urls = [ f"https://www.{domain}/{city_slug}/c110728/", # 财政信息 f"https://www.{domain}/{city_slug}/c110728/2026.shtml", f"https://www.{domain}/{city_slug}/c110728/yusuan/", # 预算 f"https://www.{domain}/{city_slug}/c110728/juesuan/", # 决算 f"https://www.{domain}/{city_slug}/cazhb/", # 财政预决算公开 f"https://www.{domain}/{city_slug}/cazhb/list.shtml", # 部门预算公开 f"https://www.{domain}/{city_slug}/c110728/bmys.shtml", f"https://www.{domain}/{city_slug}/c110728/bmzgk.shtml", # 2026年专栏 f"https://www.{domain}/{city_slug}/c110728/2026_ysgk.shtml", ] for url in budget_urls: try: html = get_page(driver, url, wait=2) if len(html) < 200: continue # 检查是否有组织部/党校相关内容 if any(k in html for k in ["组织部", "党校"]): links = find_budget_links(driver, city_slug, city_name) for dept in ["组织部", "党校"]: for link in links[dept]: if link not in found[dept]["links"]: found[dept]["links"].append(link) print(f" ✓ 找到财政栏目: {url}") print(f" 组织部: {len(links['组织部'])}项, 党校: {len(links['党校'])}项") except Exception as e: continue # 策略2: 从首页搜索导航 try: get_page(driver, f"https://www.{domain}/", wait=3) # 点击政务公开/财政信息相关链接 nav_links = driver.find_elements(By.TAG_NAME, "a") for a in nav_links: try: text = a.text.strip() href = a.get_attribute("href") or "" if any(k in text for k in ["财政", "预算", "预决算", "公开"]): print(f" → 导航到: {text} ({href})") get_page(driver, href, wait=2) links = find_budget_links(driver, city_slug, city_name) for dept in ["组织部", "党校"]: for link in links[dept]: if link not in found[dept]["links"]: found[dept]["links"].append(link) break # 只点第一个匹配的 except: continue except: pass return found def extract_pdf_from_page(driver, url): """从预算公开详情页提取PDF附件链接""" pdfs = [] try: get_page(driver, url, wait=2) # 查找附件链接 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", ".docx"]): pdfs.append({"title": text or href.split("/")[-1], "url": href}) except: continue except: pass return pdfs # 主程序 print("="*60) print("开始深度抓取5个可达市州预算数据") print("="*60) driver = setup_driver() for slug, name in cities.items(): domain = f"{slug}.gov.cn" print(f"\n{'='*40}") print(f"🔍 抓取 {name} ({domain})") print(f"{'='*40}") result = search_budget_section(driver, domain, slug, name) all_results[slug] = result org_count = len(result["组织部"]["links"]) dx_count = len(result["党校"]["links"]) print(f" 结果: 组织部{org_count}条, 党校{dx_count}条") # 如果找到链接，深入访问提取PDF for dept in ["组织部", "党校"]: for link in result[dept]["links"][:3]: # 每个部门最多深入3个 print(f" → 深入: {link['title']}") pdfs = extract_pdf_from_page(driver, link["url"]) result[dept]["pdfs"].extend(pdfs) for p in pdfs: print(f" 📄 {p['title']}: {p['url']}") driver.quit() # 保存结果 with open(RESULTS_FILE, "w") as f: json.dump(all_results, f, ensure_ascii=False, indent=2) print(f"\n结果已保存到 {RESULTS_FILE}") # 汇总 for slug, data in all_results.items(): name = cities[slug] org = len(data["组织部"]["links"]) dx = len(data["党校"]["links"]) org_pdf = len(data["组织部"]["pdfs"]) dx_pdf = len(data["党校"]["pdfs"]) print(f" {name}: 组织部{org}条({org_pdf}附件) 党校{dx}条({dx_pdf}附件)")