#!/usr/bin/env python3 # -*- coding: utf-8 -*- """简阳市人民政府 - 公示公告栏目爬虫 站点: https://www.scjy.gov.cn (四川成都, /jianyang/ 站群) 栏目: /jianyang/c123218/list_more.shtml (公示公告, 当前 5 条静态) WAF: 瑞数4代 JS 挑战 (412, $_ts) → Playwright 过挑战 列表: list_more.shtml 静态条目 标题 (无日期, 标题可能截断带...) 详情: fetch → meta ArticleTitle(完整标题)/PubDate + div.main_bd 正文 (图片型公告 img 相对路径绝对化) 入库: 自动检测 /root/search.db 存在则 INSERT OR IGNORE 直插 gov_raw (FTS 触发器自动同步 gov_search) 运行: python3 crawl_jianyang_gsgg.py [--no-db] [--sync-only] """ import json, os, re, sys, time, sqlite3 from datetime import datetime, timedelta from playwright.sync_api import sync_playwright SITE_NAME = "简阳市-公示公告" BASE = "https://www.scjy.gov.cn" LIST_1 = "https://www.scjy.gov.cn/jianyang/c123218/list_more.shtml" CUTOFF = (datetime.now() - timedelta(days=3*365)).strftime("%Y-%m-%d") UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" OUT = "/tmp/jianyang_gsgg.jsonl" DB_PATH = "/root/search.db" CATEGORY = "公示公告" def clean_title(t): t = re.sub(r"^[•··.\s]+", "", t or "").strip() t = t.replace("\u200b", "").replace("\ufeff", "") t = re.sub(r"\.{3,}$", "", t).strip() # 列表截断省略号 return t def clean_content(html): """正文清洗: span/font unwrap + 嵌套p扁平化 + 保留表格""" from bs4 import BeautifulSoup soup = BeautifulSoup(html, "html.parser") for tag in soup.find_all(lambda t: t.name and ":" in t.name): tag.decompose() for tag in soup.find_all(["span", "font"]): tag.unwrap() # 只提升含嵌套 p 的外层 p (如

段1

