#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 鄂尔多斯市生态环境局 - 审批项目受理公示 CMS: Custom CMS (TRS_Editor 正文) 列表: ul.listu > li > a (href=./202607/t...html) 分页: index.html + index_N.html (N=1..11), JS 0-indexed, 共12页/169条/15条每页 详情: div.TRS_Editor 正文 + div.article_content 附件 注意: DBApp WAF, IP 120.192.84.245 + Host header + verify=False """ import sys, os, re, time, json from datetime import datetime, timezone, timedelta sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from crawler_lib import push_to_searchdb import requests from bs4 import BeautifulSoup import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # === 自动页数控制 === _MAX_PG = None for i, a in enumerate(sys.argv): if a == '--pages' and i + 1 < len(sys.argv): _MAX_PG = int(sys.argv[i + 1]) break if _MAX_PG is not None: print("[AutoPg] max_pages=" + str(_MAX_PG)) SITE_NAME = "鄂尔多斯市生态环境局-审批项目受理公示" BASE_IP = "120.192.84.245" DOMAIN = "sthjj.ordos.gov.cn" BASE_URL = f"https://{BASE_IP}" LIST_PATH = "/ywgz_6642/hpyxpj_6651/jsxmbgbgs_161699/xmsl" TOTAL_PAGES = 12 CUTOFF = (datetime.now(timezone.utc) - timedelta(days=365*3)).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", "Host": DOMAIN, } def html_table_to_html(table, base_url=""): """保留 HTML 表格结构，仅将相对链接/图片转绝对 URL""" import urllib.parse tbl = BeautifulSoup(str(table), 'html.parser') for a in tbl.find_all('a'): href = a.get('href', '') if href and not href.startswith(('http', 'javascript', '#')): a['href'] = urllib.parse.urljoin(base_url, href) if base_url else href for img in tbl.find_all('img'): src = img.get('src', '') if src and not src.startswith(('http', '//', 'data:')): img['src'] = urllib.parse.urljoin(base_url, src) if base_url else src return str(tbl) def fetch(url, timeout=20): """通过 IP + Host header + verify=False 抓取 (DBApp WAF)""" for attempt in range(3): try: r = requests.get(url, headers=HEADERS, timeout=timeout, verify=False) r.encoding = "utf-8" return r.text except Exception as e: if attempt < 2: time.sleep(2) else: print(f" [WARN] 抓取失败 {url}: {e}") return None def make_list_url(page): """构建列表页URL (0-indexed JS pagination)""" if page == 1: return BASE_URL + LIST_PATH + "/" else: return BASE_URL + LIST_PATH + f"/index_{page - 1}.html" def parse_list_page(html): """解析列表页""" items = [] soup = BeautifulSoup(html, "html.parser") for li in soup.select("ul.listu > li"): a_tag = li.find("a", class_="clearfix", href=True) if not a_tag: continue href = a_tag["href"].strip() if href.startswith("./"): href = "https://" + DOMAIN + LIST_PATH + "/" + href[2:] elif href.startswith("/"): href = "https://" + DOMAIN + href elif not href.startswith("http"): href = "https://" + DOMAIN + LIST_PATH + "/" + href # 标题 title_p = a_tag.find("p", class_="title") title = title_p.get_text(" ", strip=True) if title_p else "" # 用 title 属性做完整标题 full_title = a_tag.get("title", "").strip() or title # 日期 date_span = a_tag.find("span", class_="sj") date = date_span.get_text(strip=True) if date_span else "" if title and href: items.append((full_title, href, date)) return items def _clean_span_spaces(text): """清洗 span 碎片导致的间距问题""" text = re.sub(r'(?<=[\u4e00-\u9fff])\s+(?=[\u4e00-\u9fff])', '', text) text = re.sub(r'(?<=[\u4e00-\u9fff])\s+(?=\d)', '', text) text = re.sub(r'(?<=\d)\s+(?=[\u4e00-\u9fff])', '', text) text = re.sub(r'\s+([，。、；：？！)】」》])', r'\1', text) text = re.sub(r'([（【「《])\s+', r'\1', text) text = re.sub(r'(?<=\d)\s+(?=\d)', '', text) return re.sub(r'\s+', ' ', text).strip() def extract_content(trs_div, page_url): """从 TRS_Editor div 中提取正文""" parts = [] for child in trs_div.children: name = getattr(child, 'name', None) if name == 'table': tbl_html = html_table_to_html(child, page_url) if tbl_html: parts.append(tbl_html) elif name == 'p': if child.find_parent("table"): continue # Images for img in child.find_all("img"): src = img.get("src", "") if src and "/filetype/" not in src: alt = img.get("alt", "") if src.startswith("/"): parts.append(f"![{alt}](https://{DOMAIN}{src})\n\n") else: parts.append(f"![{alt}]({src})\n\n") # Text txt = child.get_text(" ", strip=True) txt = _clean_span_spaces(txt) txt = re.sub(r'&#\d+;', '', txt) if txt and len(txt) > 3: parts.append(txt) parts.append("\n\n") result = ''.join(parts).strip() result = re.sub(r'\n{4,}', '\n\n', result) return result def extract_attachments(html, page_url): """从附件区提取附件""" attachments = [] soup = BeautifulSoup(html, "html.parser") article_content = soup.find("div", class_="article_content") if not article_content: return attachments for a_tag in article_content.find_all("a", href=True): href = a_tag["href"].strip() name = a_tag.get_text(" ", strip=True) # 解析相对路径 if href.startswith("./"): href = "https://" + DOMAIN + LIST_PATH + "/" + href[2:] elif href.startswith("/"): href = "https://" + DOMAIN + href elif not href.startswith("http"): href = "https://" + DOMAIN + LIST_PATH + "/" + href ext = href.lower() if any(ext.endswith(e) for e in ('.pdf', '.doc', '.docx', '.xls', '.xlsx', '.zip', '.rar', '.txt')): attachments.append(f"[{name}]({href})") return attachments def parse_detail(url): """解析详情页""" html = fetch(url) if not html: return None, None, None, [] soup = BeautifulSoup(html, "html.parser") # 标题 title = "" meta = soup.find("meta", attrs={"name": "ArticleTitle"}) if meta: title = meta.get("content", "").strip() if not title: h2 = soup.find("h2", class_="conetent-title") if h2: title = h2.get_text(strip=True) if not title: title_tag = soup.find("title") if title_tag: t = title_tag.get_text(strip=True) title = re.sub(r'_鄂尔多斯市生态环境局.*', '', t).strip() # 日期 date = "" meta_d = soup.find("meta", attrs={"name": "PubDate"}) if meta_d: d = meta_d.get("content", "") m = re.search(r'(\d{4}-\d{2}-\d{2})', d) if m: date = m.group(1) if not date: news_mes = soup.find("div", class_="news-mes2") if news_mes: m = re.search(r'(\d{4}-\d{2}-\d{2})', news_mes.get_text()) if m: date = m.group(1) # 正文 content = "" m_con = soup.find("div", class_="m-con") if m_con: trs = m_con.find("div", class_="TRS_Editor") if trs: content = extract_content(trs, url) # 附件 attachments = extract_attachments(html, url) attach_str = ";".join(attachments) if attachments else "" # 如果正文为空但附件存在 if (not content or len(content.strip()) < 10) and attachments: content = "【该文档详见附件】\n" + "\n".join(attachments) return title, content, date, attach_str def main(): pages_to_fetch = min(TOTAL_PAGES, _MAX_PG) if _MAX_PG else TOTAL_PAGES print(f"=== {SITE_NAME} ===") print(f" 总页数: {TOTAL_PAGES}, 本次: {pages_to_fetch}") all_items = [] for page in range(1, pages_to_fetch + 1): list_url = make_list_url(page) print(f" 第 {page}/{pages_to_fetch} 页: {list_url}") html = fetch(list_url) if not html: print(f" 无法获取, 停止") break items = parse_list_page(html) if not items: print(f" 空列表, 停止") break print(f" 找到 {len(items)} 条") all_items.extend(items) time.sleep(0.3) print(f"\n 列表总计: {len(all_items)} 条\n") if not all_items: print(" 无数据") return # 日期过滤 recent = [i for i in all_items if i[2] >= CUTOFF] if all_items[0][2] else all_items results = [] for idx, (title, url, date) in enumerate(recent): print(f" [{idx+1}/{len(recent)}] {title[:40]}...", end=" ", flush=True) dt, content, dd, attach_str = parse_detail(url) final_title = dt or title final_date = dd or date if not content or len(content.strip()) < 10: print("⚠ 空") continue summary = re.sub(r'<[^>]+>', '', content)[:200].strip() if not summary: summary = final_title[:200] results.append({ "site_name": SITE_NAME, "title": final_title, "url": url, "content": content, "summary": summary, "pub_date": final_date, "attachments": attach_str, }) print(f"✅ ({len(content)}字)") if (idx + 1) % 20 == 0: print(f" 进度: {idx+1}/{len(recent)}, 已入库: {len(results)}") if results: print(f"\n 入库 {len(results)} 条") push_to_searchdb(results, "ordos_xmsl") print("完成") if __name__ == "__main__": main()