#!/usr/bin/env python3 """湄潭县人民政府（meitan.gov.cn）公示公告爬虫 - TRS CMS""" import requests import re import sqlite3 import sys import time from bs4 import BeautifulSoup, Tag from urllib.parse import urljoin import urllib.parse BASE_URL = "http://www.meitan.gov.cn" LIST_URL = BASE_URL + "/mtzx/gsgg/index.html" LIST_PAGES = 5 DB_PATH = "/root/search.db" SITE_NAME = "湄潭县公示公告" SLEEP = 1.5 HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", } def get_soup(url): try: r = requests.get(url, headers=HEADERS, timeout=30) r.encoding = "utf-8" return BeautifulSoup(r.text, "html.parser") except Exception as e: print(f"[WARN] 请求失败: {url} - {e}") return None def parse_list_items(soup): items = [] news_div = soup.find("div", class_="NewsList") if not news_div: return items for li in news_div.select("ul > li"): a = li.find("a", href=True) span = li.find("span") if not a: continue href = a.get("href", "").strip() title = a.get("title", "") or a.get_text(strip=True) date_str = span.get_text(strip=True) if span else "" if href and title: full_url = urljoin(BASE_URL, href) items.append((full_url, title, date_str)) return items def parse_detail(soup, url): # Title from h1 inside body_content title = "" body_div = soup.find("div", class_="body_content") if body_div: h1 = body_div.find("h1") if h1: title = h1.get_text(strip=True) # Fallback: meta ArticleTitle if not title: meta_title = soup.find("meta", attrs={"name": "ArticleTitle"}) if meta_title: title = meta_title.get("content", "").strip() # Date from meta pub_date = "" meta_date = soup.find("meta", attrs={"name": "PubDate"}) if meta_date: content = meta_date.get("content", "") m = re.search(r"\d{4}-\d{2}-\d{2}", content) if m: pub_date = m.group() # Content from div.content > div.trs_editor_view content_parts = [] content_div = soup.find("div", class_="content") if content_div: editor_view = content_div.find("div", class_=re.compile(r"trs_editor_view|TRS_UEDITOR")) if editor_view: container = editor_view else: container = content_div else: # Try body_content directly container = body_div or soup # Find all p, table, img elements if container: for child in container.find_all(["p", "table", "img"], recursive=True): if child.name == "p": if child.find_parent("table"): continue text = child.get_text(strip=True) if text: content_parts.append(text) elif child.name == "table": md = html_table_to_html(child) if md: content_parts.append(md) elif child.name == "img": src = child.get("src", "").strip() alt = child.get("alt", "").strip() if src and not src.endswith(".gif") and "fileTypeImages" not in src: full_src = urljoin(BASE_URL, src) content_parts.append(f"![{alt}]({full_src})" if alt else f"![]({full_src})") content = "\n\n".join(content_parts) # Attachments (PDF/doc/docx links) attachments = [] container = content_div or body_div or soup for a_tag in container.find_all("a", href=re.compile(r"\.(pdf|doc|docx|xls|xlsx|zip|rar|txt)$", re.I)): href = a_tag.get("href", "").strip() title_att = a_tag.get_text(strip=True) or href.split("/")[-1] if href: full_url = urljoin(BASE_URL, href) attachments.append({"url": full_url, "title": title_att}) return title, pub_date, content, str(attachments) def html_table_to_html(table, base_url=""): """保留 HTML 表格结构，仅将相对链接/图片转绝对 URL""" from bs4 import BeautifulSoup tbl = BeautifulSoup(str(table), 'html.parser') for a in tbl.find_all('a'): href = a.get('href', '') if href and not href.startswith(('http', 'javascript', '#')): a['href'] = urllib.parse.urljoin(base_url, href) if base_url else href for img in tbl.find_all('img'): src = img.get('src', '') if src and not src.startswith(('http', '//', 'data:')): img['src'] = urllib.parse.urljoin(base_url, src) if base_url else src return str(tbl) def get_page_url(page): if page == 1: return LIST_URL else: return f"{BASE_URL}/mtzx/gsgg/index_{page}.html" def main(): pages = LIST_PAGES if len(sys.argv) > 2 and sys.argv[1] == "--pages": pages = int(sys.argv[2]) total_new = 0 for page in range(1, pages + 1): url = get_page_url(page) print(f"[列表页] 第{page}页: {url}") soup = get_soup(url) if not soup: continue items = parse_list_items(soup) if not items: print(f" 无数据，可能已到末页") break print(f" 找到 {len(items)} 条") for item_url, list_title, list_date in items: print(f" → {list_title}") if not item_url.startswith("http"): item_url = urljoin(BASE_URL, item_url) conn = sqlite3.connect(DB_PATH, timeout=60) c = conn.cursor() c.execute("SELECT id FROM gov_raw WHERE page_url=? AND site_name=?", (item_url, SITE_NAME)) if c.fetchone(): conn.close() print(f" 已存在，跳过") continue conn.close() time.sleep(SLEEP) detail_soup = get_soup(item_url) if not detail_soup: continue detail_title, pub_date, content, attachments = parse_detail(detail_soup, item_url) date = pub_date or list_date use_title = detail_title or list_title if not content: print(f" 正文为空，跳过") continue conn = sqlite3.connect(DB_PATH, timeout=60) c = conn.cursor() try: summary = content[:200] if content else "" has_table = 1 if "| ---" in content else 0 c.execute(""" INSERT OR REPLACE INTO gov_raw (page_url, title, publish_date, site_name, content, attachments, summary, has_table) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, (item_url, use_title, date, SITE_NAME, content, attachments, summary, has_table)) conn.commit() paras = len(content.split("\n\n")) print(f" 入库成功 ({paras}段, 表格={has_table})") total_new += 1 except Exception as e: print(f" 入库失败: {e}") finally: conn.close() print(f"\n✅ {SITE_NAME} 爬取完成！共入库 {total_new} 条") if __name__ == "__main__": main()