修改创建sql

.env

@@ -1,10 +1,10 @@
 ENV=dev
 
 # MySQL（业务库）
-DB_HOST=8.163.14.142
-DB_PORT=30006
+DB_HOST=192.168.31.105
+DB_PORT=3306
 DB_USER=root
-DB_PASSWORD=^CgDatabase2020
+DB_PASSWORD=123456
 DB_NAME=offerpie
 MYSQL_POOL_SIZE=10
 MYSQL_MAX_OVERFLOW=10
@@ -20,3 +20,14 @@ VOLCENGINE_BASE_URL=https://ark.cn-beijing.volces.com/api/v3
 # Claude（Anthropic 风格）
 ANTHROPIC_API_KEY=sk-43ccdb29caa7e9ebe0db8ac0958c63f6d3a2d62e59064d3d26d94332055a9bc9
 ANTHROPIC_BASE_URL=https://code.warpdevloper.cloud
+
+
+# 阿里云 OSS
+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
+
+# 公告落库并发线程数（不要超过 MYSQL_POOL_SIZE + MYSQL_MAX_OVERFLOW）
+SPIDER_MAX_WORKERS=32

.env.prod

@@ -20,3 +20,14 @@ VOLCENGINE_BASE_URL=https://ark.cn-beijing.volces.com/api/v3
 # Claude（Anthropic 风格）
 ANTHROPIC_API_KEY=sk-43ccdb29caa7e9ebe0db8ac0958c63f6d3a2d62e59064d3d26d94332055a9bc9
 ANTHROPIC_BASE_URL=https://code.warpdevloper.cloud
+
+
+# 阿里云 OSS
+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
+
+# 公告落库并发线程数（不要超过 MYSQL_POOL_SIZE + MYSQL_MAX_OVERFLOW）
+SPIDER_MAX_WORKERS=16

.env.test

@@ -20,3 +20,14 @@ VOLCENGINE_BASE_URL=https://ark.cn-beijing.volces.com/api/v3
 # Claude（Anthropic 风格）
 ANTHROPIC_API_KEY=sk-43ccdb29caa7e9ebe0db8ac0958c63f6d3a2d62e59064d3d26d94332055a9bc9
 ANTHROPIC_BASE_URL=https://code.warpdevloper.cloud
+
+
+# 阿里云 OSS
+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
+
+# 公告落库并发线程数（不要超过 MYSQL_POOL_SIZE + MYSQL_MAX_OVERFLOW）
+SPIDER_MAX_WORKERS=12

Dockerfile

@@ -0,0 +1,51 @@
+# 使用 Python 3.12 slim 镜像
+# 必须固定 bookworm（Debian 12）：
+#   1. 下面的 sources.list 写的是 bookworm，python:3.12-slim 已滚动到 trixie(Debian 13)，混用会导致包名找不到
+#   2. Playwright 1.49 的支持列表里没有 Debian 13，install --with-deps 探测到未知发行版会直接失败
+FROM python:3.12-slim-bookworm
+
+ENV TZ=Asia/Shanghai
+ENV ENV=prod
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+
+# Playwright 浏览器统一装到系统目录，避免落在 HOME 里
+ENV PLAYWRIGHT_BROWSERS_PATH=/ms-playwright
+# 浏览器下载走 npmmirror 镜像，国内构建更快（网络可直连时可去掉）
+ENV PLAYWRIGHT_DOWNLOAD_HOST=https://cdn.npmmirror.com/binaries/playwright
+
+# 时区 + 系统依赖（libgl1、libglib2.0-0 是 opencv-python 的运行时依赖）
+RUN ln -sf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone \
+    && rm -rf /etc/apt/sources.list.d/* \
+    && echo "deb https://mirrors.aliyun.com/debian/ bookworm main non-free contrib" > /etc/apt/sources.list \
+    && echo "deb https://mirrors.aliyun.com/debian-security/ bookworm-security main non-free contrib" >> /etc/apt/sources.list \
+    && echo "deb https://mirrors.aliyun.com/debian/ bookworm-updates main non-free contrib" >> /etc/apt/sources.list \
+    && apt-get clean \
+    && apt-get update \
+    && apt-get install -y --no-install-recommends curl libgl1 libglib2.0-0 \
+    && rm -rf /var/lib/apt/lists/*
+
+RUN mkdir -p /app/logs
+
+WORKDIR /app
+
+# 先拷贝依赖声明，利用 Docker 层缓存
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt \
+    -i https://mirrors.aliyun.com/pypi/simple/ \
+    --trusted-host mirrors.aliyun.com
+
+# 安装 Chromium 及其系统依赖（供 app/tool/browser.py 使用）
+RUN playwright install --with-deps chromium \
+    && rm -rf /var/lib/apt/lists/*
+
+# 预热 OCR 模型，避免首次采集时才去下载 onnx 模型
+# 参数与 app/tool/ocr.py 保持一致，确保命中同一份模型缓存
+RUN python -c "from rapidocr import RapidOCR; RapidOCR(params={'Global.use_cls': False, 'Global.text_score': 0.5})"
+
+# 拷贝应用代码和环境配置（init_*.py 存量初始化脚本本地跑，不进镜像）
+COPY app/ ./app/
+COPY .env.prod ./.env.prod
+
+# 纯后台定时任务服务，无 HTTP 端口
+CMD ["python", "-m", "app.main"]

Jenkinsfile

@@ -0,0 +1,93 @@
+/**
+ * OfferPie Spider 部署流水线
+ *
+ * Jenkins 容器挂载了宿主机 docker.sock，镜像共享。
+ * 直接在 workspace 内 build + compose up。
+ *
+ * 只负责常驻定时任务服务（app.main）；存量初始化脚本 init_*.py 在本地手动跑。
+ */
+pipeline {
+    agent any
+
+    parameters {
+        choice(name: 'BRANCH', choices: ['master', 'dev'], description: '选择要部署的分支')
+        choice(name: 'ACTION', choices: ['deploy', 'stop'], description: '操作：deploy=构建部署，stop=停止服务')
+    }
+
+    environment {
+        IMAGE_NAME = 'offerpie-spider'
+        IMAGE_TAG = 'latest'
+        CONTAINER_NAME = 'offerpie-spider'
+        // 固定 compose 项目名：默认取 workspace 目录名，并发构建时会变成 xxx@2 导致容器名冲突
+        COMPOSE_PROJECT_NAME = 'offerpie-spider'
+    }
+
+    stages {
+        stage('停止服务') {
+            when {
+                expression { params.ACTION == 'stop' }
+            }
+            steps {
+                sh "docker compose down || true"
+            }
+        }
+
+        stage('拉取代码') {
+            when {
+                expression { params.ACTION == 'deploy' }
+            }
+            steps {
+                echo "拉取 ${params.BRANCH} 分支代码"
+                git branch: "${params.BRANCH}",
+                    credentialsId: 'gitea-fab089c1-b55d-4b58-9fad',
+                    url: 'https://git.jianshixingqiu.com/offerpai/campus_spider.git'
+            }
+        }
+
+        stage('构建镜像') {
+            when {
+                expression { params.ACTION == 'deploy' }
+            }
+            steps {
+                sh "docker build -t ${IMAGE_NAME}:${IMAGE_TAG} ."
+            }
+        }
+
+        stage('部署') {
+            when {
+                expression { params.ACTION == 'deploy' }
+            }
+            steps {
+                // 停旧启新，docker-compose.yml 就在当前 workspace
+                sh "docker compose down || true"
+                sh "docker compose up -d"
+                sleep 5
+                sh "docker ps -f name=${CONTAINER_NAME} --format '{{.Status}}'"
+            }
+        }
+
+        stage('清理') {
+            when {
+                expression { params.ACTION == 'deploy' }
+            }
+            steps {
+                sh "docker image prune -f || true"
+            }
+        }
+    }
+
+    post {
+        success {
+            script {
+                if (params.ACTION == 'deploy') {
+                    echo "✅ 部署成功！容器 ${CONTAINER_NAME} 已启动"
+                } else {
+                    echo "✅ 服务已停止"
+                }
+            }
+        }
+        failure {
+            echo '❌ 操作失败，请检查日志'
+        }
+    }
+}

app/ai/extract/announcement_extract.py

@@ -0,0 +1,42 @@
+"""招聘公告信息提取：调用 CONTENT_EXTRACT 模型，输出结构化字典。"""
+
+from __future__ import annotations
+
+from langchain_core.messages import HumanMessage, SystemMessage
+
+from app.ai.extract.prompt import EXTRACT_SYSTEM_PROMPT
+from app.ai.model_config import SpiderModel
+from app.core.logger import log
+from app.tool.json_helper import parse_llm_json
+
+
+def extract_announcement(content: str) -> dict | None:
+    """从公告全文中提取结构化信息。
+
+    Args:
+        content: 公告全文（正文 + 图片 OCR 文字 + 二维码内容拼接）。
+
+    Returns:
+        提取出的字段字典；内容为空、模型调用失败或结果不是字典时返回 None。
+    """
+    if not content or not content.strip():
+        return None
+
+    messages = [
+        SystemMessage(content=EXTRACT_SYSTEM_PROMPT),
+        HumanMessage(content=content),
+    ]
+
+    try:
+        response = SpiderModel.CONTENT_EXTRACT.invoke(messages)
+    except Exception as exc:
+        log.error(f"公告信息提取失败: {exc}")
+        return None
+
+    result = parse_llm_json(str(response.content))
+    if not isinstance(result, dict):
+        log.warning(f"公告信息提取结果不是 JSON 对象: {type(result).__name__}")
+        return None
+
+    log.info(f"公告信息提取完成（字段 {len(result)} 个）: {result.get('title')}")
+    return result

