boundaries as \n\n separators.""" parts = [] inner_ps = element.find_all("p", recursive=True) if inner_ps: for p in inner_ps: txt = p.get_text(" ", strip=True) if txt: parts.append(txt) else: txt = element.get_text(" ", strip=True) if txt: parts.append(txt) return "\n\n".join(parts) def extract_list_page(page_num): if page_num == 1: url = BASE_URL + "index.shtml" else: url = f"{BASE_URL}index.shtml?page={page_num}" try: r = fetch(url) except Exception as e: print(f" [ERROR] Page {page_num}: {e}", flush=True) return [] if r.status_code != 200: print(f" [ERROR] HTTP {r.status_code} page {page_num}", flush=True) return [] soup = BeautifulSoup(r.text, "html.parser") items = [] # Find the table with article links for table in soup.find_all("table"): for tr in table.find_all("tr"): tds = tr.find_all("td") if len(tds) >= 2: a_title = tds[0].find("a", href=re.compile(r"system/\d{4}")) if not a_title: continue href = a_title.get("href", "").strip() title = a_title.get_text(strip=True) # Clean bullet prefix title = re.sub(r'^[\s•\xa0·]+', '', title).strip() if not title or not href: continue full_url = urljoin(BASE_URL, href) # Date from second td date_str = tds[1].get_text(strip=True) if len(tds) > 1 else "" items.append({"title": title, "url": full_url, "date": date_str}) print(f" Page {page_num}: {len(items)} items", flush=True) return items def extract_detail(url): try: r = fetch(url) except Exception as e: print(f" [ERROR] {url}: {e}", flush=True) return None if r.status_code != 200: return None soup = BeautifulSoup(r.text, "html.parser") # Title from meta title = "" mt = soup.find("meta", attrs={"name": "ArticleTitle"}) if mt and mt.get("content"): title = mt["content"].strip() if not title: mt = soup.find("meta", attrs={"name": "description"}) if mt and mt.get("content"): title = mt["content"].strip() if not title: h1 = soup.find("h1") if h1: title = h1.get_text(strip=True) # Date from meta pub_date = "" md = soup.find("meta", attrs={"name": "PubDate"}) if md and md.get("content"): pub_date = md["content"].strip() pub_date = pub_date.replace(" 00:00:00", "").strip() if not pub_date: md = soup.find("meta", attrs={"name": "publishdate"}) if md and md.get("content"): pub_date = md["content"].strip() # Content from div.article-main > div.article-content or div.article-main directly content_text = "" body = soup.select_one("div.article-content") if not body: body = soup.select_one("div.article-main") if body: parts = [] for child in body.children: if child.name is None: continue tn = child.name.lower() if tn == "p": txt = child.get_text(" ", strip=True) if txt: parts.append(txt) elif tn == "div": # Check for inner text txt = child.get_text(" ", strip=True) if txt: parts.append(txt) elif tn == "table": md = table_to_markdown(child) if md: parts.append(md) elif tn in ("script", "style", "br"): continue content_text = "\n\n".join(parts) content_text = re.sub(r'\n{3,}', '\n\n', content_text) content_text = re.sub(r'扫一扫在手机打开当前页面', '', content_text).strip() # Attachments attachments_list = [] for a_tag in soup.find_all("a", href=True): h = a_tag["href"] if re.search(r'\.(pdf|doc|docx|xls|xlsx|rar|zip)$', h, re.IGNORECASE): at = a_tag.get_text(strip=True) or h.split("/")[-1] fu = urljoin(url, h) attachments_list.append({"name": at, "url": fu}) if len(content_text.strip()) < 20: content_text = f'
\n（本文为PDF附件）' return { "title": title, "content": content_text, "pub_date": pub_date, "attachments": json.dumps(attachments_list, ensure_ascii=False) if attachments_list else "", } def crawl(test_mode=False, max_pages=MAX_PAGES): print(f"[{SITE_NAME}] Starting, max_pages={max_pages}", flush=True) conn = init_db() cur = conn.cursor() total_inserted = total_skipped = pages_crawled = 0 for pn in range(1, max_pages + 1): items = extract_list_page(pn) if not items: print(f" Page {pn} empty, stop", flush=True) break pages_crawled += 1 for item in items: title, url, date = item["title"], item["url"], item["date"] cur.execute("SELECT id FROM gov_raw WHERE page_url = ?", (url,)) if cur.fetchone(): total_skipped += 1 continue detail = extract_detail(url) if detail is None: total_skipped += 1 continue if detail["title"] and len(detail["title"]) > len(title): title = detail["title"] content = detail["content"] attachments = detail["attachments"] cur.execute(INSERT_SQL, ( SITE_NAME, url, title, date or detail["pub_date"], content, CATEGORY, attachments, )) total_inserted += 1 if test_mode and total_inserted >= 10: break conn.commit() if test_mode and total_inserted >= 10: print(f" Test: stop after {total_inserted}", flush=True) break time.sleep(1) conn.close() print(f"[{SITE_NAME}] Done. Inserted={total_inserted}, Skipped={total_skipped}, Pages={pages_crawled}", flush=True) return total_inserted if __name__ == "__main__": test_mode = "--test" in sys.argv mp = 3 if test_mode else MAX_PAGES if "--max-pages" in sys.argv: idx = sys.argv.index("--max-pages") if idx + 1 < len(sys.argv): mp = int(sys.argv[idx + 1]) crawl(test_mode=test_mode, max_pages=mp)