#!/usr/bin/env python3 """Crawler for hbj.hbdawu.gov.cn - 行政许可 column (JEECMS)""" import requests import sqlite3 import re import sys import os from bs4 import BeautifulSoup from urllib.parse import urljoin from datetime import datetime, timedelta BASE_URL = "http://hbj.hbdawu.gov.cn" LIST_URL = "http://hbj.hbdawu.gov.cn/xzxk/index.jhtml" DB_PATH = "/root/search.db" MAX_PAGES = 5 # latest ~75 records SITE_NAME = "hbj.hbdawu.gov.cn-行政许可" CUTOFF_DATE = (datetime.now() - timedelta(days=3 * 365)).strftime("%Y-%m-%d") 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", } session = requests.Session() session.headers.update(HEADERS) session.verify = False import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) def get_page_urls(num_pages): """Generate list page URLs: index.jhtml, index_2.jhtml, ...""" urls = [LIST_URL] for i in range(2, num_pages + 1): urls.append(f"http://hbj.hbdawu.gov.cn/xzxk/index_{i}.jhtml") return urls def parse_list(html): """Parse list page, return list of (title, url, date)""" soup = BeautifulSoup(html, "html.parser") ul = soup.find("ul", id="newslist") if not ul: print(" [WARN] ul#newslist not found") return [] items = [] for li in ul.find_all("li", recursive=False): a = li.find("a", class_="bt_link") span = li.find("span") if not a or not span: continue href = a.get("href", "") if not href or href.startswith("javascript") or href == "#": continue full_url = urljoin(BASE_URL, href) # Use title attribute for full title, strip 【行政许可】 if present in link text title = (a.get("title") or a.get_text(strip=True)).strip() date = span.get_text(strip=True) items.append((title, full_url, date)) return items def parse_detail(html, url): """Extract content from detail page. Returns (title_from_page, content_html).""" soup = BeautifulSoup(html, "html.parser") # Title from detail page title_tag = soup.find("td", align="center", style=lambda v: v and "font-size:20px" in v if v else False) detail_title = "" if title_tag: txt = title_tag.get_text(strip=True) # Between and match = re.search(r'(.*?)', str(title_tag), re.DOTALL) if match: detail_title = match.group(1).strip() else: detail_title = txt # Content content_div = soup.find("div", id="content_txt") if not content_div: print(f" [WARN] No div#content_txt in {url}") return detail_title, "" # Get inner HTML, strip the caiji markers content_html = str(content_div) content_html = re.sub(r'', '', content_html) content_html = re.sub(r'', '', content_html) # Clean up excessive whitespace content_html = re.sub(r'\s*\n\s*', '\n', content_html) content_html = content_html.strip() return detail_title, content_html def extract_attachments(html, base_url): """Extract attachment links from detail HTML.""" soup = BeautifulSoup(html, "html.parser") attachments = [] for a in soup.find_all("a", href=True): href = a["href"] if href.startswith("http") and re.search(r'\.(pdf|doc|docx|xls|xlsx|zip|rar)$', href, re.I): attachments.append(href) elif href.startswith("/") and re.search(r'\.(pdf|doc|docx|xls|xlsx|zip|rar)$', href, re.I): attachments.append(urljoin(base_url, href)) return attachments def main(): import json page_urls = get_page_urls(MAX_PAGES) all_items = [] print(f"[{SITE_NAME}] Scanning {len(page_urls)} pages...") for pu in page_urls: try: resp = session.get(pu, timeout=20) resp.encoding = "utf-8" items = parse_list(resp.text) print(f" {pu} -> {len(items)} items") all_items.extend(items) except Exception as e: print(f" [ERR] {pu}: {e}") # Dedup by URL seen = set() unique_items = [] for item in all_items: if item[1] not in seen: seen.add(item[1]) unique_items.append(item) print(f"\n[{SITE_NAME}] Total unique items: {len(unique_items)}") # Filter to 3-year window filtered = [] for title, url, date in unique_items: try: if date >= CUTOFF_DATE: filtered.append((title, url, date)) except: filtered.append((title, url, date)) print(f"[{SITE_NAME}] Items within 3 years (>{CUTOFF_DATE}): {len(filtered)}") # Connect DB conn = sqlite3.connect(DB_PATH) c = conn.cursor() # Check existing c.execute("SELECT COUNT(*) FROM gov_raw WHERE site_name = ?", (SITE_NAME,)) existing_count = c.fetchone()[0] print(f"[{SITE_NAME}] Existing records: {existing_count}") # Process each item new_count = 0 skip_count = 0 error_count = 0 for title, url, date in filtered: # Check if already exists c.execute("SELECT COUNT(*) FROM gov_raw WHERE page_url = ?", (url,)) if c.fetchone()[0] > 0: skip_count += 1 continue # Fetch detail try: resp = session.get(url, timeout=20) resp.encoding = "utf-8" detail_title, content_html = parse_detail(resp.text, url) except Exception as e: print(f" [ERR] Fetch {url}: {e}") error_count += 1 continue if not content_html: print(f" [SKIP] Empty content: {title}") skip_count += 1 continue # Use detail title if available, otherwise list title final_title = detail_title or title # Insert using correct schema columns c.execute( "INSERT OR IGNORE INTO gov_raw (site_name, title, content, page_url, publish_date, summary) VALUES (?, ?, ?, ?, ?, ?)", (SITE_NAME, final_title, content_html, url, date, "" if len(final_title) <= 200 else final_title[:200]) ) conn.commit() new_count += 1 # Add delay to be gentle import time time.sleep(0.5) print(f"\n[{SITE_NAME}] Done: {new_count} new, {skip_count} skipped, {error_count} errors") # Summary c.execute("SELECT COUNT(*) FROM gov_raw WHERE site_name = ?", (SITE_NAME,)) total = c.fetchone()[0] print(f"[{SITE_NAME}] Total in DB: {total}") conn.close() # Output JSON summary for the agent summary = { "site": SITE_NAME, "new": new_count, "skipped": skip_count, "errors": error_count, "total": total, "pages_scanned": len(page_urls), } print(f"\nJSON_OUTPUT:{json.dumps(summary, ensure_ascii=False)}") if __name__ == "__main__": main()