添加定时任务触发逻辑

.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
+OSS_DOMAIN=https://offerpie.oss-cn-guangzhou.aliyuncs.com
+
+# 公告落库并发线程数（不要超过 MYSQL_POOL_SIZE + MYSQL_MAX_OVERFLOW）
+SPIDER_MAX_WORKERS=5

.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
+OSS_DOMAIN=https://offerpie.oss-cn-guangzhou.aliyuncs.com
+
+# 公告落库并发线程数（不要超过 MYSQL_POOL_SIZE + MYSQL_MAX_OVERFLOW）
+SPIDER_MAX_WORKERS=8

.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
+OSS_DOMAIN=https://offerpie.oss-cn-guangzhou.aliyuncs.com
+
+# 公告落库并发线程数（不要超过 MYSQL_POOL_SIZE + MYSQL_MAX_OVERFLOW）
+SPIDER_MAX_WORKERS=5

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"

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()

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

app/tool/cv.py

@@ -1,4 +1,8 @@
-"""二维码检测与裁剪工具。"""
+"""二维码检测与裁剪工具。
+
+识别引擎优先用 zxing-cpp（对反色、旋转、小尺寸、艺术化二维码的容错明显更好），
+拿不到时退回 OpenCV 自带检测器，保证不装 zxing-cpp 也能跑。
+"""
 
 from __future__ import annotations
 
@@ -8,6 +12,34 @@ from pathlib import Path
 import cv2
 import numpy as np
 
+from app.core.logger import log
+
+try:  # zxing-cpp 是可选依赖，缺失时自动退化为纯 OpenCV
+    import zxingcpp
+
+    _HAS_ZXING = True
+except ImportError:  # pragma: no cover - 取决于部署环境
+    zxingcpp = None
+    _HAS_ZXING = False
+    log.warning("未安装 zxing-cpp，二维码识别将退化为 OpenCV 检测器，识别率会明显下降")
+
+
+# 要识别的二维码类型：标准 QR + 微型 QR + 矩形 QR
+_QR_FORMATS = (
+    [
+        zxingcpp.BarcodeFormat.QRCode,
+        zxingcpp.BarcodeFormat.MicroQRCode,
+        zxingcpp.BarcodeFormat.RMQRCode,
+    ]
+    if _HAS_ZXING
+    else []
+)
+
+# 放大重试的尺寸上限，避免长图被放大到内存爆掉
+_MAX_UPSCALE_SIDE = 2600
+# 裁剪区域重试时补的静默区宽度（像素）
+_QUIET_ZONE = 16
+
 
 @dataclass
 class QrRegion:
@@ -15,6 +47,7 @@ class QrRegion:
 
     points: tuple[tuple[int, int], ...]
     crop: np.ndarray | None = None
+    text: str = ""
 
 
 @dataclass
