#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 爬虫: 怀仁市人民政府 - 环境保护 http://www.zghr.gov.cn/xxgk/zdlyxxgk/hjbh/ CMS: 怀仁市政府网 """ import requests from bs4 import BeautifulSoup import sqlite3 import re import sys DB_PATH = "/root/search.db" BASE_URL = "http://www.zghr.gov.cn" LIST_URL = "http://www.zghr.gov.cn/xxgk/zdlyxxgk/hjbh/" SITE_NAME = "怀仁市人民政府" 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", "Accept-Language": "zh-CN,zh;q=0.9", } def get_list_urls(max_pages=5): """Get all article URLs from list pages""" url_data = [] for page in range(0, max_pages): if page == 0: url = LIST_URL else: url = "http://www.zghr.gov.cn/xxgk/zdlyxxgk/hjbh/index_{}.html".format(page) try: r = requests.get(url, timeout=15, headers=HEADERS) r.encoding = "utf-8" if r.status_code != 200: print(" Page {}: HTTP {}".format(page + 1, r.status_code), file=sys.stderr) break soup = BeautifulSoup(r.text, "html.parser") items = soup.select("ul.main li") if not items: print(" Page {}: no items".format(page + 1), file=sys.stderr) break count = 0 for li in items: a = li.find("a") if a and a.get("href"): href = a["href"] full_title = a.get("title", "").strip() or a.get_text(strip=True) date_span = li.find("span") pub_date = date_span.get_text(strip=True) if date_span else "" # Normalize URL if href.startswith("./"): href = BASE_URL + "/xxgk/zdlyxxgk/hjbh/" + href[2:] elif href.startswith("/"): href = BASE_URL + href elif not href.startswith("http"): href = BASE_URL + "/xxgk/zdlyxxgk/hjbh/" + href # Determine if local or external is_local = "zghr.gov.cn" in href url_data.append((href, full_title, pub_date, is_local)) count += 1 print(" Page {}: {} items".format(page + 1, count), file=sys.stderr) if len(items) < 5: break except Exception as e: print(" Page {} error: {}".format(page + 1, e), file=sys.stderr) break return url_data def parse_detail_local(url): """Parse a local detail page (zghr.gov.cn)""" try: r = requests.get(url, timeout=15, headers=HEADERS) r.encoding = "utf-8" if r.status_code != 200: return None soup = BeautifulSoup(r.text, "html.parser") # Title from h3 title = "" h3 = soup.find("h3") if h3: title = h3.get_text(strip=True) # Date from conContent date = "" cc = soup.select_one(".conContent") if cc: m = re.search(r"发布时间[：:]\s*(\d{4}-\d{2}-\d{2})", cc.get_text()) if m: date = m.group(1) # Content from TRS_Editor content = "" te = soup.select_one(".TRS_Editor") if te: parts = [] for child in te.children: if child.name == "p": txt = child.get_text(separator=" ", strip=True) if txt: parts.append(txt) elif child.name == "table": md_table = table_to_markdown(child) if md_table: parts.append(md_table) content = "\n\n".join(parts) # Attachments attachments = [] for a in soup.find_all("a", href=True): href = a["href"] txt = a.get_text(strip=True) if any(x in href.lower() for x in [".pdf", ".doc", ".xls", ".zip", ".docx", ".xlsx"]): if not href.startswith("http"): href = BASE_URL + "/" + href.lstrip("/") attachments.append({ "name": txt or href.split("/")[-1], "url": href }) # PDF/full content fallback if len(content.strip()) < 20: content = '

{}

'.format(url, title) if attachments: al = "\n".join([" - [{}]({})".format(a["name"], a["url"]) for a in attachments]) content += "\n\n附件：\n" + al content += "\n\n（原文链接查看）" return { "title": title, "url": url, "date": date, "content": content, "site_name": SITE_NAME, "source": "", "attachments": attachments, } except Exception as e: print(" Error parsing local {}: {}".format(url, e), file=sys.stderr) return None def table_to_markdown(table, *args, **kwargs): """保留 HTML 表格结构（不转 md）""" return str(table) def save_to_db(articles): """Save articles to search.db""" conn = sqlite3.connect(DB_PATH, timeout=60) c = conn.cursor() inserted = 0 for art in articles: if not art or not art["title"]: continue c.execute( "SELECT 1 FROM gov_raw WHERE page_url=? AND site_name=?", (art["url"], SITE_NAME) ) if c.fetchone(): continue attachments_json = str(art["attachments"]) if art["attachments"] else "" c.execute( """INSERT OR REPLACE INTO gov_raw (page_url, title, content, summary, site_name, publish_date, source_url, attachments, script_name) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'crawl_zghr.py')""", ( art["url"], art["title"], art["content"], art["content"][:200] if art["content"] else "", art["site_name"], art["date"], art.get("source", ""), attachments_json, ) ) inserted += 1 conn.commit() conn.close() return inserted def main(): max_pages = 5 if len(sys.argv) > 1: try: max_pages = int(sys.argv[1]) except ValueError: pass print("Starting crawl: {} - {} ({} pages)".format(SITE_NAME, LIST_URL, max_pages), file=sys.stderr) # Get all URLs from list pages print("Fetching list pages...", file=sys.stderr) url_data = get_list_urls(max_pages) print("Total URLs: {}".format(len(url_data)), file=sys.stderr) # Parse each URL print("Fetching content...", file=sys.stderr) articles = [] for i, (url, title, date, is_local) in enumerate(url_data): print(" [{}/{}] {} (local={})".format(i + 1, len(url_data), url[:80], is_local), file=sys.stderr) if is_local: art = parse_detail_local(url) else: # External link - save title+url+date, content = external link art = { "title": title, "url": url, "date": date, "content": "[{}]({})\n\n（外部链接，请点击标题查看原文）".format(title, url), "site_name": SITE_NAME, "source": "", "attachments": [], } if art: if not art["title"]: art["title"] = title if not art["date"] and date: art["date"] = date articles.append(art) print("Parsed: {}/{} articles".format(len(articles), len(url_data)), file=sys.stderr) # Save to DB inserted = save_to_db(articles) print("Inserted: {} new articles".format(inserted), file=sys.stderr) print("Done.", file=sys.stderr) print("\nArticle count for config: {}".format(len(url_data))) if __name__ == "__main__": main()