#!/usr/bin/env python3 """ EIAFans (环评爱好者) 爬虫 — Discuz! X3.4 论坛 ==================== 全量/增量两种模式： python3 crawl_eiafans.py # 增量（默认：只爬最新10页） python3 crawl_eiafans.py --full # 全量：2078页 python3 crawl_eiafans.py --pages=50 # 指定页数 python3 crawl_eiafans.py --sync # 同步已有数据到服务器 python3 crawl_eiafans.py --refetch # 重抓存量详情页（修复标题省略号+正文占位符） ⚠️ 2026-08-31 重大更新：详情页 JS 反爬已失效（实测可抓正文） - 正文：从详情页 提取，保留

段落结构 + 附件转 - 标题：列表页标题被 Discuz 服务端截断（带 ...），从正文首段《》重建完整标题 - 兜底：详情页 404/游客无权限/反爬 → 保留占位符+列表页标题 数据流：本地 quality_results.db → SSH → 服务器 search.db """ import sys, os, re, json, sqlite3, hashlib, time, ssl, urllib.request, base64, subprocess from datetime import datetime from urllib.parse import urljoin from html import unescape # ─── 配置 ─── BASE_DIR = os.path.dirname(os.path.abspath(__file__)) QUALITY_DB = os.path.join(BASE_DIR, "quality_results.db") SERVER_SSH = "root@1.94.217.116" SERVER_DB = "/root/quality_results.db" SERVER_SEARCH_DB = "/root/search.db" BASE_URL = "http://www.eiafans.com" FORUM_URL = "http://www.eiafans.com/forum-64-{page}.html" TOTAL_PAGES = 2078 PLACEHOLDER_CONTENT = "⚠️ 该数据来自环评爱好者论坛，详情页有反爬保护，请点击链接跳转原站查看正文内容。" PLACEHOLDER_SUMMARY = "环评爱好者论坛公示信息 — 详情请跳转原站查看" # ─── HTTP 工具 ─── CTX = ssl._create_unverified_context() HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "Chrome/125.0.0.0 Safari/537.36", "Referer": "http://www.eiafans.com/forum-64-1.html", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", } def http_get(url, timeout=20): """返回 (text, final_url) — GBK 自动转 UTF-8""" req = urllib.request.Request(url, headers=HEADERS) try: resp = urllib.request.urlopen(req, timeout=timeout, context=CTX) raw = resp.read() text = raw.decode("gbk", errors="replace") return text, resp.geturl() except Exception as e: return None, str(e) # ─── 列表页解析 ─── def parse_list_page(html): """从 Discuz! 列表页解析所有帖子，提取标题、URL、日期""" threads = [] seen_tids = set() # 找到 normalthread 的区域（排除 sticky/pinned 公告） # 匹配 tbody id="normalthread_{tid}" for m in re.finditer( r']*id="normalthread_(\d+)"[^>]*>(.*?)', html, re.DOTALL ): tid = m.group(1) block = m.group(2) if tid in seen_tids: continue seen_tids.add(tid) # 标题：class="s xst" title_m = re.search(r'class="s xst"[^>]*>(.*?)', block, re.DOTALL) if not title_m: continue title = re.sub(r'<[^>]+>', '', title_m.group(1)).strip() if not title or len(title) < 5: continue # URL：thread-{tid}-1-{page}.html url_m = re.search(r'href="(thread-\d+-1-\d+\.html)"', block) thread_url = urljoin(BASE_URL, url_m.group(1)) if url_m else f"{BASE_URL}/thread-{tid}-1-1.html" # 日期：从 中的 提取 date = "" date_m = re.search(r']*>\s*]*>\s*(\d{4}-\d{1,2}-\d{1,2})', block) if date_m: date = date_m.group(1) # 备选日期：无 包裹的日期格式 if not date: date_m = re.search(r'(\d{4}-\d{1,2}-\d{1,2})\s+\d{2}:\d{2}', block) if date_m: date = date_m.group(1) threads.append({ "tid": tid, "url": f"{BASE_URL}/thread-{tid}-1-1.html", "title": title, "date": date, }) return threads # ─── 详情页解析（2026-08-31 新增） ─── def parse_detail_page(html): """ 从详情页提取 (完整标题, 正文HTML, 发布日期) 返回 None 表示详情页不可用（404/游客无权限/反爬challenge） """ # 反爬 challenge 检测 if "Ka7" in html and "return Ka7" in html: return None # 游客无权限/不存在 for kw in ["无法进行此操作", "指定的主题不存在", "已被删除"]: if kw in html: return None # 正文边界： 到

]*>(.*?)(
]*>(.*?)', html, re.DOTALL) body_html = m.group(1) if m else "" if not body_html: return None # 发布日期：authorposton 发表于 date = "" dm = re.search(r'authorposton\d*[^>]*>\s*发表于\s*(\d{4}-\d{1,2}-\d{1,2})', html) if dm: date = dm.group(1) # 完整标题：优先 thread_subject（Discuz 帖子标题，多数情况完整） # 若被服务端截断（带 ...），从正文《》重建；再不行用正文首句 full_title = "" ts = re.search(r'id="thread_subject"[^>]*>(.*?)]+>', '', ts.group(1))).strip() # 去掉列表页常见的 [验收公示]/[环评公示] 前缀（若混入） tst = re.sub(r'^\[[^\]]+\]\s*', '', tst) if tst and not re.search(r'\.\.\.$|…$', tst): full_title = tst if not full_title: # 正文《》重建：非贪婪匹配避免嵌套书名号（〈〉）吞掉后续候选 books = re.findall(r'《(.{10,200}?)》', body_html, re.DOTALL) for b in books: clean = re.sub(r'\s+', '', re.sub(r'<[^>]+>', '', b)) # 过滤法规/标准名（如《建设项目环境保护管理条例》《土壤环境质量...标准（试行）》） if re.search(r'(条例|办法|规定|决定|公告|实施细则|导则|标准(（|\(|$))', clean) and len(clean) < 35: continue # 取最长的候选（项目全名通常最长，如"公司+产品+项目+竣工环境保护验收监测报告"） if len(clean) > len(full_title): full_title = clean if not full_title: # 正文首句兜底：取正文纯文本，优先"项目名称："后内容；否则取"位于/地处/坐落"前的项目主体 text = re.sub(r'<[^>]+>', ' ', body_html) text = re.sub(r'\s+', ' ', text).strip() m1 = re.search(r'项目名称[：:]\s*《?([^《》\n，。,]{8,80})》?', text) if m1: full_title = m1.group(1).strip() else: # 首句："XX项目（XX地块）位于/地处..." m2 = re.search(r'^([^。\n]{8,90}?)(?:位于|地处|坐落|建设地点[：:]|，位于)', text) if m2: cand = m2.group(1).strip() # 需含项目/地块/公司特征，避免取到法规引用开头 if re.search(r'(项目|地块|公司|厂|工程|报告|公示)', cand): full_title = cand if not full_title: # 正文首段纯文本 = 完整标题（如"XX公司XX项目...验收监测报告全本公示，详见附件"） # 取第一个

/

段落纯文本；无标签时取整段纯文本首句 first_para = "" pm = re.search(r'<(?:p|div)[^>]*>(.*?)', body_html, re.DOTALL) if pm: first_para = re.sub(r'<[^>]+>', '', pm.group(1)) first_para = re.sub(r'\s+', '', first_para).strip() else: # 无段落标签：整段纯文本首句 first_para = re.sub(r'<[^>]+>', ' ', body_html) first_para = re.sub(r'\s+', '', first_para).strip() first_para = first_para.split('。')[0] # 去掉法规引用开头（根据/按照/依据/依照...） fm = re.search(r'^(?:根据|按照|依据|依照|遵照)[^，。]{0,60}[，。]?(.*)', first_para) if fm and len(fm.group(1)) >= 10: first_para = fm.group(1) # 去掉尾随"详见附件"/"全本公示"后多余标点，截断到第一个句号/逗号（保留标题主体） em = re.match(r'^(.{8,120}?)(?:[，,。]|详见附件|$)', first_para) if em: cand = em.group(1).strip() if re.search(r'(项目|地块|公司|厂|工程|报告|公示|监测|验收)', cand) and len(cand) >= 12: full_title = cand # 正文 HTML 清理： 标签保留文字、
转

、清理 Discuz 残留 cleaned = clean_body_html(body_html) return full_title, cleaned, date def clean_body_html(body_html): """ 清理详情页正文 HTML： -

