#!/usr/bin/env python3 """ 忻州市生态环境局 - 通知公告公示 https://zwgk.sxxz.gov.cn/szfgzbm/shjbhj/gzdt_9907/tzgg/ 自定义 CMS (政府信息公开平台) """ import json, re, sys, time, requests from bs4 import BeautifulSoup from urllib.parse import urljoin DB_PATH = "/root/search.db" SITE_NAME = "忻州市生态环境局-通知公告" CATEGORY = "山西" BASE_URL = "https://zwgk.sxxz.gov.cn/szfgzbm/shjbhj/gzdt_9907/tzgg/" MAX_PAGES = 5 PER_PAGE = 20 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", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", } INSERT_SQL = """INSERT OR IGNORE INTO gov_raw (site_name, page_url, title, publish_date, summary, content, category, attachments) VALUES (?, ?, ?, ?, '', ?, ?, ?)""" def init_db(): import sqlite3 conn = sqlite3.connect(DB_PATH, timeout=60) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA busy_timeout=5000") return conn def fetch(url, timeout=30): r = requests.get(url, headers=HEADERS, timeout=timeout) r.encoding = "utf-8" return r def table_to_markdown(table, *args, **kwargs): """保留 HTML 表格结构（不转 md）""" return str(table) def extract_text_paragraphs(element): """Extract text preserving

boundaries as \n\n separators. Handles nested

(CMS wraps content in extra

).""" parts = [] inner_ps = element.find_all("p", recursive=True) if inner_ps: for p in inner_ps: txt = p.get_text(" ", strip=True) if txt: parts.append(txt) else: txt = element.get_text(" ", strip=True) if txt: parts.append(txt) return "\n\n".join(parts) def extract_detail(url): try: r = fetch(url) except Exception as e: print(f" [ERROR] {url}: {e}", flush=True) return None if r.status_code != 200: return None soup = BeautifulSoup(r.text, "html.parser") title = "" mt = soup.find("meta", attrs={"name": "ArticleTitle"}) if mt and mt.get("content"): title = mt["content"].strip() pub_date = "" md = soup.find("meta", attrs={"name": "PubDate"}) if md and md.get("content"): pub_date = md["content"].strip() pub_date = pub_date.replace(" 00:00:00", "").strip() content_text = "" body = soup.select_one("div.article-body") if body: parts = [] for child in body.children: if child.name is None: continue tn = child.name.lower() if tn in ("p", "div"): txt = extract_text_paragraphs(child) if txt: parts.append(txt) elif tn == "table": md = table_to_markdown(child) if md: parts.append(md) elif tn in ("script", "style", "br"): continue content_text = "\n\n".join(parts) content_text = re.sub(r'\n{3,}', '\n\n', content_text) content_text = re.sub(r'扫一扫在手机打开当前页面', '', content_text).strip() attachments_list = [] for a_tag in soup.find_all("a", href=True): h = a_tag["href"] if re.search(r'\.(pdf|doc|docx|xls|xlsx|rar|zip)$', h, re.IGNORECASE): at = a_tag.get_text(strip=True) or h.split("/")[-1] fu = urljoin(url, h) attachments_list.append({"name": at, "url": fu}) if len(content_text.strip()) < 20: content_text = f'

{title}

\n（本文为PDF附件）' return { "title": title, "content": content_text, "pub_date": pub_date, "attachments": json.dumps(attachments_list, ensure_ascii=False) if attachments_list else "", } def crawl(test_mode=False, max_pages=MAX_PAGES): print(f"[{SITE_NAME}] Starting, max_pages={max_pages}", flush=True) conn = init_db() cur = conn.cursor() total_inserted = total_skipped = pages_crawled = 0 for pn in range(1, max_pages + 1): items = extract_list_page(pn) if not items: print(f" Page {pn} empty, stop", flush=True) break pages_crawled += 1 for item in items: title, url, date = item["title"], item["url"], item["date"] cur.execute("SELECT id FROM gov_raw WHERE page_url = ?", (url,)) if cur.fetchone(): total_skipped += 1 continue detail = extract_detail(url) if detail is None: total_skipped += 1 continue if detail["title"] and len(detail["title"]) > len(title): title = detail["title"] title = re.sub(r'^标题[：:]\s*', '', title) content = detail["content"] attachments = detail["attachments"] cur.execute(INSERT_SQL, ( SITE_NAME, url, title, date or detail["pub_date"], content, CATEGORY, attachments, )) total_inserted += 1 if test_mode and total_inserted >= 10: break conn.commit() if test_mode and total_inserted >= 10: print(f" Test: stop after {total_inserted}", flush=True) break time.sleep(1) conn.close() print(f"[{SITE_NAME}] Done. Inserted={total_inserted}, Skipped={total_skipped}, Pages={pages_crawled}", flush=True) return total_inserted def extract_list_page(page_num): if page_num == 1: url = BASE_URL else: url = f"{BASE_URL}index_{page_num - 1}.shtml" try: r = fetch(url) except Exception as e: print(f" [ERROR] Page {page_num}: {e}", flush=True) return [] if r.status_code != 200: print(f" [ERROR] HTTP {r.status_code} page {page_num}", flush=True) return [] soup = BeautifulSoup(r.text, "html.parser") items = [] for li in soup.find_all("li", class_="govinfo-list-title-item"): a = li.find("a") if not a: continue href = a.get("href", "").strip() title = a.get("title", "").strip() if not title: title = a.get_text(strip=True) if not title or not href: continue full_url = urljoin(BASE_URL, href) date_span = li.find(class_="right") date_str = date_span.get_text(strip=True) if date_span else "" items.append({"title": title, "url": full_url, "date": date_str}) print(f" Page {page_num}: {len(items)} items", flush=True) return items if __name__ == "__main__": test_mode = "--test" in sys.argv mp = 3 if test_mode else MAX_PAGES if "--max-pages" in sys.argv: idx = sys.argv.index("--max-pages") if idx + 1 < len(sys.argv): mp = int(sys.argv[idx + 1]) crawl(test_mode=test_mode, max_pages=mp)