#!/usr/bin/env python3 """山阴县人民政府 — 公告通知 http://www.shanyin.gov.cn/sydt/ggtz/ TRS CMS，分页 index_N.html """ 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_URL = "http://www.shanyin.gov.cn/sydt/ggtz/index.html" SITE_NAME = "山阴县人民政府-公告通知" THRESHOLD = (datetime.now() - timedelta(days=3*365)).strftime("%Y-%m-%d") 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", } session = requests.Session() session.headers.update(HEADERS) session.verify = False import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) 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 fetch(url, max_retries=3): for i in range(max_retries): try: r = session.get(url, timeout=15) r.encoding = "utf-8" if r.status_code == 200: return r.text except Exception as e: print(f"[WARN] {url} 请求失败: {e}") if i < max_retries - 1: time.sleep(2) return None def parse_list_page(html, base_url): soup = BeautifulSoup(html, "html.parser") items = [] list_box = soup.find("ul", class_="common-list-box-items_box") if not list_box: return items for li in list_box.find_all("li"): a = li.find("a", href=True) if not a: continue title = a.get_text(strip=True) href = a["href"] if not title or not href: continue # Date is embedded in li text as YYYY.MM.DD at the end txt = li.get_text(strip=True) m = re.search(r'(\d{4})[.]?(\d{1,2})[.]?(\d{1,2})$', txt) if not m: continue date_str = f"{m.group(1)}-{m.group(2).zfill(2)}-{m.group(3).zfill(2)}" # Title is the link text items.append((title, date_str, urljoin(base_url, href))) return items def parse_detail_page(url): html = fetch(url) if not html: return None, None, None soup = BeautifulSoup(html, "html.parser") # 标题 title = "" t_div = soup.find("div", class_="detail-article-title") if t_div: title = t_div.get_text(strip=True) if not title: t = soup.find("title") if t: title = t.get_text(strip=True) for suffix in ["-山阴县人民政府门户网站", "-山阴县人民政府", "_山阴县人民政府"]: if suffix in title: title = title.split(suffix)[0].strip() break # 日期 date_str = "" page_text = soup.get_text() m = re.search(r'时间[：:]\s*(\d{4}[-年]\d{1,2}[-月]\d{1,2})', page_text) if m: date_str = m.group(1).replace('年', '-').replace('月', '-').replace('日', '') # 正文 content_parts = [] for cls in ["article-body", "main-content-wrap"]: body = soup.find("div", class_=lambda c: c and cls in str(c)) if body: for p in body.find_all("p"): txt = p.get_text(strip=True) if txt and len(txt) > 4: if "扫一扫" in txt or "分享到" in txt: continue content_parts.append(txt) break content = "\n\n".join(content_parts) # 附件 attachments = [] for a in soup.find_all("a", href=True): h = a["href"].lower() if re.search(r'\.(pdf|doc|docx|xls|xlsx|zip|rar|ppt|pptx)$', h): fname = a.get_text(strip=True) or os.path.basename(a["href"]) full_url = urljoin(url, a["href"]) attachments.append(f"[{fname}]({full_url})") if attachments: if content: content += "\n\n---\n**附件：**\n" + "\n".join(attachments) else: content = "\n".join(attachments) return title, date_str, content def main(): print(f"[START] 山阴县人民政府-公告通知 阈值: {THRESHOLD}") total_new = 0 total_skip = 0 total_err = 0 for pg in range(0, 5): # pages 0=index.html, 1=index_1.html ... 4=index_4.html page_url = BASE_URL if pg == 0 else f"http://www.shanyin.gov.cn/sydt/ggtz/index_{pg}.html" print(f"\n[PAGE] 第{pg+1}页: {page_url}") html = fetch(page_url) if not html: print(f" [ERR] 获取失败") continue items = parse_list_page(html, page_url) if not items: print(f" [WARN] 无条目") continue has_recent = False for title, date_str, detail_url in items: if date_str < THRESHOLD: print(f" [SKIP] {date_str} {title[:40]}") total_skip += 1 continue has_recent = True # Handle direct PDF links if detail_url.lower().endswith('.pdf'): st = store_item(title, date_str, f"[查看PDF]({detail_url})", detail_url) if st: total_new += 1 print(f" [PDF] {date_str} {title[:40]}") else: total_err += 1 print(f" [DUP] {date_str} {title[:40]}") continue t, d, c = parse_detail_page(detail_url) if t and c: st = store_item(t, d or date_str, c, detail_url) if st: total_new += 1 print(f" [OK] {d or date_str} {t[:40]}") else: total_err += 1 print(f" [DUP] {d or date_str} {t[:40]}") else: total_err += 1 print(f" [ERR] {d or date_str} {title[:40]} 详情页解析失败 {detail_url}") time.sleep(0.3) if not has_recent: print(f" [STOP] 本页无阈值内条目，结束") break time.sleep(0.5) print(f"\n[DONE] 新增: {total_new}, 跳过: {total_skip}, 失败: {total_err}") if __name__ == "__main__": main()