正文
/
/
用法:
python3 crawl_chuanshan.py # 全量
python3 crawl_chuanshan.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_PATH = os.environ.get("GOV_DB_PATH", os.path.expanduser(
"~/Nutstore Files/我的坚果云/Crawler/gov_crawler/search.db"
))
BASE_URL = "https://www.chuanshan.gov.cn"
LIST_URL = "/gxq/zwgk/list/db49e9ea4397a019d43a5057ef8cbe9e.html"
SITE_NAME = "船山区人民政府"
SOURCE = "高新区动态"
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
def log(msg):
print(f"[chuanshan] {msg}")
def fetch_list_page(page):
"""Return [(url, title, date_str), ...]"""
url = f"{BASE_URL}{LIST_URL}?page={page}" if page > 1 else f"{BASE_URL}{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
):
title = m.group(1).strip()
href = m.group(2)
if not href.startswith("/gxq/zwgk/show/"):
continue
full_url = urljoin(BASE_URL, href)
# Find the corresponding date span - it follows the inside
date_str = ""
d = re.search(rf're:\s*{re.escape(href)}["\']?>\s*(\d{{4}}-\d{{2}}-\d{{2}})', html)
# Actually let me be more precise - the contains followed by
# Find this and get the next sibling span
# Simpler: find the containing this href
li_match = re.search(
rf']*>.*?href="{re.escape(href)}"[^>]*>.*?\s*(\d{{4}}-\d{{2}}-\d{{2}})',
html, re.DOTALL
)
if li_match:
date_str = li_match.group(1)
items.append((full_url, title, date_str))
has_next = re.search(r'page=(\d+)">下一页', html) is not None
return items, has_next
def fetch_detail(url):
"""Return (content_html, date_str, title) or (None, 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, None
soup = BeautifulSoup(html, "html.parser")
# Title from meta
meta_title = soup.find("meta", attrs={"name": "ArticleTitle"})
title = meta_title["content"].strip() if meta_title and meta_title.get("content") else ""
# Date from meta
date_str = ""
meta_date = soup.find("meta", attrs={"name": "PubDate"})
if meta_date and meta_date.get("content"):
m = re.match(r"(\d{4}-\d{2}-\d{2})", meta_date["content"].strip())
if m:
date_str = m.group(1)
# Content: (or div)
content_el = soup.find(class_=lambda c: c and "article-content" in (c if isinstance(c, str) else " ".join(c)))
content_html = str(content_el) if content_el else ""
if not content_html:
log(f" WARNING: no content for {url}")
return None, date_str, title
return content_html, date_str, title
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, None)
return results
def save_to_db(items):
"""items: list of (url, title, date_str, content_html)"""
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_next = fetch_list_page(page)
if not articles:
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")
if not has_next:
log(f"Page {page}: no next page, stopping")
break
log(f"Total articles: {len(all_articles)}")
# Phase 2: Fetch details
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_title = detail_map.get(url, (None, None, None))
final_title = detail_title or title
final_date = detail_date or date_str
results.append((url, final_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()