From dc2f3203c360a70b97f027fbc47baf3fbcb46359 Mon Sep 17 00:00:00 2001 From: zk Date: Tue, 28 Jul 2026 17:56:33 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E5=AE=9A=E6=97=B6=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E8=A7=A6=E5=8F=91=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env | 5 +- .env.prod | 5 +- .env.test | 5 +- app/config/settings.py | 4 ++ app/main.py | 75 +++++++++++++++++++++++ app/service/announcement_batch_service.py | 56 +++++++++++++++++ requirements.txt | 3 + 7 files changed, 150 insertions(+), 3 deletions(-) create mode 100644 app/main.py create mode 100644 app/service/announcement_batch_service.py diff --git a/.env b/.env index 0d6c97b..96a40c7 100644 --- a/.env +++ b/.env @@ -27,4 +27,7 @@ OSS_ACCESS_KEY_ID=LTAI5tEdLKKQUKhTyUpfH5Mk OSS_ACCESS_KEY_SECRET=RjUdTrq0V5qA4b3BUElNhXqs3ZLp5k OSS_ENDPOINT=oss-cn-guangzhou.aliyuncs.com OSS_BUCKET=offerpie -OSS_DOMAIN=https://offerpie.oss-cn-guangzhou.aliyuncs.com \ No newline at end of file +OSS_DOMAIN=https://offerpie.oss-cn-guangzhou.aliyuncs.com + +# 公告落库并发线程数（不要超过 MYSQL_POOL_SIZE + MYSQL_MAX_OVERFLOW） +SPIDER_MAX_WORKERS=5 diff --git a/.env.prod b/.env.prod index 64b55db..c976160 100644 --- a/.env.prod +++ b/.env.prod @@ -27,4 +27,7 @@ OSS_ACCESS_KEY_ID=LTAI5tEdLKKQUKhTyUpfH5Mk OSS_ACCESS_KEY_SECRET=RjUdTrq0V5qA4b3BUElNhXqs3ZLp5k OSS_ENDPOINT=oss-cn-guangzhou.aliyuncs.com OSS_BUCKET=offerpie -OSS_DOMAIN=https://offerpie.oss-cn-guangzhou.aliyuncs.com \ No newline at end of file +OSS_DOMAIN=https://offerpie.oss-cn-guangzhou.aliyuncs.com + +# 公告落库并发线程数（不要超过 MYSQL_POOL_SIZE + MYSQL_MAX_OVERFLOW） +SPIDER_MAX_WORKERS=8 diff --git a/.env.test b/.env.test index 38d0231..f4eba25 100644 --- a/.env.test +++ b/.env.test @@ -27,4 +27,7 @@ OSS_ACCESS_KEY_ID=LTAI5tEdLKKQUKhTyUpfH5Mk OSS_ACCESS_KEY_SECRET=RjUdTrq0V5qA4b3BUElNhXqs3ZLp5k OSS_ENDPOINT=oss-cn-guangzhou.aliyuncs.com OSS_BUCKET=offerpie -OSS_DOMAIN=https://offerpie.oss-cn-guangzhou.aliyuncs.com \ No newline at end of file +OSS_DOMAIN=https://offerpie.oss-cn-guangzhou.aliyuncs.com + +# 公告落库并发线程数（不要超过 MYSQL_POOL_SIZE + MYSQL_MAX_OVERFLOW） +SPIDER_MAX_WORKERS=5 diff --git a/app/config/settings.py b/app/config/settings.py index 23a8282..debaadb 100644 --- a/app/config/settings.py +++ b/app/config/settings.py @@ -34,6 +34,10 @@ class Settings(BaseSettings): oss_upload_dir: str = "company/logo" oss_domain: str = "https://offerpie.oss-cn-guangzhou.aliyuncs.com" + # ──────────── 爬虫并发 ──────────── + # 公告落库并发线程数，不要超过 mysql_pool_size + mysql_max_overflow + spider_max_workers: int = 5 + # ──────────── 日志 ──────────── logging_level: str = "INFO" log_file_name: str = "spider.log" diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..670011b --- /dev/null +++ b/app/main.py @@ -0,0 +1,75 @@ +"""爬虫服务入口：初始化数据库，注册定时任务并启动调度。 + +运行（必须在项目根目录下以模块方式启动，否则包内 import 找不到 app）： + python -m app.main +""" + +from __future__ import annotations + +from apscheduler.schedulers.blocking import BlockingScheduler + +from app.config import settings +from app.core.database import close_db, init_db +from app.core.logger import log +from app.service.announcement_batch_service import save_announcements +from app.spider.offerqingbaoju import fetch_offerqingbaoju +from app.spider.offershow import fetch_offershow + + +def crawl(source: str, fetcher, limit: int) -> None: + """采集任务：抓公告地址 → 多线程落库。异常不外抛，避免调度器丢任务。 + + Args: + source: 采集源名称，仅用于日志。 + fetcher: 爬虫函数，签名 (limit: int) -> list[str]。 + limit: 本次抓取条数上限。 + """ + log.info("[{}] 任务开始，limit={}", source, limit) + try: + urls = fetcher(limit) + if urls: + save_announcements(urls) + except Exception as exc: + log.error("[{}] 任务异常: {}", source, exc) + log.info("[{}] 任务结束", source) + + +# 采集源：(名称, 爬虫函数, 抓取条数, 每天执行的时, 分) +JOBS = [ + ("offershow", fetch_offershow, 100, 0, 30), + ("offerqingbaoju", fetch_offerqingbaoju, 100, 2, 36), +] + + +def main() -> None: + """初始化数据源并启动定时任务，阻塞运行直到 Ctrl+C。""" + log.info("爬虫服务启动，环境={}", settings.env) + init_db() + + scheduler = BlockingScheduler(timezone="Asia/Shanghai") + for source, fetcher, limit, hour, minute in JOBS: + scheduler.add_job( + crawl, + "cron", + hour=hour, + minute=minute, + args=(source, fetcher, limit), + id=source, + # 上一轮没跑完则本轮跳过，防止任务堆叠 + max_instances=1, + coalesce=True, + # 错过触发时间 10 分钟内仍补跑，超过则跳过本次 + misfire_grace_time=600, + ) + log.info("[{}] 已注册，每天 {:02d}:{:02d} 执行", source, hour, minute) + + try: + scheduler.start() + except (KeyboardInterrupt, SystemExit): + log.info("服务中断，正在退出") + finally: + close_db() + + +if __name__ == "__main__": + main() diff --git a/app/service/announcement_batch_service.py b/app/service/announcement_batch_service.py new file mode 100644 index 0000000..a610536 --- /dev/null +++ b/app/service/announcement_batch_service.py @@ -0,0 +1,56 @@ +"""公告 URL 批量落库：多线程调用单条处理逻辑。""" + +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor, as_completed + +from app.config import settings +from app.core.logger import log +from app.service.recruit_announcement_service import process_announcement + + +def _safe_process(url: str) -> bool: + """处理单条 URL，异常不外抛。 + + Args: + url: 公告地址。 + + Returns: + True 表示未抛异常。 + """ + try: + process_announcement(url) + return True + except Exception as exc: + log.error("公告处理异常 [url={}]: {}", url, exc) + return False + + +def save_announcements(urls: list[str], workers: int | None = None) -> tuple[int, int]: + """多线程处理一批公告 URL。 + + Args: + urls: 公告地址列表，内部去重。 + workers: 并发线程数，默认取配置 spider_max_workers。 + + Returns: + (正常条数, 失败条数)。 + """ + targets = list(dict.fromkeys(u for u in urls if u)) + if not targets: + return 0, 0 + + workers = max(1, min(workers or settings.spider_max_workers, len(targets))) + log.info("开始处理公告 {} 条，线程数 {}", len(targets), workers) + + ok = bad = 0 + with ThreadPoolExecutor(max_workers=workers, thread_name_prefix="ann") as pool: + futures = [pool.submit(_safe_process, url) for url in targets] + for future in as_completed(futures): + if future.result(): + ok += 1 + else: + bad += 1 + + log.info("公告处理完成：正常 {} | 失败 {}", ok, bad) + return ok, bad diff --git a/requirements.txt b/requirements.txt index 79d50ae..0defb30 100644 --- a/requirements.txt +++ b/requirements.txt @@ -21,3 +21,6 @@ json-repair>=0.61 loguru>=0.7 oss2==2.19.1 snowflake-id>=1.0 + +# 定时任务 +apscheduler>=3.10,<4