#!/usr/bin/env python3 """精河县—环境影响评价爬虫 URL: https://www.xjjh.gov.cn/zwgk/zfxxgk/fdzdgknr/zdly_zdms/sthj/hjyxpj.htm CMS: VSB9 (Visual SiteBuilder 9) """ import requests, re, sqlite3, sys, os from datetime import datetime, timedelta from bs4 import BeautifulSoup from urllib.parse import urljoin SITE_NAME = "精河县-环境影响评价" BASE_URL = "https://www.xjjh.gov.cn" LIST_PAGE = "/zwgk/zfxxgk/fdzdgknr/zdly_zdms/sthj/hjyxpj.htm" DB_PATH = os.getenv("SEARCH_DB", "/root/search.db") CUTOFF_DATE = (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 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36", } def fetch_page(page_num): """Fetch list page by number. Page 1 uses the base URL, others use hjyxpj/{N}.htm""" if page_num == 1: url = BASE_URL + LIST_PAGE else: # Pages are in a subdirectory named after the base page (without .htm) base_name = LIST_PAGE.rsplit("/", 1)[-1].replace(".htm", "") list_dir = LIST_PAGE.rsplit("/", 1)[0] url = f"{BASE_URL}{list_dir}/{base_name}/{page_num}.htm" r = requests.get(url, headers=HEADERS, timeout=30) r.encoding = "utf-8" if "404错误提示" in r.text or len(r.text) < 2000: return None # No more pages return r.text def parse_list(html): """Parse list page HTML to extract items.""" soup = BeautifulSoup(html, "html.parser") items = [] table = soup.find("table", class_="winstyle27637") if not table: return items for tr in table.find_all("tr"): a = tr.find("a", class_="c27637") date_span = tr.find("span", class_="timestyle27637") if a and date_span: href = a.get("href", "") if href and "info/" in href: title = a.get("title", "") or a.get_text(strip=True) if not title: continue # Date: "2026年05月25日" -> "2026-05-25" date_str = date_span.get_text(strip=True) date_str = date_str.replace("年", "-").replace("月", "-").replace("日", "").strip() # Normalize URL full_url = urljoin(BASE_URL, href) items.append({ "title": title, "date": date_str, "url": full_url, }) return items def fetch_detail(url): """Fetch detail page and extract content.""" try: r = requests.get(url, headers=HEADERS, timeout=30) r.encoding = "utf-8" soup = BeautifulSoup(r.text, "html.parser") except Exception as e: return {"content": "", "full_title": ""} # Title from meta full_title = "" meta_title = soup.find("meta", attrs={"name": "ArticleTitle"}) or soup.find("meta", attrs={"Name": "ArticleTitle"}) if meta_title and meta_title.get("content"): full_title = meta_title["content"].strip() # Content from v_news_content content = "" content_div = soup.find("div", class_="v_news_content") if content_div: content = str(content_div) return {"content": content, "full_title": full_title} def save_to_db(items): """Insert items into DB.""" conn = sqlite3.connect(DB_PATH) c = conn.cursor() inserted = 0 for item in items: content_clean = re.sub(r'<[^>]+>', '', item["content"]).strip() c.execute("""INSERT OR IGNORE INTO gov_raw (title, page_url, site_name, summary, content, publish_date) VALUES (?, ?, ?, ?, ?, ?)""", (item["title"], item["url"], SITE_NAME, content_clean[:500], item["content"], item["date"])) if c.rowcount > 0: inserted += 1 # Also write to gov_search_v3 (trigram FTS table used by search app) c.execute( "INSERT OR IGNORE INTO gov_search_v3 (title, content, source_url, publish_date, site_name) VALUES (?, ?, ?, ?, ?)", (item["title"], content_clean, item["url"], item["date"], SITE_NAME), ) conn.commit() conn.close() return inserted def rebuild_fts(): """Rebuild FTS indexes.""" conn = sqlite3.connect(DB_PATH) c = conn.cursor() for tbl in ["gov_raw_fts", "gov_search", "gov_search_v3"]: try: c.execute(f"INSERT INTO {tbl}({tbl}) VALUES('rebuild')") except Exception: pass conn.commit() conn.close() def main(): is_incremental = len(sys.argv) >= 2 cutoff_days = int(sys.argv[1]) if is_incremental else 9999 print(f"=== {SITE_NAME} === 增量模式: {is_incremental} (cutoff={cutoff_days}天)") # Step 1-2: Collect all items from all pages all_items = [] for page in range(1, 20): # Max 20 pages safety limit html = fetch_page(page) if html is None: print(f" 第{page}页: 无更多内容（404/空页），停止翻页") break items = parse_list(html) if not items: print(f" 第{page}页: 无有效条目，停止翻页") break print(f" 第{page}页: {len(items)} 条") all_items.extend(items) print(f" 共获取 {len(all_items)} 条列表数据") # Step 3: Date filter if is_incremental: cutoff = (datetime.now() - timedelta(days=cutoff_days)).strftime("%Y-%m-%d") filtered = [it for it in all_items if it["date"] >= cutoff] else: cutoff = CUTOFF_DATE filtered = [it for it in all_items if it["date"] >= cutoff] print(f" 日期过滤后: {len(filtered)} 条 (>= {cutoff})") if not filtered: print(" 无新数据") return # Step 4: De-duplicate against DB conn = sqlite3.connect(DB_PATH) c = conn.cursor() existing = set() for item in filtered: c.execute("SELECT page_url FROM gov_raw WHERE page_url=?", (item["url"],)) if c.fetchone(): existing.add(item["url"]) conn.close() to_fetch = [it for it in filtered if it["url"] not in existing] print(f" 去重后: {len(to_fetch)} 条需抓详情页") # Step 5: Fetch detail pages success = [] for i, item in enumerate(to_fetch): detail = fetch_detail(item["url"]) if detail["content"] and len(detail["content"]) > 50: success.append({ "title": detail["full_title"] or item["title"], "url": item["url"], "date": item["date"], "content": detail["content"], }) elif detail["content"]: # Short content still save it success.append({ "title": detail["full_title"] or item["title"], "url": item["url"], "date": item["date"], "content": detail["content"], }) else: print(f" 跳过无内容: {item['title'][:40]}...") # Step 6: Insert to DB inserted = save_to_db(success) print(f" 入库: {inserted} 条 (成功/总数: {len(success)}/{len(to_fetch)})") if inserted > 0: rebuild_fts() print(" FTS 重建完成") print(f"=== {SITE_NAME} 完成，新增 {inserted} 条 ===") if __name__ == "__main__": main()