#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 新和县人民政府 - 环境保护 爬虫 CMS: 阿克苏地区行政公署网 (Boshan CMS) 列表: dataList JSON 嵌入在HTML中（10页×15条=150条） 详情: h1.main_5121 标题, div.content.main_5123#zoom 正文 """ import re import sys import json import time import subprocess import requests from bs4 import BeautifulSoup from urllib.parse import urljoin # ── 配置 ────────────────────────────────────────── BASE_URL = "https://www.xjxinhe.gov.cn" LIST_URL = BASE_URL + "/zdxx/hms/hjbh/index.html" SITE_NAME = "新和县人民政府" GROUP = "新疆阿克苏" DB_PATH = "/mnt/data/search.db" 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 = requests.Session() session.headers.update(HEADERS) def clean_text(text): if not text: return "" text = re.sub(r'(·| |\s)+', ' ', text) return text.strip() def clean_content_html(html_content, detail_url): """提取正文，表格保留HTML""" if not html_content: return "" soup = BeautifulSoup(html_content, 'html.parser') for tag in soup(['script', 'style']): tag.decompose() parts = [] for child in soup.children: if not child.name: continue if child.name == 'p': t = child.get_text(strip=True) if t: parts.append(t) elif child.name == 'table': parts.append(str(child)) elif child.name in ('div', 'section', 'article'): for p in child.find_all('p', recursive=True): if p.find_parent('table'): continue t = p.get_text(strip=True) if t: parts.append(t) for tbl in child.find_all('table', recursive=True): parts.append(str(tbl)) elif child.name == 'img': src = child.get('src', '') alt = child.get('alt', '') if src: full_src = urljoin(detail_url, src) if not src.startswith('http') else src parts.append(f'![{alt}]({full_src})') return '\n\n'.join(parts) def fetch_list(): """从HTML中提取所有嵌入的dataList数据""" try: resp = session.get(LIST_URL, timeout=30) resp.encoding = "utf-8" if resp.status_code != 200: print(f"列表页 HTTP {resp.status_code}") return [] html = resp.text match = re.search(r'var\s+dataList\s*=\s*(\[.*?\]);', html, re.DOTALL) if not match: print("列表页 未找到dataList") return [] data = json.loads(match.group(1)) result = [] for page_data in data: for item in page_data.get("infolist", []): title = clean_text(item.get("title", "")) url = item.get("url", "") date_str = item.get("daytime", "") if not url or not title: continue result.append({ "title": title, "url": url, "date": date_str, "source": item.get("source", ""), }) print(f"列表页: {len(result)} 条 (嵌入{len(data)}页)") return result except Exception as e: print(f"列表页 异常: {e}") return [] def parse_detail(html, url): """解析详情页""" soup = BeautifulSoup(html, 'html.parser') title = '' publish_date = '' source = '' content = '' # 1. 标题 - h1.main_5121 h1 = soup.find('h1', class_='main_5121') if h1: title = clean_text(h1.get_text()) # 2. 发布日期 date_li = soup.find('li', class_='main_51221') if date_li: m = re.search(r'(\d{4}-\d{2}-\d{2})', date_li.get_text()) if m: publish_date = m.group(1) # 3. 来源 src_li = soup.find('li', class_='main_51222') if src_li: m = re.search(r'来源[：:]\s*(.+)', src_li.get_text()) if m: source = m.group(1).strip() else: source = clean_text(src_li.get_text()) # 4. 正文 - div.content.main_5123#zoom > div.main_51231.content.tabcs zoom_div = soup.find('div', id='zoom') if zoom_div: tabcs = zoom_div.find('div', class_='tabcs') if tabcs: # 附件链接嵌入 for a_tag in tabcs.find_all('a', href=True): href = a_tag['href'] if re.search(r'\.(pdf|doc|docx|xls|xlsx|zip|rar)$', href, re.I): a_text = a_tag.get_text(strip=True) or href full_url = urljoin(BASE_URL, href) if not href.startswith('http') else href a_tag.replace_with(soup.new_string(f' [{a_text}]({full_url}) ')) content = clean_content_html(str(tabcs), url) if not content: content = clean_content_html(str(zoom_div), url) return title, publish_date, source, content def insert_article(item, detail_title, pub_date, source, content): page_url = item['url'].replace("'", "''") title = (detail_title or item['title']).replace("'", "''") publish_date = pub_date or item.get('date', '') source_name = source or item.get('source', SITE_NAME) source_name = source_name.replace("'", "''") content_safe = content.replace("'", "''") summary = title[:200] site_name = SITE_NAME sql_raw = f"""INSERT OR IGNORE INTO gov_raw (page_url, title, content, publish_date, source_url, site_name, summary, group_name, industry) VALUES ('{page_url}', '{title}', '{content_safe}', '{publish_date}', '{page_url}', '{site_name}', '{summary}', '{GROUP}', '政府公告'); """ combined_sql = sql_raw + """SELECT CASE WHEN changes() > 0 THEN last_insert_rowid() ELSE 0 END; """ try: result = subprocess.run( ['sqlite3', "-cmd", ".timeout 60000", DB_PATH], input=combined_sql, capture_output=True, text=True, timeout=30 ) if result.returncode == 0: out = result.stdout.strip() try: rowid = int(out.strip()) except ValueError: rowid = 0 if rowid > 0: fts_sql = f"""INSERT OR IGNORE INTO gov_search (rowid, title, site_name, summary) SELECT rowid, title, site_name, summary FROM gov_raw WHERE rowid = {rowid}; """ subprocess.run(['sqlite3', "-cmd", ".timeout 60000", DB_PATH], input=fts_sql, capture_output=True, text=True, timeout=30) return True else: if 'UNIQUE constraint' not in result.stderr: print(f" [DB] {result.stderr[:200]}") except Exception as e: print(f" [DB] 错误: {e}") return False def main(max_pages=5): print(f"=== 新和县 - 环境保护 爬虫 ===\n") articles = fetch_list() if not articles: print("无文章列表") return # 只取前N页 cutoff = max_pages * 15 articles = articles[:cutoff] print(f"首批 {max_pages} 页: {len(articles)} 条") known_urls = set() try: result = subprocess.run( ['sqlite3', "-cmd", ".timeout 60000", DB_PATH, "SELECT page_url FROM gov_raw WHERE page_url LIKE '%xjxinhe.gov.cn/zdxx/hms/hjbh%'"], capture_output=True, text=True, timeout=30 ) if result.returncode == 0 and result.stdout.strip(): known_urls = set(result.stdout.strip().split('\n')) except Exception: pass print(f"已知URL: {len(known_urls)}") added = 0 skipped = 0 for i, a in enumerate(articles): url = a['url'] if url in known_urls: skipped += 1 continue print(f" [{i+1}/{len(articles)}] {a['title'][:50]}...", end=' ') try: r = session.get(url, timeout=30) r.encoding = 'utf-8' except Exception as e: print(f"ERR {e}") time.sleep(1) continue d_title, d_date, d_source, content = parse_detail(r.text, url) if not content.strip(): print("SKIP 无正文") skipped += 1 continue if insert_article(a, d_title, d_date, d_source, content): added += 1 print(f"OK ({len(content.strip())}字)") else: skipped += 1 print("SKIP (重复或入库失败)") time.sleep(0.3) print(f"\n结果: 新增 {added} 条, 跳过 {skipped} 条") print("=== 完成 ===") if __name__ == '__main__': pages = 5 if len(sys.argv) > 1: for arg in sys.argv[1:]: if arg.startswith('--pages='): try: pages = int(arg.split('=')[1]) except ValueError: pass main(max_pages=pages)