#!/usr/bin/env python3 """大龙经济开发区-通知公告爬虫 多彩博虹科技CMS (TRS变体) 列表: /zwgk_0/tzgg/ (第1页), /zwgk_0/tzgg/index_{N}.html (N=2~67) 每页15条，共67页 详情: /zwgk_0/tzgg/YYYYMM/tYYYYMMDD_XXXXX.html 内容: div.trs_editor_view > p 日期: 列表页 span.pull-right """ import requests import re import sys import json import time 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.dalong.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): """解析列表页""" soup = BeautifulSoup(html, 'html.parser') items = [] # 每个条目在 div.media 中 for media in soup.select('div.panel-body div.media'): # 取完整标题 - div.media-left a[title] 或 div.media-body a.infoTitle[title] title = '' href = '' left_link = media.select_one('div.media-left a[title]') if left_link: title = left_link.get('title', '').strip() href = left_link.get('href', '') # 回退: 从 infoTitle 取 if not title: info_link = media.select_one('a.infoTitle') if info_link: title = info_link.get('title', '') or info_link.get_text(strip=True) href = info_link.get('href', '') if not title or not href: continue detail_url = href if href.startswith('http') else urljoin(BASE_URL, href) # 日期 date_span = media.select_one('span.pull-right') pub_date = date_span.get_text(strip=True) if date_span else '' items.append({ 'title': title, 'url': detail_url, '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') # 标题 - 详情页的 h1 h1 = soup.find('h1') title = h1.get_text(strip=True) if h1 else 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 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) + ' ' elif child.name in ('div', 'ul', 'ol', 'li'): continue 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): a_href = a.get('href', '') a_text = a.get_text(strip=True) if a_href and re.search(r'\.(pdf|doc|docx|xls|xlsx|zip|rar)$', a_href, re.I): if not a_href.startswith('http'): a_href = urljoin(url, a_href) attachments.append({'name': a_text or '附件', 'url': a_href}) content += f'\n\n

{a_text or "附件"}

' return title, pub_date, content, attachments def main(): parser = argparse.ArgumentParser(description='大龙经济开发区-通知公告') parser.add_argument('--pages', type=int, default=0, help='页数，0=全量67页') args = parser.parse_args() max_pages = args.pages if args.pages > 0 else 67 all_items = [] for page in range(1, max_pages + 1): if page == 1: url = f'{BASE_URL}/zwgk_0/tzgg/' else: url = f'{BASE_URL}/zwgk_0/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()