app/ai/extract/prompt.py

@@ -0,0 +1,155 @@
+"""招聘公告信息提取提示词"""
+
+# 招聘公告信息提取系统提示词：输入公告全文，输出结构化 JSON
+EXTRACT_SYSTEM_PROMPT = """你是招聘公告信息提取专家。用户会给你一份校园招聘公告的全文（由网页正文、图片 OCR 文字、二维码解析内容拼接而成，可能存在顺序错乱、重复、噪声）。
+
+请从中提取信息，严格按下面的 JSON 结构输出。
+
+## 输出格式
+
+```json
+{
+  "company_name": "公司名称",
+  "title": "公告标题",
+  "company_intro": "公司简介",
+  "target_audience": "面向对象",
+  "major_require": "专业要求",
+  "recruit_position": "招聘岗位",
+  "remark": "备注",
+  "written_exam": "有",
+  "apply_start_time": "2026-07-01 00:00:00",
+  "apply_end_time": "2026-09-30 23:59:59",
+  "apply_end_desc": "投递截止描述",
+  "invite_code": "内推码",
+  "apply_url": "投递地址",
+  "apply_email": "投递邮箱",
+  "source": "信息来源说明",
+  "publish_time": "2026-07-01 00:00:00",
+  "recruit_years": [2027],
+  "batches": ["暑期实习"],
+  "tags": ["六险一金", "不限专业", "带薪年假", "导师带教"],
+  "cities": ["北京市"],
+  "categories": ["后端开发", "人工智能"],
+  "industries": ["互联网", "人工智能"],
+  "educations": ["本科", "硕士"]
+}
+```
+
+## 字段说明
+
+- company_name：招聘的公司名称，取简称
+- title：公告标题，如「XX公司2027届秋季校园招聘正式启动」
+- company_intro：公司介绍段落，原文摘录，最多 300 字。公告中没有公司介绍时，可以根据你自己对该公司的了解补充一段简介；不了解该公司时填 null
+- target_audience：面向对象，如「2027届本硕博」「2026届及2027届毕业生」，最多 200 字
+- major_require：专业要求，如「计算机、电子信息、自动化等相关专业」，最多 200 字
+- recruit_position：本次招聘的具体岗位名称，多个岗位用「、」连接，如「后端开发工程师、算法工程师、产品经理」，
+  最多 500 字。只填岗位名称本身，不要带岗位职责、任职要求、招聘人数等描述。岗位过多时保留最主要的若干个
+- remark：其他值得注意的补充信息，如薪资待遇、福利、流程安排、注意事项，最多 1000 字
+- written_exam：是否有笔试，只能填「有」「无」「未明确」三者之一
+- apply_start_time：投递开始时间，格式 yyyy-MM-dd HH:mm:ss
+- apply_end_time：投递截止时间，格式 yyyy-MM-dd HH:mm:ss
+- apply_end_desc：无法解析成具体时间的截止描述，如「尽快投递」「招满即止」「长期有效」，最多 64 字
+- invite_code：专属内推码，只填码本身，不要带「内推码：」这类前缀
+- apply_url：投递地址，只填一个，必须是完整 http/https 链接。二维码解出的投递链接也算。
+  公告中出现多个链接时，根据链接路径判断最可能是网申投递入口的那一个，优先级参考：
+  路径含 campus / recruit / job / apply / talent / hr / zhaopin / xiaozhao 等招聘语义词 >
+  企业招聘域名（如 campus.xxx.com、hr.xxx.com、xxx.zhiye.com）>
+  第三方招聘平台投递页（如 mokahr、beisen、moseeker、workday、talent.liepin 等）>
+  其他链接。公告首页、公众号文章、官网首页、下载链接、图片链接一律不算投递地址
+- apply_email：投递邮箱，只填一个。出现多个邮箱时，根据邮箱用途判断最可能用于简历投递的那一个，
+  优先级参考：本地部分含 campus / recruit / hr / job / zhaopin / resume 等招聘语义词 >
+  企业自有域名邮箱 > 公共邮箱（qq.com / 163.com 等）。咨询、客服、商务合作类邮箱不算投递邮箱
+- source：信息来源说明，如公告中提到的发布方、公众号名称
+- publish_time：公告发布或更新时间，格式 yyyy-MM-dd HH:mm:ss
+- recruit_years：招聘届数数组，整数年份，如 [2027]、[2026, 2027]
+- batches：招聘批次数组，**只能从下列固定值中原样选取**，不得改写或自造：
+  「实习」「暑期实习」「寒假实习」
+  「秋招提前批」「秋招正式批」「秋招补招」
+  「春招提前批」「春招正式批」「春招补招」
+  「校园招聘」「社招」
+  判断依据是公告标题、正文中的批次表述以及投递时间所处季节，规则如下：
+  - 实习岗且能判断季节的用「暑期实习」「寒假实习」，只说实习没说季节的用「实习」
+  - 校招正式岗按季节和阶段选：提前批/内推批归「XX提前批」，主招/正式批/专场归「XX正式批」，补录/二次招聘归「XX补招」
+  - 确定是校招但分不清秋招/春招或所处阶段时，用「校园招聘」兜底
+  - 面向社会人士、要求工作经验的用「社招」
+  最多 3 个，无法判断时填 []
+- cities：工作城市数组，规范到市级，如「北京市」「深圳市」
+- categories：岗位分类数组。**只能从文末「岗位分类数据」列表中原样选取**，不得改写、合并、简化或自造分类名。
+  根据 recruit_position 中的岗位判断，可多选，最多 5 个，按相关度从高到低排列。
+  找不到精确匹配时，选该岗位所属领域下的「其他XX职位」；连所属领域都无法判断时填 []
+- industries：行业分类数组。**只能从文末「行业分类数据」列表中原样选取**，不得改写、合并、简化或自造行业名。
+  根据 company_name、company_intro 中体现的公司主营业务判断，可多选，最多 3 个，按相关度从高到低排列。
+  找不到精确匹配时，选同一领域下最接近的行业（如该领域有「其他XX」则用它）；连所属领域都无法判断时填「其他行业」
+- educations：学历要求数组，如「大专」「本科」「硕士」「博士」
+- tags：其他关键标签数组，如「竞争力薪酬」「六险一金」「可远程」「不限专业」。
+  提取 3-6 个，从公告的薪资福利、工作方式、专业/学历门槛、培养机制、流程特点、地点等维度概括，
+  允许在不改变原意的前提下把原文表述凝练成短标签，每个标签 2-8 字，不与 batches、cities、educations 重复。
+  宁缺勿滥：公告内容确实撑不起 3 个标签时就按实际数量输出，不要为了凑数拆分同一条信息或写空泛标签
+
+## 提取规则
+
+1. 只提取公告中明确出现的信息，不要推测、不要补全、不要编造（company_intro、tags 例外，
+   tags 允许对公告内容做概括提炼，但不能提炼出公告里没有依据的内容）。
+2. 字符串字段找不到时填 null，数组字段找不到时填 []。不要填「无」「未知」「暂无」这类占位文字（written_exam 例外）。
+3. 时间统一输出 yyyy-MM-dd HH:mm:ss。
+   - 只有日期没有时刻：开始时间补 00:00:00，截止时间补 23:59:59。
+   - 只写了月日没写年份：结合公告届数或发布时间推断年份；无法推断则填 null。
+   - 出现「即日起」「长期开放」这类无法定位到具体日期的表述：对应时间字段填 null，把原文写进 apply_end_desc。
+4. apply_end_time 和 apply_end_desc 可以同时有值，也可以只有一个。
+5. 数组元素去重，保持公告中出现的顺序，单个元素不超过 64 字。
+6. 一份公告涉及多家公司时，以主体招聘方为准。
+7. batches、categories 和 industries 都是封闭枚举：batches 必须是字段说明中列出的 11 个值之一，
+   categories 必须与文末「岗位分类数据」列表中的某一项完全一致（含标点「/」），
+   industries 必须与文末「行业分类数据」列表中的某一项完全一致（含标点「/」「(020)」等）。
+   出现枚举之外的值视为错误输出。
+8. 直接输出 JSON，不要输出任何解释文字。
+
+## 岗位分类数据
+
+categories 的取值范围如下（按领域分组，冒号后的名称才是合法取值）：
+
+- 技术研发：后端开发、前端/移动开发、测试、运维/技术支持、人工智能、数据、技术项目管理、销售技术支持、高端技术职位、其他技术职位
+- 硬件通信：电子/硬件开发、半导体/芯片、电气/自动化、通信
+- 产品运营：产品经理、游戏策划/制作、客服、内容运营、电商运营、业务运营、线下运营、编辑、高端运营职位、其他运营职位
+- 销售：销售、销售管理、销售行政/商务、外贸销售、教培销售、汽车销售、房地产销售/招商、服务业销售、医疗销售、广告/会展销售、金融销售、其他销售职位
+- 职能：人力资源、行政、法律服务、其他职能职位
+- 财务：会计、审计/税务、高级财务职位、其他财务岗位
+- 生产制造：普工、机械加工、技工、运输设备操作、质量管理、机械设计/制造、生产营运、生产安全、化工、服装/纺织/皮革、新能源汽车、汽车研发/制造、环保、其他生产制造职位
+- 服务业：零售、美容美发、理疗保健、家政/保洁、安保服务、维修服务、汽车服务、宠物服务、运动健身、驾驶员、其他服务业职位
+- 餐饮：前厅、后厨、餐饮管理、甜点饮品、其他餐饮岗位
+- 酒店旅游：酒店、旅游服务、其他旅游职位
+- 教育培训：教师、幼少儿教师、教育行政、文化艺术、科学探索培训、职业培训、教育产品研发、其他教育培训职位
+- 设计：视觉/交互设计、环境设计、工业设计、服装设计、美术/3D/动画、游戏设计、高端设计职位、其他设计职位
+- 房地产建筑：工程管理、装饰装修、物业管理、建筑/规划设计、房地产规划开发、建筑/装修工人、高端房地产职位、其他房地产职位
+- 传媒：直播、影视、广告、采编/写作/出版、其他传媒职位
+- 市场：市场营销、推广/投放、政府事务、公关、调研分析、其他市场职位
+- 采购物流贸易：物流/运输、配送理货、仓储、供应链、采购、进出口贸易、其他采购/贸易职位
+- 医疗健康：护士/护理、医生/医技、药店、生物医药、临床试验、医疗器械、其他医疗健康职位
+- 金融：银行、证券/基金/期货、中后台、投融资、保险、其他金融职位
+- 咨询翻译：咨询/调研、翻译、其他咨询类职位
+- 其他：能源/地质、农/林/牧/渔、高级管理职位、其他职位类别
+
+
+## 行业分类数据
+
+industries 的取值范围如下（按领域分组，冒号后的名称才是合法取值）：
+
+- 互联网/IT：互联网、生活服务(020)、游戏、云计算、大数据、新零售、电子商务、企业服务、社交网络与媒体、在线教育、广告营销、信息安全、计算机软件、医疗健康、人工智能、计算机服务、物联网
+- 电子硬件通信：半导体/芯片、智能硬件/消费电子、电子/硬件开发、运营商/增值服务、通信/网络设备、计算机硬件、电子/半导体/集成电路
+- 生活服务：餐饮、酒店/民宿、保健/养生、婚庆/摄影、美容/美发、美容、美发、休闲/娱乐、家政服务、宠物服务、运动/健身、旅游/景区、回收/维修、其他生活服务
+- 消费品零售：批发/零售、服装/纺织、日化、进出口贸易、家具/家居、家具/家电/家居、家用电器、珠宝/首饰、食品/饮料/烟酒、其他消费品
+- 房地产建筑：房地产开发经营、房地产中介/租赁、物业管理、房屋建筑工程、土木工程、工程施工、建筑设计、建筑材料、建筑工程咨询服务、机电工程、装修装饰、土地与公共设施管理
+- 教育培训：培训/辅导机构、学校/学历教育、职业培训、学前教育、学术/科研
+- 文娱传媒：文化艺术/娱乐、广播/影视、新闻/出版、广告/公关/会展、体育
+- 制造业：通用设备、专用设备、自动化设备、电气机械/器材、机械设备/机电/重工、仪器仪表、仪器仪表/工业自动化、计算机/通信/其他电子设备、铁路/船舶/航空/航天制造、金属制品、非金属矿物制品、橡胶/塑料制品、化学原料/化学制品、新材料、原材料及加工/模具、印刷/包装/造纸、其他制造业
+- 专业服务：咨询、法律、财务/审计/税务、人力资源服务、检测/认证/知识产权、翻译、其他专业服务
+- 医疗医药：医疗服务、医美服务、生物/制药、医疗器械、医药批发零售、医疗研发外包、IVD
+- 汽车：汽车研发/制造、新能源汽车、汽车零部件、汽车智能网联、汽车后市场、4S店/后市场、汽车经销商、摩托车/自行车制造
+- 物流运输：物流/仓储、交通/运输、公路物流、跨境物流、快递、即时配送、同城货运、客运服务、港口/铁路/公路/机场、装卸搬运和仓储业
+- 能源环保：新能源、光伏、风电、储能、动力电池、其他新能源、电力/热力/燃气/水利、石油/石化、化工、采掘/冶炼、矿产/地质、环保
+- 金融：银行、证券/期货、基金、保险、投资/融资、财富管理、互联网金融、信托租赁/拍卖/典当/担保、其他金融业
+- 其他：农/林/牧/渔、政府/公共事业、非盈利机构、其他行业
+
+
+
+"""

