#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 第十三师新星市 - 公示公告（含环评/生态环境） https://www.btnsss.gov.cn/info/iList.jsp?node_id=GKxxs&isSd=false&cat_id=12051 JS分页: infoSize=20, pageCount=2, 15条/页, 共~30条 """ import sys, os, re, time, json, requests from bs4 import BeautifulSoup SEARCH_DB = os.getenv("SEARCH_DB", "/mnt/data/search.db") HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} DOMAIN = "https://www.btnsss.gov.cn" LIST_URL = "https://www.btnsss.gov.cn/info/iList.jsp?node_id=GKxxs&isSd=false&cat_id=12051" SITE_NAME = "btnsss_gsgg" MAX_PAGES = 2 def extract_table_tags(html_str): tbl_soup = BeautifulSoup(html_str, "html.parser") tables = tbl_soup.find_all("table") replacements = [] for i, tb in enumerate(tables): key = "___TBL_%d___" % i replacements.append((key, str(tb))) tb.replace_with(key) return str(tbl_soup), replacements def fetch(url, encoding="utf-8"): try: r = requests.get(url, headers=HEADERS, timeout=30) r.encoding = encoding return r.text except Exception as e: print(" FETCH ERROR %s: %s" % (url, e), file=sys.stderr) return None def parse_list_items(html): items = [] soup = BeautifulSoup(html, "lxml") # Container: find ul with li > span + a.gk-br for ul in soup.find_all("ul"): for li in ul.find_all("li"): a = li.find("a", class_="gk-br") sp = li.find("span") if not a or not sp or not a.get("href"): continue href = a["href"].strip() if not href.startswith("/gk/"): continue title = a.get_text(strip=True) # Remove the bullet prefix title = re.sub(r"^[•·\s]+", "", title).strip() if not title or len(title) < 5: continue date = sp.get_text(strip=True)[:10] # "2026-06-25 09:54" -> "2026-06-25" full_url = DOMAIN + href items.append({ "url": full_url, "title": title, "date": date, }) return items def get_content(url, title_from_list): html = fetch(url) if not html: return "", "", [], title_from_list soup = BeautifulSoup(html, "lxml") # 标题 full_title = title_from_list con_tt = soup.select_one("div.con-tt") if con_tt: t = con_tt.get_text(strip=True) if t: full_title = t # 日期 - 从元数据表提取 pub_date = "" file_table = soup.select_one("div.file-table") if file_table: m = re.search(r"\u53d1\u5e03\u65e5\u671f.*?]*>(\d{4}[-/]\d{2}[-/]\d{2})", str(file_table), re.DOTALL) if m: pub_date = m.group(1).replace("/", "-") if not pub_date: m = re.search(r"(\d{4}[-/]\d{2}[-/]\d{2})\s*\d{2}:\d{2}", html) if m: pub_date = m.group(1).replace("/", "-") # 正文 body = soup.select_one("div.con-nr") content_html = "" attachments = [] if body: # 附件链接 for a_tag in body.find_all("a", href=True): h = a_tag["href"].strip() if re.search(r"\.(docx?|pdf|pptx?|xlsx?|jpg|png|gif)($|\?)", h, re.IGNORECASE): fname = a_tag.get_text(strip=True) or h.split("/")[-1] attach_url = h if h.startswith("http") else DOMAIN + h attachments.append({"url": attach_url, "name": fname}) md_link = "[%s](%s)" % (fname, attach_url) a_tag.replace_with(md_link) # 提取表格 body_html = str(body) body_html, table_markers = extract_table_tags(body_html) body_soup = BeautifulSoup(body_html, "html.parser") for tag in body_soup.find_all(["script", "style"]): tag.decompose() for tag in body_soup.find_all(["span", "font", "b", "strong", "i", "em", "u"]): tag.unwrap() for p in body_soup.find_all("p"): p.insert(0, "\n\n") p.unwrap() for d in body_soup.find_all("div"): d.insert(0, "\n") d.unwrap() for br in body_soup.find_all("br"): br.replace_with("\n") content_html = body_soup.get_text(separator="", strip=False) content_html = re.sub(r"\n{3,}", "\n\n", content_html).strip() for key, tbl_html in table_markers: content_html = content_html.replace(key, "\n\n" + tbl_html + "\n\n") if not content_html or len(content_html.strip()) < 20: content_html = '

%s

' % (url, full_title) return content_html, pub_date, attachments, full_title def main(): import sqlite3 conn = sqlite3.connect(SEARCH_DB, timeout=60) conn.execute("PRAGMA journal_mode=WAL") c = conn.cursor() c.execute("SELECT COUNT(*) FROM gov_raw WHERE site_name=?", (SITE_NAME,)) existing = c.fetchone()[0] incremental = existing > 0 max_pages = 1 if incremental else MAX_PAGES print("站点 %s: 已有 %d 条, incremental=%s, 爬取 %d 页" % (SITE_NAME, existing, incremental, max_pages), file=sys.stderr) all_new = 0 for pg in range(1, max_pages + 1): page_url = LIST_URL if pg == 1 else LIST_URL + "&cur_page=%d" % pg html = fetch(page_url) if not html: print(" 第%d页 获取失败" % pg, file=sys.stderr) continue items = parse_list_items(html) if not items: print(" 第%d页 无条目" % pg, file=sys.stderr) break print(" 第%d页: %d 条" % (pg, len(items)), file=sys.stderr) for it in items: c.execute("SELECT id FROM gov_raw WHERE page_url=? AND site_name=?", (it["url"], SITE_NAME)) if c.fetchone(): continue content, pub_date, attachments, full_title = get_content(it["url"], it["title"]) date = it["date"] or pub_date try: c.execute( """INSERT INTO gov_raw (site_name, page_url, title, content, publish_date, source_url, status, attachments) VALUES (?,?,?,?,?,?,?,?)""", (SITE_NAME, it["url"], full_title[:500], content, date[:20], it["url"], "synced", json.dumps(attachments, ensure_ascii=False)) ) all_new += 1 print(" + %s (%s) body:%dB attach:%d" % (full_title[:50], date, len(content), len(attachments)), file=sys.stderr) except Exception as e: print(" ! 入库失败: %s" % e, file=sys.stderr) time.sleep(0.3) conn.commit() print("\n完成! 新增 %d 条" % all_new, file=sys.stderr) conn.close() if __name__ == "__main__": main()