#!/usr/bin/env python3 """南票区 (npq.gov.cn) 通知公告 TRS WCM 静态分页""" import re, sys, json, time from datetime import datetime, timedelta from urllib.parse import urljoin from concurrent.futures import ThreadPoolExecutor, as_completed import requests from bs4 import BeautifulSoup IP_RESOLVE = "42.176.201.44" HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9", } CUTOFF = datetime.now() - timedelta(days=1095) MAX_PAGES = 30 SITE_NAME = "南票区通知公告" MAX_WORKERS = 8 def fetch(url, session): resolved_url = url.replace("www.npq.gov.cn", IP_RESOLVE) try: resp = session.get(resolved_url, headers={**HEADERS, "Host": "www.npq.gov.cn"}, timeout=30, verify=False) resp.encoding = "utf-8" return resp.text except Exception as e: print(" [WARN] fetch failed: " + str(e), file=sys.stderr) return None def parse_list_page(html, base_url): items = [] soup = BeautifulSoup(html, "html.parser") for li in soup.select("ul.col-list-txt li"): a = li.find("a") if not a or not a.get("href"): continue title = a.get("title", "").strip() or a.get_text(strip=True) full_url = urljoin(base_url, a["href"].strip()) date_span = li.find("span", class_="datetime") date_str = date_span.get_text(strip=True) if date_span else "" if title and full_url: items.append((title, full_url, date_str)) return items def parse_detail(url, html): soup = BeautifulSoup(html, "html.parser") title_tag = soup.find("h1") title = title_tag.get_text(strip=True) if title_tag else "" if not title: meta = soup.find("meta", attrs={"name": "ArticleTitle"}) if meta and meta.get("content"): title = meta["content"].strip() date_str = "" meta_date = soup.find("meta", attrs={"name": "PubDate"}) if meta_date and meta_date.get("content"): date_str = meta_date["content"].strip() if not date_str: time_el = soup.select_one(".content_source .time") if time_el: m = re.search(r"(\d{4}-\d{2}-\d{2})", time_el.get_text()) if m: date_str = m.group(1) content_div = soup.select_one(".content_con .TRS_Editor") or soup.select_one(".content_con") content_html = str(content_div) if content_div else "" return title, date_str, content_html def date_from_str(s): if not s: return None for fmt in ["%Y-%m-%d", "%Y-%m-%d %H:%M", "%Y/%m/%d", "%Y年%m月%d日"]: try: return datetime.strptime(s.strip()[:19], fmt) except: continue return None def main(): import sqlite3, urllib3 urllib3.disable_warnings() session = requests.Session() session.headers.update(HEADERS) all_items, empty_pages = [], 0 for page_idx in range(MAX_PAGES): url = "https://www.npq.gov.cn/xwzx/tzgg/index_{}.html".format(page_idx) if page_idx else "https://www.npq.gov.cn/xwzx/tzgg/index.html" print(" Page {}: {}".format(page_idx+1, url), file=sys.stderr) html = fetch(url, session) if not html: break items = parse_list_page(html, url) if not items: empty_pages += 1 if empty_pages >= 2: break else: empty_pages = 0 all_items.extend(items) print(" {} items (total: {})".format(len(items), len(all_items)), file=sys.stderr) time.sleep(0.3) print("Total items: {}".format(len(all_items)), file=sys.stderr) conn = sqlite3.connect("/root/search.db", timeout=60) existing = set() try: for row in conn.execute("SELECT page_url FROM gov_raw WHERE site_name=?", (SITE_NAME,)): existing.add(row[0]) except: pass to_fetch = [item for item in all_items if item[1] not in existing] skipped_dup = len(all_items) - len(to_fetch) print("Dup: {}, to fetch: {}".format(skipped_dup, len(to_fetch)), file=sys.stderr) results, skipped_date = [], 0 with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool: def process(item): title, url, list_date_str = item html = fetch(url, session) if not html: return None try: dt, dds, ch = parse_detail(url, html) ft = dt or title fds = dds or list_date_str fd = date_from_str(fds) if fd and fd < CUTOFF: return {"skip": "date"} return {"page_url": url, "title": ft, "content": ch, "publish_date": fd.strftime("%Y-%m-%d") if fd else "", "site_name": SITE_NAME, "source_url": url} except Exception as e: print(" [ERR] {}: {}".format(url, e), file=sys.stderr) return None fut_map = {pool.submit(process, item): item for item in to_fetch} for fut in as_completed(fut_map): r = fut.result() if r is None: continue if r.get("skip") == "date": skipped_date += 1; continue results.append(r) print("New: {}, Dup: {}, Date-skip: {}".format(len(results), skipped_dup, skipped_date), file=sys.stderr) if results: conn.execute("BEGIN") now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") for r in results: try: conn.execute("""INSERT OR IGNORE INTO gov_raw (site_name, source_url, page_url, title, publish_date, content, date_rank) VALUES (?,?,?,?,?,?,?)""", (r["site_name"], r["source_url"], r["page_url"], r["title"], r["publish_date"], r["content"], 0 if not r["publish_date"] else int(r["publish_date"].replace("-", "")))) except Exception as e: print(" [ERR] insert: {}".format(e), file=sys.stderr) conn.commit() conn.execute("DELETE FROM gov_search WHERE site_name=?", (SITE_NAME,)) conn.execute("""INSERT INTO gov_search(rowid, site_name, title, summary) SELECT rowid, site_name, title, content FROM gov_raw WHERE site_name=?""", (SITE_NAME,)) conn.commit() print("FTS rebuilt", file=sys.stderr) conn.close() print("Done. +{}".format(len(results)), file=sys.stderr) if __name__ == "__main__": main()