#!/usr/bin/env python3 """ 固原市 - 环保公示 爬虫 URL: https://www.nxgy.gov.cn/zwgk/zfxxgkml/jgzn_38938/hjbh/hbgs/ CMS: TRS (createPageHTML分页) 列表: ul.pageList.newsList > li > a + span 分页: index.html (p1) / index_{N-1}.html (page N), 25页 详情: ./YYYYMM/tYYYYMMDD_XXXXXX.html 正文: div.view.TRS_UEDITOR 日期: div.xl_con_qt_l span """ import re, sys, time import requests from bs4 import BeautifulSoup from urllib.parse import urljoin BASE_URL = "https://www.nxgy.gov.cn/zwgk/zfxxgkml/jgzn_38938/hjbh/hbgs/" SITE_NAME = "固原市-环保公示" GROUP = "宁夏" HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} TIMEOUT = 15 DELAY = 1.0 DB_PATH = "/root/search.db" def parse_date(text): m = re.search(r'(\d{4})-(\d{1,2})-(\d{1,2})', text) if m: return f"{m.group(1)}-{int(m.group(2)):02d}-{int(m.group(3)):02d}" return "" def get_soup(url): for retry in range(3): try: resp = requests.get(url, headers=HEADERS, timeout=TIMEOUT, verify=False) resp.encoding = 'utf-8' if resp.status_code == 200: return BeautifulSoup(resp.text, 'html.parser') except: time.sleep(2) return None def get_list_soup(url): """Get list page soup and extract total pages""" soup = get_soup(url) total_pages = 0 if soup: for s in soup.find_all('script'): if s.string and 'createPageHTML' in s.string: m = re.search(r'createPageHTML\((\d+)', s.string) if m: total_pages = int(m.group(1)) return soup, total_pages def fetch_page(page_num): if page_num == 1: url = BASE_URL else: url = f"{BASE_URL}index_{page_num-1}.html" soup, _ = get_list_soup(url) if not soup: return [] items = [] ul = soup.find('ul', class_='pageList') if not ul: ul = soup.find('ul', class_='newsList') if not ul: return items for li in ul.find_all('li', recursive=False): a = li.find('a') if not a or not a.get('href'): continue href = urljoin(BASE_URL, a['href']) title = a.text.strip() if not title: continue date_str = "" date_span = li.find('span') if date_span: date_str = parse_date(date_span.text) items.append({ 'title': title, 'url': href, 'date': date_str, }) return items def get_total_pages(): soup, total = get_list_soup(BASE_URL) return total def extract_content(soup): title = "" t = soup.find('title') if t: title = t.text.strip() title = re.sub(r'_\u56fa\u539f\u5e02\u4eba\u6c11\u653f\u5e9c$', '', title).strip() pub_date = "" date_div = soup.find('div', class_='xl_con_qt_l') if date_div: for span in date_div.find_all('span'): txt = span.text.strip() if '\u53d1\u8868\u65e5\u671f' in txt or '\u53d1\u5e03' in txt: pub_date = parse_date(txt) if pub_date: break parts = [] attachments = [] cd = soup.find('div', class_=lambda c: c and 'TRS_UEDITOR' in c) if cd: for child in cd.children: if not child.name: text = str(child).strip() if text and not all(c in ' \n\r\t\u3000\xa0' for c in text): parts.append(text) continue tag = child.name.lower() if tag in ('p', 'div'): hb = any(c.name in ('table', 'div', 'ul', 'ol') for c in child.find_all(recursive=False)) if hb: for sub in child.children: if not sub.name: t = str(sub).strip() if t: parts.append(t) continue if sub.name == 'table': parts.append(str(sub)) elif sub.name in ('p', 'div'): st = sub.get_text(strip=True) if st: parts.append(st) else: t = sub.get_text(strip=True) if t: parts.append(t) else: txt = child.get_text(strip=True) if txt: parts.append(txt) elif tag == 'table': parts.append(str(child)) elif tag == 'img': src = child.get('src', '') alt = child.get('alt', '') if src: parts.append(f"![{alt}]({urljoin(BASE_URL, src)})") elif tag in ('ul', 'ol'): txt = child.get_text(strip=True) if txt: parts.append(txt) elif tag == 'br': pass else: txt = child.get_text(strip=True) if txt: parts.append(txt) for a in cd.find_all('a', href=True): h = a['href'] if re.search(r'\.(pdf|doc|docx|xls|xlsx|zip|rar|ceb|ofd|ppt|pptx)$', h.lower()): attachments.append(f"[{a.text.strip()}]({urljoin(BASE_URL, h)})") seen = set() up = [] for p in parts: k = p[:100] if k not in seen: seen.add(k) up.append(p) return title, '\n\n'.join(up), pub_date, attachments def init_db(): import sqlite3 conn = sqlite3.connect(DB_PATH, timeout=60) c = conn.cursor() c.execute('''CREATE TABLE IF NOT EXISTS gov_raw ( id INTEGER PRIMARY KEY AUTOINCREMENT, page_url TEXT, title TEXT, content TEXT, publish_date TEXT, site_name TEXT, summary TEXT, attachments TEXT, date_rank TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )''') c.execute('''CREATE INDEX IF NOT EXISTS idx_gov_raw_url ON gov_raw(page_url, site_name)''') conn.commit() conn.close() def save_to_db(items): import sqlite3 conn = sqlite3.connect(DB_PATH, timeout=60) c = conn.cursor() n = 0 for item in items: summary = item['content'][:200] if item['content'] else '' try: c.execute('''INSERT OR REPLACE INTO gov_raw (page_url, title, content, publish_date, site_name, summary, attachments, date_rank, script_name) VALUES (?, ?, ?, ?, ?, ?, ?, ?, \'crawl_nxgy.py\')''', ( item['url'], item['title'], item['content'], item['pub_date'], SITE_NAME, summary, '\n'.join(item['attachments']) if item['attachments'] else '', item['pub_date'] or '0000-00-00', )) n += 1 except: pass conn.commit() conn.close() return n def main(): import argparse parser = argparse.ArgumentParser() parser.add_argument('--pages', type=int, default=0) parser.add_argument('--skip-db', action='store_true') args = parser.parse_args() init_db() print(f"===== {SITE_NAME} =====") total_pages = get_total_pages() print(f"Total pages: {total_pages}") pages = total_pages if args.pages == 0 else min(args.pages, total_pages) print(f"Will fetch: {pages}") all_items = [] for page in range(1, pages + 1): page_items = fetch_page(page) if not page_items: print(f"\n--- Page {page}/{pages} --- no items") continue print(f"\n--- Page {page}/{pages} ({len(page_items)} items) ---") for idx, item in enumerate(page_items): print(f" [{len(all_items)+1}] {item['title'][:40]}...") ds = get_soup(item['url']) if not ds: all_items.append(dict(item, content='', pub_date=item['date'], attachments=[])) print(f" FAIL") continue title, content, pub_date, attachments = extract_content(ds) all_items.append({ 'title': title or item['title'], 'url': item['url'], 'content': content, 'pub_date': pub_date or item['date'], 'attachments': attachments, }) if content: print(f" seg={len(content.split(chr(10)+chr(10)))} | attach={'YES' if attachments else 'no'}") else: print(f" WARN empty") time.sleep(DELAY) t = len(all_items) wb = sum(1 for i in all_items if i['content']) ta = sum(len(i['attachments']) for i in all_items) avg = sum(len(i['content'].split('\n\n')) for i in all_items if i['content']) / max(wb, 1) print(f"\n===== DONE =====") print(f"Total: {t}") if t: print(f"With body: {wb} ({wb/t*100:.1f}%)") print(f"Avg seg: {avg:.1f}") print(f"Attachments: {ta}") if not args.skip_db: print(f"Imported: {save_to_db(all_items)}") if __name__ == '__main__': main()