bitfennec
1059be1b62
- Rust core engine with HTTP ingestion gateway, hybrid search, and white-box resume matching - Python edge collectors supporting declarative TOML and specialized adapters - PostgreSQL schema migrations with pgvector embedding support - Hardened security controls: strict CSP, terminal sanitization, and input boundary validation - Multi-service container orchestration with dedicated non-root execution
230 lines
7.5 KiB
Python
230 lines
7.5 KiB
Python
import argparse
|
||
from pathlib import Path
|
||
import re
|
||
import sys
|
||
import time
|
||
from urllib.parse import urljoin
|
||
|
||
current = Path(__file__).resolve()
|
||
collector_root = (
|
||
current.parents[2]
|
||
if (current.parents[2] / "common").exists()
|
||
else current.parents[1]
|
||
)
|
||
if str(collector_root) not in sys.path:
|
||
sys.path.insert(0, str(collector_root))
|
||
|
||
from common.content import sanitize_body_text
|
||
from common.fetcher import PoliteFetcher
|
||
from common.protocol import (
|
||
AnnouncementItem,
|
||
AttachmentItem,
|
||
CollectorMeta,
|
||
IngestPayload,
|
||
emit_payload,
|
||
)
|
||
from common.security import is_safe_url
|
||
from selectolax.lexbor import LexborHTMLParser
|
||
|
||
ALLOWED_DOMAINS = [
|
||
"cdrsigc.com",
|
||
"chengdu.gov.cn",
|
||
"sc.gov.cn",
|
||
"scssfw.gov.cn",
|
||
"mnr.gov.cn",
|
||
"cdsjs.com",
|
||
]
|
||
|
||
META = {
|
||
"source_id": "sc.chengdu.rsks",
|
||
"name": "成都人事考试网",
|
||
"province": "四川省",
|
||
"city": "成都市",
|
||
"domain": "cdpta.cdrsigc.com",
|
||
"tags": ["gwy", "sydw", "sub_provincial"],
|
||
}
|
||
|
||
SOURCE_ID = META["source_id"]
|
||
COLLECTOR_NAME = "chengdu_rsks_crawler"
|
||
START_URL = "https://cdpta.cdrsigc.com/netpage/index.do"
|
||
|
||
|
||
def crawl_chengdu_rsks(max_items: int = 5) -> IngestPayload:
|
||
fetcher = PoliteFetcher()
|
||
start_time = time.time()
|
||
error_msg = None
|
||
|
||
items: list[AnnouncementItem] = []
|
||
|
||
try:
|
||
html = fetcher.get_text(START_URL, ALLOWED_DOMAINS, "cdpta.cdrsigc.com")
|
||
tree = LexborHTMLParser(html)
|
||
|
||
candidates = []
|
||
for a in tree.css("a"):
|
||
text = a.text(strip=True)
|
||
href = a.attributes.get("href", "").strip()
|
||
if not text or len(text) < 6:
|
||
continue
|
||
|
||
full_url = None
|
||
if "noticecontent.jsp" in href:
|
||
full_url = urljoin(START_URL, href)
|
||
elif "goxxinfo(" in href:
|
||
m = re.search(r"goxxinfo\([^,]+,[^,]+,'([^']+)'", href)
|
||
if m and m.group(1).startswith("/frt/"):
|
||
full_url = urljoin(START_URL, m.group(1))
|
||
elif href.startswith("http") and not href.endswith(".do"):
|
||
full_url = href
|
||
|
||
if full_url and is_safe_url(full_url, ALLOWED_DOMAINS, "cdpta.cdrsigc.com") and any(
|
||
k in text for k in ["招聘", "录用", "选拔", "考录", "公示", "选调", "选聘"]
|
||
):
|
||
candidates.append((text, full_url))
|
||
|
||
seen_urls = set()
|
||
for title, url in candidates:
|
||
if url in seen_urls:
|
||
continue
|
||
seen_urls.add(url)
|
||
|
||
body_text = None
|
||
attachments = []
|
||
pub_date = None
|
||
|
||
try:
|
||
detail_html = fetcher.get_text(url, ALLOWED_DOMAINS, "cdpta.cdrsigc.com")
|
||
detail_tree = LexborHTMLParser(detail_html)
|
||
detail_tree.strip_tags(["script", "style", "noscript", "iframe", "svg"])
|
||
|
||
title_node = detail_tree.css_first("h1, h2, div.title, span.title")
|
||
if title_node:
|
||
title_text = title_node.text(strip=True)
|
||
if len(title_text) > 4:
|
||
title = title_text
|
||
|
||
body_node = (
|
||
detail_tree.css_first("div#content")
|
||
or detail_tree.css_first("div.content")
|
||
or detail_tree.css_first("div.con")
|
||
or detail_tree.css_first("div.article")
|
||
or detail_tree.css_first("body")
|
||
)
|
||
if body_node:
|
||
extracted = body_node.text(strip=True)
|
||
extracted = re.sub(
|
||
r"^(?:首页|当前位置[：:]\s*首页|您当前所在的位置[：:]\s*首页)[^\n\r]{0,100}(?:>|»|/|·|正文|内容|详情)?\s*",
|
||
"",
|
||
extracted,
|
||
)
|
||
body_text = sanitize_body_text(extracted)
|
||
|
||
date_node = detail_tree.css_first("span.date, div.date, span.time")
|
||
if date_node:
|
||
m = re.search(r"(\d{4}[-/年]\d{1,2}[-/月]\d{1,2})", date_node.text())
|
||
if m:
|
||
pub_date = (
|
||
m.group(1)
|
||
.replace("年", "-")
|
||
.replace("月", "-")
|
||
.replace("/", "-")
|
||
)
|
||
|
||
if not pub_date and body_text:
|
||
m = re.search(r"(\d{4}[-/年]\d{1,2}[-/月]\d{1,2})", body_text[:500])
|
||
if m:
|
||
pub_date = (
|
||
m.group(1)
|
||
.replace("年", "-")
|
||
.replace("月", "-")
|
||
.replace("/", "-")
|
||
)
|
||
|
||
seen_att_urls = set()
|
||
for att_a in detail_tree.css("a"):
|
||
if len(attachments) >= 50:
|
||
break
|
||
att_href = att_a.attributes.get("href", "").strip()
|
||
att_text = att_a.text(strip=True)
|
||
if any(
|
||
att_href.lower().endswith(ext)
|
||
for ext in [".xlsx", ".xls", ".pdf", ".docx", ".doc", ".zip"]
|
||
):
|
||
full_att_url = urljoin(url, att_href)
|
||
if (
|
||
is_safe_url(full_att_url, ALLOWED_DOMAINS, "cdpta.cdrsigc.com")
|
||
and full_att_url not in seen_att_urls
|
||
):
|
||
seen_att_urls.add(full_att_url)
|
||
attachments.append(
|
||
AttachmentItem(name=att_text or "附件", url=full_att_url)
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
if not body_text and not attachments:
|
||
continue
|
||
|
||
items.append(
|
||
AnnouncementItem(
|
||
url=url,
|
||
title=title,
|
||
kind="招聘" if "公示" not in title else "拟聘用",
|
||
employer="成都市人力资源和社会保障局 / 考试评价中心",
|
||
employer_type="事业单位",
|
||
published_at=pub_date,
|
||
region_hint="四川省 成都市",
|
||
body_text=body_text,
|
||
attachments=attachments,
|
||
positions=[],
|
||
)
|
||
)
|
||
|
||
if len(items) >= max_items:
|
||
break
|
||
|
||
except Exception as exc:
|
||
error_msg = str(exc)
|
||
finally:
|
||
fetcher.close()
|
||
|
||
elapsed_ms = int((time.time() - start_time) * 1000)
|
||
|
||
meta = CollectorMeta(
|
||
name=COLLECTOR_NAME,
|
||
runtime="python/3.12",
|
||
auth_status="none",
|
||
duration_ms=elapsed_ms,
|
||
error_message=error_msg,
|
||
)
|
||
|
||
return IngestPayload(
|
||
source_id=SOURCE_ID,
|
||
batch_id=f"cd-rsks-{int(time.time())}",
|
||
collector=meta,
|
||
items=items,
|
||
)
|
||
|
||
|
||
crawl = crawl_chengdu_rsks
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(
|
||
description="Collector for Chengdu Personnel Examination Portal"
|
||
)
|
||
parser.add_argument(
|
||
"--push-url", type=str, default=None, help="HTTP ingestion endpoint URL"
|
||
)
|
||
parser.add_argument(
|
||
"--limit", type=int, default=5, help="Maximum number of items to crawl"
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
payload = crawl_chengdu_rsks(max_items=args.limit)
|
||
emit_payload(payload, push_url=args.push_url)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|