#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 仪征市乡镇人民政府 - 通知公告 ×11 (yizheng.gov.cn/zfxxgk/yqxzbmxxgkml/yqxz/<镇>/fdzdgknr/tzgg/) CMS: 翰Web/unitbuild JS 渲染 (Playwright 必须) 列表: Playwright 渲染 + 点击翻页 (共~22页/镇, 新规只抓前5页) 列表项: a href=".../tzgg/art/2026/art_{md5}.html" 详情: requests + bt-content/article-content/content 用法: python3 crawl_yizheng_xz_tzgg.py # 全部11乡镇 python3 crawl_yizheng_xz_tzgg.py --col=xjz # 单乡镇 python3 crawl_yizheng_xz_tzgg.py --col=all --pages=5 python3 crawl_yizheng_xz_tzgg.py --json out.jsonl # JSONL模式(不写DB) """ import re, sys, os, json, time, sqlite3, requests from datetime import datetime, timedelta from bs4 import BeautifulSoup from urllib.parse import urljoin from playwright.sync_api import sync_playwright BASE = "https://www.yizheng.gov.cn" DOMAIN = "www.yizheng.gov.cn" GROUP = "江苏" CUTOFF = (datetime.now() - timedelta(days=3*365)).strftime("%Y-%m-%d") DB_PATH = os.environ.get("DB_PATH", "/root/search.db") TOWNS = { "xjz": "新集镇", "mjz": "马集镇", "cjz": "陈集镇", "ljz": "刘集镇", "dyz": "大仪镇", "ytz": "月塘镇", "xcz": "新城镇", "qsz": "青山镇", "zlwlydjq": "枣林湾旅游度假区", "jjkfq": "经济开发区", "zzz": "真州镇", } HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"} def log(msg): print(msg, flush=True) def clean_title(t): t = re.sub(r" | |\u200b|\ufeff", "", t or "") t = re.sub(r"\s+", " ", t) t = re.sub(r"\.\.\.|…$", "", t).strip() return t def fetch_list_playwright(town_code, max_pages): """Playwright 渲染列表 + 点击翻页, 返回 [{title,url,date}]""" town_url = f"{BASE}/zfxxgk/yqxzbmxxgkml/yqxz/{town_code}/fdzdgknr/tzgg/index.html" items = [] seen = set() with sync_playwright() as p: browser = p.chromium.launch(headless=True, args=['--no-sandbox']) ctx = browser.new_context(user_agent=HEADERS["User-Agent"]) page = ctx.new_page() try: page.goto(town_url, wait_until='networkidle', timeout=40000) # 等待列表容器稳定出现 try: page.wait_for_selector('a[href*="/fdzdgknr/tzgg/art/"]', timeout=15000) except Exception: pass for pg in range(1, max_pages + 1): data = page.evaluate('''() => { const out = []; document.querySelectorAll('a').forEach(a => { const href = a.getAttribute('href') || ''; if (!href.includes('/fdzdgknr/tzgg/art/')) return; const t = (a.getAttribute('title') || a.textContent || '').trim(); if (t.length >= 6) out.push({t: t, h: href}); }); return out; }''') pg_new = 0 for it in data: url = it['h'] if it['h'].startswith('http') else BASE + it['h'] if url not in seen: seen.add(url) items.append({"title": clean_title(it['t']), "url": url, "date": ""}) pg_new += 1 log(f" [P{pg}] {len(data)} links, {pg_new} new, total {len(items)}") # 点击下一页 if pg < max_pages: clicked = page.evaluate('''() => { const links = Array.from(document.querySelectorAll('.page a, .pagination a, a')); const next = links.find(a => (a.textContent||'').trim() === '下一页'); if (next) { next.click(); return true; } return false; }''') if not clicked: break try: # 等列表内容变化 (URL 数量增加或 DOM 更新) page.wait_for_timeout(3000) try: page.wait_for_function('''(cnt) => { let n = 0; document.querySelectorAll('a').forEach(a => { if ((a.getAttribute('href')||'').includes('/fdzdgknr/tzgg/art/')) n++; }); return n > 0; }''', arg=None, timeout=8000) except Exception: pass except Exception: break except Exception as e: log(f" Playwright ERR: {e}") finally: browser.close() return items def parse_detail(url): try: r = requests.get(url, headers=HEADERS, timeout=30) r.encoding = 'utf-8' html = r.text except Exception: return {"title": "", "publish_date": "", "content": ""} soup = BeautifulSoup(html, 'lxml') result = {"title": "", "publish_date": "", "content": ""} ttag = soup.find("title") if ttag: result["title"] = clean_title(ttag.get_text(strip=True)) m = re.search(r"(\d{4}-\d{2}-\d{2})", html[:8000]) if m: result["publish_date"] = m.group(1) content_div = None for cls in ['bt-content', 'bt-article', 'article-content', 'content', 'page-content', 'theme-showcase']: content_div = soup.find('div', class_=cls) if content_div: break if not content_div: content_div = soup.find('div', id='zoom') or soup.find('div', id='UCAP-CONTENT') if content_div: for tag in content_div(['script', 'style']): tag.decompose() result["content"] = extract_clean_text(str(content_div), url) return result def extract_clean_text(content_html, page_url=None): if not content_html: return "" soup = BeautifulSoup(content_html, "html.parser") for a_tag in soup.find_all("a", href=True): href = a_tag.get("href", "") text = a_tag.get_text(strip=True) or "附件" full_url = urljoin(page_url or BASE, href) a_tag.replace_with(f'

