from datetime import date, datetime, timedelta, timezone from fastapi.testclient import TestClient from app.core.config import get_settings from app.db.session import init_db, reset_db_for_tests from app.main import app from app.models.core import AnalyticsPostSnapshot, ExternalPost from app.services.analytics_youtube_csv import ( YOUTUBE_CHART, YOUTUBE_DAILY_TOTALS, YOUTUBE_VIDEO_TABLE, detect_csv_type, parse_duration_seconds, parse_number, parse_percent, parse_youtube_chart_csv, parse_youtube_daily_totals_csv, parse_youtube_date, parse_youtube_video_table_csv, read_csv, ) def daily_csv() -> bytes: rows = ["Datum,Aufrufe"] values = [10] * 27 + [59] start = date(2026, 5, 8) for offset, views in enumerate(values): rows.append(f"{(start + timedelta(days=offset)).isoformat()},{views}") return "\n".join(rows).encode("utf-8-sig") def video_table_csv(views_offset: int = 0) -> bytes: return f'''Videos,Videotitel,Veröffentlichungszeitpunkt des Videos,Dauer,Aufrufe,Wiedergabezeit (Stunden),Abonnenten,Impressionen,Klickrate der Impressionen (%) Gesamt,Gesamt,,,329,0.31,-1,439,1.59 abc123,"Direct scam-warning hook, with comma","May 31, 2026",0:31,{120 + views_offset},0.12,2,180,3.20 xyz456,"Slow opening 😬","May 30, 2026",0:38,{80 + views_offset},0.03,-1,200,0.50 partial789,"Impressions only row","May 29, 2026",0:27,,,0,59,1.20 '''.encode("utf-8-sig") def chart_csv() -> bytes: return "Datum,Videos,Videotitel,Veröffentlichungszeitpunkt des Videos,Dauer,Aufrufe\n".encode("utf-8-sig") def make_client(tmp_path): settings = get_settings() settings.database_url = "sqlite://" settings.storage_root = tmp_path / "storage" engine = init_db(settings) reset_db_for_tests(engine) return TestClient(app) def test_detect_youtube_csv_types(): assert detect_csv_type(read_csv(daily_csv())[0]) == YOUTUBE_DAILY_TOTALS assert detect_csv_type(read_csv(video_table_csv())[0]) == YOUTUBE_VIDEO_TABLE assert detect_csv_type(read_csv(chart_csv())[0]) == YOUTUBE_CHART def test_parse_helpers_for_youtube_exports(): assert parse_youtube_date("May 31, 2026").isoformat() == "2026-05-31" assert parse_youtube_date("2026-05-08").isoformat() == "2026-05-08" assert parse_number("0.31") == 0.31 assert parse_percent("1.59") == 1.59 assert parse_duration_seconds("1:02") == 62 assert parse_duration_seconds("0:01:02") == 62 def test_parse_daily_totals_and_header_only_chart(): daily = parse_youtube_daily_totals_csv(daily_csv()) assert daily.detected_type == YOUTUBE_DAILY_TOTALS assert daily.row_count == 28 assert sum(row["views"] for row in daily.rows) == 329 assert daily.date_range_start.isoformat() == "2026-05-08" chart = parse_youtube_chart_csv(chart_csv()) assert chart.detected_type == YOUTUBE_CHART assert chart.row_count == 0 assert "Header-only" in chart.warnings[0] def test_parse_video_table_total_and_partial_rows(): parsed = parse_youtube_video_table_csv(video_table_csv()) assert parsed.detected_type == YOUTUBE_VIDEO_TABLE assert parsed.total_row["views"] == 329 assert parsed.total_row["watch_time_hours"] == 0.31 assert parsed.total_row["subscribers_delta"] == -1 assert parsed.total_row["impressions"] == 439 assert parsed.total_row["impression_ctr_pct"] == 1.59 assert len(parsed.rows) == 3 assert any(row["external_post_id"] == "partial789" and row["views"] is None for row in parsed.rows) assert any("partial snapshot without views" in warning for warning in parsed.warnings) def test_youtube_csv_import_creates_normalized_records_and_insights(tmp_path): client = make_client(tmp_path) response = client.post( "/api/analytics/import/youtube-csv", data={"provider": "youtube", "snapshot_at": "2026-06-06T10:00:00+00:00"}, files=[ ("files", ("Gesamtwerte.csv", daily_csv(), "text/csv")), ("files", ("Tabellendaten.csv", video_table_csv(), "text/csv")), ("files", ("Diagrammdaten.csv", chart_csv(), "text/csv")), ], ) assert response.status_code == 200, response.text payload = response.json() assert payload["created_daily_metrics"] == 28 assert payload["created_snapshots"] == 3 assert payload["created_posts"] == 3 assert any("total row views=329" in warning for warning in payload["warnings"]) overview = client.get("/api/analytics/overview").json() assert overview["kpis"]["views_last_28_days"] == 329 assert overview["kpis"]["total_views_latest"] == 200 assert len(overview["daily_views"]) == 28 assert overview["insights"] videos = client.get("/api/analytics/videos").json()["items"] assert videos[0]["views"] == 120 assert {item["external_post_id"] for item in videos} == {"abc123", "xyz456", "partial789"} def test_duplicate_file_hash_is_skipped_and_second_changed_import_creates_deltas(tmp_path): client = make_client(tmp_path) first = client.post( "/api/analytics/import/youtube-csv", data={"provider": "youtube", "snapshot_at": "2026-06-06T10:00:00+00:00"}, files=[("files", ("Tabellendaten.csv", video_table_csv(), "text/csv"))], ) assert first.status_code == 200 dup = client.post( "/api/analytics/import/youtube-csv", data={"provider": "youtube", "snapshot_at": "2026-06-06T11:00:00+00:00"}, files=[("files", ("Tabellendaten.csv", video_table_csv(), "text/csv"))], ) assert dup.status_code == 200 assert dup.json()["created_snapshots"] == 0 assert any("duplicate file hash skipped" in warning for warning in dup.json()["warnings"]) second = client.post( "/api/analytics/import/youtube-csv", data={"provider": "youtube", "snapshot_at": "2026-06-07T10:00:00+00:00"}, files=[("files", ("Tabellendaten.csv", video_table_csv(views_offset=20), "text/csv"))], ) assert second.status_code == 200 assert second.json()["created_snapshots"] == 3 detail = client.get("/api/analytics/videos/abc123").json() assert len(detail["snapshots"]) == 2 latest = detail["snapshots"][-1] assert latest["views_delta"] == 20 assert latest["trend"] == "rising"