SupplyAgent/tests/api/test_video_discovery_records.py

from __future__ import annotations

import json
from collections.abc import Generator
from contextlib import contextmanager
from datetime import datetime
from decimal import Decimal

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker

from api.services import video_discovery_records as records_service
from supply_infra.db.models.video_discovery import (
    VideoDiscoveryCandidate,
    VideoDiscoveryRun,
    VideoDiscoverySearch,
)


def _patch_sessions(monkeypatch) -> sessionmaker[Session]:
    engine = create_engine("sqlite+pysqlite:///:memory:")
    VideoDiscoveryRun.__table__.create(engine)
    VideoDiscoverySearch.__table__.create(engine)
    VideoDiscoveryCandidate.__table__.create(engine)
    factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)

    @contextmanager
    def get_test_session() -> Generator[Session, None, None]:
        session = factory()
        try:
            yield session
            session.commit()
        except Exception:
            session.rollback()
            raise
        finally:
            session.close()

    monkeypatch.setattr(records_service, "get_session", get_test_session)
    return factory


def _seed(factory: sessionmaker[Session]) -> None:
    now = datetime(2026, 7, 30, 12, 30)
    with factory.begin() as session:
        session.add_all(
            [
                VideoDiscoveryRun(
                    id=1,
                    run_id="find-001",
                    biz_dt="20260730",
                    demand_grade_id=10,
                    demand_word="老年人智能手机教程",
                    seed_video_id="seed-1",
                    seed_video_title="手机使用入门",
                    relevant_points_json=json.dumps([{"point": "大字模式"}]),
                    intent_summary="寻找步骤清楚、面向老年人的手机教程。",
                    status="finished",
                    search_count=0,
                    primary_count=0,
                    valid_primary_count=1,
                    outcome_status="partial",
                    attempt_count=1,
                    create_time=now,
                    update_time=now,
                ),
                VideoDiscoveryRun(
                    id=2,
                    run_id="find-002",
                    biz_dt="20260729",
                    demand_grade_id=11,
                    demand_word="退休生活",
                    relevant_points_json="[]",
                    status="failed",
                    search_count=0,
                    primary_count=0,
                    create_time=now,
                    update_time=now,
                ),
            ]
        )
        session.add(
            VideoDiscoverySearch(
                id=101,
                run_id="find-001",
                search_key="key-1",
                keyword="老年人 手机 教程",
                query_reason="验证教程内容",
                source_type="demand",
                provider="internal_keyword",
                provider_state_json='{"cursor":"next"}',
                content_type="视频",
                sort_type="综合排序",
                publish_time="不限",
                cursor="0",
                page_no=1,
                results_count=2,
                new_candidate_count=2,
                has_more=1,
                result_ids_json='["aweme-1","aweme-2"]',
                status="success",
                create_time=now,
                update_time=now,
            )
        )
        session.add_all(
            [
                VideoDiscoveryCandidate(
                    id=1001,
                    run_id="find-001",
                    search_id=101,
                    aweme_id="aweme-1",
                    title="教爸妈设置大字体",
                    author_name="数字生活助手",
                    tags_json='["手机教程"]',
                    relevance_score=Decimal("0.91"),
                    elder_score=Decimal("0.88"),
                    share_score=Decimal("0.70"),
                    value_score=Decimal("0.82"),
                    decision_bucket="primary",
                    create_time=now,
                    update_time=now,
                ),
                VideoDiscoveryCandidate(
                    id=1002,
                    run_id="find-001",
                    search_id=101,
                    aweme_id="aweme-2",
                    title="手机发布会",
                    decision_bucket="rejected",
                    create_time=now,
                    update_time=now,
                ),
            ]
        )


def test_lists_runs_with_live_relation_counts(monkeypatch) -> None:
    factory = _patch_sessions(monkeypatch)
    _seed(factory)

    response = records_service.list_video_discovery_runs(
        biz_dt="20260730",
        keyword="智能手机",
        limit=20,
        offset=0,
    )

    assert response["total"] == 1
    run = response["items"][0]
    assert run["run_id"] == "find-001"
    assert run["search_count"] == 1
    assert run["candidate_count"] == 2
    assert run["primary_count"] == 1
    assert run["valid_primary_count"] == 1
    assert run["outcome_status"] == "partial"
    assert run["attempt_count"] == 1
    assert run["rejected_count"] == 1
    assert run["relevant_points"] == [{"point": "大字模式"}]


def test_hides_runs_before_visible_date(monkeypatch) -> None:
    factory = _patch_sessions(monkeypatch)
    _seed(factory)

    response = records_service.list_video_discovery_runs(limit=20, offset=0)

    assert response["total"] == 1
    assert [item["run_id"] for item in response["items"]] == ["find-001"]
    assert records_service.get_video_discovery_run("find-002") is None
    assert (
        records_service.list_video_discovery_searches(
            "find-002",
            limit=20,
            offset=0,
        )
        is None
    )
    assert (
        records_service.list_video_discovery_candidates(
            "find-002",
            limit=20,
            offset=0,
        )
        is None
    )


def test_lists_searches_and_candidates_with_filters(monkeypatch) -> None:
    factory = _patch_sessions(monkeypatch)
    _seed(factory)

    searches = records_service.list_video_discovery_searches(
        "find-001",
        keyword="教程",
        limit=20,
        offset=0,
    )
    assert searches is not None
    assert searches["total"] == 1
    assert searches["items"][0]["provider_state"] == {"cursor": "next"}
    assert searches["items"][0]["result_ids"] == ["aweme-1", "aweme-2"]
    assert [item["aweme_id"] for item in searches["items"][0]["candidates"]] == [
        "aweme-1",
        "aweme-2",
    ]
    assert searches["items"][0]["candidates"][0]["title"] == "教爸妈设置大字体"
    assert searches["items"][0]["candidates"][0]["comment_count"] is None
    assert searches["items"][0]["candidates"][0]["collect_count"] is None
    assert searches["items"][0]["candidates"][0]["share_count"] is None

    candidates = records_service.list_video_discovery_candidates(
        "find-001",
        bucket="primary",
        keyword="数字生活",
        limit=20,
        offset=0,
    )
    assert candidates is not None
    assert candidates["total"] == 1
    assert candidates["items"][0]["id"] == 1001
    assert candidates["items"][0]["tags"] == ["手机教程"]
    assert candidates["items"][0]["relevance_score"] == 0.91


def test_record_children_return_none_for_missing_run(monkeypatch) -> None:
    _patch_sessions(monkeypatch)

    assert records_service.get_video_discovery_run("missing") is None
    assert (
        records_service.list_video_discovery_searches(
            "missing",
            limit=20,
            offset=0,
        )
        is None
    )
    assert (
        records_service.list_video_discovery_candidates(
            "missing",
            limit=20,
            offset=0,
        )
        is None
    )