@@ -105,34 +138,159 @@ def _normalize_points(points: np.ndarray | None) -> list[tuple[tuple[int, int],
     return result
 
 
-def _detect_points(img: np.ndarray) -> list[tuple[tuple[int, int], ...]]:
-    detector = cv2.QRCodeDetector()
+def _clip_points(
+    quad: tuple[tuple[int, int], ...], shape: tuple[int, ...]
+) -> tuple[tuple[int, int], ...]:
+    """把角点裁进图像范围内，避免透视变换取到界外。"""
+    height, width = shape[:2]
+    return tuple(
+        (min(max(x, 0), width - 1), min(max(y, 0), height - 1)) for x, y in quad
+    )
 
-    if hasattr(detector, "detectMulti"):
-        ok, points = detector.detectMulti(img)
-        if ok:
-            normalized_points = _normalize_points(points)
-            if normalized_points:
-                return normalized_points
 
-    ok, points = detector.detect(img)
-    if ok:
-        normalized_points = _normalize_points(points)
-        if normalized_points:
-            return normalized_points
+def zxing_read(
+    img: np.ndarray,
+    *,
+    binarizer: object | None = None,
+    try_downscale: bool = True,
+) -> list:
+    """用 zxing-cpp 识别图中所有二维码，失败返回空列表。
 
+    zxing-cpp 默认已开启 try_invert（反色）和 try_rotate（旋转），
+    这是它比 OpenCV 检测器兼容性好的主要原因。
+    """
+    if not _HAS_ZXING:
+        return []
+
+    kwargs = {
+        "formats": _QR_FORMATS,
+        "try_rotate": True,
+        "try_invert": True,
+        "try_downscale": try_downscale,
+    }
+    if binarizer is not None:
+        kwargs["binarizer"] = binarizer
+
+    try:
+        return list(zxingcpp.read_barcodes(img, **kwargs))
+    except Exception as exc:  # zxing 内部异常不应该打断整个流程
+        log.debug(f"zxing-cpp 识别异常: {exc}")
+        return []
+
+
+def zxing_variants(img: np.ndarray) -> list[tuple[str, np.ndarray]]:
+    """构造 zxing 的重试图像变体：原图之外再补一版放大图。
+
+    放大主要救「长图里的小二维码」和「低分辨率二维码」两类。
+    """
+    variants: list[tuple[str, np.ndarray]] = [("原图", img)]
+
+    longest = max(img.shape[:2])
+    if longest and longest * 2 <= _MAX_UPSCALE_SIDE:
+        variants.append(
+            ("放大2倍", cv2.resize(img, None, fx=2, fy=2, interpolation=cv2.INTER_CUBIC))
+        )
+    return variants
+
+
+def pad_quiet_zone(img: np.ndarray, border: int = _QUIET_ZONE) -> np.ndarray:
+    """给裁剪出来的二维码补静默区。
+
+    有些海报把二维码贴边放，裁出来没有留白，补一圈能提高解码成功率。
+    边框颜色取图像四角的中位数，反色码补深色、正常码补浅色，避免破坏极性。
+    """
+    corners = np.array(
+        [img[0, 0], img[0, -1], img[-1, 0], img[-1, -1]], dtype=np.float32
+    )
+    color = np.median(corners, axis=0)
+    value = tuple(int(round(c)) for c in np.atleast_1d(color))
+    if len(value) == 1:
+        value = value * 3
+    return cv2.copyMakeBorder(
+        img, border, border, border, border, cv2.BORDER_CONSTANT, value=value
+    )
+
+
+def _points_from_zxing(barcode: object) -> tuple[tuple[int, int], ...] | None:
+    """把 zxing 的 position 转成四角点。"""
+    position = getattr(barcode, "position", None)
+    if position is None:
+        return None
+
+    quad: list[tuple[int, int]] = []
+    for name in ("top_left", "top_right", "bottom_right", "bottom_left"):
+        point = getattr(position, name, None)
+        if point is None:
+            return None
+        quad.append((int(round(point.x)), int(round(point.y))))
+    return tuple(quad)
+
+
+def _detect_by_opencv(img: np.ndarray) -> list[tuple[tuple[int, int], ...]]:
+    """OpenCV 检测器兜底，额外补一次反色重试。"""
+    for candidate in (img, cv2.bitwise_not(img)):
+        for detector in (cv2.QRCodeDetector(), cv2.QRCodeDetectorAruco()):
+            try:
+                if hasattr(detector, "detectMulti"):
+                    ok, points = detector.detectMulti(candidate)
+                    if ok:
+                        normalized = _normalize_points(points)
+                        if normalized:
+                            return normalized
+
+                ok, points = detector.detect(candidate)
+                if ok:
+                    normalized = _normalize_points(points)
+                    if normalized:
+                        return normalized
+            except cv2.error:
+                continue
     return []
 
 
+def _detect_regions(img: np.ndarray) -> list[QrRegion]:
+    """检测二维码区域：zxing 优先（顺带拿到内容），OpenCV 兜底。"""
+    for name, variant in zxing_variants(img):
+        barcodes = zxing_read(variant)
+        if not barcodes:
+            continue
+
+        scale = variant.shape[1] / img.shape[1] if img.shape[1] else 1
+        regions: list[QrRegion] = []
+        for barcode in barcodes:
+            quad = _points_from_zxing(barcode)
+            if quad is None:
+                continue
+            if scale != 1:
+                quad = tuple(
+                    (int(round(x / scale)), int(round(y / scale))) for x, y in quad
+                )
+            quad = _clip_points(quad, img.shape)
+            regions.append(
+                QrRegion(
+                    points=quad,
+                    crop=_warp_qr_image(img, np.array(quad, dtype=np.float32)),
+                    text=getattr(barcode, "text", "") or "",
+                )
+            )
+        if regions:
+            if name != "原图":
+                log.debug(f"二维码检测命中变体: {name}")
+            return regions
+
+    return [
+        QrRegion(
+            points=quad,
+            crop=_warp_qr_image(img, np.array(quad, dtype=np.float32)),
+        )
+        for quad in _detect_by_opencv(img)
+    ]
+
+
 def scan_qr(image: bytes | np.ndarray | str | Path) -> QrDetectResult:
     """扫描图片中是否存在二维码，并返回二维码区域。"""
     img = _load_image(image)
-    points_list = _detect_points(img)
-
-    items = [
-        QrRegion(points=points, crop=_warp_qr_image(img, np.array(points, dtype=np.float32)))
-        for points in points_list
-    ]
+    items = _detect_regions(img)
     return QrDetectResult(has_qr=bool(items), items=items)
 
 

app/tool/qr_decode.py

@@ -1,4 +1,12 @@
-"""二维码识别工具。"""
+"""二维码识别工具。
+
+多级管线，从快到慢逐级重试，命中即止：
+    1. zxing-cpp 原图（默认已开反色 / 旋转 / 缩放重试）
+    2. zxing-cpp 换二值化算法（救低对比度、带纹理背景）
+    3. zxing-cpp 放大图（救长图里的小码、低分辨率码）
+    4. OpenCV 检测器 + 反色重试（zxing-cpp 缺失时的主路径）
+    5. 先裁剪二维码区域、补静默区再放大解码（救贴边、占比极小的码）
+"""
 
 from __future__ import annotations
 
@@ -7,48 +15,185 @@ from pathlib import Path
 import cv2
 import numpy as np
 
-from .cv import _load_image, _normalize_points, _warp_qr_image
+from app.core.logger import log
+
+from .cv import (
+    _HAS_ZXING,
+    _load_image,
+    _normalize_points,
+    _warp_qr_image,
+    pad_quiet_zone,
+    zxing_read,
+    zxing_variants,
+    zxingcpp,
+)
+
+# 裁剪区域重试时，把二维码放大到的目标边长
+_CROP_TARGET_SIDE = 480
 
 
-def _decode_with_detector(img: np.ndarray) -> list[str]:
-    detector = cv2.QRCodeDetector()
-    texts: list[str] = []
+def _dedupe(texts: list[str]) -> list[str]:
+    """去重且保持顺序。"""
+    seen: set[str] = set()
+    result: list[str] = []
+    for text in texts:
+        if text and text not in seen:
+            seen.add(text)
+            result.append(text)
+    return result
 
-    if hasattr(detector, "detectAndDecodeMulti"):
-        ok, decoded_info, points, _ = detector.detectAndDecodeMulti(img)
-        if ok and points is not None:
-            normalized_points = _normalize_points(points)
-            if isinstance(decoded_info, (list, tuple)):
-                decoded_iter = [item or "" for item in decoded_info]
-            else:
-                decoded_iter = [decoded_info or ""]
 
-            for idx, points_item in enumerate(normalized_points):
-                text = decoded_iter[idx] if idx < len(decoded_iter) else ""
-                if not text:
-                    crop = _warp_qr_image(img, np.array(points_item, dtype=np.float32))
-                    fallback_text, _, _ = detector.detectAndDecode(crop)
-                    text = fallback_text or ""
-                if text:
-                    texts.append(text)
-            if texts:
-                return texts
+def _texts_of(barcodes: list) -> list[str]:
+    return _dedupe([getattr(item, "text", "") or "" for item in barcodes])
 
-    text, points, _ = detector.detectAndDecode(img)
-    if text:
-        return [text]
 
-    normalized_points = _normalize_points(points)
-    if normalized_points:
-        crop = _warp_qr_image(img, np.array(normalized_points[0], dtype=np.float32))
-        fallback_text, _, _ = detector.detectAndDecode(crop)
-        if fallback_text:
-            return [fallback_text]
+def _stage_zxing_plain(img: np.ndarray) -> list[str]:
+    """原图直接交给 zxing-cpp。"""
+    return _texts_of(zxing_read(img))
 
+
+def _stage_zxing_binarizers(img: np.ndarray) -> list[str]:
+    """换二值化算法重试，应对低对比度和带图案的背景。"""
+    if not _HAS_ZXING:
+        return []
+
+    for binarizer in (
+        zxingcpp.Binarizer.GlobalHistogram,
+        zxingcpp.Binarizer.FixedThreshold,
+        zxingcpp.Binarizer.BoolCast,
+    ):
+        texts = _texts_of(zxing_read(img, binarizer=binarizer))
+        if texts:
+            log.debug(f"二维码解码命中二值化算法: {binarizer}")
+            return texts
     return []
 
 
+def _stage_zxing_upscaled(img: np.ndarray) -> list[str]:
+    """放大后重试，救小尺寸二维码。"""
+    for name, variant in zxing_variants(img):
+        if name == "原图":
+            continue
+        texts = _texts_of(zxing_read(variant, try_downscale=False))
+        if texts:
+            log.debug(f"二维码解码命中变体: {name}")
+            return texts
+    return []
+
+
+def _decode_with_opencv(img: np.ndarray) -> list[str]:
+    """OpenCV 检测器解码，含反色重试。"""
+    texts: list[str] = []
+
+    for candidate in (img, cv2.bitwise_not(img)):
+        for detector in (cv2.QRCodeDetector(), cv2.QRCodeDetectorAruco()):
+            try:
+                ok, infos, points, _ = detector.detectAndDecodeMulti(candidate)
+            except cv2.error:
+                ok, infos, points = False, None, None
+
+            if ok and infos is not None:
+                found = [item for item in infos if item]
+                if found:
+                    texts.extend(found)
+
+            if texts:
+                return _dedupe(texts)
+
+            try:
+                text, points, _ = detector.detectAndDecode(candidate)
+            except cv2.error:
+                continue
+
+            if text:
+                return [text]
+
+            # 检测到位置但没解出内容时，裁出区域再试一次
+            for quad in _normalize_points(points):
+                crop = _warp_qr_image(candidate, np.array(quad, dtype=np.float32))
+                if crop.size == 0:
+                    continue
+                try:
+                    fallback, _, _ = detector.detectAndDecode(crop)
+                except cv2.error:
+                    continue
+                if fallback:
+                    texts.append(fallback)
+
+            if texts:
+                return _dedupe(texts)
+
+    return _dedupe(texts)
+
+
+def _stage_opencv(img: np.ndarray) -> list[str]:
+    return _decode_with_opencv(img)
+
+
+def _stage_crop_retry(img: np.ndarray) -> list[str]:
+    """先定位并裁出二维码，补静默区放大后再解码。
+
+    针对二维码在长图里占比极小、或紧贴边缘没有留白的情况。
+    """
+    from .cv import _detect_regions  # 局部导入，避免循环依赖
+
+    texts: list[str] = []
+    for region in _detect_regions(img):
+        if region.text:
+            texts.append(region.text)
+            continue
+
+        crop = region.crop
+        if crop is None or crop.size == 0:
+            continue
+
+        padded = pad_quiet_zone(crop)
+        longest = max(padded.shape[:2])
+        if longest and longest < _CROP_TARGET_SIDE:
+            scale = _CROP_TARGET_SIDE / longest
+            padded = cv2.resize(
+                padded, None, fx=scale, fy=scale, interpolation=cv2.INTER_CUBIC
+            )
+
+        found = _texts_of(zxing_read(padded, try_downscale=False))
+        if not found:
+            found = _decode_with_opencv(padded)
+        texts.extend(found)
+
+    return _dedupe(texts)
+
+
+# 管线顺序：命中即返回
+_STAGES = (
+    ("zxing原图", _stage_zxing_plain),
+    ("zxing二值化", _stage_zxing_binarizers),
+    ("zxing放大", _stage_zxing_upscaled),
+    ("opencv", _stage_opencv),
+    ("裁剪重试", _stage_crop_retry),
+)
+
+
 def decode_qr(image: bytes | np.ndarray | str | Path) -> list[str]:
-    """识别图片中的二维码内容。"""
+    """识别图片中的二维码内容。
+
+    Args:
+        image: 图片字节流、BGR 图像数组或图片路径。
+
+    Returns:
+        二维码内容列表，已去重；没识别出来时返回空列表。
+    """
     img = _load_image(image)
-    return _decode_with_detector(img)
+
+    for name, stage in _STAGES:
+        try:
+            texts = stage(img)
+        except Exception as exc:  # 单级失败不影响后续重试
+            log.debug(f"二维码解码阶段异常({name}): {exc}")
+            continue
+
+        if texts:
+            if name != "zxing原图":
+                log.debug(f"二维码解码命中阶段: {name}")
+            return texts
+
+    return []

requirements.txt

@@ -3,6 +3,7 @@ httpx>=0.28
 rapidocr>=3.9
 onnxruntime>=1.17
 opencv-python>=4.5
+zxing-cpp>=3.1
 playwright>=1.49
 pydantic-settings>=2.0
 
@@ -20,3 +21,6 @@ json-repair>=0.61
 loguru>=0.7
 oss2==2.19.1
 snowflake-id>=1.0
+
+# 定时任务
+apscheduler>=3.10,<4