段2

), # 保留内层真段落 p — 勿 unwrap 内层 p 否则段落被拍平成一坨 (永丰修复) while True: nested = [p for p in soup.find_all("p") if p.find("p")] if not nested: break nested[0].unwrap() for p in soup.find_all("p"): if p.get("style"): p.attrs = {"align": p.get("align")} if p.get("align") else {} return str(soup) def extract_attachments(html_frag, detail_url): """正文内附件: 相对路径绝对化""" atts = [] for m in re.finditer(r']+href="([^"]+)"[^>]*>(.*?)', html_frag, re.I | re.S): href, name_html = m.group(1).strip(), m.group(2) if not href or href.startswith("javascript"): continue if re.search(r'\.(pdf|docx?|xlsx?|zip|rar|7z|doc)$', href, re.I): name = re.sub(r"<[^>]+>", "", name_html).strip()[:100] if href.startswith("/"): href = BASE + href elif not href.startswith("http"): href = detail_url.rsplit("/", 1)[0] + "/" + href atts.append({"name": name or href.split("/")[-1], "url": href}) seen = set() uniq = [] for a in atts: if a["url"] not in seen: seen.add(a["url"]) uniq.append(a) return uniq def fetch_html(page, url): """页面内 fetch, 失败重试""" for _ in range(2): try: return page.evaluate("""async (u) => { const r = await fetch(u); return await r.text(); }""", url) except Exception: time.sleep(2) return None def push_to_db(results): if not results: return 0, 0 if not os.path.exists(DB_PATH): print(f" 未发现 {DB_PATH}, 跳过入库 (仅写 JSONL)", flush=True) return 0, 0 conn = sqlite3.connect(DB_PATH, timeout=290) conn.execute("PRAGMA busy_timeout=290000") added = skipped = 0 cur = conn.cursor() for r in results: try: cur.execute( """INSERT OR IGNORE INTO gov_raw (site_name, source_url, page_url, title, publish_date, content, summary, category, script_name, attachments, group_name, industry) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""", (SITE_NAME, r["url"], r["url"], r["title"], r["pub_date"], r["content"], (r["content"] or "")[:500], CATEGORY, "crawl_jianyang_gsgg.py", r["attachments"], r.get("group_name", "四川"), r.get("industry", "other")) ) if cur.rowcount > 0: added += 1 else: skipped += 1 except sqlite3.Error as e: print(f" DB err: {e}", flush=True) skipped += 1 conn.commit() conn.close() return added, skipped def main(): no_db = "--no-db" in sys.argv if "--sync-only" in sys.argv: print("sync-only: 抓取+入库一体, 无需单独同步", flush=True) return print(f"站点: {SITE_NAME} CUTOFF: {CUTOFF}", flush=True) results = [] stats = {"list": 0, "detail_ok": 0, "empty": 0, "cutoff": 0} with sync_playwright() as p: browser = p.chromium.launch( headless=True, args=["--no-sandbox", "--disable-blink-features=AutomationControlled", "--disable-dev-shm-usage"]) ctx = browser.new_context(user_agent=UA, locale="zh-CN", viewport={"width": 1366, "height": 900}, ignore_https_errors=True) page = ctx.new_page() # 过瑞数挑战 try: page.goto(LIST_1, timeout=60000, wait_until="commit") except Exception: pass passed = False for i in range(20): page.wait_for_timeout(3000) try: html = page.content() except Exception: continue if "$_ts" not in html and len(html) > 5000: print(f"瑞数挑战通过 {i*3}s", flush=True) passed = True break if not passed: print("挑战失败!", flush=True) browser.close() return # 列表条目 items = re.findall(r']+href="(/jianyang/c123218/\d{4}-\d{2}/\d{2}/content_[^"]+)"[^>]*class="special_content_list"[^>]*>(.*?)', html, re.S) if not items: items = re.findall(r']+href="(/jianyang/c123218/[^"]+)"[^>]*class="special_content_list"[^>]*>(.*?)', html, re.S) print(f"列表条目: {len(items)}", flush=True) for href, t_html in items: stats["list"] += 1 list_title = clean_title(re.sub(r"<[^>]+>", "", t_html)) detail_url = href if href.startswith("http") else BASE + href detail_html = fetch_html(page, detail_url) if not detail_html: print(f" 详情失败: {list_title[:30]}", flush=True) continue stats["detail_ok"] += 1 # 完整标题: meta ArticleTitle m = re.search(r']+name="ArticleTitle"[^>]+content="([^"]+)"', detail_html) title = clean_title(m.group(1)) if m else list_title # 日期: meta PubDate m = re.search(r']+name="PubDate"[^>]+content="([^"]+)"', detail_html) pub_date = (m.group(1) if m else "")[:10] if pub_date and pub_date < CUTOFF: stats["cutoff"] += 1 continue # 正文 div.main_bd m = re.search(r']*class="main_bd"[^>]*>(.*?)\s*]*class="xgfj"', detail_html, re.S) content_html = m.group(1) if m else "" content = clean_content(content_html) if content_html else "" text = re.sub(r"<[^>]+>", "", content).strip() if len(text) < 10 and "]+src=")([^"]+)(")', lambda mm: mm.group(1) + (mm.group(2) if mm.group(2).startswith("http") else (BASE + mm.group(2) if mm.group(2).startswith("/") else detail_url.rsplit("/", 1)[0] + "/" + mm.group(2))) + mm.group(3), content) atts = extract_attachments(content_html, detail_url) rec = { "site_name": SITE_NAME, "title": title, "pub_date": pub_date, "content": content, "source_url": detail_url, "url": detail_url, "group_name": "四川", "industry": "other", "attachments": json.dumps(atts, ensure_ascii=False) if atts else "", } results.append(rec) browser.close() save_results(results, stats) if not no_db: added, skipped = push_to_db(results) print(f"=== 入库: 新增{added} 跳过{skipped} ===", flush=True) print(f"=== 完成: 列表{stats['list']} 详情OK{stats['detail_ok']} 空{stats['empty']} CUTOFF截{stats['cutoff']} ===", flush=True) def save_results(results, stats=None): with open(OUT, "w", encoding="utf-8") as f: for r in results: f.write(json.dumps(r, ensure_ascii=False) + "\n") print(f"已写 {len(results)} 条到 {OUT}", flush=True) if __name__ == "__main__": main()