#!/usr/bin/env python3 """石棉县人民政府 — 环境监测、保护与治理爬虫 http://www.shimian.gov.cn/htm/openlist.htm?sid=102 → /gongkai/kuozhan/102.html?page=N 分页, 20条/页 内容含正文HTML(p+table), 非仅PDF """ import os, sys, re, time import requests from bs4 import BeautifulSoup from datetime import datetime, timedelta from urllib.parse import urljoin DB_PATH = os.getenv("SEARCH_DB", "/root/search.db") BASE_LIST = "http://www.shimian.gov.cn/gongkai/kuozhan/102.html" SITE_NAME = "石棉县人民政府-环境监测保护与治理" THRESHOLD = (datetime.now() - timedelta(days=3*365)).strftime("%Y-%m-%d") MAX_PAGES = 35 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", "Referer": "http://www.shimian.gov.cn/", } session = requests.Session() session.headers.update(HEADERS) import sqlite3 def store_item(title, date_str, content, page_url): try: conn = sqlite3.connect(DB_PATH, timeout=10) c = conn.cursor() summary = (content.strip()[:200] if content.strip() else '') date_rank = 0 if date_str: try: date_rank = int(datetime.strptime(date_str[:10], "%Y-%m-%d").timestamp()) except: pass c.execute("INSERT OR IGNORE INTO gov_raw (title, publish_date, content, page_url, source_url, site_name, summary, date_rank) VALUES (?,?,?,?,?,?,?,?)", (title.strip(), date_str, content.strip(), page_url.strip(), page_url.strip(), SITE_NAME, summary, date_rank)) a = c.rowcount; conn.commit(); conn.close(); return a except Exception as e: print(f"[ERROR] 写入失败: {e}"); return 0 def parse_list_page(html): soup = BeautifulSoup(html, "html.parser") items = [] for cls_name in ["content-list", "data-list"]: pl = soup.find(class_=cls_name) if pl: break if not pl: return items for li in pl.find_all("li"): a = li.find("a") if not a: continue href, text = a.get("href", ""), a.get_text(strip=True) if not text or len(text) < 5: continue span = li.find("span") date_str = span.get_text(strip=True) if span else "" full_url = href if href.startswith("http") else urljoin("http://www.shimian.gov.cn", href) items.append((text, date_str, full_url)) return items def parse_detail_page(html, page_url): soup = BeautifulSoup(html, "html.parser") # Title title = "" h1 = soup.select_one("div.msg-content h1") if h1: title = h1.get_text(strip=True) if not title: mt = soup.find("meta", attrs={"name": "ArticleTitle"}) if mt and mt.get("content"): title = mt["content"] if not title: tt = soup.find("title") if tt: title = tt.get_text(strip=True).split(" - ")[0] # Date date_str = "" mp = soup.find("meta", attrs={"name": "PubDate"}) if mp and mp.get("content"): m = re.search(r"(\d{4}-\d{1,2}-\d{1,2})", mp["content"]) if m: date_str = m.group(1) if not date_str: intro = soup.find(class_="intro") if intro: for span in intro.find_all("span"): m = re.search(r"(\d{4}-\d{1,2}-\d{1,2})", span.get_text()) if m: date_str = m.group(1); break # Content from xqing-web-box content = "" xqb = soup.find(class_="xqing-web-box") if xqb: # Extract paragraph text with table content parts = [] for p in xqb.find_all("p"): t = p.get_text(strip=True) if t: parts.append(t) for table in xqb.find_all("table"): rows = [] for tr in table.find_all("tr"): cells = [td.get_text(strip=True) for td in tr.find_all(["td", "th"])] if cells: rows.append(" | ".join(cells)) if rows: parts.append("表格:\n" + "\n".join(rows)) if parts: content = "\n\n".join(parts) # If no xqing-web-box, try id_body if not content: ib = soup.find(class_="id_body") if ib: # Skip title line and metadata txt = ib.get_text(strip=True) # Remove title (first bold div) title_div = ib.find("div", style=re.compile(r"font-size.*23")) if title_div: txt = txt.replace(title_div.get_text(strip=True), "", 1) # Remove 浏览次数 line txt = re.sub(r'浏览次数.*?(?:分享|$)', '', txt, flags=re.DOTALL) txt = txt.strip() if txt: content = txt # Attachments atts = [] for a in soup.find_all("a", href=True): h = a["href"] if re.search(r'\.(pdf|doc|docx|xls|xlsx|zip|rar|caj)$', h, re.I): t = a.get_text(strip=True) or "附件" fu = h if h.startswith("http") else urljoin("http://www.shimian.gov.cn", h) atts.append(f"[{t}]({fu})") if atts: content += "\n\n---\n附件：\n" + "\n".join(atts) return title, date_str, content def main(): print(f"[INFO] 开始爬取: {SITE_NAME}") print(f"[INFO] 阈值: {THRESHOLD}, 最大页数: {MAX_PAGES}") total_new, stop = 0, False for page_num in range(1, MAX_PAGES + 1): if stop: break page_url = f"{BASE_LIST}?page={page_num}" if page_num > 1: time.sleep(1.5) try: r = session.get(page_url, timeout=15) r.encoding = "utf-8" except: time.sleep(1); continue if r.status_code != 200: print(f"[PAGE] {page_num}: HTTP {r.status_code}, 停止"); break items = parse_list_page(r.text) if not items: print(f"[PAGE] {page_num}: 无数据, 停止"); break print(f"[PAGE] 第{page_num}页: {len(items)}条 [{items[0][1]}] {items[0][0][:35]}") for title, date_str, detail_url in items: if date_str and date_str < THRESHOLD: print(f"[STOP] {date_str} 超阈值"); stop = True; break time.sleep(0.3) try: rd = session.get(detail_url, timeout=15) rd.encoding = "utf-8" except: continue if rd.status_code != 200: continue dt, dd, dc = parse_detail_page(rd.text, detail_url) ft = dt if dt else title fd = dd or date_str affected = store_item(ft, fd, dc, detail_url) if affected > 0: total_new += 1 print(f" [+] {ft[:35]}...({fd})") else: print(f" [-] 已存在") print(f"\n=== 完成, 新增: {total_new} ===") if __name__ == "__main__": main()