#!/usr/bin/env python3 """独山县人民政府-通知公告爬虫 - v2 TRS CMS (拓尔思) 列表: /xwdt/tzgg/index.html (181页, ~2713条) 第N页(>=2): /xwdt/tzgg/index_{N}.html 详情: /xwdt/tzgg/YYYYMM/tYYYYMMDD_XXXXX.html 内容: div.trs_editor_view > p 日期: 页内"时间：" 或 meta PubDate 注意：列表页HTML含注释，BeautifulSoup解析li有误，改用正则 """ import requests import re import sys import json import time import os import argparse from bs4 import BeautifulSoup from urllib.parse import urljoin sys.path.insert(0, '/root/gov_crawler') from crawler_lib import push_to_searchdb import sqlite3 DB_PATH = '/root/search.db' def get_db_connection(): try: conn = sqlite3.connect(DB_PATH, timeout=30) conn.execute("PRAGMA busy_timeout=30000") conn.execute("PRAGMA journal_mode=WAL") return conn except Exception as e: print(f' WARN DB connect: {e}') return None BASE_URL = 'https://www.dushan.gov.cn' HEADERS = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.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', } SITE_NAME = '独山县-通知公告' sess = requests.Session() sess.headers.update(HEADERS) def fetch_page(url): resp = sess.get(url, timeout=15) resp.encoding = 'utf-8' return resp.text def parse_list(html): """解析列表页 - 用正则（BS解析li有问题）""" items = [] # 用正则找所有通知公告详情链接 # pattern = re.compile( r'href="(http[s]?://www\.dushan\.gov\.cn/xwdt/tzgg/\d+[^"]*)"[^>]*title="([^"]*)"[^>]*>.*?\s*([^<]*)', re.DOTALL ) for m in pattern.finditer(html): href = m.group(1) title = m.group(2).strip() pub_date = m.group(3).strip() if title and '/tzgg/' in href: items.append({ 'title': title, 'url': href, 'pub_date': pub_date, }) return items def fetch_detail(item): """抓取详情页""" url = item['url'] try: html = fetch_page(url) except Exception as e: print(f' WARN skip: {url} - {e}') return item['title'], item.get('pub_date', ''), '', [] soup = BeautifulSoup(html, 'html.parser') # 标题 title = item['title'] # 日期 pub_date = item.get('pub_date', '') page_text = soup.get_text() m = re.search(r'时间[：:]\s*(\d{4}-\d{2}-\d{2})', page_text) if m: pub_date = m.group(1) # 正文 - div.trs_editor_view content_view = soup.find('div', class_='trs_editor_view') content = '' attachments = [] if content_view: parts = [] for elem in content_view.find_all(['p', 'table', 'img'], recursive=True): if elem.name == 'p': if elem.find('p'): continue # 跳过在 table 内部的 p（避免表格内容重复，因为 table 会整体输出） if elem.find_parent('table'): continue txt = '' for child in elem.children: if isinstance(child, str): t = child.strip() if t: txt += t + ' ' elif child.name == 'br': txt += '\n' elif child.name == 'img': src = child.get('src', '') alt = child.get('alt', '') if src: full_src = urljoin(url, src) txt += f'![{alt}]({full_src}) ' elif child.name == 'a': a_href = child.get('href', '') a_text = child.get_text(strip=True) if a_href and a_text: full_href = urljoin(url, a_href) txt += f'[{a_text}]({full_href}) ' elif child.name in ('span', 'strong', 'b', 'em'): txt += child.get_text(' ', strip=True) + ' ' txt = re.sub(r'[ \t]+', ' ', txt).strip() if txt: parts.append(txt) elif elem.name == 'table': parts.append(str(elem)) elif elem.name == 'img': src = elem.get('src', '') alt = elem.get('alt', '') if src: full_src = urljoin(url, src) parts.append(f'![{alt}]({full_src})') content = '\n\n'.join(parts) # 附件嵌入 for a in content_view.find_all('a', href=True): href = a.get('href', '') text = a.get_text(strip=True) if href and text and re.search(r'\.(pdf|doc|docx|xls|xlsx|zip|rar)$', href, re.I): if not href.startswith('http'): href = urljoin(url, href) attachments.append({'name': text, 'url': href}) content += f'\n\n

{text}

' return title, pub_date, content, attachments def main(): parser = argparse.ArgumentParser(description='独山县-通知公告') parser.add_argument('--pages', type=int, default=0, help='页数，0=全量181页') args = parser.parse_args() max_pages = args.pages if args.pages > 0 else 5 all_items = [] for page in range(1, max_pages + 1): if page == 1: url = f'{BASE_URL}/xwdt/tzgg/index.html' else: url = f'{BASE_URL}/xwdt/tzgg/index_{page}.html' print(f'Page {page}/{max_pages}...') try: html = fetch_page(url) items = parse_list(html) if not items: print(f' empty, stop') break print(f' found {len(items)} items') all_items.extend(items) except Exception as e: print(f' WARN: {e}') break time.sleep(0.3) print(f'\nTotal list: {len(all_items)} items') if not all_items: print('No items found, abort') return # 详情 records = [] for i, item in enumerate(all_items): print(f' [{i+1}/{len(all_items)}] {item["title"][:40]}...') title, pub_date, content, attachments = fetch_detail(item) if not title: title = item['title'] title = re.sub(r'\s+', ' ', title).strip() summary = content[:500] if content else '' record = { 'url': item['url'], 'source_url': item['url'], 'title': title, 'content': content, 'summary': summary, 'pub_date': pub_date, 'site_name': SITE_NAME, 'attachments': json.dumps(attachments, ensure_ascii=False) if attachments else '', } records.append(record) time.sleep(0.2) # 入库 print(f'\nInserting {len(records)} records...') if records: push_to_searchdb(records, SITE_NAME) # FTS验证 conn = get_db_connection() if conn: try: cursor = conn.execute( "SELECT COUNT(*) FROM gov_raw WHERE site_name=?", (SITE_NAME,) ) raw_count = cursor.fetchone()[0] cursor = conn.execute( "SELECT COUNT(*) FROM gov_search WHERE site_name=?", (SITE_NAME,) ) fts_count = cursor.fetchone()[0] print(f'DB: {raw_count} rows (FTS: {fts_count} rows)') except Exception as e: print(f' WARN DB query: {e}') finally: conn.close() print('DONE') if __name__ == '__main__': main()