#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Crawler for 米易县生态环境局 - 通知公告 CMS: 攀枝花市网站群（自定义） List: ul.clearfix.mt-10 > li > a + span (date) Pagination: index.shtml (p1), index_{N}.shtml (pN) Detail: h1 title, #Zoom content, meta ArticleTitle/PubDate """ import requests, re, subprocess, sys, time from bs4 import BeautifulSoup from urllib.parse import urljoin DB_PATH = "/root/search.db" SITE_NAME = "米易县生态环境局-通知公告" INDUSTRY = "环境公示" BASE_URL = "http://www.scmiyi.gov.cn" LIST_URL = "http://www.scmiyi.gov.cn/zwgk/zzjg/xjbm/hjbhj/tzgg/index.shtml" MAX_PAGES = 10 HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", } def esc(s): return (s or "").replace("'", "''") def get_list_url(page_num): if page_num == 1: return LIST_URL return f"{BASE_URL}/zwgk/zzjg/xjbm/hjbhj/tzgg/index_{page_num}.shtml" def extract_content(content_div, detail_url): """Extract text content from a div, preserving links and images""" parts = [] for child in content_div.children: if child.name == "a": href = child.get("href", "").strip() text = child.get_text(strip=True) if href and text: full = urljoin(detail_url, href) parts.append(f"[{text}]({full})") elif text: parts.append(text) elif child.name == "img": src = child.get("src", "") if src: alt = child.get("alt", "") full_src = urljoin(detail_url, src) parts.append(f"![{alt}]({full_src})" if alt else f"![]({full_src})") elif child.name == "br": parts.append("\n") elif child.name == "p": t = child.get_text(" ", strip=True) if t: parts.append(t + "\n\n") elif hasattr(child, "get_text"): t = child.get_text(" ", strip=True) if t: parts.append(t) return " ".join(parts).strip() def extract_attachments(soup, detail_url): """Extract attachment links""" attachments = [] for a in soup.find_all("a", href=True): href = a["href"].strip().lower() if href.endswith((".pdf", ".doc", ".docx", ".xls", ".xlsx", ".zip", ".rar")): name = a.get_text(strip=True) or a["href"] full = urljoin(detail_url, a["href"]) attachments.append({"name": name, "url": full}) return attachments def parse_detail(url): """Parse detail page""" try: r = requests.get(url, headers=HEADERS, timeout=15) r.encoding = "utf-8" if r.status_code != 200: return None, None, None except Exception as e: print(f" [ERR] {e}", flush=True) return None, None, None soup = BeautifulSoup(r.text, "html.parser") # Title from h1 or meta h1 = soup.find("h1") title = h1.get_text(strip=True) if h1 else "" if not title: mt = soup.find("meta", attrs={"name": "ArticleTitle"}) if mt and mt.get("content"): title = mt["content"].strip() # Date from meta or text date_str = "" mt = soup.find("meta", attrs={"name": "PubDate"}) if mt and mt.get("content"): date_str = mt["content"].strip()[:10] if not date_str: m = re.search(r"(\d{4}[-/]\d{1,2}[-/]\d{1,2})", r.text) if m: date_str = m.group(1).replace("/", "-") # Content content_div = soup.find("div", id="Zoom") content = "" if content_div: content = extract_content(content_div, url) else: # Fallback divs = soup.find_all("div") if divs: max_div = max(divs, key=lambda x: len(x.get_text(strip=True))) content = max_div.get_text(strip=True) # Attachments attachments = extract_attachments(soup, url) if attachments: for a in attachments: inline_marker = f"[{a['name']}]({a['url']})" if inline_marker not in content: content += f"\n\n附件：[{a['name']}]({a['url']})" return title, date_str, content def extract_list_page(page_num): """Extract list items from a page""" url = get_list_url(page_num) try: r = requests.get(url, headers=HEADERS, timeout=20) r.encoding = "utf-8" except Exception as e: print(f" Error fetching {url}: {e}", flush=True) return [] if r.status_code != 200: print(f" Page {page_num}: HTTP {r.status_code}", flush=True) return [] soup = BeautifulSoup(r.text, "html.parser") ul = soup.select_one("ul.clearfix.mt-10") if not ul: return [] items = [] for li in ul.find_all("li"): a = li.find("a") span = li.find("span") if not a or not a.get("href"): continue href = a["href"].strip() title = a.get("title", "").strip() or a.get_text(strip=True) page_url = urljoin(url, href) d = span.get_text(strip=True) if span else "" d = d.split()[0] if d else "" if title and page_url: items.append((title, page_url, d)) return items def insert_one(page_url, title, content, pub_date): summary = (content or "")[:200] sql = f"""INSERT INTO gov_raw (page_url, title, publish_date, content, site_name, industry, summary) VALUES ('{esc(page_url)}','{esc(title)}','{esc(pub_date)}','{esc(content)}','{esc(SITE_NAME)}','{INDUSTRY}','{esc(summary)}')""" result = subprocess.run( ["sqlite3", "-cmd", ".timeout 60000", DB_PATH, sql], capture_output=True, text=True, timeout=10, ) if result.returncode != 0 and "UNIQUE" not in result.stderr: print(f" [DB ERROR] {result.stderr}", file=sys.stderr) return False # FTS fts_sql = f"""INSERT OR REPLACE INTO gov_search(rowid, title, site_name, summary) SELECT rowid, title, site_name, summary FROM gov_raw WHERE page_url='{esc(page_url)}' AND rowid NOT IN (SELECT rowid FROM gov_search)""" subprocess.run(["sqlite3", "-cmd", ".timeout 60000", DB_PATH, fts_sql], capture_output=True, text=True, timeout=10) return True def crawl(): max_pages = int(sys.argv[1]) if len(sys.argv) > 1 else MAX_PAGES print(f"[{SITE_NAME}] Starting crawl, max_pages={max_pages}", flush=True) all_items = [] for page in range(1, max_pages + 1): items = extract_list_page(page) if not items: if page > 1: break continue print(f" Page {page}: {len(items)} items", flush=True) all_items.extend(items) if len(items) < 15: break print(f"Total list items: {len(all_items)}", flush=True) new_count = 0 dup_count = 0 for i, (title, url, date_str) in enumerate(all_items): print(f" [{i+1}/{len(all_items)}] {title[:50]}...", flush=True) detail_title, detail_date, content = parse_detail(url) final_title = detail_title or title final_date = detail_date or date_str if insert_one(url, final_title, content, final_date): new_count += 1 else: dup_count += 1 time.sleep(0.3) print(f"\nDone! New: {new_count}, Duplicates: {dup_count}", flush=True) if __name__ == "__main__": crawl()