#!/usr/bin/env python3 """土默特右旗人民政府 - 通知公告 爬虫 v3 (并发版) 列表: ul.list > li > a > span (title) + span (date) 详情: div.main_box > h1, div.cont > div.trs_editor_view (content) """ import sys, os, re, time, json, logging, argparse, sqlite3 from datetime import datetime, timedelta from urllib.parse import urljoin from concurrent.futures import ThreadPoolExecutor, as_completed import requests from bs4 import BeautifulSoup BASE_URL = "http://www.tmtyq.gov.cn/ywdt/tzgg" SITE_NAME = "土默特右旗-通知公告" MAX_PAGES = 45 DATE_CUTOFF = (datetime.now() - timedelta(days=3*365)).strftime("%Y-%m-%d") DB_PATH = "/root/search.db" NUM_THREADS = 5 logging.basicConfig(level=logging.INFO, format='[%(asctime)s] %(levelname)s %(message)s', datefmt='%H:%M:%S') log = logging.getLogger(__name__) HEADERS = {'User-Agent': 'Mozilla/5.0'} def init_db(): db = sqlite3.connect(DB_PATH, timeout=60) db.execute("""CREATE TABLE IF NOT EXISTS gov_raw ( id INTEGER PRIMARY KEY AUTOINCREMENT, site_name TEXT NOT NULL, title TEXT NOT NULL, page_url TEXT NOT NULL UNIQUE, publish_date TEXT, source_url TEXT, content TEXT, crawl_time TEXT DEFAULT (datetime('now','localtime')) )""") db.commit() return db def fetch(url): try: r = requests.get(url, timeout=20, headers=HEADERS) r.encoding = 'utf-8' return r.text except Exception as e: log.error(f"请求失败 {url[:60]}: {e}") return None def parse_list_items(html): soup = BeautifulSoup(html, 'html.parser') items = [] ul = soup.select_one('ul.list') if not ul: return items for li in ul.select('li'): a = li.find('a') if not a: continue href = a.get('href', '') if not href or href == '#': continue page_url = urljoin(BASE_URL + '/', href) spans = a.find_all('span') title = spans[0].get_text(strip=True) if spans else a.get_text(strip=True) pub_date = spans[1].get_text(strip=True) if len(spans) > 1 else '' pub_date = re.sub(r'[^\d-]', '', pub_date) items.append({'title': title, 'page_url': page_url, 'publish_date': pub_date}) return items def get_total_pages(html): m = re.search(r"total:\s*'(\d+)'", html) if m: return (int(m.group(1)) + 9) // 10 return MAX_PAGES def fetch_detail(item): """独立详情爬取函数（线程安全）""" try: html = fetch(item['page_url']) if not html: return None soup = BeautifulSoup(html, 'html.parser') h1 = soup.select_one('div.cont_items h1') title = h1.get_text(strip=True) if h1 else '' dm = re.search(r'发文时间[：:]\s*(\d{4}-\d{2}-\d{2})', html) pub_date = dm.group(1) if dm else '' cont = soup.select_one('div.cont') content_html = '' if cont: for s in cont.select('script,style'): s.decompose() content_html = str(cont).strip() else: main = soup.select_one('div.main_box') if main: for s in main.select('h1,div.time_source,div.break_up_bg,div.fun_icon,h3,script,style'): s.decompose() content_html = str(main).strip() content_html = re.sub(r'\s+', ' ', content_html).strip() item['title'] = title or item['title'] item['publish_date'] = pub_date or item['publish_date'] item['content'] = content_html item['site_name'] = SITE_NAME item['source_url'] = "土默特右旗人民政府" return item except Exception as e: log.error(f"详情失败 {item['page_url'][:50]}: {e}") return None def save_batch(db, items): saved = 0 for item in items: db.execute("INSERT OR IGNORE INTO gov_raw (site_name, title, page_url, publish_date, source_url, content) VALUES (?,?,?,?,?,?)", (item['site_name'], item['title'], item['page_url'], item.get('publish_date',''), item.get('source_url',''), item.get('content',''))) if db.cursor().rowcount: saved += 1 db.commit() return saved def crawl(full=True): db = init_db() # 第1页 html1 = fetch(BASE_URL + '/index.html') if not html1: return {"total": 0, "saved": 0, "skipped": 0} items_1 = parse_list_items(html1) total_pages = get_total_pages(html1) pages = min(total_pages, MAX_PAGES) if full else 1 log.info(f"第1页: {len(items_1)} 条, 共{pages}页") # 收集列表 all_items = [] for page in range(pages): if page == 0: items = items_1 else: h = fetch(f"{BASE_URL}/index_{page}.html") if not h: continue items = parse_list_items(h) log.info(f"列表页 {page+1}/{pages}: {len(items)} 条") all_items.extend(items) time.sleep(0.3) # 去重 seen = set() unique = [] for item in all_items: if item['page_url'] not in seen: seen.add(item['page_url']) unique.append(item) log.info(f"去重后共 {len(unique)} 条") # 过滤已存在和过期 to_crawl = [] for item in unique: if item['publish_date'] and item['publish_date'] < DATE_CUTOFF: continue if db.execute("SELECT 1 FROM gov_raw WHERE page_url=?", (item['page_url'],)).fetchone(): continue to_crawl.append(item) log.info(f"待爬详情: {len(to_crawl)} 条") # 并发详情 saved_count = 0 with ThreadPoolExecutor(max_workers=NUM_THREADS) as executor: futures = {executor.submit(fetch_detail, item): i for i, item in enumerate(to_crawl)} batch = [] for i, future in enumerate(as_completed(futures)): result = future.result() if result: batch.append(result) log.info(f"详情 [{i+1}/{len(to_crawl)}]: {result['title'][:45]}... ✓") else: log.info(f"详情 [{i+1}/{len(to_crawl)}]: ❌") # 每20条批量写入 if len(batch) >= 20: saved_count += save_batch(db, batch) batch = [] # 最后一批 if batch: saved_count += save_batch(db, batch) log.info(f"完成: crawled={len(to_crawl)}, saved={saved_count}") return {"total": len(to_crawl), "saved": saved_count, "skipped": len(unique) - len(to_crawl)} if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument('--full', action='store_true') parser.add_argument('--incremental', type=int, default=1) args = parser.parse_args() result = crawl(full=args.full or args.incremental > 1) print(json.dumps(result, ensure_ascii=False))