#!/usr/bin/env python3 import os """ Crawler for 在榆林网 - 产经 CMS: EmpireCMS, static .html pages Site: www.zaiyulin.com/chanjing/ """ import requests, re, json, sys, os, tempfile from datetime import datetime, timedelta from bs4 import BeautifulSoup import urllib.parse BASE_URL = "https://www.zaiyulin.com" LIST_PATH = "/chanjing" LIST_URL = BASE_URL + LIST_PATH + "/" SITE_NAME = "在榆林网-产经" INCREMENTAL_DAYS = 7 MAX_PAGES = 5 PER_PAGE = 10 HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.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,en;q=0.8", } TIMEOUT = 20 session = requests.Session() session.headers.update(HEADERS) def html_table_to_html(table, base_url=""): """保留 HTML 表格结构，仅将相对链接/图片转绝对 URL""" from bs4 import BeautifulSoup tbl = BeautifulSoup(str(table), 'html.parser') for a in tbl.find_all('a'): href = a.get('href', '') if href and not href.startswith(('http', 'javascript', '#')): a['href'] = urllib.parse.urljoin(base_url, href) if base_url else href for img in tbl.find_all('img'): src = img.get('src', '') if src and not src.startswith(('http', '//', 'data:')): img['src'] = urllib.parse.urljoin(base_url, src) if base_url else src return str(tbl) def fetch(url): resp = session.get(url, timeout=TIMEOUT) resp.encoding = "utf-8" return resp.text def parse_list(html): """Parse list page, return list of (title, url, date).""" soup = BeautifulSoup(html, "html.parser") items = [] for card in soup.find_all("div", class_="card-pic"): h2 = card.find("h2", class_="card-h2") if not h2: continue a = h2.find("a", href=True) if not a: continue href = a["href"].strip() title = a.get_text(strip=True) span = card.find("span", class_="time") date = span.get_text(strip=True) if span else "" # Extract just the date part YYYY-MM-DD m = re.search(r'(\d{4}-\d{2}-\d{2})', date) if m: date = m.group(1) if not href.startswith("http"): href = BASE_URL + href if href.startswith("/") else BASE_URL + "/" + href if title and href: items.append((title.strip(), href, date)) return items def parse_detail(html, url): """Parse detail page content.""" soup = BeautifulSoup(html, "html.parser") # Title from h1 in .title section title = "" section = soup.find("section", class_="title") if section: h1 = section.find("h1") if h1: title = h1.get_text(strip=True) # Date from .info .time pub_date = "" info_section = soup.find("section", class_="info") if info_section: span = info_section.find("span", class_="time") if span: text = span.get_text(strip=True) m = re.search(r'(\d{4}-\d{2}-\d{2})', text) if m: pub_date = m.group(1) # Content from .post_text.col-12 content_html = "" post_text = soup.find("div", class_="post_text") if post_text: content_html = post_text.decode_contents() # Attachments attachments = [] if content_html: csoup = BeautifulSoup(content_html, "html.parser") for a in csoup.find_all("a", href=True): ahref = a["href"].strip() if any(ext in ahref.lower() for ext in [".pdf", ".doc", ".docx", ".xls", ".xlsx", ".zip", ".rar"]): if not ahref.startswith("http"): ahref = BASE_URL + ahref if ahref.startswith("/") else BASE_URL + "/" + ahref attachments.append({ "title": a.get_text(strip=True) or "附件", "url": ahref }) return { "title": title, "date": pub_date, "content": content_html, "attachments": attachments, "source_url": url, } def render_content(content_html, title, url): """Convert HTML to readable text.""" if not content_html or len(content_html.strip()) < 50: return "[{}]({})".format(title, url) soup = BeautifulSoup(content_html, "html.parser") parts = [] for el in soup.children: if el.name is None: text = el.strip() if text: parts.append(text) elif el.name == "p": text = el.get_text(" ", strip=True) if text: parts.append(text) if el.name == 'table': parts.append(html_table_to_html(el, url)) elif el.name == "br": parts.append("") elif el.name == "div": text = el.get_text(" ", strip=True) if text: parts.append(text) result = "\n\n".join(p for p in parts if p).strip() result = re.sub(r"\n{3,}", "\n\n", result) if len(result.strip()) < 20: return "[{}]({})".format(title, url) return result def is_incremental(): return len(sys.argv) > 1 and sys.argv[1] == "incremental" def main(): incremental = is_incremental() print("[{}] Starting crawl (incremental={}, max_pages={})".format(SITE_NAME, incremental, MAX_PAGES)) all_items = [] for page in range(1, MAX_PAGES + 1): if page == 1: page_url = LIST_URL else: page_url = "{}/index_{}.html".format(LIST_URL.rstrip("/"), page) print(" Fetching page {}: {}".format(page, page_url)) try: html = fetch(page_url) except Exception as e: print(" ERROR: {}".format(e)) continue items = parse_list(html) print(" Found {} items".format(len(items))) if not items: break all_items.extend(items) print("\n Total items from list: {}".format(len(all_items))) if not all_items: print(" No items found, exiting") return cutoff_date = None if incremental: cutoff_date = datetime.now() - timedelta(days=INCREMENTAL_DAYS) print(" Incremental mode: cutoff = {}".format(cutoff_date.date())) filtered = [(t, u, d) for t, u, d in all_items if d and _date_filter(d, cutoff_date)] all_items = filtered print(" After incremental filter: {} items".format(len(all_items))) results = [] for idx, (title, url, date_from_list) in enumerate(all_items): print(" [{}/{}] Fetching: {}...".format(idx + 1, len(all_items), title[:40])) try: html = fetch(url) except Exception as e: print(" ERROR: {}".format(e)) results.append({ "title": title, "date": date_from_list, "content": "[{}]({})".format(title, url), "attachments": [], "source_url": url, }) continue detail = parse_detail(html, url) if not detail["title"]: detail["title"] = title if not detail["date"]: detail["date"] = date_from_list detail["content"] = render_content(detail["content"], detail["title"], url) results.append(detail) print("\n Total valid items: {}".format(len(results))) if not results: print(" No items to import, exiting") return tmp = tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False, dir="/tmp") label = "{} ({}条)".format(SITE_NAME, len(results)) for r in results: line = { "title": r["title"], "page_url": r["source_url"], "content": r["content"], "publish_date": r["date"], "site_name": SITE_NAME, "source_url": r["source_url"], "attachments": r.get("attachments", []), } tmp.write(json.dumps(line, ensure_ascii=False) + "\n") tmp.close() print(" JSONL written: {}".format(tmp.name)) ret = os.system("python3 /root/gov_crawler/import_jsonl.py '{}' '{}'".format(tmp.name, label)) if ret == 0: print(" OK Import complete") else: print(" FAIL Import failed (exit={})".format(ret)) sys.exit(1) def _date_filter(date_str, cutoff): try: d = datetime.strptime(date_str, "%Y-%m-%d") return d >= cutoff except ValueError: return True if __name__ == "__main__": main()