#!/usr/bin/env python3 """ 宜兴市人民政府 - 公告公示爬虫 URL: https://www.yixing.gov.cn/zgyx/zzxx/gggs/index.shtml CMS: 宜兴站群系统 列表: /zgyx/zzxx/gggs/index.shtml (page1), index_{N}.shtml (page 2-7) 详情: /doc/YYYY/MM/DD/ID.shtml 共 ~94 条 (6页x15 + 4) 日跑: --max-pages 1 (增量取最新页) """ import sys, os, re, time, sqlite3, argparse from datetime import datetime from urllib.parse import urljoin import requests from bs4 import BeautifulSoup import warnings warnings.filterwarnings("ignore") BASE_URL = "https://www.yixing.gov.cn" SITE_NAME = "宜兴市人民政府" CATEGORY = "公告公示" 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", "Referer": f"{BASE_URL}/zgyx/zzxx/gggs/index.shtml", } MAX_TOTAL_PAGES = 7 # 共7页 def clean_content_html(html): """清理HTML""" if not html: return "" cleaned = re.sub(r'\sstyle="[^"]*"', '', html) cleaned = re.sub(r']*>|', '', cleaned) cleaned = re.sub(r']*>|', '', cleaned) cleaned = re.sub(r']*>|', '', cleaned) cleaned = re.sub(r']*>|', '', cleaned) cleaned = re.sub(r']*>|', '', cleaned) cleaned = re.sub(r'\n{3,}', '\n\n', cleaned) return cleaned.strip() def get_page_url(page): """获取列表页 URL""" if page == 1: return f"{BASE_URL}/zgyx/zzxx/gggs/index.shtml" return f"{BASE_URL}/zgyx/zzxx/gggs/index_{page}.shtml" def fetch_list(page): """解析列表页""" url = get_page_url(page) items = [] try: resp = requests.get(url, headers=HEADERS, timeout=30) resp.encoding = "utf-8" soup = BeautifulSoup(resp.text, "html.parser") ul = soup.select_one("ul.contlist") if not ul: return items for li in ul.find_all("li", recursive=False): a = li.find("a") span = li.find("span") if not a: continue href = a.get("href", "") if not href or href.startswith("javascript"): continue full_url = urljoin(BASE_URL, href) title = a.get_text(strip=True) # 列表标题可能带省略号，但详情页标题是完整的 date = span.get_text(strip=True) if span else "" items.append({"url": full_url, "title": title, "date": date}) except Exception as e: print(f" [错误] 列表页 {page} 请求失败: {e}", flush=True) return items def fetch_detail(detail_url): """获取详情页正文""" try: resp = requests.get(detail_url, headers=HEADERS, timeout=30) resp.encoding = "utf-8" soup = BeautifulSoup(resp.text, "html.parser") # 正文: div.article_cont content_div = soup.select_one("div.article_cont") if not content_div: return "" # 处理图片和链接的绝对路径 for img in content_div.find_all("img"): src = img.get("src", "") if src and not src.startswith("http"): img["src"] = urljoin(BASE_URL, src) for a in content_div.find_all("a"): href = a.get("href", "") if href and not href.startswith("http") and not href.startswith("javascript"): a["href"] = urljoin(BASE_URL, href) return clean_content_html(str(content_div)) except Exception as e: print(f" [错误] 详情页获取失败 {detail_url}: {e}", flush=True) return "" def extract_detail_info(detail_url): """从详情页提取标题和日期（完整标题）""" try: resp = requests.get(detail_url, headers=HEADERS, timeout=30) resp.encoding = "utf-8" soup = BeautifulSoup(resp.text, "html.parser") # 标题 h1 = soup.select_one("div.m_article h1") title = h1.get_text(strip=True) if h1 else "" # 日期 date = "" mes = soup.select_one("div.article_mes") if mes: text = mes.get_text() m = re.search(r"日期[：:]\s*(\d{4}-\d{2}-\d{2})", text) if m: date = m.group(1) return title, date except: return "", "" def crawl(max_pages=None): """主爬取逻辑""" s = requests.Session() s.headers.update(HEADERS) db = sqlite3.connect(DB_PATH, timeout=60) db.execute("PRAGMA journal_mode=WAL") db.execute("PRAGMA synchronous=NORMAL") existing = set() cur = db.execute("SELECT page_url FROM gov_raw WHERE site_name=?", (SITE_NAME,)) for row in cur: existing.add(row[0]) print(f"已有 {len(existing)} 条宜兴记录", flush=True) pages_to_fetch = MAX_TOTAL_PAGES if max_pages: pages_to_fetch = min(max_pages, MAX_TOTAL_PAGES) total_new = 0 total_skip = 0 total_error = 0 for page in range(1, pages_to_fetch + 1): print(f"[第 {page}/{pages_to_fetch} 页] ", end="", flush=True) items = fetch_list(page) if not items: print("无数据", flush=True) continue page_new = 0 for item in items: full_url = item["url"] if full_url in existing: total_skip += 1 continue list_title = item["title"] list_date = item["date"] # 从详情页取标题补充完整（如果列表标题有省略号） detail_title, detail_date = extract_detail_info(full_url) title = list_title if "..." not in list_title else (detail_title or list_title) publish_date = detail_date or list_date if not title: total_skip += 1 continue # 正文 body = fetch_detail(full_url) if not body or len(body.strip()) < 20: body = f"

详情见原文：{title}

" summary = re.sub(r'<[^>]+>', ' ', body) summary = re.sub(r'\s+', ' ', summary).strip()[:300] # 检查是否含附件链接 has_attachments = bool(re.search(r'附件|下载|文件', title + summary)) try: cur = db.execute("""INSERT OR IGNORE INTO gov_raw (site_name, source_url, page_url, title, publish_date, summary, content, status, category, group_name, has_table) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( SITE_NAME, full_url, full_url, title[:500], publish_date, summary[:500], body, 'active', CATEGORY, GROUP, 1 if ' 0: row = db.execute("SELECT id FROM gov_raw WHERE page_url=?", (full_url,)).fetchone() if row: rid = row[0] try: # 2026-09-22: 先提交 gov_raw —— 库上触发器已维护 FTS，下面这条手动写入会因 # rowid 重复而 IntegrityError；不先 commit 会把 gov_raw 那条一并回滚（静默丢数据） db.commit() db.execute("INSERT OR REPLACE INTO gov_search(rowid, title, site_name, summary) VALUES (?, ?, ?, ?)", (rid, title[:500], SITE_NAME, summary[:200])) except sqlite3.IntegrityError: pass # 已存在 existing.add(full_url) db.commit() total_new += 1 page_new += 1 else: total_skip += 1 except Exception as e: print(f"x", end="", flush=True) total_error += 1 print(f"+{page_new} 条 (累计 {total_new})", flush=True) if page < pages_to_fetch: time.sleep(0.5) db.close() print(f"\n完成! 新增 {total_new}, 跳过 {total_skip}, 错误 {total_error}", flush=True) return total_new if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--max-pages", type=int, default=None) args = parser.parse_args() crawl(max_pages=args.max_pages)