#!/usr/bin/env python3 """ 拜城县人民政府 - 公示公告 URL: https://www.xjbc.gov.cn/gsgg/index.html CMS: 阿克苏CMS (AJAX + OpenAPI) List: dataList JSON embedded in first page (10 pages x 15 = 150 items pre-loaded) Detail: div#zoomc > p (text) + table + img + attachments """ import sys, os, re, json, time, sqlite3, html as html_mod import requests from bs4 import BeautifulSoup from urllib.parse import urljoin, urlparse import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) BASE = "https://www.xjbc.gov.cn" LIST_URL = BASE + "/gsgg/index.html" SITE = "拜城县人民政府-公示公告" COLUMN = "公示公告" PROVINCE = "新疆" PER_PAGE = 15 DB_PATH = os.getenv("SEARCH_DB", "/root/search.db") HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": BASE + "/", } session = requests.Session() session.headers.update(HEADERS) session.verify = False def log(msg): print(msg, file=sys.stderr, flush=True) def fetch(url, retries=3): for attempt in range(retries): try: r = session.get(url, timeout=30) r.encoding = "utf-8" return r.text except Exception as e: log(f" [WARN] 请求失败 (尝试 {attempt+1}/{retries}): {e}") time.sleep(2) return "" def parse_list_data(html, max_pages=5): """从第一页HTML中解析dataList JSON，获取前N页列表数据""" items = [] m = re.search(r'var dataList=(.*?);\s*var pagesData=', html, re.DOTALL) if not m: log(" [ERR] 未找到dataList数据") return items try: data = json.loads(m.group(1)) except json.JSONDecodeError as e: log(f" [ERR] dataList解析失败: {e}") return items pages_to_take = min(max_pages, len(data)) for page_idx in range(pages_to_take): page_data = data[page_idx] infolist = page_data.get("infolist", []) for entry in infolist: title = entry.get("title", "").strip() url = entry.get("url", "").strip() # 时间戳转日期 ts = entry.get("releaseTime", 0) date_str = time.strftime("%Y-%m-%d", time.localtime(ts / 1000)) if ts else "" if title and url: items.append({ "url": url, "title": title, "date": date_str, }) log(f" Page {page_idx+1}: {len(infolist)} items") return items def fetch_detail(url): """提取详情页: 标题、日期、正文、附件""" html = fetch(url) if not html: return "", "", "", [] soup = BeautifulSoup(html, "html.parser") # 标题: h1.main_5121 title = "" h1 = soup.find("h1", class_="main_5121") if h1: title = h1.get_text(strip=True) # 日期: meta PubDate date_str = "" meta_pub = soup.find("meta", attrs={"name": "PubDate"}) if meta_pub and meta_pub.get("content"): date_str = meta_pub["content"][:10] # 正文: div#zoomc content = "" zoomc = soup.find("div", id="zoomc") or soup.find("div", class_="main_51231") attachments = [] if zoomc: # 提取文本段落 parts = [] for p in zoomc.find_all("p", recursive=True): p_text = p.get_text("", strip=True) if p_text and len(p_text) > 2: parts.append(p_text) # 提取表格（保留HTML格式以便搜索展示） tables = zoomc.find_all("table") for tbl in tables: tbl_html = str(tbl) parts.append(f"\n[表格]\n{tbl_html}\n[/表格]") # 提取附件链接 for a in zoomc.find_all("a", href=True): href = a["href"] fname = a.get_text(strip=True) if re.search(r'\.(pdf|doc|docx|xls|xlsx|zip|rar|txt)$', href, re.I): full_url = urljoin(BASE, href) attachments.append({"name": fname or "附件", "url": full_url}) parts.append(f"[附件: {fname}]({full_url})") # 提取图片 imgs = zoomc.find_all("img") for img in imgs: src = img.get("src", "") if src: full_src = urljoin(BASE, src) # 如果是扫描件，标记 alt = img.get("alt", "") or "正文附图" attachments.append({"name": alt, "url": full_src}) content = "\n\n".join(parts) if parts else "" if not content and imgs: content = f"[本公告为图片格式，共{len(imgs)}张附图]" return title, date_str, content, attachments def import_to_db(record): try: db = sqlite3.connect(DB_PATH, timeout=10) title = (record.get("title") or "")[:500] page_url = (record.get("page_url") or "")[:1000] content = record.get("content") or "" publish_date = (record.get("publish_date") or "")[:20] site_name = (record.get("site_name") or "unknown")[:100] attachments_str = json.dumps(record.get("attachments") or [], ensure_ascii=False) old = db.execute("SELECT rowid FROM gov_raw WHERE page_url = ?", (page_url,)).fetchone() if old: db.execute("DELETE FROM gov_search WHERE rowid = ?", (old[0],)) db.execute("DELETE FROM gov_raw WHERE page_url = ?", (page_url,)) db.execute( "INSERT INTO gov_raw (title, page_url, content, publish_date, site_name, source_url, status, attachments) VALUES (?, ?, ?, ?, ?, ?, 'synced', ?)", (title, page_url, content, publish_date, site_name, page_url, attachments_str) ) new_rowid = db.execute("SELECT last_insert_rowid()").fetchone()[0] summary = content[:500] if content else title[:500] db.execute("INSERT OR REPLACE INTO gov_search(rowid, title, site_name, summary) VALUES (?, ?, ?, ?)", (new_rowid, title, site_name, summary)) db.commit() db.close() log(f" ✅ {title[:30]}") return True except Exception as e: log(f" [ERR] DB import failed: {e}") return False def main(): import argparse parser = argparse.ArgumentParser() parser.add_argument("--pages", type=int, default=5, help="爬取前N页（每页15条，默认5页=75条，最多10页=150条）") args = parser.parse_args() pages = min(args.pages, 10) # dataList预加载最多10页 count = 0 log(f"📄 获取列表页...") html = fetch(LIST_URL) if not html: log("❌ 无法获取列表页") return items = parse_list_data(html, max_pages=pages) log(f" → 共 {len(items)} 条（前{pages}页）") for idx, item in enumerate(items, 1): url = item["url"] log(f" ({idx}/{len(items)}) {item['title'][:40]}") title, detail_date, content, attachments = fetch_detail(url) record = { "title": title or item["title"], "page_url": url, "publish_date": detail_date or item["date"], "content": content, "attachments": attachments, "site_name": SITE, "column": COLUMN, "province": PROVINCE, } ok = import_to_db(record) if ok: count += 1 time.sleep(0.5) log(f"\n✅ {SITE} 爬取完成，共入库 {count} 条") if __name__ == "__main__": main()