app/ai/model_config.py

@@ -10,4 +10,4 @@ class SpiderModel:
     """爬虫 AI 场景模型。"""
 
     # 页面内容理解与结构化提取
-    CONTENT_EXTRACT = LLM.CLAUDE_OPUS.create(temperature=0)
+    CONTENT_EXTRACT = LLM.DOUBAO_SEED_LITE.create(temperature=0)

app/config/settings.py

@@ -26,6 +26,18 @@ class Settings(BaseSettings):
     anthropic_api_key: str = ""
     anthropic_base_url: str = ""
 
+    # ──────────── 阿里云 OSS ────────────
+    oss_access_key_id: str = ""
+    oss_access_key_secret: str = ""
+    oss_endpoint: str = "oss-cn-guangzhou.aliyuncs.com"
+    oss_bucket: str = "offerpie"
+    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/core/oss.py

@@ -0,0 +1,51 @@
+"""阿里云 OSS 图片上传工具。"""
+
+from __future__ import annotations
+
+import os
+from urllib.parse import unquote, urlparse
+from uuid import uuid4
+
+import oss2
+
+from app.config import settings
+from app.core.logger import log
+from app.tool.image_download import download_image
+
+_auth = oss2.Auth(settings.oss_access_key_id, settings.oss_access_key_secret)
+_bucket = oss2.Bucket(_auth, f"https://{settings.oss_endpoint}", settings.oss_bucket)
+
+
+def _get_file_name(image_url: str) -> str:
+    """从图片 URL 提取文件名，无扩展名时默认使用 PNG。"""
+    path = unquote(urlparse(image_url).path)
+    file_name = os.path.basename(path)
+    if not file_name or not os.path.splitext(file_name)[1]:
+        return "image.png"
+    return file_name
+
+
+def upload(data: bytes, file_name: str) -> str:
+    """上传图片字节到固定目录，返回 OSS URL。"""
+    key = f"{settings.oss_upload_dir}/{uuid4().hex[:18]}{os.path.splitext(file_name)[1]}"
+    _bucket.put_object(key, data)
+    url = f"{settings.oss_domain}/{key}"
+    log.info("OSS 上传成功 [{}] -> {}", file_name, url)
+    return url
+
+
+def upload_image_url(image_url: str) -> str | None:
+    """下载图片 URL 并上传到 OSS，成功时返回 OSS URL。"""
+    if not image_url or not image_url.strip():
+        return None
+
+    image_url = image_url.strip()
+    image = download_image(image_url)
+    if image is None:
+        return None
+
+    try:
+        return upload(image, _get_file_name(image_url))
+    except Exception as exc:
+        log.warning("图片上传 OSS 失败 [{}]: {}", image_url, exc)
+        return None

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/models/recruit_announcement.py

@@ -3,7 +3,7 @@
 from datetime import datetime
 from typing import Optional
 
-from sqlalchemy import BigInteger, DateTime, String, Text
+from sqlalchemy import BigInteger, DateTime, Integer, String, Text
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.core.database import MysqlBase
@@ -21,6 +21,7 @@ class RecruitAnnouncement(MysqlBase):
     company_intro: Mapped[Optional[str]] = mapped_column(Text, comment="公司简介")
     target_audience: Mapped[Optional[str]] = mapped_column(String(500), comment="面向对象，如2027届本硕博")
     major_require: Mapped[Optional[str]] = mapped_column(String(500), comment="专业要求")
+    recruit_position: Mapped[Optional[str]] = mapped_column(String(1000), comment="招聘岗位")
     remark: Mapped[Optional[str]] = mapped_column(String(1000), comment="备注")
     written_exam: Mapped[Optional[str]] = mapped_column(String(16), comment="是否笔试：有 / 无 / 未明确")
     apply_start_time: Mapped[Optional[datetime]] = mapped_column(DateTime, comment="投递开始时间")
@@ -32,5 +33,7 @@ class RecruitAnnouncement(MysqlBase):
     apply_email: Mapped[Optional[str]] = mapped_column(String(128), comment="投递邮箱")
     source: Mapped[Optional[str]] = mapped_column(String(255), comment="信息来源说明")
     publish_time: Mapped[Optional[datetime]] = mapped_column(DateTime, comment="公告发布/更新时间")
