#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 福建省工业和信息化厅 - 项目核准审查 爬虫 栏目: https://gxt.fujian.gov.cn/zwgk/jsxm/xmhz/ CMS: TRS WCM (list2.trsapi 动静结合) — was5 检索 JSON API 列表 API: /was5/web/search?channelid=229105&searchword=chnlid%3D61090&perpage=15&outlinepage=1&page=N&sortfield=-docreltime channelid=229105 (SiteInfo.searchId), chnlid=61090 (PageInfo.chnlid 栏目ID) total=552 条, 15条/页 (37页), sortfield=-docreltime 时间倒序 docs[] 字段: title(完整)/url/chnldocurl/content(截断~260字符)/pubdate/pubtime/file(附件P0)/filedesc/fileno(文号) ⚠️ 每页返回 16 条含 1 条坏条目 (url 非 http, date 空) → 过滤 url startswith http + pubdate 非空 详情: /zwgk/jsxm/xmhz/YYYYMM/tYYYYMMDD_ID.htm 正文容器:

...

⚠️ class=TRS_Editor 无引号; 内层 DIV 大写; style 块须剥; 段落

保留 附件: API file 字段 (P02026...pdf) + 详情页 article_attachment 区 下载：xxx.pdf ⚠️ 附件相对路径 urljoin(详情页URL, href) 绝对化 (保留栏目路径段) 标题: API title 字段完整 (福建省工业和信息化厅关于XX项目节能报告的审查意见) 用法: python3 crawl_fjgxt_xmhz.py --pages=1 python3 crawl_fjgxt_xmhz.py --pages=16 # CUTOFF 3年窗口 (2023-08-11 起) python3 crawl_fjgxt_xmhz.py # 默认 5 页增量 """ import os, sys, re, time, json, html as html_mod import requests from urllib.parse import urljoin sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from crawler_lib import push_to_searchdb BASE_URL = "https://gxt.fujian.gov.cn" API_TPL = (BASE_URL + "/was5/web/search?channelid=229105&searchword=chnlid%%3D61090" "&perpage=15&outlinepage=1&page=%d&sortfield=-docreltime") SITE_NAME = "福建省工信厅-项目核准审查" CATEGORY = "项目核准审查" CUTOFF = "2023-08-11" # 3 年窗口 MAX_PAGES = 37 # total=552, 15条/页 HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36", "Accept": "application/json, text/javascript, */*; q=0.01", "Accept-Language": "zh-CN,zh;q=0.9", "Referer": BASE_URL + "/zwgk/jsxm/xmhz/", } TIMEOUT = 30 def parse_args(): pages = 0 for a in sys.argv[1:]: if a.startswith("--pages="): pages = int(a.split("=")[1]) elif a.startswith("--pages") and len(sys.argv) > sys.argv.index(a) + 1: pages = int(sys.argv[sys.argv.index(a) + 1]) return pages def clean_title(t): t = html_mod.unescape(t or "") t = re.sub(r"[\u200b\u200e\u200f\ufeff\xa0]", "", t) t = re.sub(r"\s+", " ", t) t = re.sub(r"^[•·\-—]\s*", "", t) return t.strip() def fetch_api(page): url = API_TPL % page for attempt in range(3): try: r = requests.get(url, headers=HEADERS, timeout=TIMEOUT) if r.status_code == 200: return json.loads(r.text, strict=False) # was5 返回含控制字符, 须宽松解析 except Exception as e: if attempt == 2: print(f" [WARN] API p{page} failed: {e}", file=sys.stderr) return None time.sleep(1.5) return None def _abs_url(u, page_url): if u.startswith(("http://", "https://", "javascript:", "mailto:", "tel:", "data:", "#")): return u return urljoin(page_url, u) def clean_content(raw_html, page_url): """正文清洗: 保留

