#!/usr/bin/env python3 """ 莱阳市-建设项目环评审批 爬虫 (JPAAS Publish-Server) """ import requests, json, re, sqlite3, time, sys from bs4 import BeautifulSoup from urllib.parse import quote import warnings warnings.filterwarnings('ignore') from datetime import datetime, timedelta import os SEARCH_DB = os.getenv("SEARCH_DB", "/root/search.db") SITE_NAME = "莱阳-建设项目环评审批" DOMAIN = "www.laiyang.gov.cn" API = f"https://{DOMAIN}/api-gateway/jpaas-publish-server/front/page/build/unit" PAGE_SIZE = 25 CUTOFF = (datetime.now() - timedelta(days=3*365)).strftime("%Y-%m-%d") HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": f"https://{DOMAIN}/col/col43923/index.html"} BASE_PARAMS = { "parseType": "bulidstatic", "webId": "162", "tplSetId": "dNL352FK70O2rT296SQOw", "pageType": "column", "tagId": "当前栏目列表", "editType": "null", "pageId": "43923", } def fetch_list_page(page): params = dict(BASE_PARAMS) params["paramJson"] = json.dumps({"pageNo": page, "pageSize": PAGE_SIZE}, ensure_ascii=False) for retry in range(3): try: r = requests.get(API, params=params, headers=HEADERS, timeout=30) if r.status_code == 200: d = r.json() if not d.get("success"): return [], 0 html = d["data"]["html"] soup = BeautifulSoup(html, 'html.parser') items = [] for li in soup.find_all('li'): a = li.find('a') if not a: continue href = a.get('href', '').strip() title = a.get_text(strip=True) span = li.find('span') date = span.get_text(strip=True) if span else "" if not href.startswith('http'): if href.startswith('/'): href = f"https://{DOMAIN}{href}" else: href = f"https://{DOMAIN}/{href}" items.append({"url": href, "title": title, "date": date}) # Total count from pagination div cm = re.search(r'count="(\d+)"', html) count = int(cm.group(1)) if cm else len(items) total_pages = (count + PAGE_SIZE - 1) // PAGE_SIZE return items, total_pages except Exception as e: if retry < 2: time.sleep(2) return [], 0 def extract_content(html): soup = BeautifulSoup(html, 'html.parser') zoom = soup.find('div', id='zoom') if zoom: for tag in zoom.find_all(['script', 'style']): tag.decompose() text = zoom.get_text(strip=True) if len(text) > 30: return str(zoom) bf = soup.find('div', id='barrierfree_container') if bf: for tag in bf.find_all(['script', 'style']): tag.decompose() text = bf.get_text(strip=True) if len(text) > 30: return str(bf) return "" def extract_meta(html): soup = BeautifulSoup(html, 'html.parser') title = "" tm = soup.find('meta', attrs={"name": "ArticleTitle"}) if tm and tm.get('content'): title = tm['content'].strip() if not title: tt = soup.find('title') if tt: title = tt.get_text(strip=True) date = "" pm = soup.find('meta', attrs={"name": "PubDate"}) if pm and pm.get('content'): d = pm['content'].strip() cm = re.match(r'(\d{4})-(\d{1,2})-(\d{1,2})', d) if cm: date = f"{cm.group(1)}-{cm.group(2).zfill(2)}-{cm.group(3).zfill(2)}" if not date: dm = re.search(r'(\d{4}-\d{2}-\d{2})', html) if dm: date = dm.group(1) return title, date def insert_article(url, title, date, content): db = sqlite3.connect(SEARCH_DB, timeout=60) db.execute("PRAGMA journal_mode=WAL") try: db.execute( "INSERT OR IGNORE INTO gov_raw(site_name,source_url,page_url,title,publish_date,content,summary,status,category) VALUES(?,?,?,?,?,?,?,?,?)", (SITE_NAME, url, url, title, date, content, title[:200], "active", "环评审批")) affected = db.total_changes db.commit() if affected: db.execute( "INSERT OR REPLACE INTO gov_search(rowid,title,site_name,summary) SELECT r.id,r.title,r.site_name,r.summary FROM gov_raw r WHERE r.page_url=? AND r.id NOT IN (SELECT rowid FROM gov_search)", (url,)) db.commit() db.close() return affected except Exception as e: db.close() print(f" DB error: {e}") return 0 def run(max_pages=40): print(f"\n{'='*50}") print(f"🚀 {SITE_NAME}") print(f"{'='*50}") items, total_pages = fetch_list_page(1) if not items: print("❌ 列表页无数据") return total_pages = min(max_pages, total_pages) print(f" 共{total_pages}页") all_items = list(items) for pg in range(2, total_pages + 1): its, _ = fetch_list_page(pg) if not its: break all_items.extend(its) dates = [it["date"] for it in its if it["date"]] if dates: print(f" 第{pg}页: {len(its)} 条 ({min(dates)} ~ {max(dates)})") else: print(f" 第{pg}页: {len(its)} 条") time.sleep(0.2) print(f"\n📊 共 {len(all_items)} 条") new = skip = no_body = err = 0 for i, item in enumerate(all_items, 1): if item["date"] and item["date"] < CUTOFF: skip += 1 continue for retry in range(3): try: r = requests.get(item["url"], headers=HEADERS, timeout=30) if r.status_code == 200: r.encoding = "utf-8" html = r.text break except: if retry < 2: time.sleep(2) html = None if not html: err += 1 continue pt, pd = extract_meta(html) real_title = pt if pt else item["title"] real_date = pd if pd else item["date"] if real_date and real_date < CUTOFF: skip += 1 continue content = extract_content(html) text_len = len(re.sub(r'<[^>]+>', '', content).strip()) if content else 0 if text_len < 10: no_body += 1 continue if insert_article(item["url"], real_title, real_date, content): new += 1 else: skip += 1 if i % 5 == 0: print(f" ...{i}/{len(all_items)}") time.sleep(0.2) print(f"\n📊 FTS同步...") db = sqlite3.connect(SEARCH_DB, timeout=60) # QC20260926 去掉手写 gov_search 整站删除(抢锁源; FTS 由 gov_raw 触发器维护) # db.execute("DELETE FROM gov_search WHERE site_name=?", (SITE_NAME,)) db.execute("INSERT OR REPLACE INTO gov_search(rowid,title,site_name,summary) SELECT rowid,title,site_name,summary FROM gov_raw WHERE site_name=?", (SITE_NAME,)) db.commit() c = db.execute("SELECT COUNT(*) FROM gov_search WHERE site_name=?", (SITE_NAME,)) cnt = c.fetchone()[0] db.close() print(f" FTS: {cnt} 条") print(f"\n✅ {SITE_NAME}: 新增{new}, 跳过{skip}, 空正文{no_body}, 错误{err}, FTS共{cnt}条") if __name__ == "__main__": mp = int(sys.argv[1]) if len(sys.argv) > 1 else 40 run(mp)