#!/usr/bin/env python3 """zjshjkj.com - 项目公示 (浙江省环境科技股份有限公司) 列表: /?mod=publicity&page=N 详情: /?mod=hpgs-in&id=N -> div.cd-list__title (h3+span) + a.file (PDF) """ import sys, os, re, json, time from datetime import datetime, timedelta import requests from bs4 import BeautifulSoup BASE = "https://www.zjshjkj.com" SITE = "zjshjkj.com-项目公示" COLUMN = "项目公示" PROVINCE = "企业" TOTAL_PAGES = 8 HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} session = requests.Session() session.headers.update(HEADERS) MARKER = "\x00P\x00" def log(msg): sys.stderr.write(msg + "\n") sys.stderr.flush() def fetch_list(page): url = f"{BASE}/?mod=publicity&page={page}" try: r = session.get(url, timeout=20, verify=False) r.encoding = "utf-8" return r.text except Exception as e: log(f" [ERROR] list page {page}: {e}") return None def parse_list(html): soup = BeautifulSoup(html, "html.parser") items = [] for a in soup.select("ul.public-view a"): href = a.get("href", "") if not href.startswith("http"): href = BASE + href title = a.get_text(strip=True) if not title or not href: continue # Extract date from DDMM.YYYY prefix date_str = "" m = re.match(r'(\d{2})(\d{2})\.(\d{4})', title) if m: day, month, year = m.group(1), m.group(2), m.group(3) date_str = f"{year}-{month}-{day}" items.append({"title": title, "url": href, "date": date_str}) return items def extract_content(detail_url, title): try: r = session.get(detail_url, timeout=20, verify=False) r.encoding = "utf-8" except Exception as e: return f"[{title}]({detail_url})", title, "" soup = BeautifulSoup(r.text, "html.parser") pub_date = "" # Get title and date from cd-list__title title_div = soup.select_one("div.cd-list__title") if title_div: h3 = title_div.find("h3") if h3: title = h3.get_text(strip=True) span = title_div.find("span") if span: dm = re.search(r'(\d{4}-\d{2}-\d{2})', span.get_text()) if dm: pub_date = dm.group(1) # Content from cd-list__full content_el = soup.select_one("div.cd-list__full.fl") if not content_el: return f"[{title}]({detail_url})", title, pub_date # Find PDF links pdf_links = [] for a_tag in content_el.find_all("a", class_="file"): href = a_tag.get("href", "") if not href.startswith("http"): href = BASE + href txt = a_tag.get_text(strip=True) or title if href: pdf_links.append(f"[{txt}]({href})") # Save and remove tables tables_html = [] for table in content_el.find_all("table"): tables_html.append(str(table)) table.decompose() # Paragraph markers for tag in content_el.find_all(["p", "h1", "h2", "h3", "h4", "h5", "h6"]): tag.insert(0, MARKER) tag.append(MARKER) for br in content_el.find_all("br"): br.replace_with(MARKER) # Unwrap inline for tag in content_el.find_all(["span", "b", "strong", "font", "em", "i", "u", "s"]): tag.unwrap() text = content_el.get_text(separator="", strip=True) text = re.sub(r"\x00P\x00(\s*\x00P\x00)+", "\x00P\x00", text) text = text.replace("\x00P\x00", "\n\n") text = re.sub(r"\n{3,}", "\n\n", text) text = text.strip() for tbl_html in tables_html: text += f"\n\n{tbl_html}" # If content is just the title/date/sidebar noise, use PDF links if not text.strip() or len(text.strip()) < 30: if pdf_links: text = "\n\n".join(pdf_links) else: text = f"[{title}]({detail_url})" elif pdf_links: # Append PDF links if not already in text for link in pdf_links: if link not in text: text += f"\n\n{link}" return text, title, pub_date def crawl_all(months_back=36): cutoff = datetime.now() - timedelta(days=months_back * 30) if months_back else None all_items = [] seen_urls = set() for page in range(1, TOTAL_PAGES + 1): html = fetch_list(page) if not html: break items = parse_list(html) if not items: break new_count = 0 for item in items: if item["url"] not in seen_urls: if cutoff and item["date"]: try: d = datetime.strptime(item["date"], "%Y-%m-%d") if d < cutoff: continue except ValueError: pass seen_urls.add(item["url"]) all_items.append(item) new_count += 1 log(f" [PAGE {page}/{TOTAL_PAGES}] {len(items)} items, +{new_count} new") if new_count == 0 and page > 1: break time.sleep(1) log(f"\n共 {len(all_items)} 条待爬详情") for idx, item in enumerate(all_items, 1): content, title, pub_date = extract_content(item["url"], item["title"]) if pub_date: item["date"] = pub_date if title: item["title"] = title item["content"] = content output_item(item, idx) if idx % 10 == 0: log(f" [PROGRESS] {idx}/{len(all_items)}") log(f"\n[DONE] 共爬取 {len(all_items)} 条") def crawl_incremental(): html = fetch_list(1) if not html: return items = parse_list(html) log(f"[LIST] page 1 -> {len(items)} 条") for idx, item in enumerate(items, 1): content, title, pub_date = extract_content(item["url"], item["title"]) if pub_date: item["date"] = pub_date if title: item["title"] = title item["content"] = content output_item(item, idx) log(f"[DONE] 增量爬取 {len(items)} 条") def output_item(item, idx): record = { "title": item.get("title", ""), "page_url": item.get("url", ""), "publish_date": item.get("date", ""), "content": item.get("content", ""), "site_name": SITE, "column": COLUMN, "province": PROVINCE, } print(json.dumps(record, ensure_ascii=False)) if __name__ == "__main__": mode = sys.argv[1] if len(sys.argv) > 1 else "incremental" if mode == "incremental": crawl_incremental() elif mode == "full": months = int(sys.argv[2]) if len(sys.argv) > 2 else 36 crawl_all(months) elif mode == "list": html = fetch_list(1) items = parse_list(html) log(f"共 {len(items)} 条") for it in items[:5]: log(f" {it['date']} | {it['title'][:50]} | {it['url']}") else: log(f"Usage: {sys.argv[0]} [incremental|full [months]]")