{text}

') for tag in soup.find_all(['span', 'b', 'strong', 'font', 'em', 'i', 'u', 's']): tag.unwrap() parts = [] for el in soup.find_all(['table', 'p']): if el.name == 'table': parts.append(str(el)) elif el.name == 'p' and not el.find_parent('table'): t = el.get_text(separator='', strip=True) if t: parts.append(f'

{t}

') return '\n\n'.join(parts) if parts else content_html.strip() def crawl(towns=None, pages=5, json_out=None): if towns is None: towns = list(TOWNS.keys()) all_enriched = [] for tc in towns: tname = TOWNS.get(tc, tc) site_name = f"仪征市{tname}-通知公告" log(f"\n===== {site_name} =====") items = fetch_list_playwright(tc, pages) log(f"列表共 {len(items)} 条, 开始抓详情...") for idx, it in enumerate(items, 1): d = parse_detail(it["url"]) if d["publish_date"] and d["publish_date"] < CUTOFF: continue all_enriched.append({ "title": d["title"] or it["title"], "page_url": it["url"], "publish_date": d["publish_date"] or it["date"], "content": d["content"], "site_name": site_name, "column": "通知公告", }) if idx % 10 == 0: log(f" {idx}/{len(items)}") time.sleep(0.2) log(f"\n共 {len(all_enriched)} 条") if json_out: with open(json_out, "w", encoding="utf-8") as f: for item in all_enriched: f.write(json.dumps(item, ensure_ascii=False) + "\n") log(f"JSONL: {len(all_enriched)} items -> {json_out}") return conn = sqlite3.connect(DB_PATH, timeout=60) conn.execute("PRAGMA busy_timeout=10000") c = conn.cursor() stored = skipped = 0 for item in all_enriched: try: c.execute( "INSERT OR IGNORE INTO gov_raw " "(site_name, source_url, page_url, title, publish_date, content, summary, category, script_name)" " VALUES (?,?,?,?,?,?,?,?,?)", (item["site_name"], item["page_url"], item["page_url"], item["title"], item["publish_date"], item["content"], (item["content"] or "")[:500], item["column"], "crawl_yizheng_xz_tzgg.py")) if c.rowcount > 0: stored += 1 else: skipped += 1 except Exception as e: log(f"DB error: {e}") skipped += 1 conn.commit() conn.close() log(f"Result: {stored} new, {skipped} skipped") if __name__ == "__main__": args = sys.argv[1:] pages = 5 json_out = None towns = None if "--pages=" in " ".join(args): pages = int(re.search(r"--pages=(\d+)", " ".join(args)).group(1)) if "--col=" in " ".join(args): col = re.search(r"--col=([\w,]+)", " ".join(args)).group(1) towns = col.split(",") if col != "all" else list(TOWNS.keys()) if "--json" in args: json_out = args[args.index("--json") + 1] crawl(towns=towns, pages=pages, json_out=json_out)