+    clean_status: Mapped[int] = mapped_column(Integer, nullable=False, default=0, comment="清洗状态：0-清洗中 1-清洗完成")
+    status: Mapped[int] = mapped_column(Integer, nullable=False, default=1, comment="状态：0-失效 1-有效")
     create_time: Mapped[datetime] = mapped_column(DateTime, nullable=False, comment="创建时间")
     update_time: Mapped[datetime] = mapped_column(DateTime, nullable=False, comment="更新时间")

app/models/recruit_announcement_batch.py

@@ -15,5 +15,5 @@ class RecruitAnnouncementBatch(MysqlBase):
 
     id: Mapped[int] = mapped_column(BigInteger, primary_key=True, comment="主键ID(雪花)")
     announcement_id: Mapped[int] = mapped_column(BigInteger, nullable=False, comment="公告id")
-    batch_name: Mapped[str] = mapped_column(String(64), nullable=False, comment="批次值，如实习/暑期实习/寒假实习/秋招专场/春招提前批/春招补招")
+    batch_name: Mapped[str] = mapped_column(String(64), nullable=False, comment="批次值，枚举：实习/暑期实习/寒假实习/秋招提前批/秋招正式批/秋招补招/春招提前批/春招正式批/春招补招/校园招聘/社招")
     create_time: Mapped[datetime] = mapped_column(DateTime, nullable=False, comment="创建时间")

app/models/recruit_announcement_category.py

@@ -15,5 +15,5 @@ class RecruitAnnouncementCategory(MysqlBase):
 
     id: Mapped[int] = mapped_column(BigInteger, primary_key=True, comment="主键ID(雪花)")
     announcement_id: Mapped[int] = mapped_column(BigInteger, nullable=False, comment="公告id")
-    category_name: Mapped[str] = mapped_column(String(64), nullable=False, comment="岗位大类值，如IT技术/人工智能/通信")
+    category_name: Mapped[str] = mapped_column(String(64), nullable=False, comment="岗位分类值，取自固定分类表，如后端开发/人工智能/产品经理")
     create_time: Mapped[datetime] = mapped_column(DateTime, nullable=False, comment="创建时间")

app/models/recruit_announcement_industry.py

@@ -0,0 +1,19 @@
+"""MySQL: bg_recruit_announcement_industry 招聘公告-行业关联表模型"""
+
+from datetime import datetime
+
+from sqlalchemy import BigInteger, DateTime, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.database import MysqlBase
+
+
+class RecruitAnnouncementIndustry(MysqlBase):
+    """招聘公告-行业关联表"""
+
+    __tablename__ = "bg_recruit_announcement_industry"
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, comment="主键ID(雪花)")
+    announcement_id: Mapped[int] = mapped_column(BigInteger, nullable=False, comment="公告id")
+    industry_name: Mapped[str] = mapped_column(String(64), nullable=False, comment="行业分类值，取自固定分类表，如互联网/人工智能/半导体/芯片")
+    create_time: Mapped[datetime] = mapped_column(DateTime, nullable=False, comment="创建时间")

app/models/recruit_announcement_tag.py

@@ -15,5 +15,5 @@ class RecruitAnnouncementTag(MysqlBase):
 
     id: Mapped[int] = mapped_column(BigInteger, primary_key=True, comment="主键ID(雪花)")
     announcement_id: Mapped[int] = mapped_column(BigInteger, nullable=False, comment="公告id")
-    tag_name: Mapped[str] = mapped_column(String(64), nullable=False, comment="标签值，如秋招提前批/竞争力薪酬/校招")
+    tag_name: Mapped[str] = mapped_column(String(64), nullable=False, comment="标签值，如竞争力薪酬/六险一金/不限专业，不与批次、城市、学历重复")
     create_time: Mapped[datetime] = mapped_column(DateTime, nullable=False, comment="创建时间")

app/service/__init__.py

@@ -0,0 +1 @@
+"""业务服务层。"""

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/service/company_service.py

@@ -0,0 +1,73 @@
+"""公司业务服务：查找或创建公司基础记录，上传 Logo。"""
+
+from __future__ import annotations
+
+import threading
+from datetime import datetime
+
+from sqlalchemy import insert, text
+
+from app.core.database import MysqlSession
+from app.core.id_gen import next_id
+from app.core.logger import log
+from app.core.oss import upload_image_url
+from app.models.company import Company
+
+# 公司创建锁（防止并发重复插入同一公司）
+_company_lock = threading.Lock()
+
+
+def find_or_create_company(company_name: str, logo_url: str | None = None) -> int:
+    """查找或创建公司，上传 Logo 并回填地址。
+
+    Args:
+        company_name: 公司名称（用于查重和创建）。
+        logo_url: Logo 图片原始地址，非空时下载并上传 OSS。
+
+    Returns:
+        公司 ID。
+    """
+    if not company_name:
+        company_name = "未知公司"
+
+    with _company_lock:
+        with MysqlSession() as session:
+            row = session.execute(
+                text("SELECT id FROM bg_company WHERE short_name = :name LIMIT 1"),
+                {"name": company_name},
+            )
+            existing = row.scalar()
+            if existing:
+                return existing
+
+            company_id = next_id()
+            now = datetime.now()
+            session.execute(
+                insert(Company).values(
+                    id=company_id,
+                    name=company_name,
+                    short_name=company_name,
+                    status=0,
+                    create_time=now,
+                    update_time=now,
+                )
+            )
+            session.commit()
+            log.info("公司创建成功 [id={}]: {}", company_id, company_name)
+
+    # 锁外处理 Logo 上传
+    if logo_url:
+        try:
+            oss_url = upload_image_url(logo_url)
+            if oss_url:
+                with MysqlSession() as session:
+                    session.execute(
+                        text("UPDATE bg_company SET logo_url = :url, update_time = :t WHERE id = :id"),
+                        {"url": oss_url, "t": datetime.now(), "id": company_id},
+                    )
+                    session.commit()
+                log.info("公司 Logo 上传成功 [id={}]: {}", company_id, oss_url)
+        except Exception as exc:
+            log.warning("公司 Logo 上传失败 [id={}]: {}", company_id, exc)
+
+    return company_id

app/service/recruit_announcement_service.py

