正文 |
日期
用法:
python3 crawl_gaotang.py # 全量
python3 crawl_gaotang.py 1 # 增量（只爬首页）
"""
import os
import re
import sys
import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin
from datetime import datetime, timedelta
from concurrent.futures import ThreadPoolExecutor, as_completed
# ── DB ──────────────────────────────────────────────
DB_PATH = os.environ.get("GOV_DB_PATH", os.path.expanduser(
"~/Nutstore Files/我的坚果云/Crawler/gov_crawler/search.db"
))
# ── URLs ────────────────────────────────────────────
BASE_URL = "http://www.gaotang.gov.cn"
LIST_URL = BASE_URL + "/channel_t_275_15948/"
SITE_NAME = "高唐县人民政府"
SOURCE = "高唐县公告公示"
# ── Headers ─────────────────────────────────────────
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",
}
MAX_WORKERS = 20
MAX_PAGES = 100 # safety
# ── helpers ─────────────────────────────────────────
def log(msg):
print(f"[gaotang] {msg}")
def clean_title(title):
return title.replace(" | 高唐县人民政府", "").strip()
def fetch_list_page(page):
"""Return [(url, title, date_str), ...]"""
url = f"{LIST_URL}?page={page}" if page > 1 else LIST_URL
try:
resp = requests.get(url, headers=HEADERS, timeout=30)
resp.encoding = "utf-8"
html = resp.text
except Exception as e:
log(f" ERROR page {page}: {e}")
return [], False
items = []
for m in re.finditer(
r'
]*>(.*?)',
html, re.DOTALL
):
href = m.group(1)
inner = m.group(2)
text = re.sub(r'<[^>]+>', '', inner).strip()
text = re.sub(r'\s+', ' ', text).strip()
# Skip footer links (郑重声明/隐私保护/关于我们)
if any(kw in text for kw in ['郑重声明', '隐私保护', '关于我们']):
continue
date_str = ""
d = re.search(r'(\d{4}-\d{2}-\d{2})', text)
if d:
date_str = d.group(1)
title = re.sub(r'\d{4}-\d{2}-\d{2}', '', text).strip()
title = clean_title(title)
full_url = href if href.startswith("http") else urljoin(BASE_URL, href)
items.append((full_url, title, date_str))
return items, len(items) > 0
def fetch_detail(url):
"""Return (content_html, date_str) or (None, None)"""
try:
resp = requests.get(url, headers=HEADERS, timeout=30)
resp.encoding = "utf-8"
html = resp.text
except Exception as e:
log(f" ERROR {url}: {e}")
return None, None
soup = BeautifulSoup(html, "html.parser")
# Content: or class variants
content_html = ""
for selector in [
("div", {"id": "sxNewsDetailContent"}),
("div", {"class": lambda c: c and "sxNewsDetailContent" in (c if isinstance(c, str) else ' '.join(c))}),
]:
div = soup.find(selector[0], selector[1])
if div:
content_html = str(div)
break
# Fallback
if not content_html:
for cls in ["content", "gtxIxCont", "main"]:
div = soup.find("div", class_=lambda c: c and cls in (c if isinstance(c, str) else ' '.join(c)))
if div:
content_html = str(div)
break
# Date
date_str = ""
meta = soup.find("meta", attrs={"name": "PubDate"})
if meta and meta.get("content"):
m = re.match(r"(\d{4}-\d{2}-\d{2})", meta["content"].strip())
if m:
date_str = m.group(1)
if not content_html:
log(f" WARNING: no content for {url}")
return None, date_str
return content_html, date_str
def fetch_detail_batch(urls):
results = {}
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as ex:
fut_map = {ex.submit(fetch_detail, url): url for url in urls}
for fut in as_completed(fut_map):
url = fut_map[fut]
try:
results[url] = fut.result()
except Exception as e:
results[url] = (None, None)
return results
def save_to_db(items):
import sqlite3
if not items:
return 0
conn = sqlite3.connect(DB_PATH)
c = conn.cursor()
inserted = 0
for url, title, date_str, content_html in items:
try:
c.execute(
"""INSERT OR REPLACE INTO gov_raw
(site_name, page_url, title, publish_date, content, category)
VALUES (?, ?, ?, ?, ?, ?)""",
(SITE_NAME, url, title, date_str, content_html, SOURCE),
)
inserted += 1
except Exception as e:
log(f" DB error for {url}: {e}")
conn.commit()
conn.close()
return inserted
def main():
incremental = len(sys.argv) > 1 and sys.argv[1] == "1"
three_years_ago = (datetime.now() - timedelta(days=365 * 3)).strftime("%Y-%m-%d")
log(f"3-year cutoff: {three_years_ago}")
# Phase 1: Scan list pages
all_articles = []
for page in range(1, MAX_PAGES + 1):
articles, has_data = fetch_list_page(page)
if not has_data:
log(f"Page {page}: no data, stopping")
break
filtered = [(u, t, d) for u, t, d in articles if d and d >= three_years_ago]
all_articles.extend(filtered)
if incremental:
log(f"Incremental: page 1, {len(filtered)} articles")
break
if articles and not filtered:
log(f"Page {page}: all before cutoff, stopping")
break
if page % 10 == 0:
log(f" Page {page}: {len(all_articles)} articles so far")
log(f"Total articles: {len(all_articles)}")
# Phase 2: Fetch details in parallel batches
results = []
batch_size = MAX_WORKERS * 3
for batch_start in range(0, len(all_articles), batch_size):
batch = all_articles[batch_start:batch_start + batch_size]
batch_urls = [u for u, _, _ in batch]
log(f" Details [{batch_start+1}-{batch_start+len(batch)}/{len(all_articles)}]...")
detail_map = fetch_detail_batch(batch_urls)
for url, title, date_str in batch:
content_html, detail_date = detail_map.get(url, (None, None))
final_date = detail_date or date_str
results.append((url, title, final_date, content_html or ""))
# Save
saved = save_to_db(results)
log(f"\n{'='*50}")
log(f"Done. Total: {len(results)} articles (saved: {saved})")
if results:
dates = sorted([r[2] for r in results if r[2]])
log(f"Date range: {dates[0]} ~ {dates[-1]}")
# FTS rebuild
try:
import sqlite3
conn = sqlite3.connect(DB_PATH)
conn.execute("""
INSERT OR REPLACE INTO gov_search(rowid, title, site_name, summary)
SELECT rowid, title, site_name,
CASE WHEN length(content) > 200 THEN substr(content, 1, 200) ELSE content END
FROM gov_raw WHERE category = ?
""", (SOURCE,))
conn.commit()
conn.close()
log("FTS index rebuilt")
except Exception as e:
log(f"FTS rebuild note: {e}")
if __name__ == "__main__":
main()