#!/usr/bin/env python3
"""
crawl_hanchuan_tzgg.py - 汉川市-通知公告 (www.hanchuan.gov.cn)
站点: 湖北孝感汉川市 (JPAAS 系统 .jhtml, JS混淆挑战 JbU — 真实浏览器执行后过盾)
过盾: playwright-stealth, goto 后 DOM 即真实页面
列表: index.jhtml (第1页) + index_N.jhtml (N=2..), total=160 条, 每页 10 条
列表项: li > a[href=/tzgg/{id}.jhtml] 标题 + 日期/来源
详情: /tzgg/{id}.jhtml, meta ArticleTitle/PubDate/ContentSource, 正文 div.p-xwxq-content
附件: /u/cms/hanchuan/... 相对路径转绝对
"""
import sys, os, re, asyncio, fcntl, time
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from crawler_lib import push_to_searchdb
BASE = "http://www.hanchuan.gov.cn"
LIST_DIR = "/tzgg"
LOCK_FILE = "/tmp/hanchuan_tzgg.lock"
SITE_NAME = "汉川市-通知公告"
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36"
def acquire_lock(max_wait=1800):
fd = open(LOCK_FILE, 'w')
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
return fd
except BlockingIOError:
print(f" [LOCK] 另一 hanchuan 实例运行中, 等待释放 (最多 {max_wait}s)...")
deadline = time.time() + max_wait
while time.time() < deadline:
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
return fd
except BlockingIOError:
time.sleep(10)
print(" [LOCK] 等待超时, 放弃")
return None
async def run_async(pages=5):
from playwright.async_api import async_playwright
from playwright_stealth import Stealth
async with async_playwright() as p:
browser = await p.chromium.launch(
headless=True,
executable_path='/root/.cache/ms-playwright/chromium-1228/chrome-linux64/chrome',
args=['--no-sandbox', '--disable-setuid-sandbox', '--disable-blink-features=AutomationControlled']
)
ctx = await browser.new_context(user_agent=UA, locale='zh-CN',
viewport={'width': 1440, 'height': 900}, timezone_id='Asia/Shanghai')
stealth = Stealth()
await stealth.apply_stealth_async(ctx)
page = await ctx.new_page()
# 过盾: 列表第一页
first_url = BASE + LIST_DIR + "/index.jhtml"
try:
resp = await page.goto(first_url, wait_until='domcontentloaded', timeout=40000)
await page.wait_for_timeout(7000)
html = await page.content()
m = re.search(r'([^<]*)', html)
print(f" [LOAD] status={resp.status if resp else '?'} title={m.group(1) if m else '?'} len={len(html)}")
if len(html) < 8000 or ('JbU' in html[:5000]):
print(" [WAF] 挑战未过")
await page.wait_for_timeout(8000)
html = await page.content()
m = re.search(r'([^<]*)', html)
print(f" 再等: title={m.group(1) if m else '?'} len={len(html)}")
if len(html) < 8000:
await browser.close()
return 0
except Exception as e:
print(f" [LOAD ERR] {str(e)[:120]}")
await browser.close()
return 0
# 列表页
items_all = []
seen = set()
for pg in range(1, pages + 1):
if pg == 1:
url = first_url
else:
url = BASE + LIST_DIR + f"/index_{pg}.jhtml"
try:
resp = await page.goto(url, wait_until='domcontentloaded', timeout=30000)
await page.wait_for_timeout(2500)
html = await page.content()
except Exception as e:
print(f" [LIST ERR p{pg}] {str(e)[:100]}")
break
# 提取: news-list 容器内 + 时间 div
rows = []
mseg = re.search(r'class="news-list"[\s\S]*?', html)
if mseg:
seg = mseg.group(0)
links = re.findall(r']*href="([^"]*\.jhtml)"[^>]*title="([^"]*)"', seg)
dates = re.findall(r'(\d{4}-\d{2}-\d{2})', seg)
for i, (u, t) in enumerate(links):
d = dates[i] if i < len(dates) else ''
rows.append((u, t, d))
new_rows = 0
for u, t, d in rows:
url = u if u.startswith('http') else BASE + u
url = url.replace(':80', '')
title = t.strip()
if title and url not in seen:
seen.add(url)
items_all.append({"url": url, "title": title, "date": d})
new_rows += 1
print(f" 第{pg}页: 解析 {len(rows)} 条, 新增 {new_rows} (累计 {len(items_all)})")
if len(rows) < 5:
print(f" [p{pg}] 列表结束")
break
print(f" [LIST] 去重后 {len(items_all)} 条")
# 详情页
records = []
for idx, it in enumerate(items_all):
print(f" [{idx+1}/{len(items_all)}] {it['title'][:40]}...")
try:
resp = await page.goto(it['url'], wait_until='domcontentloaded', timeout=30000)
await page.wait_for_timeout(2500)
dh = await page.content()
except Exception as e:
print(f" [DETAIL ERR] {str(e)[:100]}")
continue
title = it['title']
m = re.search(r']*>([\s\S]*?)', dh)
if m:
content = m.group(1).strip()
if not content.strip():
print(" [SKIP] 正文空")
continue
content = re.sub(r'href="/', 'href="' + BASE + '/', content)
content = re.sub(r'src="/', 'src="' + BASE + '/', content)
records.append({
"title": title,
"url": it['url'],
"pub_date": pub_date,
"site_name": SITE_NAME,
"content": content,
"summary": "",
})
await browser.close()
print(f" 共 {len(records)} 条有效")
if records:
push_to_searchdb(records, "hanchuan_tzgg")
return len(records)
def main():
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--pages", type=int, default=5)
args = parser.parse_args()
lock = acquire_lock()
if not lock:
return
try:
n = asyncio.run(run_async(pages=args.pages))
print(f"Done: {n} records")
finally:
os.close(lock.fileno())
if __name__ == "__main__":
main()