@@ -0,0 +1,198 @@
+"""招聘公告业务服务：编排公告爬取、AI 信息提取与数据库保存。"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import insert, text
+
+from app.ai.extract.announcement_extract import extract_announcement
+from app.core.database import MysqlSession
+from app.core.id_gen import next_id
+from app.core.logger import log
+from app.models.recruit_announcement import RecruitAnnouncement
+from app.models.recruit_announcement_batch import RecruitAnnouncementBatch
+from app.models.recruit_announcement_category import RecruitAnnouncementCategory
+from app.models.recruit_announcement_city import RecruitAnnouncementCity
+from app.models.recruit_announcement_education import RecruitAnnouncementEducation
+from app.models.recruit_announcement_industry import RecruitAnnouncementIndustry
+from app.models.recruit_announcement_tag import RecruitAnnouncementTag
+from app.models.recruit_announcement_year import RecruitAnnouncementYear
+from app.service.company_service import find_or_create_company
+from app.tool.page_extract import extract_page
+
+# 微信公众号文章域名，页面提取逻辑只适配了这一种页面结构
+_WECHAT_DOMAIN = "mp.weixin.qq.com"
+
+
+def _parse_datetime(value: str | None) -> datetime | None:
+    """将 yyyy-MM-dd HH:mm:ss 字符串解析为 datetime，失败返回 None。"""
+    if not value:
+        return None
+    try:
+        return datetime.strptime(value, "%Y-%m-%d %H:%M:%S")
+    except (ValueError, TypeError):
+        return None
+
+
+def _truncate(value: object, limit: int) -> str | None:
+    """转成字符串并按上限截断，空值返回 None。"""
+    if value is None or value == "":
+        return None
+    return str(value)[:limit]
+
+
+def _save_announcement(announcement_id: int, company_id: int, url: str, data: dict) -> None:
+    """保存公告主表和七张关联表，单事务提交。"""
+    now = datetime.now()
+
+    with MysqlSession() as session:
+        # 主表
+        session.execute(
+            insert(RecruitAnnouncement).values(
+                id=announcement_id,
+                company_id=company_id,
+                company_name=data.get("company_name") or "",
+                title=data.get("title") or "",
+                company_intro=data.get("company_intro"),
+                target_audience=data.get("target_audience"),
+                major_require=data.get("major_require"),
+                recruit_position=_truncate(data.get("recruit_position"), 1000),
+                remark=data.get("remark"),
+                written_exam=data.get("written_exam"),
+                apply_start_time=_parse_datetime(data.get("apply_start_time")),
+                apply_end_time=_parse_datetime(data.get("apply_end_time")),
+                apply_end_desc=data.get("apply_end_desc"),
+                invite_code=data.get("invite_code"),
+                announcement_url=url,
+                apply_url=data.get("apply_url"),
+                apply_email=data.get("apply_email"),
+                source=data.get("source"),
+                publish_time=_parse_datetime(data.get("publish_time")),
+                clean_status=1,
+                status=1,
+                create_time=now,
+                update_time=now,
+            )
+        )
+
+        # 招聘届数
+        for year in data.get("recruit_years") or []:
+            session.execute(
+                insert(RecruitAnnouncementYear).values(
+                    id=next_id(),
+                    announcement_id=announcement_id,
+                    recruit_year=int(year),
+                    create_time=now,
+                )
+            )
+
+        # 批次
+        for batch in data.get("batches") or []:
+            session.execute(
+                insert(RecruitAnnouncementBatch).values(
+                    id=next_id(),
+                    announcement_id=announcement_id,
+                    batch_name=str(batch)[:64],
+                    create_time=now,
+                )
+            )
+
+        # 标签
+        for tag in data.get("tags") or []:
+            session.execute(
+                insert(RecruitAnnouncementTag).values(
+                    id=next_id(),
+                    announcement_id=announcement_id,
+                    tag_name=str(tag)[:64],
+                    create_time=now,
+                )
+            )
+
+        # 城市
+        for city in data.get("cities") or []:
+            session.execute(
+                insert(RecruitAnnouncementCity).values(
+                    id=next_id(),
+                    announcement_id=announcement_id,
+                    city_name=str(city)[:64],
+                    create_time=now,
+                )
+            )
+
+        # 岗位大类
+        for category in data.get("categories") or []:
+            session.execute(
+                insert(RecruitAnnouncementCategory).values(
+                    id=next_id(),
+                    announcement_id=announcement_id,
+                    category_name=str(category)[:64],
+                    create_time=now,
+                )
+            )
+
+        # 行业
+        for industry in data.get("industries") or []:
+            session.execute(
+                insert(RecruitAnnouncementIndustry).values(
+                    id=next_id(),
+                    announcement_id=announcement_id,
+                    industry_name=str(industry)[:64],
+                    create_time=now,
+                )
+            )
+
+        # 学历要求
+        for edu in data.get("educations") or []:
+            session.execute(
+                insert(RecruitAnnouncementEducation).values(
+                    id=next_id(),
+                    announcement_id=announcement_id,
+                    education_name=str(edu)[:32],
+                    create_time=now,
+                )
+            )
+
+        session.commit()
+
+
+def process_announcement(url: str) -> None:
+    """处理单条公告 URL 的完整流程。"""
+    # 1. 只处理微信公众号文章，页面提取逻辑依赖公众号页面结构
+    if _WECHAT_DOMAIN not in url:
+        log.info("非微信公众号文章，跳过: {}", url)
+        return
+
+    # 2. URL 去重
+    with MysqlSession() as session:
+        row = session.execute(
+            text("SELECT id FROM bg_recruit_announcement WHERE announcement_url = :url LIMIT 1"),
+            {"url": url},
+        )
+        if row.scalar():
+            log.info("公告已存在，跳过: {}", url)
+            return
+
+    # 3. 页面内容提取
+    result = extract_page(url)
+    if not result.content or len(result.content) < 50:
+        log.info("公告页内容过短（{}字），跳过: {}", len(result.content) if result.content else 0, url)
+        return
+
+    # 4. AI 信息提取
+    data = extract_announcement(result.content)
+    if data is None:
+        log.warning("AI 信息提取失败，跳过: {}", url)
+        return
+
+    # 5. 公司处理
+    company_name = data.get("company_name") or ""
+    company_id = find_or_create_company(company_name, result.logo_url)
+
+    # 6. 保存公告
+    announcement_id = next_id()
+    try:
+        _save_announcement(announcement_id, company_id, url, data)
+        log.info("公告入库成功 [id={}]: {} | 公司={}", announcement_id, data.get("title"), company_name)
+    except Exception as exc:
+        log.error("公告入库失败 [url={}]: {}", url, exc)

app/tool/browser.py

@@ -1,24 +1,47 @@
 """Playwright 页面抓取工具。
 
-对外仅暴露两个动作：
+对外暴露三个动作：
 - open_page(url, wait_ms): 打开页面，返回 page_id
 - query(page_id, selector): 选中节点并提取文字、图片 URL 与属性
+- close_page(page_id): 关闭页面，释放浏览器上下文
 
 实现上内部常驻一个 Browser；每次 open_page 创建独立 BrowserContext 和 Page。
