#!/usr/bin/env python3 """DrissionPage批量抓取成都区县组织部/党校2026年预算""" import json, time, os, re from DrissionPage import ChromiumPage, ChromiumOptions from urllib.parse import urlparse OUT_DIR = "/data/www/files/budget_pdfs/chengdu_districts" SS_DIR = "/data/www/files/budget_pdfs/screenshots" RESULT_FILE = "/data/www/files/budget_data/cd_districts_v2.json" os.makedirs(OUT_DIR, exist_ok=True) os.makedirs(SS_DIR, exist_ok=True) DISTRICTS = { "青羊区": {"domain": "www.qingyang.gov.cn", "code": "qingyang"}, "武侯区": {"domain": "www.wuhou.gov.cn", "code": "wuhou"}, "成华区": {"domain": "www.chenghua.gov.cn", "code": "chenghua"}, "新都区": {"domain": "www.xindu.gov.cn", "code": "xindu"}, "温江区": {"domain": "www.wenjiang.gov.cn", "code": "wenjiang"}, "郫都区": {"domain": "www.pidu.gov.cn", "code": "pidu"}, "新津区": {"domain": "www.xinjin.gov.cn", "code": "xinjin"}, "青白江区": {"domain": "www.qbj.gov.cn", "code": "qingbaijiang"}, "都江堰市": {"domain": "www.djy.gov.cn", "code": "dujiangyan"}, "彭州市": {"domain": "www.pengzhou.gov.cn", "code": "pengzhou"}, "邛崃市": {"domain": "www.qionglai.gov.cn", "code": "qionglai"}, "崇州市": {"domain": "www.chongzhou.gov.cn", "code": "chongzhou"}, "简阳市": {"domain": "www.jianyang.gov.cn", "code": "jianyang"}, "金堂县": {"domain": "www.jintang.gov.cn", "code": "jintang"}, "大邑县": {"domain": "www.dayi.gov.cn", "code": "dayi"}, "蒲江县": {"domain": "www.pujiang.gov.cn", "code": "pujiang"}, } results = {} co = ChromiumOptions() co.set_argument("--no-sandbox") co.set_argument("--disable-dev-shm-usage") co.set_argument("--disable-gpu") co.set_argument("--window-size=1920,1080") co.headless(True) page = ChromiumPage(co) print("浏览器启动成功") for name, cfg in DISTRICTS.items(): print(f"\n=== {name} ({cfg['domain']}) ===") domain = cfg["domain"] code = cfg["code"] base = f"https://{domain}" district_result = {"组织部": [], "党校": [], "screenshots": [], "status": "scanning"} # Step 1: 访问首页建立WAF Cookie try: page.get(base) time.sleep(3) title = page.title or "" print(f" 首页: {title[:40]}") except Exception as e: print(f" 首页失败: {e}") district_result["status"] = "unreachable" results[name] = district_result with open(RESULT_FILE, "w") as f: json.dump(results, f, ensure_ascii=False, indent=2) continue # Step 2: 搜索财政信息栏目 found_any = False for col_code in ["c110728", "c110729", "c100043", "c111709", "c115761"]: for pn in range(1, 4): suffix = f"_{pn}" if pn > 1 else "" list_url = f"{base}/{code}/{col_code}/list{suffix}.shtml" try: page.get(list_url) time.sleep(1.5) html = page.html or "" if len(html) < 500: continue # 提取链接 all_links = re.findall(r']*href=["\']([^"\']+)["\'][^>]*>([^<]{4,80})', html) for href, text in all_links: text = text.strip() if not text: continue for kw in ["组织部", "党校"]: if kw in text and ("2026" in text or "预算" in text or "2026" in href): full_href = href if href.startswith("/"): full_href = f"{base}{href}" elif not href.startswith("http"): full_href = f"{base}/{code}/{href}" existing_titles = [x["title"] for x in district_result[kw]] if text not in existing_titles: district_result[kw].append({"title": text, "url": full_href}) print(f" ✅ {kw}: {text[:50]}") found_any = True except: continue # Step 3: 如果列表页没找到，尝试搜索 if not found_any: for kw in ["组织部", "党校"]: search_url = f"{base}/search.html?q={kw}+2026+预算" try: page.get(search_url) time.sleep(2) html = page.html or "" all_links = re.findall(r']*href=["\']([^"\']+)["\'][^>]*>([^<]{4,80})', html) for href, text in all_links: text = text.strip() if kw in text and ("2026" in text or "预算" in text): full_href = href if href.startswith("/"): full_href = f"{base}{href}" elif not href.startswith("http"): full_href = f"{base}/{code}/{href}" existing_titles = [x["title"] for x in district_result[kw]] if text not in existing_titles: district_result[kw].append({"title": text, "url": full_href}) print(f" 🔍 {kw}: {text[:50]}") except: continue # Step 4: 截图附件下载 for kw in ["组织部", "党校"]: for item in district_result[kw][:3]: try: page.get(item["url"]) time.sleep(2) # 截图 ss_name = f"{code}_{kw}_detail.png" page.get_screenshot(path=f"{SS_DIR}/{ss_name}", full_page=True) district_result["screenshots"].append({"file": f"budget_pdfs/screenshots/{ss_name}", "title": item["title"]}) print(f" 📸 截图: {ss_name}") # 附件 html = page.html or "" att_links = re.findall(r'href=["\']([^"\']*\.(?:pdf|xlsx?|docx?))["\']', html, re.IGNORECASE) downloaded = [] for att_url in att_links[:3]: if att_url.startswith("/"): att_url = f"{base}{att_url}" elif not att_url.startswith("http"): att_url = f"{base}/{code}/{att_url}" fname = att_url.split("/")[-1].split("?")[0] save_path = f"{OUT_DIR}/{code}_{kw}_{fname}" try: js = f"return await fetch('{att_url}').then(r=>r.arrayBuffer()).then(b=>Array.from(new Uint8Array(b)))" data = page.run_js(js) if data and len(data) > 200: with open(save_path, "wb") as f: f.write(bytes(data)) sz = os.path.getsize(save_path) downloaded.append({"url": att_url, "local": f"budget_pdfs/chengdu_districts/{code}_{kw}_{fname}", "size": sz}) print(f" 📎 {fname} ({sz/1024:.0f}KB)") except: pass item["attachments"] = downloaded except Exception as e: print(f" 详情页失败: {e}") district_result["status"] = "completed" results[name] = district_result with open(RESULT_FILE, "w") as f: json.dump(results, f, ensure_ascii=False, indent=2) print(f" 💾 进度已保存") try: page.quit() except: pass total_org = sum(len(r.get("组织部", [])) for r in results.values()) total_dx = sum(len(r.get("党校", [])) for r in results.values()) total_files = sum(len(a) for r in results.values() for k in ["组织部", "党校"] for i in r.get(k, []) for a in i.get("attachments", [])) total_ss = sum(len(r.get("screenshots", [])) for r in results.values()) completed = sum(1 for r in results.values() if r.get("status") == "completed") print(f"\n===== 完成统计 =====") print(f"扫描区县: {len(results)}") print(f"完成: {completed}") print(f"组织部条目: {total_org}") print(f"党校条目: {total_dx}") print(f"下载文件: {total_files}") print(f"截图: {total_ss}")