#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 兰州洁华环境评价咨询有限公司 lzjhhp.com — 公示公告栏目 (Index/news?sec_id=269) CMS: UminiCms (meta author) 列表: /Index/news?sec_id=269&p=N, 7条/页, 449条/65页 (P65=末页, 越界翻页显示同页) 列表项:
标题YYYY-MM-DD
详情: /Index/news?sec_id=269&id=NNN 标题: 列表标题 (详情页无 h1, meta title 是站名) 日期: 列表 span.date (详情页【发布时间：YYYY-MM-DD】备用) 正文:
内

段落 (保留

HTML) 附件: /Public/euploads/file/YYYYMMDD/...doc (相对路径 → urljoin 绝对化) CUTOFF: P1~P10 (P10 2023-09-08~2023-07-12 跨边界, 保留 2023-08-10 后) """ import argparse, json, os, re, sqlite3, subprocess, sys, time from datetime import datetime, timedelta from urllib.parse import urljoin UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36" SITE_NAME = "兰州洁华-公示公告" BASE_URL = "https://lzjhhp.com" LIST_URL = "https://lzjhhp.com/Index/news?sec_id=269" DB_PATH = "/root/search.db" CUTOFF = (datetime.now() - timedelta(days=365 * 3)).strftime("%Y-%m-%d") print(f"[i] CUTOFF 边界: {CUTOFF}", file=sys.stderr) def curl_get(url): for attempt in range(3): try: r = subprocess.run( ["curl", "-sk", "-m", "25", "-A", UA, "-L", url], capture_output=True, text=True, timeout=35) if r.stdout: return r.stdout except Exception as e: print(f"[!] curl err {e}", file=sys.stderr) time.sleep(1.5) return "" def parse_detail(url): """返回 (title, pub_date, content)""" h = curl_get(url) if not h: return None, None, "" # 日期: 详情页【发布时间】 pub_date = "" dm = re.search(r"【发布时间：(\d{4}-\d{2}-\d{2})", h) if dm: pub_date = dm.group(1) # 正文: div.res-info 内

段落 (保留 HTML) content = "" m = re.search(r'

(.*?)
\s*(.*?)
', h, re.S) if m: seg = m.group(1) imgs = [u for u in re.findall(r']+src="([^"]+)"', seg)] parts = [] for p in re.findall(r"]*>(.*?)

", seg, re.S): # 段落内附件链接 保留（提取后恢复），其他标签剥掉 atts = [] def _a_collect(mm): ah = mm.group(1) at = re.sub(r"<[^>]+>", "", mm.group(2)).strip() if "euploads" in ah: if not ah.startswith("http"): ah = urljoin(BASE_URL, ah) if not at or at.startswith("/"): at = os.path.basename(ah.split("?")[0]) atts.append((ah, at)) return f"\u0001{len(atts)-1}\u0001" # 占位符 return f'{at}' tmp = re.sub(r']+href="([^"]+)"[^>]*>(.*?)', _a_collect, p, flags=re.S) tmp = re.sub(r"<[^>]+>", "", tmp) tmp = re.sub(r"[ \t\u3000]+", " ", tmp) tmp = re.sub(r"\s+", " ", tmp).strip() # 恢复附件链接 for i, (ah, at) in enumerate(atts): tmp = tmp.replace(f"\u0001{i}\u0001", f'{at}') if tmp: parts.append(f"

{tmp}

") content = "\n".join(parts) # 无

但有图 → 图片型 if not content and imgs: for src in imgs: if not src.startswith("http"): src = urljoin(BASE_URL, src) content += f'

\n' content = content.strip() return "", pub_date, content def push_to_searchdb(records): conn = sqlite3.connect(DB_PATH, timeout=60) cur = conn.cursor() cur.execute("""CREATE TABLE IF NOT EXISTS gov_raw ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, publish_date TEXT, content TEXT, page_url TEXT, source_url TEXT, site_name TEXT, summary TEXT, date_rank INTEGER, UNIQUE(page_url, site_name))""") new = skip = 0 for rec in records: cur.execute( "INSERT OR IGNORE INTO gov_raw (title, publish_date, content, page_url, source_url, site_name, summary, date_rank) VALUES (?,?,?,?,?,?,?,?)", (rec["title"], rec["pub_date"], rec["content"], rec["url"], rec["url"], SITE_NAME, re.sub(r"<[^>]+>", "", rec["content"])[:200] if rec["content"] else "", rec["date_rank"]), ) if cur.rowcount: new += 1 else: skip += 1 conn.commit() conn.close() return new, skip def main(): ap = argparse.ArgumentParser() ap.add_argument("--pages", type=int, default=10) ap.add_argument("--start", type=int, default=1) args = ap.parse_args() total_new = total_skip = 0 for pg in range(args.start, args.start + args.pages): url = LIST_URL if pg == 1 else f"{LIST_URL}&p={pg}" h = curl_get(url) if not h: print(f"[P{pg}] 空响应 -> stop", file=sys.stderr) break items = re.findall( r'
\s*(.*?)\s*(\d{4}-\d{2}-\d{2})', h, re.S) if not items: print(f"[P{pg}] 无列表项 -> stop", file=sys.stderr) break print(f"[P{pg}] {len(items)} 条 ({items[0][3]} ~ {items[-1][3]})", file=sys.stderr) records = [] for href, idv, title_html, date_str in items: if date_str < CUTOFF: print(f"[P{pg}] 跳过窗口外 {date_str}", file=sys.stderr) continue title = re.sub(r"<[^>]+>", "", title_html).strip() detail_url = urljoin(BASE_URL, href) _, dpub, content = parse_detail(detail_url) if not dpub: dpub = date_str records.append({ "title": title, "pub_date": dpub, "content": content, "url": detail_url, "date_rank": int(datetime.strptime(dpub, "%Y-%m-%d").timestamp()), }) time.sleep(0.6) if records: n, s = push_to_searchdb(records) total_new += n total_skip += s print(f"[P{pg}] new={n} skip={s}", file=sys.stderr) time.sleep(0.8) print(f"DONE new={total_new} skip={total_skip}") if __name__ == "__main__": main()