#!/usr/bin/env python3 # -*- coding: utf-8 -*- """琅琊新闻网 — 公示公告爬虫 URL: http://www.langya.cn/lyxw/gsgg/ CMS: linyi.gov.cn 系统资源 (pagedown/sys.css) 列表: div.article-item > h3.article-title > a[title] 第1页: index.html 第N页: index_{N}.html 共19页 (每页条数不固定 12/9/4/2, 20页起重复返回末尾) 详情: /a/4033/{id}/{longid}.html 标题: meta ArticleTitle 日期: meta PubDate (2026-07-29 16:08) 来源: meta ContentSource 正文: div.v_news_content (多p段落, 环评公示为主) """ import requests, sqlite3, re, sys, time from bs4 import BeautifulSoup from urllib.parse import urljoin DB = "/root/search.db" BASE = "http://www.langya.cn" LIST_DIR = "/lyxw/gsgg/" SITE = "琅琊新闻网-公示公告" GROUP = "山东" SCRIPT = "crawl_langya_gsgg.py" MAX_PAGES = 19 ATTACH_RE = re.compile(r"\.(pdf|doc|docx|xls|xlsx|zip|rar|wps|txt|et|ofd)$", re.I) HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36", } def log(msg): print(msg, file=sys.stderr, flush=True) def crawl_list(pg): """抓列表页, 返回 [{title, url, date}]""" if pg == 1: url = BASE + LIST_DIR + "index.html" else: url = BASE + LIST_DIR + f"index_{pg-1}.html" try: r = requests.get(url, headers=HEADERS, timeout=30) r.encoding = "utf-8" except Exception as e: log(f" [WARN] 第{pg}页请求失败: {e}") return [] soup = BeautifulSoup(r.text, "html.parser") items = [] for item in soup.find_all("div", class_="article-item"): h3 = item.find("h3", class_="article-title") if not h3: continue a = h3.find("a", href=True) if not a: continue href = a["href"].strip() if not href or href == "#" or "javascript" in href: continue if not href.startswith("http"): href = urljoin(BASE + LIST_DIR, href) title = a.get("title", "").strip() or a.get_text(strip=True) date = "" meta_el = item.find("div", class_=re.compile("article-meta")) if meta_el: dm = re.search(r"(20\d{2}-\d{2}-\d{2})", meta_el.get_text()) if dm: date = dm.group(1) items.append({"title": title, "url": href, "date": date}) return items def extract_detail(url): """返回 (title, date, content)""" html = "" for attempt in range(3): try: r = requests.get(url, headers=HEADERS, timeout=30) r.encoding = "utf-8" html = r.text if len(html) > 2000: break log(f" [WARN] 详情响应过小({len(html)}), 第{attempt+1}次重试") except Exception as e: log(f" [WARN] 详情失败 {url}: {e}") time.sleep(2) if not html: return None, None, None soup = BeautifulSoup(html, "html.parser") # 标题 title = "" mt = soup.find("meta", attrs={"name": "ArticleTitle"}) if mt and mt.get("content"): title = mt["content"].strip() if not title: h = soup.find("h1") if h: title = h.get_text(strip=True) # 日期 date = "" md = soup.find("meta", attrs={"name": "PubDate"}) if md and md.get("content"): m = re.search(r"(20\d{2}-\d{2}-\d{2})", md["content"]) if m: date = m.group(1) # 正文 content = "" dc = soup.find("div", class_="v_news_content") if not dc: dc = soup.find("div", id=re.compile("^content_")) if dc: # 附件 attachments = [] seen_att = set() for a in dc.find_all("a", href=True): href = a["href"].strip() if ATTACH_RE.search(href): if not href.startswith("http"): href = urljoin(BASE, href) name = a.get("title", "").strip() or 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): # 跳过表格内部节点(表格整体保留) if node.name != "table" and node.find_parent("table"): continue # 跳过含已提取附件的p(避免正文+附件区重复) if node.name == "p": has_att = any( a.get("href", "").strip() in seen_att for a in node.find_all(["a"], href=True)) if has_att: continue html_str = str(node).strip() if not html_str: continue # 去嵌套: 外层p包含表格则跳过; p内再套p只取最内层 if node.name == "p": if node.find("table"): continue inner_p = node.find("p", recursive=False) if inner_p: html_str = str(inner_p).strip() if node.name == "div" and node.find_parent("div") and not node.find(["p", "table"]): continue if node.name in ("p", "table"): parts.append(html_str) content = "\n\n".join(parts).strip() # 去重 + 过滤空段落 uniq_parts = [] for part in content.split("\n\n"): part_stripped = part.strip() if not part_stripped: continue 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(re.sub(r"<[^>]+>", "", content).strip()) < 10: 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"