#!/usr/bin/env python3 """批量抓取四川剩余市州组织部和党校预算数据""" import json, time, os, re, subprocess 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 # 杀残留进程 subprocess.run(["pkill", "-9", "chrome"], capture_output=True) subprocess.run(["pkill", "-9", "chromedriver"], capture_output=True) time.sleep(1) SAVE_DIR = "/data/www/files/budget_pdfs" os.makedirs(SAVE_DIR, exist_ok=True) # 市州配置 cities = { "乐山": "leshan.gov.cn", "达州": "dazhou.gov.cn", "巴中": "cnbz.gov.cn", "资阳": "ziyang.gov.cn", "内江": "neijiang.gov.cn", "凉山": "lsz.gov.cn", "甘孜": "gzz.gov.cn", "阿坝": "abazhou.gov.cn", "雅安": "yaan.gov.cn", } # 已有市州的关键字模式 url_patterns = [ "/c110728/", # 区级部门预决算（金牛区模式） "/c110737/", # 党校类 "区级部门及相关单位财政预决算", "部门预决算", "财政预决算", "财政信息", ] keywords_budget = ["预算", "预决算", "2026"] keywords_dept = ["组织部", "党校", "组织"] keywords_match = ["培训", "网络培训", "能力提升", "视频拍摄", "党员教育片", "课程制作", "直播", "视频", "课件", "系统", "信息化"] results = {} def init_driver(): opts = Options() opts.add_argument('--headless=new') opts.add_argument('--no-sandbox') opts.add_argument('--disable-gpu') 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"]) driver = webdriver.Chrome(options=opts) driver.execute_cdp_cmd('Page.addScriptToEvaluateOnNewDocument', { 'source': 'Object.defineProperty(navigator, "webdriver", {get: () => undefined})' }) driver.set_page_load_timeout(30) return driver def find_budget_links(driver, city, domain): """从首页导航到财政预决算栏目""" base_url = f"https://www.{domain}" found = {"org": [], "school": []} # 策略1: 从首页找"政务公开"链接 try: driver.get(base_url) time.sleep(6) links = driver.find_elements(By.TAG_NAME, "a") # 收集所有链接文本和href all_links = [] for a in links: text = (a.text or "").strip() href = a.get_attribute("href") or "" if text and href: all_links.append({"text": text, "href": href}) # 找政务公开/信息公开链接 gov_links = [l for l in all_links if any(kw in l['text'] for kw in ['政务公开', '信息公开'])] # 策略2: 直接尝试常见路径 paths_to_try = [ f"{base_url}/{domain.split('.')[0]}/c110728/", # 财政信息 f"{base_url}/{domain.split('.')[0]}/zwgk/", # 政务公开 ] budget_page_url = None for path in paths_to_try: try: driver.get(path) time.sleep(5) if len(driver.page_source) > 2000: budget_page_url = path break except: continue # 策略3: 从政务公开页面找财政信息 if gov_links: for gl in gov_links[:2]: try: driver.get(gl['href']) time.sleep(5) page_links = driver.find_elements(By.TAG_NAME, "a") for a in page_links: text = (a.text or "").strip() href = a.get_attribute("href") or "" if any(kw in text for kw in ['财政信息', '财政预决算', '部门预算', '预决算公开']): try: driver.get(href) time.sleep(5) if len(driver.page_source) > 2000: budget_page_url = href except: pass break except: continue if not budget_page_url: return found # 在财政信息页搜索组织部和党校 driver.get(budget_page_url) time.sleep(5) # 获取所有链接 page_links = driver.find_elements(By.TAG_NAME, "a") for a in page_links: text = (a.text or "").strip() href = a.get_attribute("href") or "" if '组织部' in text and href: found["org"].append({"text": text, "href": href}) elif '党校' in text and href: found["school"].append({"text": text, "href": href}) # 如果财政页面没有直接列出部门，搜索子页面 if not found["org"] and not found["school"]: # 尝试搜索 search_links = [l for l in page_links if any(kw in (l.text or '') for kw in ['区级部门', '部门预算', '部门预决算', '部门及'])] for sl in search_links[:3]: try: driver.get(sl.get_attribute('href')) time.sleep(5) sub_links = driver.find_elements(By.TAG_NAME, "a") for a in sub_links: text = (a.text or "").strip() href = a.get_attribute("href") or "" if '组织部' in text and href: found["org"].append({"text": text, "href": href}) elif '党校' in text and href: found["school"].append({"text": text, "href": href}) except: continue except Exception as e: print(f" 搜索失败: {str(e)[:100]}") return found def get_budget_detail(driver, url, city, dept_type): """访问组织部/党校预算页面，提取附件和预算数据""" detail = {"url": url, "attachments": [], "budget_items": []} try: driver.get(url) time.sleep(6) # 截图 screenshot_path = f"{SAVE_DIR}/screenshot_{city}_{dept_type}.png" driver.save_screenshot(screenshot_path) detail["screenshot"] = screenshot_path # 找附件链接（PDF/xlsx/doc） links = driver.find_elements(By.TAG_NAME, "a") for a in links: href = a.get_attribute("href") or "" text = (a.text or "").strip() if any(href.endswith(ext) for ext in ['.pdf', '.xlsx', '.xls', '.doc', '.docx']): detail["attachments"].append({"text": text, "href": href}) # 提取页面文本中的预算信息 page_text = driver.find_element(By.TAG_NAME, "body").text for line in page_text.split('\n'): line = line.strip() if any(kw in line for kw in keywords_match) and any(kw in line for kw in ['万', '元', '经费', '支出', '项目']): detail["budget_items"].append(line) except Exception as e: detail["error"] = str(e)[:200] return detail def download_file(driver, url, filepath): """通过浏览器Fetch API下载文件（绕过WAF）""" try: result = driver.execute_async_script(""" var url = arguments[0]; var callback = arguments[1]; fetch(url) .then(r => r.arrayBuffer()) .then(buf => { var arr = new Uint8Array(buf); var binary = ''; for (var i = 0; i < arr.length; i++) { binary += String.fromCharCode(arr[i]); } callback({status: 'ok', size: arr.length, data: btoa(binary)}); }) .catch(e => callback({status: 'error', msg: e.message})); """, url) if result and result.get('status') == 'ok': import base64 data = base64.b64decode(result['data']) with open(filepath, 'wb') as f: f.write(data) return True except Exception as e: print(f" 下载失败: {str(e)[:100]}") return False # ============ 主流程 ============ driver = init_driver() all_results = {} for city, domain in cities.items(): print(f"\n{'='*50}") print(f"🔍 正在抓取: {city} ({domain})") print(f"{'='*50}") city_result = {"city": city, "domain": domain, "org": None, "school": None, "status": "processing"} # 找预算公开链接 found = find_budget_links(driver, city, domain) print(f" 找到: 组织部{len(found['org'])}个, 党校{len(found['school'])}个链接") for dept_type in ["org", "school"]: if found[dept_type]: for link in found[dept_type][:1]: # 只取第一个 print(f" 📄 访问 {dept_type}: {link['text']}") detail = get_budget_detail(driver, link['href'], city, dept_type) # 下载附件 for att in detail.get("attachments", []): ext = os.path.splitext(att['href'])[1] or '.pdf' filename = f"{city}_{dept_type}_2026{ext}" filepath = os.path.join(SAVE_DIR, filename) if download_file(driver, att['href'], filepath): att['local_path'] = filepath att['filename'] = filename print(f" ✅ 下载: {filename} ({os.path.getsize(filepath)//1024}KB)") else: print(f" ❌ 下载失败: {att['href'][:80]}") city_result[dept_type] = detail if city_result["org"] or city_result["school"]: city_result["status"] = "✅ 已获取" else: city_result["status"] = "⚠️ 未找到预算公开" all_results[city] = city_result print(f" 状态: {city_result['status']}") driver.quit() # 保存结果 with open(f"{SAVE_DIR}/remaining_cities_results.json", 'w') as f: json.dump(all_results, f, ensure_ascii=False, indent=2) print(f"\n✅ 抓取完成！结果保存至 {SAVE_DIR}/remaining_cities_results.json") for city, r in all_results.items(): org_items = len(r.get('org', {}).get('budget_items', [])) if r.get('org') else 0 school_items = len(r.get('school', {}).get('budget_items', [])) if r.get('school') else 0 print(f" {city}: {r['status']} | 组织部{org_items}项 党校{school_items}项")