#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
邗江区人民政府 - 槐泗镇 - 通知公告
http://www.hj.gov.cn/zfxxgk/zjdzfxxgk/hsz/fdzdgknr/tzgg/index.html
CMS: 汉网 hanweb + jpaas API 动态渲染
列表: GET /api-gateway/jpaas-publish-server/front/page/build/unit
queryData: parseType=bulidstatic&webId=ohSFVDZIBzRtWYBHxfUyu&tplSetId=zMuCIPr1ttFFjmmbNGvm6
&pageType=column&tagId=列表一list&editType=null&pageId=BXwc2t58cn9zucOm2KYjO
分页: paramJson={"pageNo":N,"pageSize":15} (JSON URL 编码)
data.html: 标题YYYY-MM-DD
总数 621 条 / 42 页 (count=621 rows=15)
详情: meta ArticleTitle + meta PubDate (YYYY-MM-DD HH:MM:SS) + div.bt-content zoom clearfix 正文
"""
import sys
import os
import re
import time
import json
import html as html_lib
import urllib.parse
import sqlite3
try:
from bs4 import BeautifulSoup
except ImportError:
BeautifulSoup = None
# ---------------- config ----------------
BASE_URL = "http://www.hj.gov.cn"
API_URL = BASE_URL + "/api-gateway/jpaas-publish-server/front/page/build/unit"
LIST_PARAMS = {
"parseType": "bulidstatic",
"webId": "ohSFVDZIBzRtWYBHxfUyu",
"tplSetId": "zMuCIPr1ttFFjmmbNGvm6",
"pageType": "column",
"tagId": "列表一list",
"editType": "null",
"pageId": "BXwc2t58cn9zucOm2KYjO",
}
SITE_NAME = "邗江槐泗镇-通知公告"
GROUP_NAME = "江苏"
SCRIPT_NAME = "crawl_hj_hsz_tzgg.py"
DB_PATH = "/mnt/data/search.db"
PAGE_SIZE = 15
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",
}
# crawler_lib 所在目录 (classify_industry 行业分类复用)
for _p in ("/root/gov_crawler", "/root/gov_crawler"):
if os.path.isdir(_p) and _p not in sys.path:
sys.path.insert(0, _p)
# 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 · prefix + strip zero-width + collapse whitespace."""
if not t:
return ""
t = html_lib.unescape(t)
t = re.sub(r"^\s*[·•]+\s*", "", t)
t = t.replace("\xa0", " ").replace("\u200b", "").replace("\u200c", "").replace("\u200d", "").replace("\ufeff", "")
t = re.sub(r"\s+", " ", t)
return t.strip()
def fetch(url, session=None, timeout=30, retries=3):
import requests
last = None
for i in range(retries):
try:
r = (session or requests).get(url, headers=HEADERS, timeout=timeout)
if r.status_code == 200:
r.encoding = r.apparent_encoding or "utf-8"
return r
last = r
except Exception as e:
last = e
time.sleep(1.5 * (i + 1))
if isinstance(last, Exception):
raise last
return last
def html_to_text(content, page_url):
"""Convert article 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 fetch_list_page(session, page_no):
"""Fetch one list page via jpaas API. Returns (html_fragment, total_count)."""
params = dict(LIST_PARAMS)
params["paramJson"] = json.dumps({"pageNo": page_no, "pageSize": PAGE_SIZE}, ensure_ascii=False)
try:
r = fetch(API_URL + "?" + urllib.parse.urlencode(params), session)
except Exception:
return None, 0
if not r:
return None, 0
try:
data = r.json()
if not data.get("success"):
return None, 0
html_frag = data["data"]["html"]
total = 0
m = re.search(r'count="(\d+)"', html_frag)
if m:
total = int(m.group(1))
return html_frag, total
except Exception:
return None, 0
def parse_list_fragment(html_frag, page_url):
"""Extract (url, title, date) from API list fragment."""
items = []
for m in re.finditer(
r']*href="([^"]+)"[^>]*>(.*?)\s*]*>([^<]*)',
html_frag, re.S,
):
href = m.group(1).strip()
title = clean_title(m.group(2))
date = m.group(3).strip()
if not href or not title:
continue
abs_url = urllib.parse.urljoin(BASE_URL, html_lib.unescape(href))
items.append((abs_url, title, date))
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")
# title: meta ArticleTitle
mt = soup.find("meta", attrs={"name": "ArticleTitle"})
if mt and mt.get("content"):
title = clean_title(mt["content"])
# date: meta PubDate (YYYY-MM-DD HH:MM:SS)
md = soup.find("meta", attrs={"name": "PubDate"})
if md and md.get("content"):
m = re.match(r"(\d{4}-\d{2}-\d{2})", md["content"].strip())
if m:
date = m.group(1)
# content: div.bt-content (inner html only, no outer div wrapper)
bc = soup.find("div", class_=re.compile(r"bt-content"))
if bc:
content_html = bc.decode_contents()
if not title:
m = re.search(r'([^<]*)", html_text)
if m:
title = clean_title(m.group(1))
if not date:
m = re.search(r'(.*?)\s*\s*(?:)', html_text, re.S)
if m:
content_html = m.group(1)
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:
industry = "other"
try:
from crawler_lib import classify_industry
industry = classify_industry(title)
except Exception:
pass
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, industry),
)
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():
import requests
s = requests.Session()
s.headers.update(HEADERS)
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()
# 首页获取总数
frag, total = fetch_list_page(s, 1)
if not frag:
print("新增: 0")
conn.close()
return
total_pages = (total + PAGE_SIZE - 1) // PAGE_SIZE if total else _PAGES
pages_to_fetch = min(_PAGES, total_pages) if total else _PAGES
for page in range(1, pages_to_fetch + 1):
frag, _ = fetch_list_page(s, page)
if not frag:
continue
items = parse_list_fragment(frag, API_URL)
if not items:
continue
for url, title, date in items:
if url in seen_urls:
continue
seen_urls.add(url)
try:
dr = fetch(url, s)
except Exception as e:
print(f" [ERR] {url}: {e}")
skip_count += 1
continue
if not dr:
skip_count += 1
continue
d_title, d_date, d_html = parse_detail(dr.text, url)
if not d_title:
d_title = title
if not d_date:
d_date = date
content, has_table, attachments = html_to_text(d_html, url)
if not content:
skip_count += 1
continue
if store_record(conn, url, d_title, content, d_date, has_table):
new_count += 1
else:
skip_count += 1
conn.commit()
conn.close()
print(f"新增: {new_count}")
if __name__ == "__main__":
main()