-Playwright 只在一个后台线程里运行，但外层 API 保持同步。
+Playwright 用 async API 跑在一个后台事件循环线程里，多个调用方线程可以真正并发
+（页面加载彼此重叠，不会互相排队）；外层 API 仍保持同步。
 """
 
 from __future__ import annotations
 
+import asyncio
 import atexit
-import queue
 import threading
 import uuid
-from concurrent.futures import Future
+from collections.abc import Coroutine
 from dataclasses import dataclass
-from typing import Any, Callable
+from typing import Any
 
-from playwright.sync_api import Browser, BrowserContext, Page, sync_playwright
+from playwright.async_api import Browser, BrowserContext, Page, async_playwright
+
+# 页面打开超时（毫秒）
+_GOTO_TIMEOUT = 30000
+
+# Chromium 启动参数：
+#   --no-sandbox           容器内以 root 运行时 Chrome 沙箱不可用，不加会直接启动失败
+#   --disable-dev-shm-usage 容器 /dev/shm 偏小时改用磁盘，避免渲染进程崩溃
+_LAUNCH_ARGS = ["--no-sandbox", "--disable-dev-shm-usage"]
+
+# 浏览器上下文固定参数：不显式指定的话，UA 会跟着宿主系统变
+# （Windows 开发机是 Windows UA，Linux 容器是 X11 UA），目标站会按 UA 返回不同页面模板，
+# 导致同一套选择器在容器里全部落空
+_CONTEXT_OPTIONS: dict[str, Any] = {
+    "user_agent": (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+        "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
+    ),
+    "viewport": {"width": 1920, "height": 1080},
+    "locale": "zh-CN",
+    "timezone_id": "Asia/Shanghai",
+}
 
 
 @dataclass
@@ -37,114 +60,8 @@ class _PageSession:
     page: Page
 
 
-class _BrowserRuntime:
-    """在单独线程中管理 Playwright 生命周期。"""
-
-    def __init__(self) -> None:
-        self._thread: threading.Thread | None = None
-        self._ready = threading.Event()
-        self._closed = threading.Event()
-        self._job_queue: queue.Queue[
-            tuple[Callable[..., Any], tuple[Any, ...], dict[str, Any], Future[Any]]
-        ] = queue.Queue()
-        self._lock = threading.RLock()
-        self._playwright = None
-        self._browser: Browser | None = None
-        self._pages: dict[str, _PageSession] = {}
-        atexit.register(self.close)
-
-    def _thread_main(self) -> None:
-        with sync_playwright() as playwright:
-            self._playwright = playwright
-            self._browser = playwright.chromium.launch(headless=True)
-            self._ready.set()
-
-            while True:
-                job = self._job_queue.get()
-                if job is None:
-                    break
-
-                fn, args, kwargs, fut = job
-                try:
-                    result = fn(*args, **kwargs)
-                except BaseException as exc:  # pragma: no cover - propagate failures
-                    fut.set_exception(exc)
-                else:
-                    fut.set_result(result)
-
-            for session in self._pages.values():
-                try:
-                    session.page.close()
-                except Exception:
-                    pass
-                try:
-                    session.context.close()
-                except Exception:
-                    pass
-            self._pages.clear()
-
-            if self._browser is not None:
-                try:
-                    self._browser.close()
-                finally:
-                    self._browser = None
-
-            self._playwright = None
-            self._closed.set()
-
-    def _ensure_thread(self) -> None:
-        with self._lock:
-            if self._thread and self._thread.is_alive():
-                return
-
-            self._thread = threading.Thread(
-                target=self._thread_main,
-                name="offerpie-playwright",
-                daemon=True,
-            )
-            self._thread.start()
-            self._ready.wait()
-
-    def _run(self, fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
-        self._ensure_thread()
-        fut: Future[Any] = Future()
-        self._job_queue.put((fn, args, kwargs, fut))
-        return fut.result()
-
-    def close(self) -> None:
-        with self._lock:
-            if self._closed.is_set():
-                return
-            if self._thread is None:
-                return
-            self._job_queue.put(None)
-            self._closed.wait(timeout=10)
-
-    def _ensure_browser(self) -> Browser:
-        if self._browser is None:
-            raise RuntimeError("Browser runtime is not ready.")
-        return self._browser
-
-    def _open_page_impl(self, url: str, wait_ms: int) -> str:
-        browser = self._ensure_browser()
-        context = browser.new_context()
-        page = context.new_page()
-        page.goto(url, wait_until="domcontentloaded", timeout=30000)
-        if wait_ms > 0:
-            page.wait_for_timeout(wait_ms)
-
-        page_id = uuid.uuid4().hex
-        self._pages[page_id] = _PageSession(context=context, page=page)
-        return page_id
-
-    def _query_impl(self, page_id: str, selector: str) -> list[NodeSnapshot]:
-        session = self._pages.get(page_id)
-        if session is None:
-            raise KeyError(f"Unknown page_id: {page_id}")
-
-        locator = session.page.locator(selector)
-        payload = locator.evaluate_all(
-            r"""(elements) => elements.map((el) => {
+# 节点提取脚本：取标签名、可见文字、图片地址（含懒加载）与全部属性
+_EXTRACT_JS = r"""(elements) => elements.map((el) => {
     const attrs = {};
     for (const attr of Array.from(el.attributes || [])) {
         attrs[attr.name] = attr.value;
@@ -236,8 +153,103 @@ class _BrowserRuntime:
         image_urls: Array.from(new Set(imageUrls)),
         attrs,
     };
-            })"""
+})"""
+
+
+class _BrowserRuntime:
+    """在后台事件循环线程中管理 Playwright 生命周期。
+
+    所有 Playwright 调用都以协程形式提交到同一个事件循环，
+    因此多个业务线程的页面操作可以并发交叠执行。
+    """
+
+    def __init__(self) -> None:
+        self._thread: threading.Thread | None = None
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._launch_lock: asyncio.Lock | None = None
+        self._ready = threading.Event()
+        self._lock = threading.RLock()
+        self._playwright: Any = None
+        self._browser: Browser | None = None
+        self._pages: dict[str, _PageSession] = {}
+        atexit.register(self.close)
+
+    # ──────────── 事件循环线程 ────────────
+
+    def _thread_main(self) -> None:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        self._loop = loop
+        # 在循环所属线程内创建，避免多协程各自新建锁
+        self._launch_lock = asyncio.Lock()
+        self._ready.set()
+        try:
+            loop.run_forever()
+        finally:
+            loop.close()
+
+    def _ensure_loop(self) -> asyncio.AbstractEventLoop:
+        with self._lock:
+            if self._thread is None or not self._thread.is_alive():
+                self._ready.clear()
+                self._thread = threading.Thread(
+                    target=self._thread_main,
+                    name="offerpie-playwright",
+                    daemon=True,
                 )
+                self._thread.start()
+                self._ready.wait()
+
+        loop = self._loop
+        if loop is None:
+            raise RuntimeError("Browser runtime loop is not ready.")
+        return loop
+
+    def _submit(self, coro: Coroutine[Any, Any, Any]) -> Any:
+        """把协程提交到事件循环并同步等待结果。"""
+        loop = self._ensure_loop()
+        return asyncio.run_coroutine_threadsafe(coro, loop).result()
+
+    # ──────────── Playwright 操作 ────────────
+
+    async def _ensure_browser(self) -> Browser:
+        if self._browser is not None:
+            return self._browser
+
+        assert self._launch_lock is not None
+        async with self._launch_lock:
+            if self._browser is None:
+                self._playwright = await async_playwright().start()
+                self._browser = await self._playwright.chromium.launch(
+                    headless=True, args=_LAUNCH_ARGS
+                )
+        return self._browser
+
+    async def _open_page_impl(self, url: str, wait_ms: int) -> str:
+        browser = await self._ensure_browser()
+        context = await browser.new_context(**_CONTEXT_OPTIONS)
+        page = await context.new_page()
+
+        try:
+            await page.goto(url, wait_until="domcontentloaded", timeout=_GOTO_TIMEOUT)
+            if wait_ms > 0:
+                await page.wait_for_timeout(wait_ms)
+        except BaseException:
+            # 打开失败也要释放上下文，避免泄漏
+            await self._dispose(context, page)
+            raise
+
+        page_id = uuid.uuid4().hex
+        self._pages[page_id] = _PageSession(context=context, page=page)
+        return page_id
+
+    async def _query_impl(self, page_id: str, selector: str) -> list[NodeSnapshot]:
+        session = self._pages.get(page_id)
+        if session is None:
+            raise KeyError(f"Unknown page_id: {page_id}")
+
+        locator = session.page.locator(selector)
+        payload = await locator.evaluate_all(_EXTRACT_JS)
         return [
             NodeSnapshot(
                 tag=item.get("tag", ""),
@@ -248,18 +260,77 @@ class _BrowserRuntime:
             for item in payload
         ]
 
+    @staticmethod
+    async def _dispose(context: BrowserContext, page: Page | None) -> None:
+        if page is not None:
+            try:
+                await page.close()
+            except Exception:
+                pass
+        try:
+            await context.close()
+        except Exception:
+            pass
+
+    async def _close_page_impl(self, page_id: str) -> None:
+        session = self._pages.pop(page_id, None)
+        if session is None:
+            return
+        await self._dispose(session.context, session.page)
+
+    async def _shutdown_impl(self) -> None:
+        for page_id in list(self._pages):
+            await self._close_page_impl(page_id)
+
+        if self._browser is not None:
+            try:
+                await self._browser.close()
+            except Exception:
+                pass
+            self._browser = None
+
+        if self._playwright is not None:
+            try:
+                await self._playwright.stop()
+            except Exception:
+                pass
+            self._playwright = None
+
+    # ──────────── 生命周期 ────────────
+
+    def close(self) -> None:
+        """关闭全部页面与浏览器，停止事件循环。"""
+        with self._lock:
+            thread = self._thread
+            loop = self._loop
+            if thread is None or not thread.is_alive() or loop is None:
+                return
+
+            try:
+                asyncio.run_coroutine_threadsafe(
+                    self._shutdown_impl(), loop
+                ).result(timeout=15)
+            except Exception:
+                pass
+
+            loop.call_soon_threadsafe(loop.stop)
+            thread.join(timeout=10)
+            self._thread = None
+            self._loop = None
+            self._launch_lock = None
+
 
 _runtime = _BrowserRuntime()
 
 
-def open_page(url: str, wait_ms: int = 3000) -> str:
+def open_page(url: str, wait_ms: int = 8000) -> str:
     """打开页面并返回 page_id。"""
-    return _runtime._run(_runtime._open_page_impl, url, wait_ms)
+    return _runtime._submit(_runtime._open_page_impl(url, wait_ms))
 
 
 def query(page_id: str, selector: str) -> list[NodeSnapshot]:
     """按选择器抓取节点，提取文本、图片 URL 与属性。"""
-    items = _runtime._run(_runtime._query_impl, page_id, selector)
+    items = _runtime._submit(_runtime._query_impl(page_id, selector))
     seen_urls: set[str] = set()
 
     for item in items:
@@ -272,3 +343,8 @@ def query(page_id: str, selector: str) -> list[NodeSnapshot]:
         item.image_urls = unique_urls
 
     return items
+
+
+def close_page(page_id: str) -> None:
+    """关闭页面，释放对应的浏览器上下文。"""
+    _runtime._submit(_runtime._close_page_impl(page_id))

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/image_download.py

@@ -0,0 +1,36 @@
+"""图片下载工具。
+
+把图片 URL 下载成字节流，供 OCR（app.tool.ocr）与二维码识别
+（app.tool.cv / app.tool.qr_decode）直接消费，不落盘。
+"""
+
+from __future__ import annotations
+
+import httpx
+
+from app.core.logger import log
+
+# 请求超时（秒）
+_TIMEOUT = 20
+
+
+def download_image(url: str) -> bytes | None:
+    """下载图片。
+
+    Args:
+        url: 图片地址。
+
+    Returns:
+        图片字节流；下载失败时返回 None。
+    """
+    if not url or not url.startswith("http"):
+        return None
+
+    try:
+        with httpx.Client(timeout=_TIMEOUT, follow_redirects=True) as client:
+            response = client.get(url)
+            response.raise_for_status()
+            return response.content or None
+    except httpx.HTTPError as exc:
+        log.warning(f"图片下载失败: {url} | {exc}")
+        return None

app/tool/json_helper.py

@@ -0,0 +1,32 @@
+"""AI 输出 JSON 解析工具
+
+将 LLM 返回的可能带 markdown 代码块、思考标签等包裹的文本解析为 Python 对象。
+"""
+
+import re
+
+from json_repair import repair_json
+
+# 匹配 <think>任意内容</think>，用于剥离 DeepSeek R1 等推理模型的思考过程
+# re.DOTALL 让 . 匹配换行，re.IGNORECASE 忽略大小写
+# .*? 非贪婪匹配，避免跨多个 think 标签
+_THINK_RE = re.compile(r"<think>.*?</think>", re.DOTALL | re.IGNORECASE)
+
+# 匹配 ```json ... ``` 代码块，提取中间的 JSON 内容
+# (?:json\w*)? — 可选的语言标记，兼容 json / JSON / jsonc 等，也兼容无标记的裸 ```
+# \s*\n?       — 跳过语言标记后的空白和换行
+# (.*?)        — 非贪婪捕获代码块内容（第1组）
+# \n?\s*```    — 匹配结尾的 ```
+_CODE_BLOCK_RE = re.compile(r"```(?:json\w*)?\s*\n?(.*?)\n?\s*```", re.DOTALL | re.IGNORECASE)
+
+
+def parse_llm_json(text: str):
+    """解析 AI 输出的 JSON，自动去除思考标签、markdown 代码块，容错处理"""
+    # 1. 去掉 <think>...</think> 思考内容
+    cleaned = _THINK_RE.sub("", text).strip()
+    # 2. 如果有 ```json ... ``` 代码块，只取代码块里的内容
+    match = _CODE_BLOCK_RE.search(cleaned)
+    if match:
+        cleaned = match.group(1).strip()
+    # 3. repair_json 容错解析：修复不规范的 JSON（多余逗号、缺引号等）
+    return repair_json(cleaned, return_objects=True)

app/tool/page_extract.py

@@ -0,0 +1,115 @@
+"""公告页内容提取工具。
+
+流程：
+1. 打开公告页，取 `#page-content` 的正文文字与图片地址；
+2. 通过 `.wx_follow_avatar` 单独提取 Logo 地址；
+3. 正文图片下载成字节流；
+4. 每张图片 OCR 提取文字；
+5. 每张图片检测二维码，有则解出内容；
+6. 返回合并后的公告文本与 Logo 地址。
+
+图片下载、OCR、二维码识别都允许失败，单张出错只记日志并跳过。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from app.core.logger import log
+from app.tool.browser import close_page, open_page, query
+from app.tool.cv import has_qr
+from app.tool.image_download import download_image
+from app.tool.ocr import ocr
+from app.tool.qr_decode import decode_qr
+
+
+@dataclass
+class PageExtractResult:
+    """公告页提取结果。"""
+
+    content: str
+    """正文 + 图片 OCR + 二维码内容合并后的文本"""
+
+    logo_url: str | None
+    """公众号 Logo 图片地址"""
+
+# 正文容器选择器
+_SELECTOR = "#page-content"
+
+# 底部公众号信息区域中的 Logo 容器选择器
+_LOGO_SELECTOR = "#js_article_bottom_bar .wx_follow_avatar"
+
+# 打开页面后的等待时间（毫秒），等懒加载图片就位
+_WAIT_MS = 3000
+
+
+def extract_page(url: str) -> PageExtractResult:
+    """提取公告页的全部文字信息和 Logo 地址。
+
+    Args:
+        url: 公告页地址。
+
+    Returns:
+        PageExtractResult 对象，包含 content 和 logo_url。
+    """
+    if not url or not url.startswith("http"):
+        return PageExtractResult(content="", logo_url=None)
+
+    page_id: str | None = None
+    try:
+        page_id = open_page(url, _WAIT_MS)
+        nodes = query(page_id, _SELECTOR)
+        logo_nodes = query(page_id, _LOGO_SELECTOR)
+    except Exception as exc:
+        log.error(f"公告页打开失败: {url} | {exc}")
+        return PageExtractResult(content="", logo_url=None)
+    finally:
+        # 页面数据已取出，尽早释放浏览器上下文，不占着资源等后续 OCR
+        if page_id is not None:
+            close_page(page_id)
+
+    logo_url = next(
+        (
+            image_url
+            for node in logo_nodes
+            for image_url in node.image_urls
+            if image_url
+        ),
+        None,
+    )
+
+    texts: list[str] = []
+    image_urls: list[str] = []
+
+    for node in nodes:
+        if node.text:
+            texts.append(node.text)
+        image_urls.extend(node.image_urls)
+
+    for image_url in image_urls:
+        image = download_image(image_url)
+        if image is None:
+            continue
+
+        # OCR 提取图片文字
+        try:
+            image_text = ocr(image)
+        except Exception as exc:
+            log.warning(f"图片 OCR 失败: {image_url} | {exc}")
+        else:
+            if image_text:
+                texts.append(image_text)
+
+        # 有二维码则解出内容
+        try:
+            if has_qr(image):
+                texts.extend(decode_qr(image))
+        except Exception as exc:
+            log.warning(f"二维码识别失败: {image_url} | {exc}")
+
+    content = "\n".join(text for text in texts if text)
+    log.info(
+        f"公告页提取完成（图片 {len(image_urls)} 张，文本 {len(content)} 字，"
+        f"Logo {'已提取' if logo_url else '未提取'}）: {url}"
+    )
+    return PageExtractResult(content=content, logo_url=logo_url)

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)
+def _texts_of(barcodes: list) -> list[str]:
+    return _dedupe([getattr(item, "text", "") or "" for item in barcodes])
+
+
+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
-
-    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]
-
     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 []

