#!/usr/bin/env python3 import os """ crawl_zgzhijiang.py — 枝江市政府信息公开（行政许可/环评公示） ========================================================= URL: http://www.zgzhijiang.gov.cn/zfxxgk/list.html?depid=172&catid=663&t=0 API: /show/lists (列表) /show/detail (详情) 写入 /root/search.db gov_raw 表 """ import sys, re, time, json import requests from bs4 import BeautifulSoup from datetime import datetime, timedelta BASE_URL = "http://www.zgzhijiang.gov.cn" LIST_API = BASE_URL + "/show/lists" DETAIL_API = BASE_URL + "/show/detail" REFERER = BASE_URL + "/zfxxgk/list.html?depid=172&catid=663&t=0" DB_PATH = os.getenv("SEARCH_DB", "/root/search.db") SITE_NAME = "枝江市政府信息公开 - 行政许可" AREAID = 9 WEBID = 172 CID = 663 # 行政许可（含环评公示） HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36", "Referer": REFERER, } THREE_YEARS_AGO = (datetime.now() - timedelta(days=3*365)).strftime("%Y-%m-%d") session = requests.Session() session.headers.update(HEADERS) session.verify = False requests.packages.urllib3.disable_warnings() def parse_date(date_str): """Parse various date formats to YYYY-MM-DD""" if not date_str: return "" date_str = str(date_str).strip() for fmt in ["%Y-%m-%d", "%Y/%m/%d", "%Y年%m月%d日"]: try: return datetime.strptime(date_str[:10], fmt).strftime("%Y-%m-%d") except ValueError: continue m = re.search(r'(\d{4}[-/]\d{1,2}[-/]\d{1,2})', date_str) return m.group(1) if m else "" def fetch_list_page(page=1, pagenums=20): """Call list API, return (items, total)""" params = f"?areaid={AREAID}&webid={WEBID}&cid={CID}&page={page}&pagenums={pagenums}&orderby=0" url = LIST_API + params try: r = session.get(url, timeout=20) if r.status_code != 200: return None, 0 data = r.json() total = data.get("allnums", 0) items = data.get("lists", []) return items, total except Exception as e: print(f" API error: {e}") return None, 0 def get_detail(n_id): """Call detail API, return (content_html, pub_date)""" url = f"{DETAIL_API}?areaid={AREAID}&id={n_id}" try: r = session.get(url, timeout=20) if r.status_code != 200: return "", "" data = r.json() if isinstance(data, list) and len(data) > 0: detail = data[0] elif isinstance(data, dict): detail = data.get("data", data) else: return "", "" content = detail.get("content", "") pub_date = parse_date(detail.get("vc_inputtime", "")) if not pub_date: pub_date = parse_date(detail.get("createtime", "")) return content, pub_date except Exception as e: print(f" Detail API error ({n_id}): {e}") return "", "" def get_all_list_items(incremental=False): """Fetch list pages with smart cutoff""" all_items = [] seen_ids = set() page = 0 empty_recent_streak = 0 total_announced = False # For incremental, only fetch page 1 max_pages = 1 if incremental else 999 while True: page += 1 if page > max_pages: break items, total = fetch_list_page(page) if items is None: break if not total_announced: total_pages = (total + 19) // 20 print(f" Total: {total} records, ~{total_pages} pages") total_announced = True if not items: break page_items = [] page_recent = 0 for it in items: nid = it.get("n_id") if nid in seen_ids: continue seen_ids.add(nid) title = it.get("title", "") pub_date = parse_date(it.get("vc_inputtime", "")) if not pub_date: pub_date = parse_date(it.get("createtime", "")) detail_url = f"{DETAIL_API}?areaid={AREAID}&id={nid}" entry = { "title": title, "url": detail_url, "n_id": nid, "pub_date": pub_date, "vc_number": it.get("vc_number", ""), "vc_department": it.get("vc_department", ""), "class_name": it.get("class_name", ""), } page_items.append(entry) all_items.append(entry) if pub_date and pub_date >= THREE_YEARS_AGO: page_recent += 1 print(f" Page {page}: {len(page_items)} records (recent: {page_recent})") if incremental: break # Smart cutoff: 2 consecutive pages with no recent data + at least 5 pages if page_recent == 0: empty_recent_streak += 1 else: empty_recent_streak = 0 if page >= 5 and empty_recent_streak >= 2: print(f" 2 consecutive pages with no recent data, stopping at page {page}") break time.sleep(0.3) # Filter to only recent items, or fallback to first 5 pages recent = [it for it in all_items if it["pub_date"] and it["pub_date"] >= THREE_YEARS_AGO] if recent: print(f" Recent (3yr): {len(recent)} from {len(all_items)} total") return recent else: result = all_items[:100] print(f" All data >3yr old, fallback to first {len(result)} records") return result def insert_to_db(records, conn): """Insert records into gov_raw""" cursor = conn.cursor() inserted = 0 skipped = 0 for rec in records: if rec["pub_date"] and rec["pub_date"] < THREE_YEARS_AGO: skipped += 1 continue try: cursor.execute(""" INSERT OR IGNORE INTO gov_raw (page_url, source_url, title, summary, content, publish_date, site_name) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( rec["url"], BASE_URL, rec["title"], f"文号: {rec.get('vc_number','')} | 部门: {rec.get('vc_department','')}", rec.get("content", ""), rec.get("pub_date", ""), SITE_NAME, )) if cursor.rowcount > 0: inserted += 1 except Exception as e: print(f" DB error: {e}") conn.commit() return inserted, skipped def main(): incremental = 'incremental' in sys.argv or sys.argv[-1] == '1' import sqlite3 conn = sqlite3.connect(DB_PATH, timeout=60) print(f"=== {SITE_NAME} ===") if incremental: print("Mode: incremental (page 1 only)") # Get list items = get_all_list_items(incremental=incremental) if not items: print("No items found") conn.close() return print(f"Fetching {len(items)} details...") for i, item in enumerate(items): if i % 10 == 0: print(f" {i}/{len(items)}", flush=True) content, date_from_detail = get_detail(item["n_id"]) if content: item["content"] = content if date_from_detail: item["pub_date"] = date_from_detail # Insert inserted, skipped = insert_to_db(items, conn) conn.close() print(f"\n=== SUMMARY ===") print(f"Inserted: {inserted}, Skipped (old): {skipped}") if __name__ == "__main__": main()