#!/usr/bin/env python3 """Crawler: yfyunchengqu.gov.cn - 腰古镇政府信息公开平台 (广东统一gkmlpt)""" import argparse, json, os, re, sys, subprocess, ssl, time, traceback from urllib.request import Request, urlopen from urllib.error import HTTPError, URLError from datetime import datetime SITE_NAME = "云浮市云城区腰古镇" GROUP = "广东" INDUSTRY = "政府公告" DOMAIN = "www.yfyunchengqu.gov.cn" IP = "113.104.21.155" SITE_PREFIX = "yfycyg" SID = "766048" COLUMN_ID = "3637" SEARCH_DB = os.getenv("SEARCH_DB", "/mnt/data/search.db") DELAY = 0.5 _CTX = ssl.create_default_context() def http_get(url, headers=None, retries=3): for attempt in range(retries): try: req = Request(url, headers=headers or { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "http://{}/{}/gkmlpt/".format(DOMAIN, SITE_PREFIX), }) resp = urlopen(req, timeout=20, context=_CTX) return resp.read().decode("utf-8", errors="replace") except Exception as e: if attempt < retries - 1: time.sleep(2 ** attempt) continue raise def parse_date(raw): """Parse date from gkmlpt API format (Unix timestamp or formatted string)""" if not raw: return "" raw = str(raw).strip() # Unix timestamp (10 digits) m = re.match(r"^(\d{10})$", raw) if m: ts = int(m.group(1)) return datetime.fromtimestamp(ts).strftime("%Y-%m-%d") # Already formatted YYYY-MM-DD m = re.match(r"(\d{4})-(\d{2})-(\d{2})", raw) if m: return "{}-{:02d}-{:02d}".format(int(m.group(1)), int(m.group(2)), int(m.group(3))) return raw[:10] def extract_content(html): """Extract article content from detail page""" from bs4 import BeautifulSoup soup = BeautifulSoup(html, "html.parser") # Title: meta ArticleTitle or second h1 title = "" meta = soup.find("meta", attrs={"name": "ArticleTitle"}) if meta and meta.get("content"): title = meta["content"].strip() if not title: h1s = soup.find_all("h1") if h1s and len(h1s) >= 2: title = h1s[1].get_text(strip=True) elif h1s: title = h1s[0].get_text(strip=True) # Date: meta PubDate pub_date = "" meta = soup.find("meta", attrs={"name": "PubDate"}) if meta and meta.get("content"): pub_date = parse_date(meta["content"]) # Content: div.article-content content = "" content_div = soup.find("div", class_="article-content") if not content_div: content_div = soup.find("div", class_="content") if content_div: parts = [] for elem in content_div.find_all(["p", "table"], recursive=True): if elem.name == "p": if elem.find_parent("table"): continue text = elem.get_text("", strip=True) if text: parts.append(text) elif elem.name == "table": parts.append(str(elem)) content = "\n\n".join(parts) # Summary summary = content[:300].replace("\n", " ") if content else title[:300] return title, pub_date, content, summary def fetch_articles(max_pages=5): """Fetch article list from gkmlpt API""" all_articles = [] for page in range(1, max_pages + 1): api_url = "http://{}/{}/gkmlpt/api/all/{}?page={}&sid={}".format( DOMAIN, SITE_PREFIX, COLUMN_ID, page, SID) try: data = http_get(api_url) result = json.loads(data) articles = result.get("articles", []) total = result.get("total", 0) if not articles: print(f" [PAGE {page}] No articles (total={total})") break all_articles.extend(articles) print(f" [PAGE {page}] Got {len(articles)} articles (running: {len(all_articles)}/{total})") if len(all_articles) >= total: break time.sleep(DELAY) except Exception as e: print(f" [PAGE {page}] ERROR: {e}") if page == 1: raise break return all_articles def process_article(article): """Fetch detail page and extract info""" url = article.get("url", "") if not url: return None title = article.get("title", "").strip() # Use API date as fallback pub_date = parse_date(article.get("date") or article.get("create_time") or article.get("created_at", "")) # Fetch detail page for full title/content try: html = http_get(url) detail_title, detail_date, content, summary = extract_content(html) if detail_title: title = detail_title if detail_date: pub_date = detail_date if not content and not summary: print(f" [SKIP] No content: {title[:50]}") return None except Exception as e: print(f" [ERROR] Detail: {url[:60]} - {e}") return None return { "title": title, "url": url, "pub_date": pub_date, "content": content, "summary": summary, "site_name": SITE_NAME, "group_name": GROUP, "industry": INDUSTRY, } def push_to_db(items, batch_label="crawl"): """Write to search.db - gov_raw only (FTS sync handled separately)""" if not items: print(" [SKIP] no data") return lines = [] for item in items: esc = lambda s: str(s).replace("'", "''") if s else "" url = esc(item["url"]) title = esc(item["title"]) content = esc(item["content"]) summary = esc(item["summary"]) site = esc(item["site_name"]) group_ = esc(item["group_name"]) industry = esc(item["industry"]) pub_date = esc(item["pub_date"]) lines.append( f"INSERT OR IGNORE INTO gov_raw(source_url,page_url,title,content,summary,site_name,group_name,industry,publish_date) " f"VALUES('{url}','{url}','{title}','{content}','{summary}','{site}','{group_}','{industry}','{pub_date}');" ) all_sql = "\n".join(lines) for attempt in range(3): r = subprocess.run(["sqlite3", "-cmd", ".timeout 60000", SEARCH_DB], input=all_sql, capture_output=True, text=True, timeout=30) if r.returncode == 0: break if "locked" in r.stderr: import time; time.sleep(3) continue break else: print(f" [DB ERROR] {r.stderr.strip()[:200]}") return print(f" [DB] {len(items)} records written") def main(max_pages, limit): print(f"[腰古镇gkmlpt] Start (max_pages={max_pages})") articles = fetch_articles(max_pages=max_pages) print(f" API returned {len(articles)} articles") if limit and limit < len(articles): articles = articles[:limit] print(f" Limited to {limit}") items = [] ok, err = 0, 0 for i, article in enumerate(articles): print(f" [{i+1}/{len(articles)}] Processing...", end=" ") item = process_article(article) if item: items.append(item) ok += 1 print(f"OK: {item['title'][:50]}") else: err += 1 print("SKIP") time.sleep(DELAY) print(f"\nDone: {ok} OK, {err} skipped") push_to_db(items, batch_label="yfyunchengqu") if __name__ == "__main__": parser = argparse.ArgumentParser(description="腰古镇gkmlpt爬虫") parser.add_argument("--pages", type=int, default=5, help="最大页数") parser.add_argument("--limit", type=int, default=0, help="限制条数") args = parser.parse_args() main(max_pages=args.pages, limit=args.limit)