#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 枣庄国家高新技术产业开发区 -- 生态环境分局 -- 建设项目环境影响评价 列表: POST http://www.zzctp.gov.cn/irs/front/search (TRS 政务公开 IRS 检索 API) body: code=18e334699d2, dataTypeId=2857, filters=[{aggrName:f_2024327849703, aggrValues:建设项目环境影响评价}, {aggrName:f_2024327355010, aggrValues:生态环境分局}] pageNo/pageSize=20, orderBy=time 详情: meta ArticleTitle/PubDate + div.view.TRS_UEDITOR 正文 + div.attachment 附件 样本: http://www.zzctp.gov.cn/zwgk/xxgkml/sgdw/sthjfj/202608/t20260807_2314926.html 用法: python3 crawl_zzctp_hpgs.py [--pages=N | N] """ import requests import json import re import sys import os import html as html_mod from bs4 import BeautifulSoup SITE_NAME = "枣庄高新区-建设项目环境影响评价" BASE = "http://www.zzctp.gov.cn" SEARCH_URL = "http://www.zzctp.gov.cn/irs/front/search" 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", "Referer": "http://www.zzctp.gov.cn/list_6548.html?siteid=179&classinfoid=2627", "Content-Type": "application/json", "X-Requested-With": "XMLHttpRequest", } PAGE_SIZE = 20 CUTOFF = "2023-01-01" QUALITY_DB = os.path.join(os.path.dirname(os.path.abspath(__file__)), "quality_results.db") SEARCH_DB = "/root/search.db" # ---------- crawler_lib ---------- def push_to_searchdb(records, batch_label): """写入本地 quality_results.db（domain 隔离），供 sync_to_server 同步到 search.db""" import sqlite3 conn = sqlite3.connect(QUALITY_DB, timeout=60) conn.execute("PRAGMA busy_timeout=300000") cur = conn.cursor() cur.execute("""CREATE TABLE IF NOT EXISTS crawl_results ( id INTEGER PRIMARY KEY AUTOINCREMENT, site_id INTEGER DEFAULT 0, title TEXT, url TEXT, content TEXT, publish_date TEXT, summary TEXT, domain TEXT, category TEXT DEFAULT '', content_hash TEXT, crawl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, industry TEXT DEFAULT '', group_name TEXT DEFAULT '', script_name TEXT DEFAULT '', attachments TEXT DEFAULT '[]', UNIQUE(domain, url) )""") n_new = 0 for r in records: try: cur.execute( "INSERT OR IGNORE INTO crawl_results (domain, title, url, content, publish_date, summary, industry, group_name, script_name, attachments) VALUES (?,?,?,?,?,?,?,?,?,?)", (r.get("domain", ""), r.get("title", ""), r.get("url", ""), r.get("content", ""), r.get("pub_date", ""), r.get("summary", ""), r.get("industry", ""), r.get("group_name", ""), r.get("script_name", ""), r.get("attachments", "[]")), ) n_new += cur.rowcount except Exception as e: print(f" [push] ERR {r.get('url','')}: {e}") conn.commit() conn.close() print(f"[push_to_searchdb] new={n_new} skip={len(records)-n_new}") return n_new def sync_to_server(): """把本地 quality_results.db 该 domain 的记录同步到 search.db（UPDATE-or-INSERT 保 id + FTS 增量）""" import sqlite3 try: conn = sqlite3.connect(QUALITY_DB, timeout=60) conn.execute("PRAGMA busy_timeout=300000") rows = conn.execute( "SELECT title, url, content, publish_date, summary, industry, group_name, script_name, attachments FROM crawl_results WHERE domain=? ORDER BY id", (DOMAIN,), ).fetchall() conn.close() except Exception as e: print(f"[sync] local read ERR: {e}") return if not rows: print("[sync] 本地没有数据") return dst = sqlite3.connect(SEARCH_DB, timeout=300) dst.execute("PRAGMA journal_mode=WAL") dst.execute("PRAGMA busy_timeout=300000") upd = ins = 0 try: for r in rows: title, url, content, pub_date, summary, ind, grp, sn, att = r cur = dst.execute("SELECT id FROM gov_raw WHERE page_url=?", (url,)) row = cur.fetchone() try: if row: dst.execute( "UPDATE gov_raw SET title=?, content=?, publish_date=?, summary=?, industry=?, group_name=?, script_name=?, attachments=? WHERE id=?", (title, (content or "")[:500000], pub_date or "", (summary or "")[:300], ind or "", grp or "", sn or "", att or "[]", row[0])) upd += 1 else: dst.execute( "INSERT INTO gov_raw (title, page_url, content, publish_date, summary, industry, group_name, script_name, attachments, site_name, source_url) VALUES (?,?,?,?,?,?,?,?,?,?,?)", (title, url, (content or "")[:500000], pub_date or "", (summary or "")[:300], ind or "", grp or "", sn or "", att or "[]", SITE_NAME, url)) ins += 1 except Exception as e: print(f" [sync] ERR {url}: {e}") dst.commit() # FTS 增量（触发器 trg_gov_raw_fts_ins/upd/del 会自动同步，无需手动；此处兜底补漏） dst.execute( "INSERT OR REPLACE INTO gov_search(rowid,title,site_name,summary) " "SELECT r.id,r.title,r.site_name,r.summary FROM gov_raw r " "WHERE r.id NOT IN (SELECT rowid FROM gov_search) AND r.site_name=?", (SITE_NAME,)) dst.commit() total = dst.execute("SELECT COUNT(*) FROM gov_raw WHERE site_name=?", (SITE_NAME,)).fetchone()[0] finally: dst.close() print(f"[sync] OK updated={upd} inserted={ins} DB共{total}条") # ---------- 提取 ---------- def clean_title(t): t = re.sub(r"·| | |\u200b|\ufeff", "", t) t = re.sub(r"^[•·]\s*", "", t) return re.sub(r"\s+", " ", t).strip() def fetch_list(page): body = { "appendixType": "", "beginDateTime": "", "code": "18e334699d2", "codes": "", "configCode": "", "dataTypeId": "2857", "endDateTime": "", "filters": [ {"aggrName": "f_2024327849703", "aggrValues": "建设项目环境影响评价", "filterId": "1"}, {"aggrName": "f_2024327355010", "aggrValues": "生态环境分局", "filterId": "1"}, ], "granularity": "ALL", "isSearchForced": 0, "historySearchWords": ["建设项目环境影响评价"], "orderBy": "time", "pageNo": page, "pageSize": PAGE_SIZE, "searchBy": "all", } try: r = requests.post(SEARCH_URL, headers=HEADERS, data=json.dumps(body, ensure_ascii=False).encode("utf-8"), timeout=30) r.raise_for_status() d = r.json() data = d.get("data") or {} middle = data.get("middle") or {} pager = data.get("pager") or middle.get("pager") or {} lst = middle.get("listAndBox") or [] items = [] for it in lst: if it.get("type") != "DATA": continue dd = it.get("data") or {} if not dd.get("title"): continue items.append({ "title": clean_title(re.sub(r"", "", str(dd.get("title", "")))), "url": dd.get("url") or "", "pub_date": str(dd.get("time") or "")[:10], }) total = pager.get("total") or 0 page_count = pager.get("pageCount") or 1 return items, total, page_count except Exception as e: print(f" [list] page {page} ERR: {e}") return [], 0, 1 def parse_detail(url): """抓详情页: 标题(meta ArticleTitle) + 正文(div.view TRS_UEDITOR 保留

