SSL: --curves secp384r1 -k (服务器端OpenSSL兼容性) """ import sys import os import re import subprocess import json from bs4 import BeautifulSoup # ---------- 数据库配置 ---------- DB_PATH = "/mnt/data/search.db" if os.path.exists("/mnt/data/search.db") else os.path.expanduser("~/data/search.db") # ---------- 站点配置 ---------- BASE_URL = "https://sthjt.hunan.gov.cn" LIST_URL = "/sthjt/xxgk/xzgs/jsxm/hpgs/spjdgk_2/index.html" DOMAIN = "sthjt.hunan.gov.cn" IP = "113.246.57.9" # DNS解析结果 PAGES = 5 # 默认每5页（每页25条） GROUP_NAME = "湖南省" INDUSTRY = "环评公示" SITE_NAME = "湖南省生态环境厅" def fetch_page(url, use_curves=True): """用curl获取页面，处理SSL兼容性和DNS问题""" cmd = [ "curl", "-sS", "--connect-timeout", "10", "-m", "30", ] # 本地DNS无法解析时用--resolve cmd.extend(["--resolve", f"{DOMAIN}:443:{IP}"]) # 服务器端OpenSSL需要指定曲线 if use_curves: cmd.extend(["--curves", "secp384r1", "-k"]) cmd.append(url) try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=35) if result.returncode != 0: # 重试不带曲线 if use_curves: return fetch_page(url, use_curves=False) print(f" [WARN] curl返回码 {result.returncode}: {result.stderr[:100]}") return None return result.stdout except subprocess.TimeoutExpired: print(f" [WARN] 超时: {url}") return None def fetch_detail(detail_url): """获取详情页并提取标题、日期、正文""" full_url = BASE_URL + detail_url if detail_url.startswith("/") else detail_url html = fetch_page(full_url) if not html: return None, None, None soup = BeautifulSoup(html, "html.parser") # 标题: meta ArticleTitle title = "" for m in soup.select('meta[name=ArticleTitle]'): title = m.get("content", "").strip() break if not title: t = soup.find("title") if t: title = t.get_text(strip=True).replace(" - 湖南省生态环境厅", "").strip() # strip · 实体 title = title.replace("·", "").replace(" ", "").replace("\xa0", "").replace("\u3000", "").strip() # 日期: meta PubDate date_str = "" for m in soup.select('meta[name=PubDate]'): date_str = m.get("content", "").strip() break # 取前10位 YYYY-MM-DD if date_str: date_str = date_str[:10] # 正文: div.main_con_zw#j-show-body content_div = soup.select_one("div.main_con_zw#j-show-body") if not content_div: content_div = soup.select_one("div.main_con_zw") if not content_div: # 再尝试其他通用容器 content_div = soup.select_one("#j-show-body") text = "" if content_div: # 保留表格HTML，提取p标签文本 paragraphs = [] for child in content_div.children: name = getattr(child, "name", None) if name == "p": # 检查p内是否有table，有则分段处理 inner_table = child.find("table") if inner_table: for node in child.children: nname = getattr(node, "name", None) if nname is None: t = str(node).strip() if t: paragraphs.append(t) elif nname == "table": paragraphs.append(str(node)) elif nname == "br": pass else: t2 = node.get_text(strip=True) if t2: paragraphs.append(t2) else: p_text = child.get_text(strip=True) if p_text: paragraphs.append(p_text) elif name == "table": # 表格保留HTML paragraphs.append(str(child)) elif name is None: # 纯文本 t = str(child).strip() if t: paragraphs.append(t) if paragraphs: text = "\n\n".join(paragraphs) # 如果正文过短，扫描附件链接 if len(text) < 500: links = [] for img in content_div.select("img[src]") if content_div else []: src = img.get("src", "") if src and not src.startswith("http"): src = BASE_URL + src if src.startswith("/") else src links.append(f"[图片] {src}") for a in (content_div.select("a[href]") if content_div else []): href = a.get("href", "") if any(href.lower().endswith(ext) for ext in [".pdf", ".doc", ".docx", ".xls", ".xlsx", ".jpg", ".png", ".gif"]): if not href.startswith("http"): href = BASE_URL + href if href.startswith("/") else href fname = a.get_text(strip=True) or href.split("/")[-1] links.append(f"[附件: {fname}] {href}") if links: text = "\n".join(links) if not text else text + "\n" + "\n".join(links) return title, date_str, text def parse_list_page(html, page_num): """解析列表页，返回 (url, title, date) 元组列表""" soup = BeautifulSoup(html, "html.parser") items = [] rows = soup.select("table.table_list tbody tr") if not rows: # 兼容某些页可能tbody被注释掉 rows = soup.select("table.table_list tr") for row in rows: cells = row.find_all("td") if len(cells) < 3: continue a_tag = cells[1].find("a") if len(cells) > 1 else None if not a_tag: continue href = a_tag.get("href", "").strip() list_title = a_tag.get_text(strip=True) date_cell = cells[2].get_text(strip=True) if len(cells) > 2 else "" if href and list_title: items.append((href, list_title, date_cell)) print(f" 第{page_num}页: {len(items)}条") return items def main(): import argparse parser = argparse.ArgumentParser(description="湖南省生态环境厅-建设项目审批决定公开爬虫") parser.add_argument("--pages", type=int, default=PAGES, help=f"爬取页数(每页25条,默认{PAGES})") args = parser.parse_args() max_pages = args.pages # ---------- 1. 爬取列表页 ---------- print(f"🌐 爬取列表页 1-{max_pages}...") all_items = [] for page in range(1, max_pages + 1): if page == 1: list_url = BASE_URL + LIST_URL else: list_url = f"{BASE_URL}/sthjt/xxgk/xzgs/jsxm/hpgs/spjdgk_2/index_{page}.html" html = fetch_page(list_url) if not html: print(f" [ERROR] 无法获取第{page}页") continue items = parse_list_page(html, page) all_items.extend(items) print(f"\n📋 共获取 {len(all_items)} 条列表数据") # ---------- 2. 去重(按URL) ---------- seen_urls = set() unique_items = [] for href, list_title, date in all_items: if href not in seen_urls: seen_urls.add(href) unique_items.append((href, list_title, date)) print(f"🔍 去重后 {len(unique_items)} 条") # ---------- 3. 爬取详情 ---------- print("\n📄 爬取详情页...") results = [] for idx, (href, list_title, list_date) in enumerate(unique_items, 1): # 构建完整详情URL detail_url = href if href.startswith("http") else (BASE_URL + href) print(f" [{idx}/{len(unique_items)}] {list_title[:30]}...", end=" ") sys.stdout.flush() title, date_str, text = fetch_detail(detail_url) if not title: title = list_title title = title.replace("·", "").replace(" ", "").replace("\xa0", "").replace("\u3000", "").strip() if not date_str: date_str = list_date if text: print(f"✅ {len(text)}字") else: print(f"⚠️ 正文为空") # 即使正文为空也要记录，后面可能会做附件嵌入 results.append({ "title": title, "date": date_str, "content": text, "url": detail_url, "source_url": detail_url, "list_title": list_title, }) # ---------- 4. 入库 ---------- print(f"\n💾 入库 {len(results)} 条到 {DB_PATH}...") if not os.path.exists(DB_PATH): print(f" [ERROR] 数据库不存在: {DB_PATH}") print("\n📊 结果预览:") for r in results[:5]: print(f" {r['date']} | {r['title'][:40]} | {len(r['content'])}字") print(f"\n ...共 {len(results)} 条 (未入库)") return import sqlite3 conn = sqlite3.connect(DB_PATH, timeout=60) c = conn.cursor() added = 0 for r in results: try: c.execute( "INSERT OR IGNORE INTO gov_raw (title, content, publish_date, page_url, source_url, site_name, group_name, industry) VALUES (?,?,?,?,?,?,?,?)", (r["title"], r["content"], r["date"], r["url"], r["source_url"], SITE_NAME, GROUP_NAME, INDUSTRY) ) if c.rowcount > 0: added += 1 except Exception as e: print(f" [ERROR] 入库失败: {e}") conn.commit() # ---------- 5. FTS同步 ---------- if added > 0: print(f"\n🔍 FTS同步 {added} 条...") # 使用subprocess防止sqlite3 WAL损坏 import subprocess as sp script = """ import sqlite3, sys db = sys.argv[1] conn = sqlite3.connect(db) c = conn.cursor() # 获取最新插入的行(按id倒序取added条) added = int(sys.argv[2]) c.execute("SELECT id, title, content FROM gov_raw ORDER BY id DESC LIMIT ?", (added,)) rows = c.fetchall() for rowid, title, content in rows: c.execute("INSERT OR IGNORE INTO gov_search (rowid, title, site_name, summary) VALUES (?,?,?,?)", (rowid, title, '""" + SITE_NAME + """', (content or '')[:500])) conn.commit() c.execute("SELECT COUNT(*) FROM gov_search") print(f"FTS条数: {c.fetchone()[0]}") conn.close() """ result = sp.run(["python3", "-c", script, DB_PATH, str(added)], capture_output=True, text=True, timeout=30) print(f" FTS: {result.stdout.strip()}") if result.stderr: print(f" FTS stderr: {result.stderr[:100]}") else: print("\n ⚠️ 无新增记录，跳过FTS同步") conn.close() # ---------- 6. 输出结果 ---------- print(f"\n📊 结果总览:") empty_count = sum(1 for r in results if not r["content"]) print(f" 总条数: {len(results)}") print(f" 新增入库: {added}") print(f" 空正文: {empty_count}") print(f" 站点: {SITE_NAME}") print(f" 栏目: 建设项目审批决定公开") print(f"\n{'─' * 60}") print(f"{'日期':<14} {'标题':<40} {'字数':>6}") print(f"{'─' * 60}") for r in results[:5]: title_short = r['title'][:38] if len(r['title']) > 38 else r['title'] print(f"{r['date']:<14} {title_short:<40} {len(r['content']):>6}") if len(results) > 5: print(f" ... 共 {len(results)} 条") print(f"{'─' * 60}") if __name__ == "__main__": main()