#!/usr/bin/env python3 """兴隆台区人民政府 - 公示公告""" import json, re, sys, requests from bs4 import BeautifulSoup from urllib.parse import urljoin DB_PATH = "/root/search.db" SITE_NAME = "兴隆台区-公示公告" CATEGORY = "公示公告" BASE_URL = "https://www.xlt.gov.cn/13547/" MAX_PAGES = 32 HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9", } SQL = """INSERT OR IGNORE INTO gov_raw (site_name, source_url, page_url, title, publish_date, summary, content, category, attachments, group_name) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""" def init_db(): import sqlite3 conn = sqlite3.connect(DB_PATH, timeout=60) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA busy_timeout=5000") return conn def decode_resp(r): raw = r.content ct = r.headers.get("Content-Type", "") if "gb2312" in ct.lower() or "gbk" in ct.lower(): return raw.decode("gbk", errors="replace") raw_lower = raw[:2000].lower() if b"gb2312" in raw_lower or b"gbk" in raw_lower or b"charset=gb" in raw_lower: return raw.decode("gbk", errors="replace") return raw.decode("utf-8", errors="replace") def table_to_md(table, *args, **kwargs): """保留 HTML 表格结构（不转 md）""" return str(table) def extract_content(artcont_div, detail_url): parts = [] processed = set() for el in artcont_div.find_all(["p", "table", "img", "h1", "h2", "h3", "h4"], recursive=True): if id(el) in processed: continue processed.add(id(el)) if el.name == "table": md = table_to_md(el) if md: parts.append(md) elif el.name == "img": src = el.get("src", "") if src: alt = el.get("alt", "") full_src = urljoin(detail_url, src) if not src.startswith("http") else src parts.append(f"![{alt}]({full_src})" if alt else f"![]({full_src})") for sub_img in el.find_all("img", recursive=True): processed.add(id(sub_img)) elif el.name == "p": txt = el.get_text(" ", strip=True) if txt: txt = re.sub(r'[^\S\n]+', ' ', txt) parts.append(txt) else: txt = el.get_text(" ", strip=True) if txt: parts.append(txt) return "\n\n".join(parts) def extract_attachments(soup, detail_url): attachments = [] for a in soup.select('div.Q25_artcont a[href]'): href = a["href"].strip() if not re.search(r'\.(pdf|doc|docx|xls|xlsx|zip|rar)$', href, re.I): continue name = a.get_text(strip=True) or href.split("/")[-1] full_url = urljoin(detail_url, href) if not href.startswith("http") else href if not any(aa["url"] == full_url for aa in attachments): attachments.append({"name": name, "url": full_url}) return attachments def extract_list_page(page_num): if page_num == 1: url = BASE_URL else: url = f"https://www.xlt.gov.cn/13547/list-{page_num}.html" try: r = requests.get(url, headers=HEADERS, timeout=30) r.raise_for_status() except Exception as e: print(f" [ERR] page {page_num}: {e}", file=sys.stderr) return [] html = decode_resp(r) soup = BeautifulSoup(html, "html.parser") items = [] # List container: Q25_ListCont > ul.Q25_ListContlist ul = soup.find("ul", class_="Q25_ListContlist") if not ul: print(f" [WARN] page {page_num}: no Q25_ListContlist found", file=sys.stderr) return [] for li in ul.find_all("li", recursive=False): a_tag = li.find("a", href=True) if not a_tag: continue href = a_tag["href"].strip() title = a_tag.get("title", "") or a_tag.get_text(strip=True) if not title: continue spans = a_tag.find_all("span") date_str = spans[-1].get_text(strip=True) if len(spans) >= 2 else "" items.append({"title": title, "url": href, "date": date_str}) print(f" Page {page_num}: {len(items)} items") return items def extract_detail(url): try: r = requests.get(url, headers=HEADERS, timeout=30) r.raise_for_status() except Exception as e: return None, None, None, [] html = decode_resp(r) soup = BeautifulSoup(html, "html.parser") # Title title_p = soup.find("p", class_="Q25_articleconttitle") title = title_p.get_text(strip=True) if title_p else "" # Date date_str = "" date_div = soup.find("div", class_="Q25_articlecontdateleft") if date_div: spans = date_div.find_all("span") for span in spans: txt = span.get_text(strip=True) m = re.search(r'(\d{4}-\d{2}-\d{2})', txt) if m: date_str = m.group(1) break # Content artcont = soup.find("div", class_="Q25_artcont") content = "" if artcont: content = extract_content(artcont, url) # Attachments attachments = extract_attachments(soup, url) # Empty content fallback if len(content.strip()) < 20: fallback = f"[{title or url.split('/')[-1]}]({url})" if attachments: attach_links = [f"附件：[{a['name']}]({a['url']})" for a in attachments] fallback += "\n\n" + "\n\n".join(attach_links) content = fallback return title, date_str, content, attachments def crawl(test_mode=False, max_pages=MAX_PAGES): conn = init_db() cur = conn.cursor() total = 0 errors = 0 total_listed = 0 for pn in range(1, max_pages + 1): items = extract_list_page(pn) total_listed += len(items) if not items: break for item in items: url = item["url"] cur.execute("SELECT id FROM gov_raw WHERE page_url = ?", (url,)) if cur.fetchone(): continue title, date_str, content, attachments = extract_detail(url) if not title: errors += 1 continue final_title = title or item["title"] final_date = date_str or item["date"] att_json = json.dumps(attachments, ensure_ascii=False) if attachments else "" summary = (content[:200] if content else final_title).strip() if not summary: summary = final_title try: cur.execute(SQL, ( SITE_NAME, url, url, final_title.strip(), final_date, summary, content, CATEGORY, att_json, CATEGORY, )) conn.commit() if cur.rowcount > 0: total += 1 except Exception as e: print(f" DB error: {e}", file=sys.stderr) errors += 1 if test_mode and total >= 5: break if test_mode and total >= 5: break conn.close() print(f"\n[DONE] {SITE_NAME}") print(f" New: {total}, Errors: {errors}, Listed total: {total_listed}") return total if __name__ == "__main__": mode = "test" if "--test" in sys.argv else "full" max_p = 5 if mode == "test" else MAX_PAGES if len(sys.argv) > 1 and sys.argv[1].isdigit(): max_p = int(sys.argv[1]) crawl(test_mode=(mode == "test"), max_pages=max_p)