/) + 附件(div.attachment)""" try: r = requests.get(url, headers=HEADERS, timeout=30) r.raise_for_status() r.encoding = "utf-8" soup = BeautifulSoup(r.text, "html.parser") except Exception as e: print(f" [detail] {url} ERR: {e}") return None # 标题 title = "" m = soup.find("meta", attrs={"name": "ArticleTitle"}) if m and m.get("content"): title = clean_title(m["content"]) if not title: h = soup.find("h1") if h: title = clean_title(h.get_text()) if not title: m = re.search(r"(.*?)", r.text, re.S) if m: t = m.group(1).strip() t = re.sub(r"^枣庄国家高新技术产业开发区--?", "", t) title = clean_title(t) # 日期 pub_date = "" m = soup.find("meta", attrs={"name": "PubDate"}) if m and m.get("content"): pub_date = m["content"].strip()[:10] if not pub_date: m = re.search(r"(\d{4}-\d{2}-\d{2})", r.text) if m: pub_date = m.group(1) # 正文容器 body_el = soup.select_one("div.view.TRS_UEDITOR") or soup.select_one("div.view") or soup.select_one("div.TRS_UEDITOR") if body_el is None: # fallback 任意含 TRS_UEDITOR class 的元素 for el in soup.find_all(class_=re.compile("TRS_UEDITOR")): body_el = el break if body_el is None: return {"title": title, "content": "", "pub_date": pub_date, "attachments": []} # 附件（div.attachment 区，绝对 URL） atts = [] att_div = body_el.find_parent("div") # 先找页面级 attachment div page_att = soup.select_one("div.attachment") if page_att: for a in page_att.find_all("a", href=True): href = a["href"].strip() if re.search(r"\.(pdf|doc|docx|xls|xlsx|zip|rar|wps|7z)$", href, re.I): name = clean_title(a.get_text(strip=True)) if not name: name = href.split("/")[-1] atts.append({"name": name, "url": href}) # 正文内附件链接 for a in body_el.find_all("a", href=True): href = a["href"].strip() if re.search(r"\.(pdf|doc|docx|xls|xlsx|zip|rar|wps|7z)$", href, re.I): abs_url = href if href.startswith("http") else "http://www.zzctp.gov.cn" + (href if href.startswith("/") else "/" + href) name = clean_title(a.get_text(strip=True)) if not name: name = href.split("/")[-1] if not any(x["url"] == abs_url for x in atts): atts.append({"name": name, "url": abs_url}) # 正文序列化：保留

/

body_html = str(body_el.decode_contents()) # 附件段落从正文移除（统一末尾追加） for a in body_el.find_all("a", href=True): href = a["href"].strip() if re.search(r"\.(pdf|doc|docx|xls|xlsx|zip|rar|wps|7z)$", href, re.I): p = a.find_parent("p") if p: p.decompose() # 重新序列化 body_html = str(body_el.decode_contents()) # 清洗: 剥 script/style body_html = re.sub(r"]*>.*?", "", body_html, flags=re.S | re.I) body_html = re.sub(r"]*>.*?", "", body_html, flags=re.S | re.I) # 附件链接绝对化 for a in body_el.find_all("a", href=True): href = a["href"].strip() if re.search(r"\.(pdf|doc|docx|xls|xlsx|zip|rar|wps|7z)$", href, re.I): abs_url = href if href.startswith("http") else "http://www.zzctp.gov.cn" + (href if href.startswith("/") else "/" + href) a["href"] = abs_url body_html = str(body_el.decode_contents()) # 规范化 p 标签 body_html = re.sub(r"]*>", "

", body_html) # 图片绝对化 for img in body_el.find_all("img"): src = img.get("src", "") if src and not src.startswith("http"): img["src"] = "http://www.zzctp.gov.cn" + (src if src.startswith("/") else "/" + src) body_html = str(body_el.decode_contents()) body_html = re.sub(r"]*>", "

", body_html) # 移除空段落 body_html = re.sub(r"

\s*

", "", body_html) # 正文尾部 docAppendix/video 隐藏区清理 body_html = re.sub(r'\s*
{html_mod.escape(a["name"])}

' for a in atts) body_html = body_html.rstrip() + "\n" + att_html # 纯文本长度（用于空正文判定，图片型放行） text_len = len(re.sub(r"<[^>]+>", "", body_html).strip()) has_img = " max_pages: break items, total, page_count = fetch_list(pg) print(f"[list] page {pg}: items={len(items)} total={total} pageCount={page_count}") if not items: break for it in items: if it["pub_date"] and it["pub_date"] < CUTOFF: print(f" [cutoff] {it['pub_date']} {it['title'][:40]}") continue det = parse_detail(it["url"]) if det is None: print(f" [fail] {it['title'][:50]}") # 详情 404/失败：标题+URL 占位入库（正文空，用户可点开看原始站） det = {"title": it["title"], "content": "", "pub_date": it["pub_date"], "attachments": []} rec = { "domain": DOMAIN, "site_name": SITE_NAME, "title": det["title"] or it["title"], "url": it["url"], "content": det["content"], "pub_date": det["pub_date"] or it["pub_date"], "summary": re.sub(r"<[^>]+>", "", det["content"])[:200], "industry": "power" if re.search(r"风电|风力发电|海上风电", det["title"] or it["title"]) else "", "group_name": "山东", "script_name": os.path.basename(sys.argv[0]), "attachments": json.dumps(det["attachments"], ensure_ascii=False) if det["attachments"] else "[]", } valid.append(rec) if pg >= page_count: break print(f"[total] valid={len(valid)}") if valid: push_to_searchdb(valid, os.path.basename(sys.argv[0])) # 交互模式可同步；调度器模式由服务器端 sync 处理 if "--sync" in sys.argv: sync_to_server() DOMAIN = "www.zzctp.gov.cn" if __name__ == "__main__": main()