#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
庐江县人民政府 - 公示公告 (www.lj.gov.cn/zwdt/gsgg/)
CMS: TRS (div.listnews > ul.doc_list > li > a[title][class=left] + span.right.date 列表,
Ls.pagination JS -> /content/column/{id}?pageIndex=N 分页 API)
⚠️ 全站 JSL 反爬 (HTTP 521, __jsl_clearance_s cookie) → 必须用 chromium --dump-dom 渲染
列表: ul.doc_list > li > a[href=/zwdt/gsgg/{数字ID}.html][title] + span.right.date (18条/页, pageCount=28)
详情: meta ArticleTitle/PubDate/ContentSource 齐全
正文 div#zoom.j-fontContent.newscontnet (Word转换 逐段 + table 保留)
附件: 大多无文件链接(征地示意图为正文文字), 有 时通用提取
尾部: sharebox 打印/关闭 + reference-tip 引用提示 (取 #zoom 容器即可截断)
"""
import sys
import os
import re
import time
import html as html_lib
import urllib.parse
import subprocess
import sqlite3
try:
from bs4 import BeautifulSoup
except ImportError:
BeautifulSoup = None
# ---------------- config ----------------
BASE_URL = "https://www.lj.gov.cn"
LIST_PATH = "/zwdt/gsgg/"
LIST_URL = BASE_URL + LIST_PATH
SITE_NAME = "庐江县人民政府-公示公告"
GROUP_NAME = "安徽"
SCRIPT_NAME = "crawl_lj_gsgg.py"
DB_PATH = "/mnt/data/search.db"
CHROME = "/usr/bin/chromium-browser"
CHROME_DATA = f"/tmp/chrome_lj_{os.getpid()}" # fresh dir per run: stale dirs get WAF-blocked
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9",
}
# default: daily incremental = 1 page
_MAX_PAGES = 1
for i, a in enumerate(sys.argv):
if a.startswith("--pages="):
try:
_MAX_PAGES = int(a.split("=", 1)[1])
except ValueError:
pass
elif a == "--pages" and i + 1 < len(sys.argv):
try:
_MAX_PAGES = int(sys.argv[i + 1])
except ValueError:
pass
_PAGES = _MAX_PAGES if _MAX_PAGES >= 1 else 1
# ---------------- helpers ----------------
def clean_title(t):
"""unescape entities + strip zero-width + collapse whitespace."""
if not t:
return ""
t = html_lib.unescape(t)
t = t.replace("\u200b", "").replace("\u200c", "").replace("\u200d", "").replace("\ufeff", "")
t = re.sub(r"\s+", " ", t)
return t.strip()
def fetch_chrome(url, timeout=45):
"""Fetch page via chromium --dump-dom (bypasses JSL 521 challenge / WAF 403)."""
cmd = [
CHROME, "--headless", "--no-sandbox", "--disable-gpu",
"--virtual-time-budget=15000",
f"--user-data-dir={CHROME_DATA}",
"--dump-dom", url,
]
last = ""
for attempt in range(3):
try:
r = subprocess.run(cmd, capture_output=True, timeout=timeout)
html = r.stdout.decode("utf-8", errors="replace")
# success = real page (has doctype + not WAF block)
if html and "403 Forbidden" not in html[:500] and "document.cookie=" not in html[:2000]:
return html
last = html
except subprocess.TimeoutExpired:
last = ""
# WAF rate-limit cooldown: back off between chromium launches
# (lj.gov.cn实测: 连续请求60-90s内触发403, 冷却后恢复)
time.sleep(15 * (attempt + 1))
return last
def fetch(url, session=None, timeout=30, retries=2):
"""lj.gov.cn: 全站 WAF (JSL 521 + nginx 403). requests 直连 TLS 指纹被识别即触发封禁.
统一走 chromium --dump-dom 渲染 (cookie 持久化在 user-data-dir, 每次运行全新目录)."""
html_text = fetch_chrome(url)
if html_text:
return _FakeResp(html_text)
raise RuntimeError(f"chromium fetch failed: {url}")
class _FakeResp:
"""Minimal requests.Response stand-in for chromium output."""
def __init__(self, text):
self.text = text
self.content = text.encode("utf-8", errors="replace")
self.status_code = 200
def html_to_text(content, page_url):
"""Convert zoom inner HTML to stored format:
- keep
", _tbl_repl, content2, flags=re.S | re.I)
has_table = 1 if re.search(r"]*>", content, re.I) else 0
content2 = re.sub(r"", "\n\n", content2, flags=re.I)
content2 = re.sub(r"", "\n", content2, flags=re.I)
content2 = re.sub(r"]*>", "", content2, flags=re.I)
parts = []
for block in content2.split("\n\n"):
b = block.strip()
if not b:
continue
plain = re.sub(r"<[^>]+>", "", b)
plain = html_lib.unescape(plain)
plain = plain.replace("\xa0", " ").replace(" ", " ")
plain = re.sub(r"[\s\u200b\u200c\u200d\ufeff]+", "", plain)
if not plain:
continue
for k, v in link_protect.items():
b = b.replace(k, v)
for i, tbl in enumerate(table_protect):
b = b.replace(f"__TBL__{i}__", tbl)
parts.append(b.strip())
out = "\n\n".join(parts)
for k, v in link_protect.items():
out = out.replace(k, v)
for i, tbl in enumerate(table_protect):
out = out.replace(f"__TBL__{i}__", tbl)
out = re.sub(r"__LINK__\d+__", "", out)
out = re.sub(r"__TBL__\d+__", "", out)
out = html_lib.unescape(out)
out = re.sub(r"[ \t]+\n", "\n", out)
out = re.sub(r"\n{3,}", "\n\n", out)
out = re.sub(r"[ \t]{2,}", " ", out)
# dedupe paragraphs
seen = set()
final = []
for p in out.split("\n\n"):
key = re.sub(r"\s+", "", re.sub(r"<[^>]+>", "", p))
if not key:
continue
if key in seen:
continue
seen.add(key)
final.append(p)
out = "\n\n".join(final).strip()
return out, has_table, attachments
def parse_list(html_text, page_url):
"""Extract (url, title, date) from list page."""
items = []
if BeautifulSoup:
soup = BeautifulSoup(html_text, "html.parser")
ul = soup.find("ul", class_=re.compile(r"doc_list"))
if ul:
for li in ul.find_all("li"):
a = li.find("a", href=True)
if not a:
continue
href = a.get("href", "").strip()
if not re.search(r"/\d+\.html$", href) or "javascript" in href:
continue
abs_url = urllib.parse.urljoin(BASE_URL, href)
title = clean_title(a.get("title", "") or a.get_text(" ", strip=True))
span = li.find("span", class_=re.compile(r"date"))
date = span.get_text(strip=True) if span else ""
if not re.match(r"^20\d{2}-\d{2}-\d{2}$", date):
date = ""
items.append((abs_url, title, date))
if not items:
# regex fallback: li > a[title] + span.right.date
for m in re.finditer(
r']*>\s*]+href="([^"]*?/\d+\.html)"[^>]*title="([^"]*)"[^>]*>.*?\s*]*class="[^"]*date[^"]*"[^>]*>([^<]*)',
html_text, re.S,
):
abs_url = urllib.parse.urljoin(BASE_URL, m.group(1))
items.append((abs_url, clean_title(m.group(2)), m.group(3).strip()))
return items
def parse_detail(html_text, page_url):
"""Extract title, date, content from detail page."""
title = ""
date = ""
content_html = ""
if BeautifulSoup:
soup = BeautifulSoup(html_text, "html.parser")
mt = soup.find("meta", attrs={"name": re.compile(r"ArticleTitle", re.I)})
if mt and mt.get("content"):
title = clean_title(mt["content"])
h1 = soup.find("h1", class_=re.compile(r"newstitle"))
if not title and h1:
title = clean_title(h1.get_text(" ", strip=True))
md = soup.find("meta", attrs={"name": re.compile(r"PubDate", re.I)})
if md and md.get("content"):
d = md["content"].strip()
dm = re.match(r"(\d{4})-(\d{2})-(\d{2})", d)
if dm:
date = dm.group(0)
# content: div#zoom.j-fontContent.newscontnet
zoom = soup.find("div", id="zoom")
if zoom:
content_html = str(zoom)
else:
nc = soup.find("div", class_=re.compile(r"newscontnet|j-fontContent"))
if nc:
content_html = str(nc)
if not title or not content_html:
# regex fallback
m = re.search(r']*id="zoom"[^>]*>(.*?)]*class="sharebox"', html_text, re.S)
if m:
content_html = m.group(1)
# hard cutoff: drop everything after sharebox marker
cut = content_html.find("sharebox")
if cut > 0:
content_html = content_html[:cut]
return title, date, content_html
def store_record(conn, url, title, content, date, has_table):
cur = conn.execute(
"SELECT id FROM gov_raw WHERE page_url=?",
(url,),
)
if cur.fetchone():
return False
try:
cur = conn.execute(
"INSERT OR IGNORE INTO gov_raw "
"(page_url, source_url, title, content, publish_date, site_name, group_name, script_name, has_table, industry) "
"VALUES (?,?,?,?,?,?,?,?,?,?)",
(url, url, title, content, date, SITE_NAME, GROUP_NAME, SCRIPT_NAME, has_table, "other"),
)
if cur.rowcount > 0:
summary = re.sub(r"<[^>]+>", "", content)
summary = html_lib.unescape(summary)
summary = re.sub(r"\s+", " ", summary).strip()[:200]
conn.execute(
"INSERT OR IGNORE INTO gov_search (rowid, title, site_name, summary) VALUES (?,?,?,?)",
(cur.lastrowid, title, SITE_NAME, summary),
)
return True
return False
except sqlite3.OperationalError as e:
if "locked" in str(e):
time.sleep(3)
return store_record(conn, url, title, content, date, has_table)
raise
def main():
conn = sqlite3.connect(DB_PATH, timeout=60)
conn.execute("PRAGMA busy_timeout=60000")
conn.execute("PRAGMA journal_mode=WAL")
new_count = 0
skip_count = 0
seen_urls = set()
for page in range(1, _PAGES + 1):
if page == 1:
url = LIST_URL
else:
url = f"{BASE_URL}/content/column/11245557?pageIndex={page}"
try:
r = fetch(url)
html_text = r.text if hasattr(r, "text") else r.content.decode("utf-8", errors="replace")
except Exception as e:
print(f"Page {page} fetch error: {e}", flush=True)
break
items = parse_list(html_text, url)
print(f"Page {page}: found {len(items)} items", flush=True)
if not items and page > 1:
print("Empty page, stop pagination", flush=True)
break
for abs_url, title, date in items:
if abs_url in seen_urls:
continue
seen_urls.add(abs_url)
try:
r = fetch(abs_url)
dhtml = r.text if hasattr(r, "text") else r.content.decode("utf-8", errors="replace")
except Exception as e:
print(f" detail error {abs_url}: {e}", flush=True)
skip_count += 1
continue
dtitle, ddate, content_html = parse_detail(dhtml, abs_url)
if not dtitle:
dtitle = title
if not ddate:
ddate = date
content, has_table, _ = html_to_text(content_html, abs_url)
plain_len = len(re.sub(r"<[^>]+>", "", content).strip())
if plain_len < 10:
print(f" skip empty: {dtitle[:30]}", flush=True)
skip_count += 1
continue
ok = store_record(conn, abs_url, dtitle, content, ddate, has_table)
if ok:
new_count += 1
print(f" + {dtitle[:40]} ({ddate})", flush=True)
else:
skip_count += 1
# WAF rate-limit protection: slow crawl for chromium-rendered sites
time.sleep(8)
time.sleep(3)
conn.commit()
conn.close()
print(f"新增: {new_count}", flush=True)
print(f"跳过: {skip_count}", flush=True)
print(f"=== {SITE_NAME} done ===", flush=True)
if __name__ == "__main__":
main()