]*>(.*)', html_text, re.S)
if not cm:
return title, date, "", []
body = cm.group(1)
# 附件
atts = []
for a in re.finditer(r'
]+href="([^"]+)"[^>]*>(.*?)', body, re.S):
href, txt = a.group(1), re.sub(r"<[^>]+>", "", a.group(2)).strip()
if re.search(r"\.(?:pdf|docx?|xlsx?|zip|rar|wps|et|dps)(?:\?|$)", href, re.I) or txt.startswith("附件"):
atts.append((urljoin(url, href), txt or href.split("/")[-1]))
return title, date, body, atts
def html_to_text(html_text, attachments):
"""HTML -> 文本, 保留表格、\\n\\n 分段、附件嵌入"""
tables = []
def _save_tbl(m):
tables.append(m.group(0))
return f"@@TABLE{len(tables) - 1}@@"
html_text = re.sub(r"", _save_tbl, html_text)
html_text = re.sub(r"]*>", "\n", html_text)
html_text = re.sub(r"]*>", "", html_text)
html_text = html_text.replace(" ", " ").replace("\xa0", " ")
html_text = re.sub(r"\n{3,}", "\n\n", html_text)
lines = [ln.strip() for ln in html_text.split("\n")]
out = [ln for ln in lines if ln]
text = "\n\n".join(out)
for i, t in enumerate(tables):
text = text.replace(f"@@TABLE{i}@@", t)
text = re.sub(r"\n{3,}", "\n\n", text)
# 附件嵌入: 附件名
for href, txt in attachments:
if href not in text:
text += f'\n\n{txt}
'
return text.strip()
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--pages", type=int, default=1, help="抓取页数")
ap.add_argument("--db", default="/mnt/data/search.db", help="SQLite 数据库路径")
ap.add_argument("--script-name", default=SCRIPT_NAME)
args = ap.parse_args()
db_path = args.db
script_name = args.script_name
if not db_path.startswith("/mnt/"):
db_path = os.path.expanduser(db_path)
conn = sqlite3.connect(db_path, timeout=60)
cur = conn.cursor()
new_count = 0
dup_count = 0
err_count = 0
session = requests.Session()
# 先访问栏目页获取 cookie
try:
session.get(BASE_URL + "/gsgg/c/" + COLUMN_ID, headers=HEADERS, timeout=30, verify=False)
except Exception:
pass
# 第一页确定总页数
_, total = fetch_list(session, current=1, size=15)
total_pages = (total + 14) // 15 if total else 1
pages_to_crawl = min(args.pages, total_pages)
print(f"共 {total} 条 / {total_pages} 页, 本次爬 {pages_to_crawl} 页")
for page in range(1, pages_to_crawl + 1):
records, _ = fetch_list(session, current=page, size=15)
if records is None:
print(f"[{page}] 列表请求失败")
err_count += 1
continue
print(f"[{page}] 列表 {len(records)} 条")
if not records:
continue
for rec in records:
try:
article_id = rec.get("id")
title = clean_title(rec.get("title", ""))
pub_date = (rec.get("releaseTime") or rec.get("time") or "")[:10]
if not title or not article_id:
continue
art_url = f"{BASE_URL}/gsgg/article/{article_id}"
cur.execute("SELECT COUNT(*) FROM gov_raw WHERE page_url=? AND script_name=?", (art_url, script_name))
if cur.fetchone()[0] > 0:
dup_count += 1
continue
dr = session.get(art_url, headers={**HEADERS, "Content-Type": "text/html"}, timeout=30, verify=False)
dr.encoding = "utf-8"
d_title, d_date, content_html, atts = parse_detail(dr.text, art_url)
if not content_html.strip():
print(f" 跳过空正文: {title[:30]}")
dup_count += 1
continue
if d_title:
title = d_title
if d_date:
pub_date = d_date
content = html_to_text(content_html, atts)
att_txt = "\n\n".join(h for h, _ in atts) if atts else ""
md5 = hashlib.md5((title + content[:500]).encode("utf-8")).hexdigest()
summary = content[:500] if content else ""
cur.execute(
"INSERT INTO gov_raw (site_name, source_url, page_url, title, publish_date, date_rank, summary, status, category, visits, content, tags, industry, attachments, group_name, has_table, script_name) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
(SITE_NAME, art_url, art_url, title, pub_date, 0, summary, "", "", 0, content, "",
"other", att_txt, GROUP_NAME, 1 if "