#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 富源县 - 环评管理 https://www.qjfy.gov.cn/fuyuan/zdlygk/list/283.html 分页: ?cate=283&page=N, 14条/页, 共19页(~266条) """ 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.qjfy.gov.cn" LIST_BASE = "https://www.qjfy.gov.cn/fuyuan/zdlygk/list/283.html" SITE_NAME = "qjfy_hpgl" MAX_PAGES = 19 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") for ul in soup.find_all("ul"): for li in ul.find_all("li"): a = li.find("a") if not a or not a.get("href"): continue href = a["href"].strip() if "show" not in href: continue sp = a.find("span") if not sp: continue title = sp.get_text(strip=True) if not title or len(title) < 5: continue # Date is text after span a_text = a.get_text() date = a_text.replace(sp.get_text(), "", 1).strip()[:10] full_url = href if href.startswith("http") else 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 for m in soup.find_all("meta"): n = m.get("name", "") if n and "articletitle" in n.lower(): t = m.get("content", "").strip() if t: full_title = t break if not full_title or full_title == title_from_list: h3 = soup.select_one("h3.public_web_title") if h3: t = h3.get_text(strip=True) if t: full_title = t # 日期 pub_date = "" m = re.search(r"\u53d1\u5e03\u65f6\u95f4[：:]\s*(\d{4}[-/]\d{2}[-/]\d{2})", html) if m: pub_date = m.group(1).replace("/", "-") # 正文 - ly-box > public_web_con.des body = soup.select_one("div.ly-box div.public_web_con.des") if not body: body = soup.select_one("div.public_web_con.des") 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?)($|\?)", 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 min(5, 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_BASE if pg == 1 else "%s?cate=283&page=%d" % (LIST_BASE, 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()