#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ crawl_penglai_tzgg.py - 蓬莱区-通知公告 CMS: 蓬莱区政府 JPAAS CMS 列表: GET /api-gateway/jpaas-publish-server/front/page/build/unit paramJson={"pageNo":N, "pageSize":25}, pageId=13426 详情: /col/col13426/art/YYYY/art_{hash}.html 标题: meta[ArticleTitle] 日期: meta[PubDate] 内容: div#zoom (p + table 混合) """ import requests import re import sys import os import time import json as _json from bs4 import BeautifulSoup DB_PATH = "/root/search.db" BASE_URL = "https://www.penglai.gov.cn" API_URL = BASE_URL + "/api-gateway/jpaas-publish-server/front/page/build/unit" SITE_NAME = "蓬莱区-通知公告" PAGE_ID = "13426" 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", } API_PARAMS = { "parseType": "bulidstatic", "webId": "152", "tplSetId": "awU93spIZ5G7CYnk12KGt", "pageType": "column", "tagId": "当前栏目列表", "editType": "null", "pageId": PAGE_ID, } GROUP = "山东" INDUSTRY = "其他" def fetch_list_page(session, page_no, page_size=25): """Fetch one page of list data from JPAAS API""" params = dict(API_PARAMS) params["paramJson"] = _json.dumps({"pageNo": page_no, "pageSize": page_size}, ensure_ascii=False) try: r = session.get(API_URL, params=params, headers=HEADERS, timeout=30) data = r.json() html = data.get("data", {}).get("html", "") # Extract total count from pagination div count_m = re.search(r'count="(\d+)"', html) total = int(count_m.group(1)) if count_m else 0 return html, total except Exception as e: print(f" [API ERROR] page {page_no}: {e}") return None, 0 def parse_list(html): """Parse JPAAS list HTML to extract (url, title, date) tuples""" items = [] soup = BeautifulSoup(html, "html.parser") content_div = soup.find(id="当前栏目列表") if not content_div: return items for li in content_div.select("li"): a = li.find("a") span = li.find("span") if not a or not a.get("href"): continue href = a.get("href", "").strip() title = a.get("title", "").strip() or a.get_text(strip=True) date_text = span.get_text(strip=True) if span else "" date_m = re.match(r"(\d{4}-\d{2}-\d{2})", date_text) date_str = date_m.group(1) if date_m else "" if href.startswith("/"): href = BASE_URL + href items.append((href, title, date_str)) return items def fetch_detail(session, url): """Fetch detail page HTML""" try: r = session.get(url, headers=HEADERS, timeout=30) r.encoding = "utf-8" return r.text except Exception as e: print(f"ERR:{e}") return None def extract_detail(html, url): """Extract title, date, content from detail page""" soup = BeautifulSoup(html, "html.parser") # Title from meta[ArticleTitle] title = "" meta_title = soup.find("meta", attrs={"name": re.compile(r"ArticleTitle", re.I)}) if meta_title and meta_title.get("content"): title = meta_title["content"].strip() if not title: title_tag = soup.find("title") if title_tag: title = title_tag.get_text(strip=True) # Date from meta[PubDate] date_str = "" for name in ["PublishDate", "PubDate", "publishdate", "pubdate"]: meta_date = soup.find("meta", attrs={"name": re.compile(name, re.I)}) if meta_date and meta_date.get("content"): raw = meta_date["content"].strip() m = re.match(r"(\d{4}-\d{2}-\d{2})", raw) if m: date_str = m.group(1) break # Content from div#zoom content = "" has_table = False zoom = soup.find("div", id="zoom") if zoom: parts = [] for elem in zoom.find_all(["p", "table", "div", "span"], recursive=True): if elem.name == "p": text = elem.get_text(strip=True) if text and text not in ("\xa0", ""): parts.append(text) elif elem.name == "table": has_table = True parts.append(str(elem)) elif elem.name in ("div", "span"): # Only capture if it contains text (non-empty divs) text = elem.get_text(strip=True) if text and text not in ("\xa0", ""): # Check this isn't just a wrapper for table/p already captured has_nested = elem.find(["p", "table"]) if not has_nested: parts.append(text) content = "\n\n".join(parts) # Summary (plain text) summary = re.sub(r"<[^>]+>", "", content)[:200] if content else "" return title, date_str, content, summary, has_table def save_to_db(items_data): """Batch insert into gov_raw and gov_search (FTS)""" import sqlite3 conn = sqlite3.connect(DB_PATH, timeout=10) c = conn.cursor() inserted = 0 fts_batch = [] # (rowid, title, site_name, summary) for title, date_str, content, summary, has_table, page_url in items_data: try: # Check duplicate c.execute("SELECT id FROM gov_raw WHERE page_url = ?", (page_url,)) existing = c.fetchone() if existing: continue # skip duplicate c.execute( """INSERT INTO gov_raw (page_url, title, site_name, publish_date, content, summary, source_url, category, industry, has_table, group_name, script_name) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", (page_url, title, SITE_NAME, date_str, content, summary, page_url, "政府公告", INDUSTRY, 1 if has_table else 0, GROUP, "crawl_penglai_tzgg.py"), ) new_id = c.lastrowid if new_id: fts_batch.append((new_id, title, SITE_NAME, summary or "")) inserted += 1 except Exception as e: print(f" [DB ERROR] {title[:30]}: {e}") conn.commit() # FTS sync after commit (avoid DB lock) if fts_batch: for rid, title, site, summary in fts_batch: try: c.execute( "INSERT OR IGNORE INTO gov_search(rowid, title, site_name, summary) VALUES (?, ?, ?, ?)", (rid, title, site, summary), ) except Exception as e: print(f" [FTS ERROR] id={rid}: {e}") conn.commit() print(f" FTS同步: {len(fts_batch)}条") conn.close() return inserted def main(): import argparse parser = argparse.ArgumentParser(description=f"{SITE_NAME}爬虫") parser.add_argument("--pages", type=int, default=5, help="爬取页数") parser.add_argument("--full", action="store_true", help="全量爬取") args = parser.parse_args() max_pages = 999 if args.full else args.pages session = requests.Session() total_new = 0 for page_no in range(1, max_pages + 1): print(f"[{time.strftime('%H:%M:%S')}] 第{page_no}页...", end=" ", flush=True) html, total = fetch_list_page(session, page_no) if not html: print("失败") break items = parse_list(html) if not items: print("无数据，停止") break print(f"{len(items)}条", flush=True) batch = [] for href, title, date_str in items: detail_html = fetch_detail(session, href) if not detail_html: continue detail_title, detail_date, content, summary, has_table = extract_detail(detail_html, href) final_title = detail_title or title final_date = detail_date or date_str batch.append((final_title, final_date, content, summary, has_table, href)) time.sleep(0.3) saved = save_to_db(batch) total_new += saved print(f" 入库{saved}/{len(batch)}条 (累计{total_new})", flush=True) # Incremental mode: if all items from this page were already in DB, stop if not args.full and saved == 0 and len(batch) > 0: print(" 全部已存在，增量停止") break print(f"\n===== 完成 =====") print(f"新增入库: {total_new} 条") print(f"站点: {SITE_NAME}") print(f"分组: {GROUP}") if __name__ == "__main__": main()