#!/usr/bin/env python3 import re, sys, os, json import urllib.request, urllib.error from bs4 import BeautifulSoup import sqlite3 BASE_URL = "https://www.xw.gov.cn" LIST_URL = "https://www.xw.gov.cn/gov/public/special/shx.html" SITE_NAME = "\u5ba3\u5a01\u5e02-\u516c\u793a\u516c\u544a" SITE_DISPLAY = "\u5ba3\u5a01\u5e02\u4eba\u6c11\u653f\u5e9c" DB = os.environ.get("SEARCH_DB", "/root/search.db") HEADERS = { "User-Agent": "Mozilla/5.0", "Referer": "https://www.xw.gov.cn/", "Accept": "text/html,application/xhtml+xml" } _MAX_PG = None for i, a in enumerate(sys.argv): if a == "--pages" and i + 1 < len(sys.argv): _MAX_PG = int(sys.argv[i + 1]) break def fetch(url): req = urllib.request.Request(url, headers=HEADERS) return urllib.request.urlopen(req, timeout=30).read().decode() def parse_list(html): soup = BeautifulSoup(html, "html.parser") items = [] ul = soup.select_one("ul.p-list") if not ul: return items for li in ul.find_all("li", recursive=False): a = li.find("a", href=True) if not a: continue href = a["href"].strip() title = a.get_text(strip=True) if not title: continue after = li.get_text(" ", strip=True) date_text = "" if after: parts = after.split(title, 1) if len(parts) > 1: m = re.search(r"(\d{4}-\d{2}-\d{2})", parts[1]) if m: date_text = m.group(1) items.append({"title": title, "url": href, "date": date_text}) return items def get_total_pages(html): soup = BeautifulSoup(html, "html.parser") page_div = soup.select_one("div.page") if not page_div: return None max_p = 1 for a in page_div.find_all("a", href=True): m = re.search(r"page=(\d+)", a["href"]) if m: p = int(m.group(1)) if p > max_p: max_p = p return max_p def parse_detail(html, url): soup = BeautifulSoup(html, "html.parser") title_tag = soup.select_one("h3.public_web_title") title = title_tag.get_text(strip=True) if title_tag else "" date_text = "" date_li = soup.select_one("li.time") if date_li: m = re.search(r"(\d{4}-\d{2}-\d{2})", date_li.get_text()) if m: date_text = m.group(1) content_div = soup.select_one("div.public_web_con.des") if not content_div: return {"title": title, "date": date_text, "content": "", "attachments": []} parts = [] attachments = [] for child in content_div.find_all(["p", "table", "img"], recursive=True): if child.name == "p": if child.find_parent("table"): continue text = child.get_text(" ", strip=True) a_tag = child.find("a", href=True) if a_tag: ahref = a_tag["href"] if any(ahref.lower().endswith(ext) for ext in [".pdf", ".doc", ".docx", ".xls", ".xlsx", ".zip", ".rar"]): if ahref.startswith("/"): ahref = BASE_URL + ahref attachments.append({"name": a_tag.get_text(strip=True) or ahref.split("/")[-1], "url": ahref}) if text: parts.append(text) elif child.name == "table": if child.find_parent("table"): continue md_rows = [] rows = child.find_all("tr") for ri, tr in enumerate(rows): cells = tr.find_all(["td", "th"]) row_data = [c.get_text(" ", strip=True).replace("\n", " ").replace("|", "\\|") for c in cells] if row_data: md_rows.append("| " + " | ".join(row_data) + " |") if len(md_rows) >= 2: sep = "| " + " | ".join(["---"] * max(len(cells) if cells else 1, 1)) + " |" md_table = "\n".join([md_rows[0], sep] + md_rows[1:]) elif md_rows: md_table = "\n".join(md_rows) else: continue parts.append("\n" + md_table + "\n") elif child.name == "img": src = child.get("src", "") alt = child.get("alt", "") if src: if src.startswith("/"): src = BASE_URL + src parts.append("![" + (alt or "image") + "](" + src + ")") content = "\n\n".join(parts) return {"title": title, "date": date_text, "content": content, "attachments": attachments} def main(): conn = sqlite3.connect(DB) c = conn.cursor() html = fetch(LIST_URL) total_pages = get_total_pages(html) if not total_pages: total_pages = 89 pages_to_fetch = min(total_pages, _MAX_PG) if _MAX_PG else total_pages print("Total pages:", total_pages, "Fetch:", pages_to_fetch) existing = set() for row in c.execute("SELECT page_url FROM gov_raw WHERE site_name=?", (SITE_NAME,)): existing.add(row[0]) print("Existing:", len(existing)) all_items = [] for pg in range(1, pages_to_fetch + 1): try: pg_html = html if pg == 1 else fetch(LIST_URL + "?page=" + str(pg)) items = parse_list(pg_html) print(" Page", pg, ":", len(items)) all_items.extend(items) except Exception as e: print(" Page", pg, "error:", e) print("Total items:", len(all_items)) new_count = 0 skip_count = 0 empty_count = 0 for item in all_items: url = item["url"] if url in existing: skip_count += 1 continue try: html = fetch(url) detail = parse_detail(html, url) title = detail["title"] or item["title"] date_text = detail["date"] or item["date"] content = detail["content"] attachments = detail["attachments"] if not content: empty_count += 1 attachments_json = json.dumps(attachments, ensure_ascii=False) if attachments else "[]" text_only = re.sub(r"\s+", " ", content) text_only = re.sub(r"!\[.*?\]\(.*?\)", "", text_only) text_only = re.sub(r"\|.*?\|", "", text_only) summary = (text_only[:200] or title).strip() c.execute("INSERT OR REPLACE INTO gov_raw (title, site_name, page_url, publish_date, content, summary, attachments, date_rank) VALUES (?,?,?,?,?,?,?,?)", (title, SITE_NAME, url, date_text, content, summary, attachments_json, date_text)) conn.commit() new_count += 1 print(" OK", title[:30], "|", date_text) except Exception as e: print(" ERR", item["title"][:30], "|", e) print("\n" + "=" * 40) print(SITE_DISPLAY, "(", SITE_NAME, ")") print("New:", new_count, "Skip:", skip_count, "Empty:", empty_count, "Pages:", pages_to_fetch) conn.close() if __name__ == "__main__": main()