#!/usr/bin/env python3 """仙游经济开发区-通知公告爬虫 http://www.xianyou.gov.cn/jjkfq/tzgg/ TRS CMS, 5条/ul, 多ul组 详情: h1标题, meta PubDate/ContentSource, div.TRS_Editor正文 """ import requests import sqlite3 import json import os import re import argparse from urllib.parse import urljoin from bs4 import BeautifulSoup BASE_URL = "http://www.xianyou.gov.cn" LIST_URL = "http://www.xianyou.gov.cn/jjkfq/tzgg/" DB_PATH = os.environ.get("DB_PATH", "/root/search.db") SITE_NAME = "仙游经济开发区-通知公告" HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/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", } def get_soup(url, timeout=30): resp = requests.get(url, headers=HEADERS, timeout=timeout) resp.encoding = "utf-8" return BeautifulSoup(resp.text, "html.parser") def parse_list_page(soup): """ul > li > a + span (multi-ul groups)""" items = [] seen_urls = set() for ul in soup.find_all("ul"): for li in ul.find_all("li"): a = li.find("a") if not a: continue href = a.get("href", "") if not href or href.startswith("#") or href.startswith("javascript"): continue full_url = urljoin(LIST_URL, href) if full_url in seen_urls: continue seen_urls.add(full_url) title = a.get_text(strip=True) span = li.find("span") pub_date = span.get_text(strip=True) if span else "" # Only include items with a date (skip nav/footer links) if not re.search(r"\d{4}[-/]\d{1,2}[-/]\d{1,2}", pub_date): continue items.append((full_url, title, pub_date)) return items def extract_detail(soup, url): title = "" pub_date = "" source = "" content = "" attachments = [] # Title - h1 h1 = soup.find("h1") if h1: title = h1.get_text(strip=True) # Date/Source - meta tags meta_pd = soup.find("meta", attrs={"name": "PubDate"}) if meta_pd and meta_pd.get("content"): pub_date = meta_pd["content"].strip()[:10] meta_src = soup.find("meta", attrs={"name": "ContentSource"}) if meta_src and meta_src.get("content"): source = meta_src["content"].strip() # Content - div.TRS_Editor article = soup.select_one("div.TRS_Editor") if not article: article = soup.select_one("div.xl_con1") if not article: article = soup.select_one("div.myzj_con") if article: # Attachments for a_tag in article.find_all("a"): a_href = a_tag.get("href", "") if a_href.endswith((".pdf", ".doc", ".docx", ".xls", ".xlsx", ".zip", ".rar")): if not a_href.startswith("http"): a_href = urljoin(BASE_URL, a_href) attachments.append({ "name": a_tag.get_text(strip=True) or os.path.basename(a_href), "url": a_href, }) # Build content preserving tables and paragraphs text_parts = [] for child in article.children: if not hasattr(child, "name") or not child.name: continue if child.name in ("script", "style"): continue if child.name == "table": text_parts.append(str(child)) text_parts.append("") elif child.name in ("p", "div"): inner_table = child.find("table") if inner_table: text_parts.append(str(inner_table)) text_parts.append("") continue txt = child.get_text(strip=True) if txt: text_parts.append(txt) elif child.name in ("br",): text_parts.append("") content = "\n\n".join(text_parts) # Image-only fallback if len(content.strip()) < 20: imgs = article.find_all("img") if imgs: img_lines = [] for img in imgs: src = img.get("src", "") if src and not src.startswith("http"): src = urljoin(BASE_URL, src) alt = img.get("alt", "") or img.get("title", "") or "" if src: img_lines.append("![%s](%s)" % (alt, src)) content = '

%s

\n\n%s' % (url, title, "\n".join(img_lines)) if title else "\n".join(img_lines) return title, pub_date, source, content, attachments def crawl(max_pages=None): conn = sqlite3.connect(DB_PATH, timeout=60) cur = conn.cursor() # Check pagination - try index_N.htm or index_N.html all_items = [] pages_to_try = [LIST_URL] if max_pages is None or max_pages > 1: limit = max_pages if max_pages else 99 for n in range(2, limit + 1): pages_to_try.append("http://www.xianyou.gov.cn/jjkfq/tzgg/index_%d.htm" % n) for page_url in pages_to_try: try: soup = get_soup(page_url) items = parse_list_page(soup) if not items: print("[列表] %s -> 0 条" % page_url) break print("[列表] %s -> %d 条" % (page_url, len(items))) all_items.extend(items) except Exception as e: print("[列表] %s -> 异常: %s" % (page_url, str(e))) break print("\n列表共 %d 条，开始抓详情..." % len(all_items)) new_count = 0 error_count = 0 skip_count = 0 for idx, (page_url, list_title, list_date) in enumerate(all_items, 1): cur.execute( "SELECT id FROM gov_raw WHERE page_url=? AND site_name=?", (page_url, SITE_NAME), ) if cur.fetchone(): skip_count += 1 print(" [%d/%d] 跳过: %s..." % (idx, len(all_items), list_title[:30])) continue print(" [%d/%d] %s..." % (idx, len(all_items), list_title[:40])) try: soup = get_soup(page_url) title, pub_date, source, content, attachments = extract_detail(soup, page_url) if not title: title = list_title if not pub_date: pub_date = list_date att_json = json.dumps(attachments, ensure_ascii=False) if attachments else "" cur.execute( """INSERT OR REPLACE INTO gov_raw (page_url, site_name, title, publish_date, source_url, content, attachments, script_name) VALUES (?, ?, ?, ?, ?, ?, ?, 'crawl_xianyou.py')""", (page_url, SITE_NAME, title, pub_date, page_url, content, att_json), ) conn.commit() new_count += 1 print(" -> 新增 (%d chars)" % len(content)) except Exception as e: error_count += 1 print(" -> 异常: %s" % str(e)) conn.close() return new_count, skip_count, len(all_items), error_count if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--max-pages", type=int, default=None) args = parser.parse_args() new, skip, total, errors = crawl(args.max_pages) print("\n完成: 新增 %d, 跳过 %d, 列表 %d, 异常 %d" % (new, skip, total, errors))