#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ crawl_ningguo_qtxx.py - 宁国市生态环境分局-其他信息 CMS: 自定义ASP.NET MVC 列表: /XxgkContent/showList/383/22474/page_N.html (15条/页, ~10页) ul > li > a[href] + date文本 详情: /OpennessContent/show/{id}.html 标题: 标签 (取" - "之前) 日期: <meta PubDate> 正文: div#zoom > p """ import requests import re import sys import os import time from bs4 import BeautifulSoup DB_PATH = "/root/search.db" BASE_URL = "https://www.ningguo.gov.cn" SITE_NAME = "宁国市-其他信息" GROUP = "安徽" INDUSTRY = "环境公示" 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", } SESSION = None def get(url): global SESSION if SESSION is None: SESSION = requests.Session() SESSION.headers.update(HEADERS) # Bypass JS cookie challenge try: r = SESSION.get(url, timeout=60, verify=False) if 'token_verified=true' in r.text: SESSION.cookies.set('token_verified', 'true') except: pass try: r = SESSION.get(url, timeout=60, verify=False) r.encoding = "utf-8" return r.text except Exception as e: print(f"ERR:{e}") return None def fetch_list_page(page_no): url = f"{BASE_URL}/XxgkContent/showList/383/22474/page_{page_no}.html" return get(url) def parse_list(html): """Extract (url, title, date) from list page""" items = [] soup = BeautifulSoup(html, "html.parser") # Find the 15-item ul - look for links to /OpennessContent/show/ for ul in soup.find_all("ul"): lis = ul.find_all("li", recursive=False) if not (10 <= len(lis) <= 20): continue # Check if links are content links (not nav links) openness_count = sum( 1 for li in lis if li.find("a", href=True) and "/OpennessContent/show/" in li.find("a", href=True).get("href", "") ) if openness_count >= 5: for li in lis: a = li.find("a", href=True) if not a: continue href = a.get("href", "").strip() # Title from a tag - strip trailing date if present title = a.get_text(strip=True) # Remove trailing date like "2026-07-02" from title title = re.sub(r'\d{4}-\d{2}-\d{2}$', '', title).strip() if not title or not href: continue # Date from li text - find YYYY-MM-DD txt = li.get_text(strip=True) date_match = re.search(r"(\d{4}-\d{2}-\d{2})", txt) date_str = date_match.group(1) if date_match else "" if href.startswith("/"): href = BASE_URL + href items.append((href, title, date_str)) break return items def extract_detail(html, url): soup = BeautifulSoup(html, "html.parser") # Title from <title> tag title = "" title_tag = soup.find("title") if title_tag: raw = title_tag.get_text(strip=True) parts = raw.split(" - ") title = parts[0].strip() if parts else raw # Date from <meta PubDate> date_str = "" pubdate_meta = soup.find("meta", {"name": "PubDate"}) or soup.find("meta", {"name": "pubdate"}) if pubdate_meta: content = pubdate_meta.get("content", "") m = re.search(r"(\d{4}-\d{2}-\d{2})", content) if m: date_str = m.group(1) # Fallback: search body for "发布时间：" if not date_str: body = soup.find("body") if body: m = re.search(r"发布时间：(\d{4}-\d{2}-\d{2})", body.get_text()) if m: date_str = m.group(1) # Content from div#zoom content = "" zoom = soup.find("div", id="zoom") if not zoom: # Fallback: try div.g-detailbox zoom = soup.find("div", class_="g-detailbox") if zoom: paragraphs = [] for p in zoom.find_all("p"): txt = p.get_text(strip=True) if txt: paragraphs.append(txt) if paragraphs: content = "\n\n".join(paragraphs) else: content = zoom.get_text(separator="\n").strip() if content: content = re.sub(r" ", "", content) summary = content[:200] if content else "" return title, date_str, content, summary def save_to_db(items_data): import sqlite3 conn = sqlite3.connect(DB_PATH, timeout=10) c = conn.cursor() inserted = 0 fts_batch = [] for title, date_str, content, summary, page_url in items_data: try: c.execute("SELECT id FROM gov_raw WHERE page_url = ?", (page_url,)) if c.fetchone(): continue c.execute( """INSERT INTO gov_raw (page_url, title, site_name, publish_date, content, summary, source_url, category, industry, group_name, script_name) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", (page_url, title, SITE_NAME, date_str, content, summary, page_url, "环境公示", INDUSTRY, GROUP, "crawl_ningguo_qtxx.py"), ) new_id = c.lastrowid if new_id: fts_batch.append((new_id, title, SITE_NAME, summary or "")) inserted += 1 except Exception as e: print(f" [DB ERROR] {title[:30]}: {e}") conn.commit() if fts_batch: for rid, title, site, summary in fts_batch: try: c.execute( "INSERT OR IGNORE INTO gov_search(rowid, title, site_name, summary) VALUES (?, ?, ?, ?)", (rid, title, site, summary), ) except Exception as e: print(f" [FTS ERROR] id={rid}: {e}") conn.commit() print(f" FTS同步: {len(fts_batch)}条") conn.close() return inserted def main(): import argparse parser = argparse.ArgumentParser(description=f"{SITE_NAME}爬虫") parser.add_argument("--pages", type=int, default=1, help="爬取页数(共约10页)") parser.add_argument("--full", action="store_true", help="全量爬取(10页)") args = parser.parse_args() max_pages = 10 if args.full else args.pages total_new = 0 for page_no in range(1, max_pages + 1): print(f"[{time.strftime('%H:%M:%S')}] 第{page_no}页...", end=" ", flush=True) html = fetch_list_page(page_no) if not html: print("请求失败，停止") break items = parse_list(html) if not items: print("无数据，停止") break print(f"{len(items)}条", flush=True) batch = [] for href, title, date_str in items: detail_html = get(href) if not detail_html: continue dt, dd, content, summary = extract_detail(detail_html, href) batch.append((dt or title, dd or date_str, content, summary, href)) time.sleep(0.5) saved = save_to_db(batch) total_new += saved print(f" 入库{saved}/{len(batch)}条 (累计{total_new})", flush=True) if not args.full and saved == 0 and len(batch) > 0: print(" 全部已存在，增量停止") break print(f"\n===== 完成 =====") print(f"新增入库: {total_new} 条") print(f"站点: {SITE_NAME}") if __name__ == "__main__": main()