/ HTML, 附件/图片绝对化 (以详情页 URL 为基址保留栏目路径)""" if not raw_html: return "" raw_html = re.sub(r"", "", raw_html) raw_html = re.sub(r"", "", raw_html) # 剥 TRS 容器标签
等 (大写+无引号属性) raw_html = re.sub(r"]*>", "", raw_html, flags=re.I) # 附件/图片相对路径绝对化 raw_html = re.sub(r' 标签 raw_html = re.sub(r"]*>", "

", raw_html) # 剥 span 标签 (保留文本) raw_html = re.sub(r"]*>", "", raw_html) # 其余标签去属性 raw_html = re.sub(r"<(table|tbody|thead|tr|td|th|br)([^>]*)>", lambda m: f"<{m.group(1)}>", raw_html) raw_html = re.sub(r"", lambda m: f"", raw_html) # 剥残留标签 (⚠️ \b 边界防误伤 /) raw_html = re.sub(r"]*>", "", raw_html, flags=re.I) # 空白清理 raw_html = re.sub(r"[\u3000\xa0]", " ", raw_html) raw_html = re.sub(r"

\s*(?: )?\s*

", "", raw_html) raw_html = re.sub(r"\r\n", "\n", raw_html) return raw_html.strip() def extract_attachments(content, page_url): """从正文提取附件链接, 返回 JSON 数组字符串""" atts = [] seen = set() for am in re.finditer(r']*>(.*?)', content, re.S): ahref, atext = am.group(1), re.sub(r"<[^>]+>", "", am.group(2)).strip() if re.search(r"\.(pdf|doc|docx|xls|xlsx|zip|rar|wps|7z|txt)(\?|$)", ahref, re.I) or "download" in ahref.lower(): u = _abs_url(ahref, page_url) if u not in seen: seen.add(u) atts.append({"name": atext or u.split("/")[-1], "url": u}) return json.dumps(atts, ensure_ascii=False) if atts else "[]" def fetch_detail(url): """抓详情页: 返回 (完整正文, 附件列表) — API content 截断时补全; 附件从 article_attachment 区 + 正文提取""" for attempt in range(3): try: r = requests.get(url, headers=HEADERS, timeout=TIMEOUT) if r.status_code != 200: return "", [] r.encoding = "utf-8" html_text = r.text atts = [] # 附件区:
... 下载：xxx.pdf am = re.search(r'\s*
\s*', html_text) if am: for a2 in re.finditer(r']*>([^<]*)', am.group(0)): ahref = a2.group(1) if re.search(r"\.(pdf|doc|docx|xls|xlsx|zip|rar|wps|7z)(\?|$)", ahref, re.I): atext = re.sub(r"^下载[：:]\s*", "", a2.group(2).strip()) or ahref.split("/")[-1] atts.append({"name": atext, "url": _abs_url(ahref, url)}) # 正文容器:
... (class 无引号, 内层 DIV 大写) m = re.search(r'(.*?)
', html_text, re.S) if m: body = clean_content(m.group(1), url) # 正文内附件合并 (去重) for a3 in re.finditer(r']*>(.*?)', body, re.S): ahref = a3.group(1) if re.search(r"\.(pdf|doc|docx|xls|xlsx|zip|rar|wps|7z)(\?|$)", ahref, re.I): atext = re.sub(r"<[^>]+>", "", a3.group(2)).strip() or ahref.split("/")[-1] atts.append({"name": atext, "url": ahref}) # 去重附件 seen = set(); uniq = [] for a4 in atts: if a4["url"] not in seen: seen.add(a4["url"]); uniq.append(a4) return body, uniq return "", atts except Exception as e: if attempt == 2: print(f" [WARN] detail failed {url}: {e}", file=sys.stderr) return "", [] time.sleep(1.0) return "", [] def main(): pages = parse_args() if not pages: pages = 5 all_items = [] seen = set() for pg in range(1, pages + 1): data = fetch_api(pg) if not data: print(f"[PAGE {pg}] API failed, stopping") break docs = data.get("docs", []) # 过滤坏条目: url 非 http 或日期空 (was5 每页尾混 1 条垃圾) valid = [d for d in docs if str(d.get("url", "")).startswith("http") and d.get("pubdate")] if not valid: print(f"[PAGE {pg}] 0 valid items, stopping") break for d in valid: url = d["url"].strip().replace("http://", "https://") # 统一 https if url in seen: continue seen.add(url) title = clean_title(d.get("title", "")) if not title: continue pub_date = "" pm = re.search(r"(\d{4}-\d{2}-\d{2})", str(d.get("pubdate", ""))) if pm: pub_date = pm.group(1) if pub_date and pub_date < CUTOFF: continue # CUTOFF 3年窗口 content = clean_content(d.get("content", ""), url) attachments = [] # 正文过短 → 抓详情页补全 (API content 常截断 ~250 字符摘要; 详情页正文普遍 >800) if len(re.sub(r"<[^>]+>", "", content)) < 300: body, atts = fetch_detail(url) if body and len(re.sub(r"<[^>]+>", "", body)) > len(re.sub(r"<[^>]+>", "", content)): content = body attachments = atts if not attachments: # API file/filedesc 附件字段 f = d.get("file", "") if f and f.startswith("P0"): fname = (d.get("filedesc", "") or "").replace("下载：", "").strip() or f.split("/")[-1] attachments = [{"name": fname, "url": _abs_url(f, url)}] att_json = json.dumps(attachments, ensure_ascii=False) if attachments else "[]" all_items.append({ "site_name": SITE_NAME, "title": title, "pub_date": pub_date, "content": content, "source_url": url, "url": url, "attachments": att_json, "industry": "", }) print(f"[PAGE {pg}] +{len(valid)} valid (total {len(all_items)})") time.sleep(0.6) print(f"\n有效 {len(all_items)} 条, 入库...") push_to_searchdb(all_items, CATEGORY) print(f"完成! 新增/更新 {len(all_items)} 条") if __name__ == "__main__": main()