#!/usr/bin/env python3 # -*- coding: utf-8 -*- """山东联盟化工集团 — 新闻中心爬虫 URL: http://www.leaguechem.com/news/ CMS: 定制PHP站(自研模板, nginx) 列表: ul.bbnxwfl > li > a[title], 6条/页, 共79页473条 分页: /news/p{N}.html (第1页=/news/) 详情: /news/{ID}.html 标题: h1.xypg-detail-title 日期: div.detail-info-time (2026-02-10 10:13:28) 正文: div.xypg-detail-con (内联样式p/span, 保留HTML) 过滤: 电子报外链 book.yunzhan365.com 跳过, 只抓站内 /news/\d+.html """ import requests, sqlite3, re, sys, time from bs4 import BeautifulSoup from urllib.parse import urljoin DB = "/root/search.db" BASE = "http://www.leaguechem.com" LIST_DIR = BASE + "/news" SITE = "联盟化工-新闻中心" GROUP = "企业" SCRIPT = "crawl_leaguechem.py" MAX_PAGES = 79 HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36", "Referer": LIST_DIR + "/", } def log(msg): print(msg, file=sys.stderr, flush=True) def fetch(url, retries=3): for attempt in range(retries): try: r = requests.get(url, headers=HEADERS, timeout=30) r.encoding = "utf-8" return r.text except Exception as e: log(f" [WARN] {url} 请求失败({attempt+1}/{retries}): {e}") time.sleep(2) return "" def crawl_list(pg): """抓列表页, 返回 [{title, url, date}]; 过滤电子报外链""" if pg <= 1: url = LIST_DIR + "/" else: url = f"{LIST_DIR}/p{pg}.html" html = fetch(url) if not html: return [] soup = BeautifulSoup(html, "html.parser") ul = soup.find("ul", class_="bbnxwfl") if not ul: return [] items = [] for li in ul.find_all("li"): a = li.find("a", href=True) if not a: continue href = a["href"] if not href.startswith("http"): href = urljoin(BASE, href) # 只抓站内新闻详情, 跳过电子报外链(yunzhan365)等 if not re.search(r"/news/\d+\.html$", href): continue title = a.get("title", "") or a.get_text(strip=True) # strip ·  实体前缀 title = re.sub(r"^(?:·|·| |\s|•)+", "", title).strip() # 日期: div.xwsj h2=日 h3=年月 date = "" xsj = li.find("div", class_="xwsj") if xsj: h3 = xsj.find("h3") h2 = xsj.find("h2") if h3 and h2: ym = h3.get_text(strip=True) d = h2.get_text(strip=True).zfill(2) m = re.match(r"(\d{4})-(\d{2})", ym) if m: date = f"{m.group(1)}-{m.group(2)}-{d}" items.append({"title": title, "url": href, "date": date}) return items def extract_detail(url): """返回 (title, date, content)""" html = fetch(url) if not html: return None, None, None soup = BeautifulSoup(html, "html.parser") # 标题 title = "" h1 = soup.find("h1", class_="xypg-detail-title") if h1: title = h1.get_text(strip=True) # 日期 date = "" dt = soup.find("div", class_="detail-info-time") if dt: m = re.search(r"(20\d{2}-\d{2}-\d{2})", dt.get_text()) if m: date = m.group(1) # 正文 content = "" dc = soup.find("div", class_="xypg-detail-con") if dc: # 附件: a[href] 指向文件 attachments = [] seen_att = set() for a in dc.find_all("a", href=True): href = a["href"] if re.search(r"\.(pdf|doc|docx|xls|xlsx|zip|rar|wps|txt)$", href, re.I): if not href.startswith("http"): href = urljoin(BASE, href) name = a.get_text(strip=True) or "附件" if href not in seen_att: seen_att.add(href) attachments.append({"name": name, "url": href}) parts = [] # 排除相关阅读/二维码/分享区块 for node in dc.find_all(["p", "div", "table"], recursive=True): # 关键: 跳过表格内部节点(表格已整体保留, 内部 p/div 不拍平, 避免重复) if node.name != "table" and node.find_parent("table"): continue if node.name == "div" and node.get("id") in ("xgwdP", "qrcode"): continue if node.find_parent("div", id="xgwdP"): continue # 跳过含已提取附件的p(避免正文+附件区重复) if node.name == "p": def _att_url(el): u = "" if el.name == "a": u = el.get("href", "") if u and not u.startswith("http"): u = urljoin(BASE, u) return u has_att = any( _att_url(el) in seen_att for el in node.find_all(["a"]) if _att_url(el)) if has_att: continue html_str = str(node).strip() if not html_str: continue # 去嵌套: p 内再套 p 只取最内层文本段落 if node.name == "p": inner_p = node.find("p") if inner_p: html_str = str(inner_p).strip() if node.name == "div" and node.find_parent("div") and not node.find(["p", "table"]): parts.append(html_str) elif node.name in ("p", "table"): parts.append(html_str) content = "\n\n".join(parts).strip() # 去重: 相同段落只保留一次 + 过滤空/纯
段落(含 span data-type 包裹) uniq_parts = [] for part in content.split("\n\n"): part_stripped = part.strip() if not part_stripped: continue # 剥标签后无文本(仅空白/br) → 过滤 if not re.sub(r"<[^>]+>", "", part_stripped).strip(): continue if part_stripped not in uniq_parts: uniq_parts.append(part_stripped) content = "\n\n".join(uniq_parts).strip() if attachments: att_links = "\n".join(f'

{att["name"]}

' for att in attachments) content = content + "\n\n" + att_links if content else att_links return title, date, content def main(): import argparse ap = argparse.ArgumentParser() ap.add_argument("--pages", type=int, default=1) args = ap.parse_args() pages = max(1, min(args.pages, MAX_PAGES)) conn = sqlite3.connect(DB, timeout=15) conn.execute("PRAGMA busy_timeout=10000") c = conn.cursor() c.execute("CREATE TABLE IF NOT EXISTS gov_raw (id INTEGER PRIMARY KEY, site_name TEXT, source_url TEXT, page_url TEXT, title TEXT, publish_date TEXT, date_rank INTEGER DEFAULT 0, summary TEXT, status TEXT, category TEXT DEFAULT '', visits INTEGER DEFAULT 0, content TEXT DEFAULT '', tags TEXT DEFAULT '', industry TEXT DEFAULT 'other', attachments TEXT DEFAULT '', group_name TEXT, has_table INTEGER DEFAULT 0, script_name TEXT DEFAULT '')") c.execute("CREATE VIRTUAL TABLE IF NOT EXISTS gov_search USING fts5(title, site_name, summary, tokenize='trigram')") total_new = total_dup = total_skip = 0 seen = set() for pg in range(1, pages + 1): log(f"--- 第{pg}页 ---") items = crawl_list(pg) if not items: log("列表为空, 停止翻页") break log(f"找到{len(items)}条") for item in items: if item["url"] in seen: continue seen.add(item["url"]) c.execute("SELECT id FROM gov_raw WHERE page_url=?", (item["url"],)) if c.fetchone(): total_dup += 1 continue log(f" [{item['date']}] {item['title'][:45]}") title, date, content = extract_detail(item["url"]) if not title: title = item["title"] if not date: date = item["date"] if not content or len(content) < 50: log(f" 正文过短({len(content) if content else 0}), 跳过") total_skip += 1 continue summary = re.sub(r"<[^>]+>", "", content)[:200] summary = re.sub(r"\s+", " ", summary).strip() has_table = 1 if re.search(r"