#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 济南市商河县人民政府 - 通知公告爬虫列表: http://www.shanghe.gov.cn/col/col22822/index.html Hanweb 6.4.2.4 动态构建: GET /api-gateway/jpaas-publish-server/front/page/build/unit queryData: parseType=bulidstatic, webId=87, tplSetId=NcANWdo2pAbHvfVcDfN0x, pageType=column, tagId=当前列表, editType=null, pageId=22822 分页: paramJson={"pageNo":N,"pageSize":M} (默认15/页) 列表 HTML:

标题YYYY-MM-DD

详情: /col/col22822/art/YYYY/art_xxx.html 标题:

([^<]+)

", html) if m: title = clean_title(m.group(1)) if not title: m = re.search(r"([^<]+)", html) if m: title = clean_title(m.group(1).split("_")[0]) # 日期 pub_date = "" m = re.search(r"时间[：:]\s*(\d{4}-\d{1,2}-\d{1,2})", html) if m: pub_date = m.group(1) if not pub_date: m = re.search(r"(\d{4})-(\d{1,2})-(\d{1,2})", html) if m: pub_date = "%s-%02d-%02d" % (m.group(1), int(m.group(2)), int(m.group(3))) # 正文: div#zoom body = "" i = html.find('id="zoom"') if i >= 0: start = html.find(">", i) + 1 depth = 1 j = start while j < len(html) and depth > 0: if html[j:j+4] == "": depth -= 1 j += 6 else: j += 1 raw = html[start:j-6] raw = re.sub(r"", "", raw) raw = re.sub(r"", "", raw) # 附件: 用 data-link 真实路径优先, 否则 download-m 加密 URL def fix_attach(m): href = m.group(1) att = m.group(0) dl = re.search(r'data-link="([^"]+)"', att) if dl: real = dl.group(1) if not real.startswith("http"): real = urljoin(BASE_URL, real) return att.replace(m.group(1), real) if not href.startswith(("http", "#", "javascript")): return att.replace(href, urljoin(url, href)) return att raw = re.sub(r']*href="([^"]*)"[^>]*>', fix_attach, raw) raw = re.sub(r'href="([^"]*)"', lambda m: 'href="%s"' % (urljoin(url, m.group(1)) if not m.group(1).startswith(("http", "#", "javascript")) else m.group(1)), raw) raw = re.sub(r'src="([^"]*)"', lambda m: 'src="%s"' % (urljoin(url, m.group(1)) if not m.group(1).startswith(("http", "data:", "javascript")) else m.group(1)), raw) raw = re.sub(r'\sstyle="[^"]*"', "", raw) raw = re.sub(r"]*>||]*>||]*>||]*>|", "", raw) body = raw.strip() if not body or len(body) < 20: body = "" return title, pub_date, body except Exception as e: print(f" [ERR] detail {url}: {e}", flush=True) return "", "", "" def main(): pages, limit = parse_args() print(f"[SHANGHE] SITE={SITE_NAME} pages={pages} limit={limit}", flush=True) session = requests.Session() session.headers.update({"User-Agent": UA}) all_items = [] seen = set() if pages <= 0: pages = 1 for pg in range(1, pages + 1): items = fetch_page(session, pg) new_count = 0 for it in items: if it["href"] in seen: continue seen.add(it["href"]) all_items.append(it) new_count += 1 print(f" [page {pg}] {len(items)}条(新{new_count}), 累计{len(all_items)}", flush=True) if not items: print(f" [page {pg}] 空页, 停止翻页", flush=True) break time.sleep(0.4) before = len(all_items) all_items = [it for it in all_items if (it.get("date") or "9999") >= CUTOFF] if len(all_items) < before: print(f" CUTOFF {CUTOFF} 过滤 {before - len(all_items)} 条", flush=True) print(f"列表完成: {len(all_items)} 条, 抓详情...", flush=True) if limit and len(all_items) > limit: all_items = all_items[:limit] results = [] for i, it in enumerate(all_items): d_title, d_date, body = fetch_detail(session, it["href"]) title = d_title or it["title"] pub_date = d_date or it.get("date", "") summary = re.sub(r"<[^>]+>", " ", body) if body else title summary = re.sub(r"\s+", " ", summary).strip()[:300] atts = [] if body: # 附件: 优先从 fileName= URL 参数提取文件名 (空文本链接无文字可提取) for m in re.finditer(r']*href="([^"]*\.(?:pdf|doc|docx|xls|xlsx|zip|rar|ofd|wps)[^"]*)"[^>]*>([^<]*)', body, re.I): href = m.group(1) name = m.group(2).strip() if not name: fm = re.search(r"[?&]fileName=([^&]+)", href) if fm: from urllib.parse import unquote name = unquote(fm.group(1)) if not name: continue # 空名附件跳过 (重复链接) atts.append({"name": name, "url": href}) results.append({ "site_name": SITE_NAME, "source_url": it["href"], "url": it["href"], "title": title, "pub_date": pub_date, "content": body, "summary": summary, "category": CATEGORY, "attachments": json.dumps(atts, ensure_ascii=False) if atts else "", }) if (i + 1) % 10 == 0: print(f" detail {i+1}/{len(all_items)}", flush=True) time.sleep(0.3) print(f" pushing {len(results)} items to searchdb...", flush=True) push_to_searchdb(results, batch_label=SITE_NAME) empty = sum(1 for r in results if not (r.get("content") or "").strip()) print(f" DONE. pushed={len(results)} 空正文={empty}", flush=True) if __name__ == "__main__": main()