doc/recruit_announcement.sql

@@ -1,7 +1,7 @@
 -- ============================================================
 -- 招聘公告数据表设计
 -- 说明：分析对象为「招聘公告」，多值属性（招聘届数、批次、标签组、
---       热招城市、岗位大类、学历要求）直接拆关联表存原始值，
+--       热招城市、岗位大类、行业、学历要求）直接拆关联表存原始值，
 --       不引入任何字典表，避免归一化对公告解析结果的限制。
 -- ============================================================
 
@@ -10,13 +10,14 @@
 -- 招聘公告主表
 -- ============================================================
 CREATE TABLE `bg_recruit_announcement` (
-  `id`                BIGINT        NOT NULL AUTO_INCREMENT COMMENT '主键',
+  `id`                BIGINT        NOT NULL COMMENT '主键ID(雪花)',
   `company_id`        BIGINT        NULL COMMENT '关联企业主表id',
   `company_name`      VARCHAR(128)  NOT NULL DEFAULT '' COMMENT '公司名称',
   `title`             VARCHAR(255)  NOT NULL DEFAULT '' COMMENT '公告标题',
   `company_intro`     TEXT          NULL COMMENT '公司简介',
   `target_audience`   VARCHAR(500)  NULL COMMENT '面向对象，如2027届本硕博',
   `major_require`     VARCHAR(500)  NULL COMMENT '专业要求',
+  `recruit_position`  VARCHAR(1000) NULL COMMENT '招聘岗位',
   `remark`            VARCHAR(1000) NULL COMMENT '备注',
   `written_exam`      VARCHAR(16)   NULL COMMENT '是否笔试：有 / 无 / 未明确',
   `apply_start_time`  DATETIME      NULL COMMENT '投递开始时间',
@@ -28,6 +29,8 @@ CREATE TABLE `bg_recruit_announcement` (
   `apply_email`       VARCHAR(128)  NULL COMMENT '投递邮箱',
   `source`            VARCHAR(255)  NULL COMMENT '信息来源说明',
   `publish_time`      DATETIME      NULL COMMENT '公告发布/更新时间',
+  `clean_status`      INT           NOT NULL DEFAULT 0 COMMENT '清洗状态：0-清洗中 1-清洗完成',
+  `status`            INT           NOT NULL DEFAULT 1 COMMENT '状态：0-失效 1-有效',
   `create_time`       DATETIME      NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
   `update_time`       DATETIME      NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
   PRIMARY KEY (`id`),
@@ -35,7 +38,11 @@ CREATE TABLE `bg_recruit_announcement` (
   KEY `idx_company_name` (`company_name`),
   KEY `idx_written_exam` (`written_exam`),
   KEY `idx_apply_start_time` (`apply_start_time`),
-  KEY `idx_announcement_url` (`announcement_url`)
+  KEY `idx_apply_end_time` (`apply_end_time`),
+  KEY `idx_publish_time` (`publish_time`),
+  KEY `idx_announcement_url` (`announcement_url`),
+  KEY `idx_clean_status` (`clean_status`),
+  KEY `idx_status` (`status`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='招聘公告主表';
 
 
@@ -43,7 +50,7 @@ CREATE TABLE `bg_recruit_announcement` (
 -- 招聘届数关联表
 -- ============================================================
 CREATE TABLE `bg_recruit_announcement_year` (
-  `id`               BIGINT   NOT NULL AUTO_INCREMENT COMMENT '主键',
+  `id`               BIGINT   NOT NULL COMMENT '主键ID(雪花)',
   `announcement_id`  BIGINT   NOT NULL COMMENT '公告id',
   `recruit_year`     SMALLINT NOT NULL COMMENT '招聘届数，如2027',
   `create_time`      DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
@@ -57,9 +64,9 @@ CREATE TABLE `bg_recruit_announcement_year` (
 -- 批次关联表
 -- ============================================================
 CREATE TABLE `bg_recruit_announcement_batch` (
-  `id`               BIGINT      NOT NULL AUTO_INCREMENT COMMENT '主键',
+  `id`               BIGINT      NOT NULL COMMENT '主键ID(雪花)',
   `announcement_id`  BIGINT      NOT NULL COMMENT '公告id',
-  `batch_name`       VARCHAR(64) NOT NULL COMMENT '批次值，如实习/暑期实习/寒假实习/秋招专场/春招提前批/春招补招',
+  `batch_name`       VARCHAR(64) NOT NULL COMMENT '批次值，枚举：实习/暑期实习/寒假实习/秋招提前批/秋招正式批/秋招补招/春招提前批/春招正式批/春招补招/校园招聘/社招',
   `create_time`      DATETIME    NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
   PRIMARY KEY (`id`),
   KEY `idx_announcement_id` (`announcement_id`),
@@ -71,9 +78,9 @@ CREATE TABLE `bg_recruit_announcement_batch` (
 -- 标签组关联表
 -- ============================================================
 CREATE TABLE `bg_recruit_announcement_tag` (
-  `id`               BIGINT      NOT NULL AUTO_INCREMENT COMMENT '主键',
+  `id`               BIGINT      NOT NULL COMMENT '主键ID(雪花)',
   `announcement_id`  BIGINT      NOT NULL COMMENT '公告id',
-  `tag_name`         VARCHAR(64) NOT NULL COMMENT '标签值，如秋招提前批/竞争力薪酬/校招',
+  `tag_name`         VARCHAR(64) NOT NULL COMMENT '标签值，如竞争力薪酬/六险一金/不限专业，不与批次、城市、学历重复',
   `create_time`      DATETIME    NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
   PRIMARY KEY (`id`),
   KEY `idx_announcement_id` (`announcement_id`),
@@ -85,7 +92,7 @@ CREATE TABLE `bg_recruit_announcement_tag` (
 -- 热招城市关联表
 -- ============================================================
 CREATE TABLE `bg_recruit_announcement_city` (
-  `id`               BIGINT      NOT NULL AUTO_INCREMENT COMMENT '主键',
+  `id`               BIGINT      NOT NULL COMMENT '主键ID(雪花)',
   `announcement_id`  BIGINT      NOT NULL COMMENT '公告id',
   `city_name`        VARCHAR(64) NOT NULL COMMENT '城市值，如北京市/天津市',
   `create_time`      DATETIME    NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
@@ -99,9 +106,9 @@ CREATE TABLE `bg_recruit_announcement_city` (
 -- 岗位大类关联表
 -- ============================================================
 CREATE TABLE `bg_recruit_announcement_category` (
-  `id`               BIGINT      NOT NULL AUTO_INCREMENT COMMENT '主键',
+  `id`               BIGINT      NOT NULL COMMENT '主键ID(雪花)',
   `announcement_id`  BIGINT      NOT NULL COMMENT '公告id',
-  `category_name`    VARCHAR(64) NOT NULL COMMENT '岗位大类值，如IT技术/人工智能/通信',
+  `category_name`    VARCHAR(64) NOT NULL COMMENT '岗位分类值，取自固定分类表，如后端开发/人工智能/产品经理',
   `create_time`      DATETIME    NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
   PRIMARY KEY (`id`),
   KEY `idx_announcement_id` (`announcement_id`),
@@ -109,11 +116,25 @@ CREATE TABLE `bg_recruit_announcement_category` (
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='招聘公告-岗位大类关联表';
 
 
+-- ============================================================
+-- 行业关联表
+-- ============================================================
+CREATE TABLE `bg_recruit_announcement_industry` (
+  `id`               BIGINT      NOT NULL COMMENT '主键ID(雪花)',
+  `announcement_id`  BIGINT      NOT NULL COMMENT '公告id',
+  `industry_name`    VARCHAR(64) NOT NULL COMMENT '行业分类值，取自固定分类表，如互联网/人工智能/半导体/芯片',
+  `create_time`      DATETIME    NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+  PRIMARY KEY (`id`),
+  KEY `idx_announcement_id` (`announcement_id`),
+  KEY `idx_industry_name` (`industry_name`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='招聘公告-行业关联表';
+
+
 -- ============================================================
 -- 学历要求关联表
 -- ============================================================
 CREATE TABLE `bg_recruit_announcement_education` (
-  `id`               BIGINT      NOT NULL AUTO_INCREMENT COMMENT '主键',
+  `id`               BIGINT      NOT NULL COMMENT '主键ID(雪花)',
   `announcement_id`  BIGINT      NOT NULL COMMENT '公告id',
   `education_name`   VARCHAR(32) NOT NULL COMMENT '学历值，如本科/硕士/博士',
   `create_time`      DATETIME    NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
@@ -121,3 +142,16 @@ CREATE TABLE `bg_recruit_announcement_education` (
   KEY `idx_announcement_id` (`announcement_id`),
   KEY `idx_education_name` (`education_name`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='招聘公告-学历要求关联表';
+
+
+-- ============================================================
+-- 截断数据 
+-- ============================================================
+truncate bg_recruit_announcement;
+truncate bg_recruit_announcement_batch;
+truncate bg_recruit_announcement_category;
+truncate bg_recruit_announcement_city;
+truncate bg_recruit_announcement_education;
+truncate bg_recruit_announcement_industry;
+truncate bg_recruit_announcement_tag;
+truncate bg_recruit_announcement_year;

docker-compose.yml

@@ -0,0 +1,22 @@
+services:
+  spider:
+    image: offerpie-spider:latest
+    container_name: offerpie-spider
+    restart: unless-stopped
+    network_mode: host
+    environment:
+      - ENV=prod
+      - TZ=Asia/Shanghai
+      # OCR 推理线程数，与下面的 cpus 配额保持一致
+      - OMP_NUM_THREADS=8
+    volumes:
+      - /opt/offerpie/spider/logs:/app/logs
+    # Chromium 共享内存，页面渲染走内存比落盘快
+    shm_size: 2gb
+    deploy:
+      resources:
+        limits:
+          # 宿主 32G，实测全量采集峰值约 4G，上限给足避免 OOM kill
+          memory: 8G
+          # 宿主 12 核
+          cpus: '8'

init_offerqingbaoju.py

@@ -0,0 +1,38 @@
+"""offerqingbaoju 存量数据初始化脚本（一次性执行）。
+
+与定时任务走同一套逻辑，区别仅在于 limit 给得很大，用于首次全量灌数。
+公告 URL 在 process_announcement 内部会查重，重复执行不会产生脏数据。
+
+运行（项目根目录下）：
+    python init_offerqingbaoju.py          # 用默认 limit
+    python init_offerqingbaoju.py 3000     # 指定本次抓取条数
+"""
+
+from __future__ import annotations
+
+import sys
+
+from app.core.database import close_db, init_db
+from app.core.logger import log
+from app.main import crawl
+from app.spider.offerqingbaoju import fetch_offerqingbaoju
+
+# 默认抓取条数，该接口 limit 直接映射到 per_page，实测能一次返回 6000+ 条
+DEFAULT_LIMIT = 6000
+
+
+def main() -> None:
+    """初始化数据源并跑一轮大批量采集。"""
+    limit = int(sys.argv[1]) if len(sys.argv) > 1 else DEFAULT_LIMIT
+
+    log.info("offerqingbaoju 存量初始化开始，limit={}", limit)
+    init_db()
+    try:
+        crawl("offerqingbaoju-init", fetch_offerqingbaoju, limit)
+    finally:
+        close_db()
+    log.info("offerqingbaoju 存量初始化结束")
+
+
+if __name__ == "__main__":
+    main()

init_offershow.py

@@ -0,0 +1,38 @@
+"""offershow 存量数据初始化脚本（一次性执行）。
+
+与定时任务走同一套逻辑，区别仅在于 limit 给得很大，用于首次全量灌数。
+公告 URL 在 process_announcement 内部会查重，重复执行不会产生脏数据。
+
+运行（项目根目录下）：
+    python init_offershow.py          # 用默认 limit
+    python init_offershow.py 500      # 指定本次抓取条数
+"""
+
+from __future__ import annotations
+
+import sys
+
+from app.core.database import close_db, init_db
+from app.core.logger import log
+from app.main import crawl
+from app.spider.offershow import fetch_offershow
+
+# 默认抓取条数，接口每页 20 条、翻页上限 2000 页
+DEFAULT_LIMIT = 5000
+
+
+def main() -> None:
+    """初始化数据源并跑一轮大批量采集。"""
+    limit = int(sys.argv[1]) if len(sys.argv) > 1 else DEFAULT_LIMIT
+
+    log.info("offershow 存量初始化开始，limit={}", limit)
+    init_db()
+    try:
+        crawl("offershow-init", fetch_offershow, limit)
+    finally:
+        close_db()
+    log.info("offershow 存量初始化结束")
+
+
+if __name__ == "__main__":
+    main()

init_xiaozhaoya.py

@@ -0,0 +1,54 @@
+"""读取校招鸭公告 URL 文件并批量处理（一次性执行）。
+
+运行（项目根目录下）：
+    python init_xiaozhaoya.py
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from app.core.database import close_db, init_db
+from app.core.logger import log
+from app.service.announcement_batch_service import save_announcements
+
+
+URLS_PATH = Path(__file__).resolve().parent / "doc" / "xiaozhaoya_announcement_links.json"
+
+
+def load_urls() -> list[str]:
+    """从本地 JSON 读取公告地址，过滤无效值并保持原始顺序去重。"""
+    with URLS_PATH.open("r", encoding="utf-8") as json_file:
+        payload = json.load(json_file)
+
+    if not isinstance(payload, list):
+        raise ValueError(f"公告地址 JSON 必须是数组：{URLS_PATH}")
+
+    urls: list[str] = []
+    seen: set[str] = set()
+    for value in payload:
+        if not isinstance(value, str):
+            continue
+        url = value.strip()
+        if not url.startswith(("http://", "https://")) or url in seen:
+            continue
+        seen.add(url)
+        urls.append(url)
+
+    log.info("从文件读取到 {} 条公告地址（原始 {} 条）", len(urls), len(payload))
+    return urls
+
+
+def main() -> None:
+    """读取 URL，初始化数据库并直接调用批量处理服务。"""
+    urls = load_urls()
+    init_db()
+    try:
+        save_announcements(urls)
+    finally:
+        close_db()
+
+
+if __name__ == "__main__":
+    main()

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
 
@@ -16,5 +17,10 @@ langchain-anthropic>=0.3
 langchain-core>=0.3
 
 # 工具
+json-repair>=0.61
 loguru>=0.7
+oss2==2.19.1
 snowflake-id>=1.0
+
+# 定时任务
+apscheduler>=3.10,<4