#!/usr/bin/env python3 """中站区人民政府-公示公告 爬虫""" import re, sys, sqlite3 import requests from bs4 import BeautifulSoup from urllib.parse import urljoin import os SITE_NAME = "中站区-公示公告" BASE = "https://www.jzzzq.gov.cn" LIST = "/ywdt/gsgg/" SEARCH_DB = os.getenv("SEARCH_DB", "/root/search.db") MAX_PAGES = 5 HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Referer": BASE + "/", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9", } def fetch(url): try: r = requests.get(url, headers=HEADERS, timeout=30) r.encoding = "utf-8" return r.text except Exception as e: print(f" ⚠️ 请求失败 {url}: {e}") return None def parse_list(html): soup = BeautifulSoup(html, "lxml") items = [] for td in soup.find_all("td", class_="wz14_h2"): a = td.find("a", class_="wz14_h2") if not a: continue href = a.get("href", "") title = a.get_text(strip=True) if not href or not title: continue # 只保留本站链接 if "jzzzq.gov.cn" not in href and not href.startswith("/"): continue full_url = urljoin(BASE + LIST, href) items.append({"title": title, "url": full_url, "date": ""}) # 提取日期（相邻td） all_tds = soup.find_all("td", class_="wz14_h2") for i, item in enumerate(items): for td in all_tds: if td.get("align") == "right": t = td.get_text(strip=True) if re.match(r"\d{4}-\d{2}-\d{2}", t): # 匹配最近的日期 pass # 更可靠的方法：直接匹配列表区域 rows = soup.find_all("tr") result = [] for row in rows: tds = row.find_all("td", class_="wz14_h2") for td in tds: a = td.find("a", class_="wz14_h2") if a: href = a.get("href", "") title = a.get_text(strip=True) if href and title and ("jzzzq.gov.cn" in href or href.startswith("/")): full_url = urljoin(BASE + LIST, href) # 从URL提取日期: /2026/05-27/604128.html m = re.search(r"/(\d{4})/(\d{2}-\d{2})/", href) date_str = f"{m.group(1)}-{m.group(2)}" if m else "" result.append({"title": title, "url": full_url, "date": date_str}) break return result def parse_detail(html, url): soup = BeautifulSoup(html, "lxml") title = "" h1 = soup.find("h1") or soup.find("h2") if h1: title = h1.get_text(strip=True) if not title: t = soup.find("title") if t: title = re.sub(r"[_\-|].*", "", t.get_text(strip=True)) date = "" for meta in soup.find_all("meta"): if meta.get("name", "").lower() in ("publish_date", "pubdate"): date = meta.get("content", "") break if not date: m = re.search(r"(\d{4}[-/]\d{1,2}[-/]\d{1,2})", str(soup)) if m: date = m.group(1) if not date: m = re.search(r"/(\d{4})/(\d{2}-\d{2})/", url) if m: date = f"{m.group(1)}-{m.group(2)}" # 正文 content = "" for cls in ["article-content", "TRS_Editor", "zoom", "content", "article"]: div = soup.find("div", class_=cls) if div: for tag in div(["script", "style", "iframe"]): tag.decompose() content = str(div) break if not content: best, best_count = None, 0 for div in soup.find_all("div"): ps = div.find_all("p") if len(ps) > best_count: best, best_count = div, len(ps) if best: for tag in best(["script", "style", "iframe"]): tag.decompose() content = str(best) content = re.sub(r'\s*style="[^"]*"', "", content) content = re.sub(r'\s*class="[^"]*"', "", content) return title, date, content def run(max_pages=None): if max_pages is None: max_pages = MAX_PAGES db = sqlite3.connect(SEARCH_DB) known = set( r[0] for r in db.execute( "SELECT page_url FROM gov_raw WHERE site_name=?", (SITE_NAME,) ).fetchall() ) db.close() all_items = [] for page in range(1, max_pages + 1): if page == 1: url = BASE + LIST else: url = f"{BASE}{LIST}index_{page}.html" print(f" 📄 第 {page} 页...", end=" ") html = fetch(url) if not html: print("❌") break items = parse_list(html) new_items = [it for it in items if it["url"] not in known] print(f"✅ {len(items)} 条 (新增 {len(new_items)})") if len(items) == 0: break all_items.extend(new_items) if len(new_items) == 0 and len(items) < 10: break if not all_items: print(" ⏭ 无新数据") return print(f" 🔍 获取 {len(all_items)} 条详情...") results = [] for i, item in enumerate(all_items): html = fetch(item["url"]) if not html: continue title, date_str, content = parse_detail(html, item["url"]) if not title: title = item["title"] if not date_str: date_str = item["date"] results.append({ "site_name": SITE_NAME, "source_url": item["url"][:500], "page_url": item["url"], "title": title[:500], "publish_date": date_str[:10] if date_str else "", "summary": title[:500], "content": content, "status": "active", "category": "", "tags": "", }) if (i + 1) % 10 == 0: print(f" [{i+1}/{len(all_items)}]") db = sqlite3.connect(SEARCH_DB) db.execute("PRAGMA journal_mode=WAL") db.execute("PRAGMA synchronous=NORMAL") ok, skip = 0, 0 for item in results: try: db.execute( "INSERT OR IGNORE INTO gov_raw (site_name, source_url, page_url, title, publish_date, summary, content, status, category, tags) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ( item["site_name"][:200], item["source_url"], item["page_url"], item["title"], item["publish_date"], item["summary"], item["content"], item["status"], item["category"], item["tags"], ), ) if db.total_changes > 0: ok += 1 else: skip += 1 except: skip += 1 db.commit() db.execute( "INSERT INTO gov_search(rowid, title, site_name, summary) " "SELECT r.id, r.title, r.site_name, r.summary FROM gov_raw r " "WHERE r.id NOT IN (SELECT rowid FROM gov_search) AND r.site_name=?", (SITE_NAME,), ) db.commit() db.close() print(f" 💾 入库: 新增{ok}, 跳过{skip}, FTS已同步") if __name__ == "__main__": max_p = int(sys.argv[1]) if len(sys.argv) > 1 and sys.argv[1].isdigit() else None run(max_p)