→

（保留段落结构，勿拍平） - 剥掉标签保留文字 - 清理空 div、多余空白 - 附件区在正文外（pattl），由调用方单独处理 """ # div → p html = re.sub(r']*align="left"[^>]*>', '

', body_html) html = re.sub(r']*>', '

', html) html = re.sub(r'

', '

', html) # 去 font 标签（保留文字） html = re.sub(r']*>', '', html) # br → 保留 html = re.sub(r'', '\n', html) # 清空段落 html = re.sub(r'

\s*

', '', html) # 合并连续空白 html = re.sub(r'[ \t]+', ' ', html) html = re.sub(r'\n\s*\n+', '\n', html) html = html.strip() return html def extract_attachments(html): """ 从详情页提取附件列表：[(附件名, 完整下载URL)] 附件在
→ → """ atts = [] for m in re.finditer( r']*>(.*?)', html, re.DOTALL ): url = m.group(1).replace('&', '&') name = re.sub(r'<[^>]+>', '', m.group(2)).strip() if name: atts.append((name, f"{BASE_URL}/{url}")) return atts def fetch_detail(tid): """ 抓取详情页 → (完整标题, 正文HTML, 附件HTML, 日期) 失败返回 None """ url = f"{BASE_URL}/thread-{tid}-1-1.html" html, _ = http_get(url, timeout=25) if not html: return None parsed = parse_detail_page(html) if not parsed: return None full_title, body_clean, date = parsed atts = extract_attachments(html) return full_title, body_clean, atts, date # ─── 数据库操作 ─── def init_db(): conn = sqlite3.connect(QUALITY_DB, timeout=60) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA synchronous=NORMAL") conn.executescript(""" CREATE TABLE IF NOT EXISTS quality_results ( id INTEGER PRIMARY KEY AUTOINCREMENT, site_name TEXT NOT NULL, domain TEXT, title TEXT, url TEXT UNIQUE, content TEXT, publish_date TEXT, summary TEXT, crawled_at TEXT DEFAULT (datetime('now','localtime')) ); CREATE TABLE IF NOT EXISTS eiafans_threads ( tid TEXT PRIMARY KEY, title TEXT, url TEXT, publish_date TEXT, crawled_at TEXT DEFAULT (datetime('now','localtime')) ); """) conn.commit() conn.close() import queue, threading _write_queue = queue.Queue() _write_thread_started = False def _write_worker(): while True: task = _write_queue.get() if task is None: break func, args, kwargs = task try: func(*args, **kwargs) except Exception as e: print(f" ❌ DB写入错误: {e}") _write_queue.task_done() def _ensure_write_thread(): global _write_thread_started if not _write_thread_started: t = threading.Thread(target=_write_worker, daemon=True) t.start() _write_thread_started = True def _db_save_quality(name, domain, title, url, content, date, summary): conn = sqlite3.connect(QUALITY_DB, timeout=60) conn.execute("PRAGMA journal_mode=WAL") try: conn.execute("DELETE FROM quality_results WHERE url=?", (url,)) conn.execute( "INSERT INTO quality_results (site_name, domain, title, url, content, publish_date, summary) VALUES (?,?,?,?,?,?,?)", (name, domain, title, url, content or "", date or "", summary or ""), ) conn.commit() except Exception as e: print(f" ❌ DB: {e}") finally: conn.close() def _db_mark_thread(tid, title, url, date): conn = sqlite3.connect(QUALITY_DB, timeout=60) conn.execute("PRAGMA journal_mode=WAL") try: conn.execute( "INSERT OR IGNORE INTO eiafans_threads (tid, title, url, publish_date) VALUES (?,?,?,?)", (tid, title, url, date), ) conn.commit() except Exception as e: print(f" ❌ DB: {e}") finally: conn.close() def _db_update_thread(tid, title, url, date): conn = sqlite3.connect(QUALITY_DB, timeout=60) conn.execute("PRAGMA journal_mode=WAL") try: conn.execute( "UPDATE eiafans_threads SET title=?, url=?, publish_date=? WHERE tid=?", (title, url, date, tid), ) conn.commit() except Exception as e: print(f" ❌ DB: {e}") finally: conn.close() def is_thread_crawled(tid): conn = sqlite3.connect(QUALITY_DB, timeout=60) try: conn.execute("PRAGMA journal_mode=WAL") r = conn.execute("SELECT 1 FROM eiafans_threads WHERE tid=?", (tid,)).fetchone() return r is not None finally: conn.close() def mark_thread_crawled(tid, title, url, date): _ensure_write_thread() _write_queue.put((_db_mark_thread, (tid, title, url, date), {})) def update_thread(tid, title, url, date): _ensure_write_thread() _write_queue.put((_db_update_thread, (tid, title, url, date), {})) def save_to_quality(name, domain, title, url, content, date, summary): _ensure_write_thread() _write_queue.put((_db_save_quality, (name, domain, title, url, content, date, summary), {})) # ─── 同步到服务器 ─── def sync_to_server(): """将本地数据直接写入 search.db（服务器本地模式）""" print("\n📤 同步到 search.db...") conn = sqlite3.connect(QUALITY_DB, timeout=60) # ⚠️ 2026-08-11 修复: 之前误读 crawl_results 表（佛冈/新疆/荆州共用表，120条被误标成 eiafans） # 本脚本写入的是 quality_results 表，且必须 WHERE domain 过滤只同步自己的数据 rows = conn.execute( "SELECT title, url, content, publish_date, summary FROM quality_results WHERE domain='www.eiafans.com' ORDER BY id" ).fetchall() conn.close() if not rows: print(" 本地没有数据") return dst = sqlite3.connect("/root/search.db", timeout=60) dst.execute("PRAGMA journal_mode=WAL") site_name = "eiafans" new_count = 0 upd_count = 0 for r in rows: title, url, content, pub_date, summary = r try: # 已存在则 UPDATE（重抓修复），不存在则 INSERT exists = dst.execute("SELECT id FROM gov_raw WHERE page_url=? AND site_name=?", (url, site_name)).fetchone() if exists: dst.execute( "UPDATE gov_raw SET title=?, content=?, publish_date=?, summary=? WHERE id=?", (title, (content or "")[:500000], pub_date or "", (summary or "")[:300], exists[0]) ) upd_count += 1 else: dst.execute( "INSERT INTO gov_raw (title, page_url, content, publish_date, summary, site_name, tags, industry) " "VALUES (?,?,?,?,?,?,?,?)", (title, url, (content or "")[:500000], pub_date or "", (summary or "")[:300], site_name, "", "环保") ) new_count += 1 except Exception as e: print(f" Error: {e}") if new_count > 0 or upd_count > 0: dst.commit() # Update FTS: 全量重建该站 FTS（内容变了必须重建） # QC20260926 去掉手写 gov_search 整站删除(抢锁源; FTS 由 gov_raw 触发器维护) # dst.execute("DELETE FROM gov_search WHERE site_name=?", (site_name,)) dst.execute( "INSERT OR REPLACE INTO gov_search(rowid,title,site_name,summary) " "SELECT r.id,r.title,r.site_name,r.summary FROM gov_raw r WHERE r.site_name=?", (site_name,)) dst.commit() total = dst.execute( "SELECT COUNT(*) FROM gov_raw WHERE site_name=?", (site_name,)).fetchone()[0] dst.close() print(f" OK 新增{new_count}/更新{upd_count} 条同步到 search.db (DB共{total}条)") # ─── 站点爬取 ─── def crawl_one_page(page_num, out_file=None, fetch_detail_pages=True): """爬取单页列表 — 可选抓详情页正文""" list_url = FORUM_URL.format(page=page_num) html, _ = http_get(list_url) if not html: print(f" ❌ 列表页 {page_num} 失败") return 0 threads = parse_list_page(html) if not threads: print(f" ⚠️ 列表页 {page_num} 无帖子") return 0 # 修正URL：始终指向 thread-{tid}-1-1.html for t in threads: t["url"] = f"{BASE_URL}/thread-{t['tid']}-1-1.html" count = 0 for t in threads: is_new = not is_thread_crawled(t["tid"]) # 详情页抓取（2026-08-31: 反爬已失效） detail = None if fetch_detail_pages: detail = fetch_detail(t["tid"]) if detail: full_title, body_clean, atts, det_date = detail # 完整标题优先用重建的 title = full_title if full_title else t["title"] # 正文：正文HTML + 附件段 content = body_clean if atts: att_html = "\n".join( f'

{n}

' for n, u in atts ) content = f"{content}\n{att_html}" if content else att_html date = det_date or t["date"] summary = f"{PLACEHOLDER_SUMMARY.replace(' — 详情请跳转原站查看', '')}\n📎 原帖链接：{t['url']}" if not content else f"环评爱好者论坛公示信息\n📎 原帖链接：{t['url']}" save_to_quality( "环评爱好者", "www.eiafans.com", title, t["url"], content, date, summary, ) if is_new: mark_thread_crawled(t["tid"], title, t["url"], date) else: update_thread(t["tid"], title, t["url"], date) count += 1 print(f" ✅ {title[:40]}... ({date})") else: # 详情页不可用 → 占位符兜底 save_to_quality( "环评爱好者", "www.eiafans.com", t["title"], t["url"], PLACEHOLDER_CONTENT, t["date"], f"{PLACEHOLDER_SUMMARY}\n📎 原帖链接：{t['url']}", ) if is_new: mark_thread_crawled(t["tid"], t["title"], t["url"], t["date"]) else: update_thread(t["tid"], t["title"], t["url"], t["date"]) count += 1 print(f" ⚠️ {t['title'][:40]}... (详情不可用)") else: # 纯列表模式（旧行为） save_to_quality( "环评爱好者", "www.eiafans.com", t["title"], t["url"], PLACEHOLDER_CONTENT, t["date"], f"{PLACEHOLDER_SUMMARY}\n📎 原帖链接：{t['url']}", ) if is_new: mark_thread_crawled(t["tid"], t["title"], t["url"], t["date"]) else: update_thread(t["tid"], t["title"], t["url"], t["date"]) count += 1 print(f" ✅ {t['title'][:35]}...{(' | ' + t['date']) if t['date'] else ''}") if out_file: out_file.write(json.dumps({ "site_name": "环评爱好者", "title": t["title"], "publish_date": t["date"], "page_url": t["url"], "content": t["title"], "content_text": f"{PLACEHOLDER_SUMMARY} — {t['url']}", }, ensure_ascii=False) + "\n") out_file.flush() return count def crawl(pages_to_crawl, page_start=1, fetch_detail_pages=True): """全量/增量爬取所有页""" conn = sqlite3.connect(QUALITY_DB, timeout=60) existing = conn.execute("SELECT COUNT(*) FROM eiafans_threads").fetchone()[0] conn.close() print(f"📊 已有记录: {existing} 个帖子") start_time = time.time() total_new = 0 for page in range(page_start, page_start + pages_to_crawl): if page > TOTAL_PAGES: print(f"\n⏹ 已达最大页数 {TOTAL_PAGES}") break print(f"\n📃 第 {page}/{TOTAL_PAGES} 页...", end=" ") sys.stdout.flush() n = crawl_one_page(page, fetch_detail_pages=fetch_detail_pages) total_new += n if n > 0: print(f" ➕+{n} 条") else: print(" ➖ 无新增") # 进度报告 if page % 100 == 0 or page == page_start + pages_to_crawl - 1: elapsed = time.time() - start_time rate = total_new / max(elapsed, 1) print(f"\n📊 【进度】第{page}页 | 累计{total_new}条 | {elapsed:.0f}s | {rate:.0f}条/秒") elapsed = time.time() - start_time print(f"\n{'='*50}") print(f"✅ 完成！共处理 {min(pages_to_crawl, TOTAL_PAGES - page_start + 1)} 页，新增 {total_new} 条") print(f"⏱ 耗时: {elapsed:.0f}秒 ({elapsed/60:.1f}分钟)") print(f"⚡ 速率: {total_new/max(elapsed,1):.1f} 条/秒") return total_new def refetch_existing(): """重抓存量详情页 — 修复标题省略号 + 正文占位符""" print("\n🔄 重抓存量详情页...") conn = sqlite3.connect(QUALITY_DB, timeout=60) rows = conn.execute( "SELECT tid FROM eiafans_threads ORDER BY tid" ).fetchall() conn.close() print(f"📊 存量帖子: {len(rows)} 个") ok = 0 fail = 0 for (tid,) in rows: detail = fetch_detail(tid) if detail: full_title, body_clean, atts, det_date = detail title = full_title if full_title else None content = body_clean if atts: att_html = "\n".join( f'

{n}

' for n, u in atts ) content = f"{content}\n{att_html}" if content else att_html # 更新 quality_results conn = sqlite3.connect(QUALITY_DB, timeout=60) conn.execute("PRAGMA journal_mode=WAL") r = conn.execute("SELECT title, url, publish_date FROM eiafans_threads WHERE tid=?", (tid,)).fetchone() old_title, url, old_date = r new_title = title or old_title new_date = det_date or old_date conn.execute( "UPDATE quality_results SET title=?, content=?, publish_date=? WHERE url=?", (new_title, content or "", new_date, url), ) conn.execute( "UPDATE eiafans_threads SET title=?, publish_date=? WHERE tid=?", (new_title, new_date, tid), ) conn.commit() conn.close() update_thread(tid, new_title, url, new_date) ok += 1 if ok <= 5 or ok % 20 == 0: print(f" ✅ {tid}: {new_title[:45]}...") else: fail += 1 if fail <= 5: print(f" ⚠️ {tid}: 详情不可用（保留占位符）") time.sleep(0.8) print(f"\n🔄 重抓完成: 成功 {ok}, 失败 {fail}") return ok # ─── 主入口 ─── if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="环评爱好者爬虫") parser.add_argument("--full", action="store_true", help="全量爬取所有2078页") parser.add_argument("--pages", type=int, default=10, help="爬取页数 (默认10)") parser.add_argument("--start", type=int, default=1, help="起始页数 (默认1)") parser.add_argument("--sync", action="store_true", help="仅同步已有数据到服务器") parser.add_argument("--list-only", action="store_true", help="纯列表模式（不抓详情页）") parser.add_argument("--refetch", action="store_true", help="重抓存量详情页") args = parser.parse_args() if args.sync: sync_to_server() sys.exit(0) init_db() if args.refetch: ok = refetch_existing() if ok > 0: sync_to_server() sys.exit(0) pages = TOTAL_PAGES if args.full else args.pages fetch_details = not args.list_only print(f"\n{'='*50}") print(f"🚀 环评爱好者爬虫 — {'全量' if args.full else f'增量({pages}页)'}") print(f"{'='*50}") print(f" 模式: {'全量' if args.full else '增量'}") print(f" 页数: {pages} (第{args.start}页起)") print(f" 详情页: {'抓取正文+重建标题' if fetch_details else '仅列表'}") print(f"{'='*50}") total_new = crawl(pages, page_start=args.start, fetch_detail_pages=fetch_details) if total_new > 0: sync_to_server() print(f"\n{'='*50}") print(f"🏁 完成")