#!/usr/bin/env python3 """ 世科生态环境科技(苏州)有限公司 - 公示公告 爬虫 http://www.sk-env.com/index.php/list-article-cid-238.html CMS: Custom PHP (thinkcmf-like) 分页: /index.php/list-article-cid-238.html + /index.php/list-article-cid-238-p-{N}.html 列表: ul > li > a > div.news-int > span (标题) + span (日期) 详情: div.nr-content > p 登录墙: 发布日期 >1个月 的详情页 302 → /index.php/membernews-login-gid-{id}.html 策略: --days N (默认31) 只抓最近 N 天, 列表日期早于 cutoff 直接跳过(避免撞登录墙), 整页全旧则停止翻页 """ import sys import time import re import requests import sqlite3 from datetime import datetime, timedelta from bs4 import BeautifulSoup from urllib.parse import urljoin DB_PATH = "/mnt/data/search.db" BASE_URL = "http://www.sk-env.com/index.php/list-article-cid-238.html" SITE_NAME = "世科生态环境-公示公告" 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", } TIMEOUT = 30 def crawl_page(page_url): """Parse list page and return items""" r = requests.get(page_url, headers=HEADERS, timeout=TIMEOUT) r.encoding = 'utf-8' if r.status_code != 200: print(f" HTTP {r.status_code}, stopping") return None soup = BeautifulSoup(r.text, 'html.parser') news = soup.select_one('div.ny-news') if not news: print(" No ny-news found") return None ul = news.find('ul') if not ul: print(" No ul found") return None items = [] for li in ul.find_all('li'): a = li.find('a', href=True) if not a: continue href = a['href'].strip() if '/show-article-cid-238-id-' not in href: continue spans = a.find_all('span') title = '' date = '' for s in spans: txt = s.get_text(strip=True) if re.match(r'\d{4}-\d{2}-\d{2}', txt): date = txt elif len(txt) > 5: title = txt if not title: title = a.get_text(strip=True).replace(date, '').strip() full_url = urljoin(page_url, href) items.append({'title': title, 'url': full_url, 'date': date}) return items def fetch_detail(item): """Fetch and parse detail page. Handle login redirect gracefully.""" sess = requests.Session() # First visit the list page to establish session list_resp = sess.get(BASE_URL, headers=HEADERS, timeout=TIMEOUT) # Immediately fetch detail with referrer resp = sess.get(item['url'], headers={ **HEADERS, "Referer": BASE_URL, }, timeout=TIMEOUT, allow_redirects=True) resp.encoding = 'utf-8' # Check if redirected to login page if resp.status_code != 200 or 'class="login"' in resp.text or 'membernews-login' in resp.url: print(f" Login required: {item['title'][:40]}...") return None soup = BeautifulSoup(resp.text, 'html.parser') content_div = soup.select_one('div.nr-content') if not content_div: print(f" No nr-content") return None for tag in content_div.find_all(['script', 'style', 'iframe']): tag.decompose() # Extract content preserving tables parts = [] def extract_content(node): for child in node.children: tag = getattr(child, 'name', None) if tag == 'p': txt = child.get_text(separator='', strip=True) if txt: parts.append(txt) elif tag == 'table': parts.append(str(child)) elif tag == 'div': # Skip prev/next/return navigation blocks cls = ' '.join(child.get('class', [])) if any(k in cls for k in ('prev', 'next', 'returnbt')): continue extract_content(child) elif tag is None and isinstance(child, str): txt = child.strip() if txt and len(txt) > 3: parts.append(txt) extract_content(content_div) content = '\n\n'.join(parts) if parts else content_div.get_text(separator='', strip=True) if not content or len(content) < 20: return None # Title from detail page title tag title_tag = soup.find('title') raw_title = title_tag.get_text(strip=True) if title_tag else '' full_title = re.sub(r'\s*[-_―]\s*.*', '', raw_title).strip() if not full_title or len(full_title) < len(item['title']): full_title = item['title'] # Date from publish date if not item['date']: info_div = soup.select_one('div.nr-info') if info_div: m = re.search(r'(\d{4}-\d{2}-\d{2})', info_div.get_text()) if m: item['date'] = m.group(1) return { 'title': full_title, 'content': content, 'date': item['date'], 'url': item['url'], 'site_name': SITE_NAME, } def save_to_db(records, db_path=DB_PATH): if not records: return 0 conn = sqlite3.connect(db_path, timeout=60) c = conn.cursor() inserted = 0 for rec in records: try: c.execute( "INSERT OR IGNORE INTO gov_raw (title, content, page_url, publish_date, site_name) " "VALUES (?, ?, ?, ?, ?)", (rec['title'], rec['content'], rec['url'], rec['date'], rec['site_name']) ) if c.rowcount > 0: inserted += 1 except Exception as e: print(f" DB error: {e}") conn.commit() conn.commit() conn.close() return inserted def main(): import argparse parser = argparse.ArgumentParser() parser.add_argument('--start', type=int, default=1, help='Start page') parser.add_argument('--pages', type=int, default=66, help='Max pages to crawl') parser.add_argument('--days', type=int, default=None, help='Crawl items within last N days (default: current calendar month)') parser.add_argument('--db', type=str, default=None, help='SQLite DB path (default: server path)') args = parser.parse_args() db_path = args.db if args.db else DB_PATH # Date filter: sk-env.com requires login for articles older than current calendar month. # Verified: Aug items 200 OK, Jul items 302 -> membernews-login. Use natural month by default. if args.days is None: cutoff = datetime.now().strftime('%Y-%m-01') cutoff_desc = f"current calendar month (>= {cutoff})" else: cutoff = (datetime.now() - timedelta(days=args.days)).strftime('%Y-%m-%d') cutoff_desc = f"last {args.days} days (>= {cutoff})" print(f"=== {SITE_NAME} ===") print(f"Pages: {args.start} to {args.start + args.pages - 1}") print(f"Date cutoff: {cutoff_desc} (older items need login, 302)") total_inserted = 0 total_fresh = 0 for page_num in range(args.start, args.start + args.pages): if page_num == 1: page_url = BASE_URL else: page_url = f"http://www.sk-env.com/index.php/list-article-cid-238-p-{page_num}.html" print(f"\nPage {page_num}...") items = crawl_page(page_url) if items is None: print(f" Stopped at page {page_num}") break if not items: print(f" Empty page {page_num}") continue print(f" Found {len(items)} items") # Filter by date: skip items older than cutoff (they require login anyway) fresh_items = [it for it in items if it['date'] >= cutoff] skipped = len(items) - len(fresh_items) if skipped: print(f" Skipped {skipped} older than {cutoff} (login-protected)") if not fresh_items: # List is date-descending; once a page has no fresh items, stop print(f" No items within cutoff ({cutoff}), stopping (list is date-descending)") break print(f" Fresh: {len(fresh_items)} items") total_fresh += len(fresh_items) batch = [] for item in fresh_items: detail = fetch_detail(item) if detail: batch.append(detail) time.sleep(0.3) if batch: inserted = save_to_db(batch, db_path) total_inserted += inserted print(f" Inserted {inserted}/{len(batch)} new") else: print(f" 0 inserted (all login-protected)") time.sleep(0.5) print(f"\n{'='*40}") print(f"Fresh items found: {total_fresh}") print(f"Total inserted: {total_inserted}") if __name__ == '__main__': main()