#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 许昌网 - 社会新闻 (21xc.com/shehui_list.html) CMS: 定制 (content/YYYYMM/DD/c{id}.html) ⚠️ 分页 API 222.143.158.138:8081 已失效，仅抓首页静态 33 条 列表: a[href] /content/YYYYMM/DD/c{id}.html 详情: 容器 class=text 用法: python3 crawl_21xc_shehui.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.21xc.com" LIST_URL = "http://www.21xc.com/shehui_list.html" DOMAIN = "www.21xc.com" 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_page(p): if p > 1: return "" r = requests.get(LIST_URL, headers=HEADERS, timeout=30) r.encoding = r.apparent_encoding or "utf-8" return r.text def fetch_detail(url): r = requests.get(url, headers=HEADERS, timeout=30, verify=False) r.encoding = r.apparent_encoding or "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 not re.search(r"/content/\d{6}/\d{2}/c\d+\.html", href): continue title = clean_title(a.get_text(strip=True)) if not title or len(title) < 6: continue pub_date = "" parent = a.find_parent(["li", "div"]) if parent: m = re.search(r"20\d{2}[-/年]\d{1,2}[-/月]\d{1,2}", parent.get_text()) if m: pub_date = m.group(0).replace("年", "-").replace("月", "-").replace("/", "-")[:10] 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") h = soup.find("h1", class_="content-title") if h: result["title"] = clean_title(h.get_text(strip=True)) if not result["title"]: tt = soup.find("title") if tt: t = tt.get_text(strip=True) t = re.sub(r"[_\-—]\s*许昌网\s*$", "", t) result["title"] = clean_title(t) m = re.search(r"20\d{2}[-/年]\d{1,2}[-/月]\d{1,2}", html[:8000]) if m: result["publish_date"] = m.group(0).replace("年", "-").replace("月", "-").replace("/", "-")[:10] div = soup.find("div", class_="article-content") or soup.find("div", class_="article-detail") or soup.find("div", class_="text") if div: result["content"] = extract_clean_text(str(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 f"{LIST_URL}_{p}" try: html = fetch_page(p) 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_detail(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)" " VALUES (?,?,?,?,?,?,?,?)", (item["site_name"], item["page_url"], item["page_url"], item["title"], item["publish_date"], item["content"], (item["content"] or "")[:500], item["column"])) 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)