#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 崇义县人民政府 - 通知公告 (chongyi.gov.cn/cyxrmzf/tzgg) CMS: TRS createPageHTML (list_N.shtml) 列表: li > a[href=/cyxrmzf/tzgg/YYYYMM/hash.shtml] + span 日期 分页: list_2.shtml ... 详情: 正文 #zoomcon 用法: python3 crawl_chongyi_tzgg.py [--pages=N] [--json out.jsonl] """ import re, sys, os, json, time, sqlite3, requests from datetime import datetime, timedelta from bs4 import BeautifulSoup from urllib.parse import urljoin SITE_NAME = "崇义县人民政府-通知公告" BASE_URL = "http://www.chongyi.gov.cn" LIST_URL = "http://www.chongyi.gov.cn/cyxrmzf/tzgg/list.shtml" DOMAIN = "www.chongyi.gov.cn" GROUP = "江西" CUTOFF = (datetime.now() - timedelta(days=3*365)).strftime("%Y-%m-%d") DB_PATH = os.environ.get("DB_PATH", "/root/search.db") 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"} def log(msg): print(f"[{SITE_NAME}] {msg}", flush=True) def clean_title(t): t = re.sub(r" | |\u200b|\ufeff", "", t or "") t = re.sub(r"\s+", " ", t) t = re.sub(r"\.\.\.|…$", "", t).strip() return t def fetch(url, timeout=30): r = requests.get(url, headers=HEADERS, timeout=timeout) r.encoding = "utf-8" return r.text def parse_list(html): soup = BeautifulSoup(html, "html.parser") items = [] for a in soup.find_all("a", href=True): href = a.get("href", "") if "tzgg" not in href or ".shtml" not in href: continue if href.startswith("/cyxrmzf/tzgg/list"): continue title = clean_title(a.get_text(strip=True)) if not title: continue pub_date = "" parent = a.find_parent(["li", "div"]) if parent: m = re.search(r"20\d{2}-\d{2}-\d{2}", parent.get_text()) if m: pub_date = m.group(0) items.append({"title": title, "url": urljoin(BASE_URL, href), "pub_date": pub_date}) seen = set() uniq = [] for it in items: if it["url"] not in seen: seen.add(it["url"]) uniq.append(it) return uniq def parse_detail(html): result = {"title": "", "publish_date": "", "content": ""} soup = BeautifulSoup(html, "html.parser") title_tag = soup.find("title") if title_tag: t = title_tag.get_text(strip=True) t = re.sub(r"\s*\|\s*崇义县人民政府\s*$", "", t) t = re.sub(r"[_\-—]\s*(公示公告|通知公告|崇义县人民政府)\s*$", "", t) result["title"] = t.strip() meta_date = soup.find("meta", attrs={"name": "PubDate"}) if meta_date and meta_date.get("content"): result["publish_date"] = meta_date["content"][:10] if not result["publish_date"]: m = re.search(r"20\d{2}-\d{2}-\d{2}", html[:8000]) if m: result["publish_date"] = m.group(0) content_div = soup.find("div", id="zoomcon") if not content_div: content_div = soup.find("div", class_="article-con") if content_div: result["content"] = extract_clean_text(str(content_div)) return result def extract_clean_text(content_html): if not content_html: return "" soup = BeautifulSoup(content_html, "html.parser") for a_tag in soup.find_all("a", href=True): href = a_tag.get("href", "") text = a_tag.get_text(strip=True) or "附件" full_url = urljoin(BASE_URL, href) a_tag.replace_with(f"[{text}]({full_url})") for tag in soup.find_all(['span', 'b', 'strong', 'font', 'em', 'i', 'u', 's']): tag.unwrap() parts = [] for el in soup.find_all(['table', 'p']): if el.name == 'table': parts.append(str(el)) elif el.name == 'p' and not el.find_parent('table'): t = el.get_text(separator='', strip=True) if t: parts.append(t) return '\n\n'.join(parts) if parts else content_html.strip() def crawl(pages=1, json_out=None): log(f"Crawl pages={pages} cutoff={CUTOFF}") items_all = [] seen = set() for p in range(1, pages + 1): url = LIST_URL if p == 1 else re.sub(r"list\.shtml$", f"list_{p}.shtml", LIST_URL) try: html = fetch(url) except Exception as e: log(f"P{p} ERR: {e}") break items = parse_list(html) new = 0 for it in items: if it["url"] not in seen: seen.add(it["url"]) items_all.append(it) new += 1 log(f"P{p}: {len(items)} items ({new} new), total {len(items_all)}") if not items: break time.sleep(0.8) enriched = [] for idx, it in enumerate(items_all, 1): if it["pub_date"] and it["pub_date"] < CUTOFF: continue try: html = fetch(it["url"]) detail = parse_detail(html) except Exception as e: log(f" detail {it['url'][-40:]} ERR: {e}") detail = {"title": "", "publish_date": "", "content": ""} enriched.append({ "title": detail["title"] or it["title"], "page_url": it["url"], "publish_date": detail["publish_date"] or it["pub_date"], "content": detail["content"], "site_name": SITE_NAME, "column": "通知公告", }) if idx % 10 == 0: log(f" {idx}/{len(items_all)}") time.sleep(0.4) if json_out: with open(json_out, "w", encoding="utf-8") as f: for item in enriched: f.write(json.dumps(item, ensure_ascii=False) + "\n") log(f"JSONL: {len(enriched)} items -> {json_out}") return conn = sqlite3.connect(DB_PATH, timeout=60) conn.execute("PRAGMA busy_timeout=10000") c = conn.cursor() stored = skipped = 0 for item in enriched: try: c.execute( "INSERT OR IGNORE INTO gov_raw " "(site_name, source_url, page_url, title, publish_date, content, summary, category, script_name)" " VALUES (?,?,?,?,?,?,?,?,?)", (item["site_name"], item["page_url"], item["page_url"], item["title"], item["publish_date"], item["content"], (item["content"] or "")[:500], item["column"], "crawl_chongyi_tzgg.py")) if c.rowcount > 0: stored += 1 else: skipped += 1 except Exception as e: log(f"DB error: {e}") skipped += 1 conn.commit() conn.close() log(f"Result: {stored} new, {skipped} skipped") if __name__ == "__main__": args = sys.argv[1:] pages = 5 json_out = None if "--pages=" in " ".join(args): pages = int(re.search(r"--pages=(\d+)", " ".join(args)).group(1)) if "--json" in args: json_out = args[args.index("--json") + 1] crawl(pages=pages, json_out=json_out)