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, 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["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 )