campus_spider/app/service/recruit_announcement_service.py

"""招聘公告业务服务：编排公告爬取、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_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


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 _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"),
                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 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. 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

    # 2. 页面内容提取
    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

    # 3. AI 信息提取
    data = extract_announcement(result.content)
    if data is None:
        log.warning("AI 信息提取失败，跳过: {}", url)
        return

    # 4. 公司处理
    company_name = data.get("company_name") or ""
    company_id = find_or_create_company(company_name, result.logo_url)

    # 5. 保存公告
    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)