(.*?)<', html) if ttm: title = ttm.group(1).strip() date = "" for pm in re.finditer(r'PubDate[^>]*content="([^"]*)"', html): d = pm.group(1).strip() cm = re.match(r'(\d{4})-(\d{1,2})-(\d{1,2})', d) if cm: date = f"{cm.group(1)}-{cm.group(2).zfill(2)}-{cm.group(3).zfill(2)}" break if not date: dm = re.search(r'(\d{4}-\d{2}-\d{2})', html) if dm: date = dm.group(1) return title, date def fetch_url(url): for retry in range(3): try: r = requests.get(url, headers=HEADERS, timeout=30, verify=False) if r.status_code == 200: r.encoding = "utf-8" return r.text except Exception as e: if retry < 2: time.sleep(2) return None def insert_article(url, title, date, content): db = sqlite3.connect(SEARCH_DB, timeout=60) db.execute("PRAGMA journal_mode=WAL") try: db.execute( "INSERT OR IGNORE INTO gov_raw(site_name,source_url,page_url,title,publish_date,content,summary,status,category) VALUES(?,?,?,?,?,?,?,?,?)", (SITE_NAME, url, url, title, date, content, title[:200], "active", "通知公告")) affected = db.total_changes db.commit() if affected: db.execute( "INSERT OR REPLACE INTO gov_search(rowid,title,site_name,summary) SELECT r.id,r.title,r.site_name,r.summary FROM gov_raw r WHERE r.page_url=? AND r.id NOT IN (SELECT rowid FROM gov_search)", (url,)) db.commit() db.close() return affected except Exception as e: db.close() print(f" ❌ DB error: {e}") return 0 def run(max_pages=200): print(f"\n{'='*50}") print(f"🚀 {SITE_NAME}") print(f"{'='*50}") # 先看API共多少页 items, total = fetch_list(1) if not items: print("⚠ API无返回") return print(f" 共{total}条, 15条/页 = {(total+14)//15}页") total_pages = min(max_pages, (total + 14) // 15) all_list_items = list(items) for pg in range(2, min(total_pages, _MAX_PG or total_pages) + 1): its, _ = fetch_list(pg) if not its: print(f" 第{pg}页: 空 -> 结束") break all_list_items.extend(its) dates = [it["date"] for it in its] print(f" 第{pg}页: {len(its)} 条 ({dates[0]} ~ {dates[-1]})") # 如果该页最晚日期已经早于3年截断，停止翻页 if max(dates) < CUTOFF: print(f" 该页已全部早于截断日{CUTOFF}，停止翻页") break time.sleep(0.15) print(f"\n📊 API共 {len(all_list_items)} 条") new = skip = no_body = err = 0 for i, item in enumerate(all_list_items, 1): if item["date"] and item["date"] < CUTOFF: skip += 1 continue html = fetch_url(item["url"]) if not html: err += 1 continue pt, pd = extract_meta(html) real_title = pt if pt else item["title"] real_date = pd if pd else item["date"] if real_date and real_date < CUTOFF: skip += 1 continue content = extract_content(html) text_len = len(re.sub(r'<[^>]+>', '', content).strip()) if content else 0 if text_len < 10: no_body += 1 if text_len == 0: print(f" ⚠ 空正文: {real_title[:30]}") continue if insert_article(item["url"], real_title, real_date, content): new += 1 else: skip += 1 if i % 5 == 0: print(f" ...{i}/{len(all_list_items)}") time.sleep(0.3) print(f"\n📊 FTS同步...") db = sqlite3.connect(SEARCH_DB, timeout=60) # QC20260926 去掉手写 gov_search 整站删除(抢锁源; FTS 由 gov_raw 触发器维护) # db.execute("DELETE FROM gov_search WHERE site_name=?", (SITE_NAME,)) db.execute("INSERT OR REPLACE INTO gov_search(rowid,title,site_name,summary) SELECT rowid,title,site_name,summary FROM gov_raw WHERE site_name=?", (SITE_NAME,)) db.commit() c = db.execute("SELECT COUNT(*) FROM gov_search WHERE site_name=?", (SITE_NAME,)) cnt = c.fetchone()[0] db.close() print(f" FTS: {cnt} 条") print(f"\n✅ {SITE_NAME}: 新增{new}, 跳过{skip}, 空正文{no_body}, 错误{err}, FTS共{cnt}条") if __name__ == "__main__": mp = int(sys.argv[1]) if len(sys.argv)

#!/usr/bin/env python3 """ 扬州经济技术开发区-通知公告独立爬虫站点: kfq.yangzhou.gov.cn 栏目: 通知公告 (tzgg, JPAAS publish-system) """ import requests, re, sqlite3, time, sys, json, warnings warnings.filterwarnings('ignore') from datetime import datetime, timedelta import os import sys as _SYS _MAX_PG = int(_SYS.argv[1]) if len(_SYS.argv) > 1 and _SYS.argv[1].isdigit() else None if _MAX_PG is not None: print('[AutoPg] max_pages=' + str(_MAX_PG)) # END AUTO PAGES SEARCH_DB = os.getenv("SEARCH_DB", "/root/search.db") SITE_NAME = "扬州经开区-通知公告" DOMAIN = "kfq.yangzhou.gov.cn" CUTOFF = (datetime.now() - timedelta(days=3*365)).strftime("%Y-%m-%d") HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} API_URL = "https://kfq.yangzhou.gov.cn/api-gateway/jpaas-publish-server/front/page/build/unit" API_PARAMS = { "parseType": "bulidstatic", "webId": "l3juUa1slUgnLQOQgtJwy", "tplSetId": "CU3LRJa5b4SPXybhicnZl", "pageType": "column", "tagId": "列表列表", "editType": "null", "pageId": "Lw8wtupr8vbNtGQMW3yCr", } def fetch_list(page, page_size=15): """获取API列表页""" params = dict(API_PARAMS) params["paramJson"] = json.dumps({"pageNo": page, "pageSize": page_size}) for retry in range(3): try: r = requests.get(API_URL, params=params, headers=HEADERS, timeout=15, verify=False) if r.status_code == 200: d = r.json() html = d.get("data", {}).get("html", "") # Parse items items = [] for li in re.findall(r']*>(.*?)', html, re.DOTALL): m = re.search( r'title="([^"]*)"[^>]*href="([^"]+)"[^>]*>([^<]+)\s*]*>(\d{4}-\d{2}-\d{2})', li) if m: items.append({ "url": "https://" + DOMAIN + m.group(2), "title": m.group(1), "date": m.group(4), }) # Check total count count_m = re.search(r'count="(\d+)"', html) total_count = int(count_m.group(1)) if count_m else 0 return items, total_count except Exception as e: if retry < 2: time.sleep(2) return [], 0 def extract_content(html): """提取正文 - id=zoom""" for pat, tag in [('id="zoom"', 'div'), ('class="contentShow"', 'div')]: i = html.find(pat) if i < 0: continue ds = html.rfind(f"<{tag}", 0, i) if ds < 0: continue s = html[ds:] d = 0 for j in range(len(s)): if s[j:j+4] == f"<{tag}" and (j+4 >= len(s) or s[j+4] in " >\n\r\t"): d += 1 elif s[j:j+3+len(tag)] == f"": d -= 1 if d == 0: gt = s.find(">", 0, j) c = s[gt+1:j] if gt > 0 else s[7:j] c = re.sub(r'<(script|style)[^>]*>.*?', '', c, flags=re.DOTALL|re.I) text = re.sub(r'<[^>]+>', '', c).strip() if len(text) > 50: return c.strip() return "" def extract_meta(html): """提取标题和日期""" title = "" tm = re.search(r']*name="ArticleTitle"[^>]*content="([^"]*)"', html) if tm: title = tm.group(1) if not title: ttm = re.search(r'(.*?)<', html) if ttm: title = ttm.group(1).strip() date = "" for pm in re.finditer(r'PubDate[^>]*content="([^"]*)"', html): d = pm.group(1).strip() cm = re.match(r'(\d{4})-(\d{1,2})-(\d{1,2})', d) if cm: date = f"{cm.group(1)}-{cm.group(2).zfill(2)}-{cm.group(3).zfill(2)}" break if not date: dm = re.search(r'(\d{4}-\d{2}-\d{2})', html) if dm: date = dm.group(1) return title, date def fetch_url(url): for retry in range(3): try: r = requests.get(url, headers=HEADERS, timeout=30, verify=False) if r.status_code == 200: r.encoding = "utf-8" return r.text except Exception as e: if retry < 2: time.sleep(2) return None def insert_article(url, title, date, content): db = sqlite3.connect(SEARCH_DB, timeout=60) db.execute("PRAGMA journal_mode=WAL") try: db.execute( "INSERT OR IGNORE INTO gov_raw(site_name,source_url,page_url,title,publish_date,content,summary,status,category) VALUES(?,?,?,?,?,?,?,?,?)", (SITE_NAME, url, url, title, date, content, title[:200], "active", "通知公告")) affected = db.total_changes db.commit() if affected: db.execute( "INSERT OR REPLACE INTO gov_search(rowid,title,site_name,summary) SELECT r.id,r.title,r.site_name,r.summary FROM gov_raw r WHERE r.page_url=? AND r.id NOT IN (SELECT rowid FROM gov_search)", (url,)) db.commit() db.close() return affected except Exception as e: db.close() print(f" ❌ DB error: {e}") return 0 def run(max_pages=200): print(f"\n{'='*50}") print(f"🚀 {SITE_NAME}") print(f"{'='*50}") # 先看API共多少页 items, total = fetch_list(1) if not items: print("⚠ API无返回") return print(f" 共{total}条, 15条/页 = {(total+14)//15}页") total_pages = min(max_pages, (total + 14) // 15) all_list_items = list(items) for pg in range(2, min(total_pages, _MAX_PG or total_pages) + 1): its, _ = fetch_list(pg) if not its: print(f" 第{pg}页: 空 -> 结束") break all_list_items.extend(its) dates = [it["date"] for it in its] print(f" 第{pg}页: {len(its)} 条 ({dates[0]} ~ {dates[-1]})") # 如果该页最晚日期已经早于3年截断，停止翻页 if max(dates) < CUTOFF: print(f" 该页已全部早于截断日{CUTOFF}，停止翻页") break time.sleep(0.15) print(f"\n📊 API共 {len(all_list_items)} 条") new = skip = no_body = err = 0 for i, item in enumerate(all_list_items, 1): if item["date"] and item["date"] < CUTOFF: skip += 1 continue html = fetch_url(item["url"]) if not html: err += 1 continue pt, pd = extract_meta(html) real_title = pt if pt else item["title"] real_date = pd if pd else item["date"] if real_date and real_date < CUTOFF: skip += 1 continue content = extract_content(html) text_len = len(re.sub(r'<[^>]+>', '', content).strip()) if content else 0 if text_len < 10: no_body += 1 if text_len == 0: print(f" ⚠ 空正文: {real_title[:30]}") continue if insert_article(item["url"], real_title, real_date, content): new += 1 else: skip += 1 if i % 5 == 0: print(f" ...{i}/{len(all_list_items)}") time.sleep(0.3) print(f"\n📊 FTS同步...") db = sqlite3.connect(SEARCH_DB, timeout=60) # QC20260926 去掉手写 gov_search 整站删除(抢锁源; FTS 由 gov_raw 触发器维护) # db.execute("DELETE FROM gov_search WHERE site_name=?", (SITE_NAME,)) db.execute("INSERT OR REPLACE INTO gov_search(rowid,title,site_name,summary) SELECT rowid,title,site_name,summary FROM gov_raw WHERE site_name=?", (SITE_NAME,)) db.commit() c = db.execute("SELECT COUNT(*) FROM gov_search WHERE site_name=?", (SITE_NAME,)) cnt = c.fetchone()[0] db.close() print(f" FTS: {cnt} 条") print(f"\n✅ {SITE_NAME}: 新增{new}, 跳过{skip}, 空正文{no_body}, 错误{err}, FTS共{cnt}条") if __name__ == "__main__": mp = int(sys.argv[1]) if len(sys.argv) > 1 else 200 run(mp)