Blame for backend/tests/test_analytics.py

"""Test theo dõi người dùng: ghi log thô -> suy ra 9 chỉ số (A1-A5, B1-B4).

Cô lập bằng cách trỏ STORAGE_DIR vào thư mục tạm TRƯỚC khi import storage, và
ép DATABASE_URL rỗng để luôn chạy nhánh SQLite (không đụng Postgres thật).
"""
from __future__ import annotations

import os
import tempfile
import time
import uuid
from datetime import datetime, timedelta, timezone

_TEST_STORAGE = tempfile.mkdtemp(prefix="tnt_an_")
os.environ["STORAGE_DIR"] = _TEST_STORAGE

from app.config import settings           # noqa: E402

settings.STORAGE_DIR = _TEST_STORAGE      # .env của máy load override -> ép lại
settings.DATABASE_URL = ""                # ép SQLite
os.makedirs(_TEST_STORAGE, exist_ok=True)

from app.services.analytics import etl, storage as store   # noqa: E402

TZ = timezone(timedelta(hours=7))
T0 = time.time() - 3600


def _ev(off, name, step, *, feature="dub", task="task_1", sess="sess_1",
        user="an", lat=None, value="", **props):
    """Một dòng log đã chuẩn hoá, đặt tại mốc T0+off giây."""
    ts = T0 + off
    row = store.normalize({
        "event_id": uuid.uuid4().hex,
        "event_time": datetime.fromtimestamp(ts, TZ).isoformat(timespec="milliseconds"),
        "user_id": user, "session_id": sess, "task_id": task, "feature": feature,
        "step": step, "event_name": name, "event_value": value,
        "latency_ms": lat, "properties": props,
    }, ts)
    row["event_ts"] = ts                  # giữ nguyên trục thời gian giả lập
    # storage.normalize() trả dạng GHI (properties_json). Khi đọc lại từ DB,
    # _row_to_dict() bung thành `properties` dict — ETL dùng dạng đó. Cho dòng
    # mang cả hai để test được cả nhánh ghi lẫn nhánh tính chỉ số.
    row["properties"] = dict(props)
    return row


def _one_full_task():
    """Một task dub hoàn chỉnh: 2 lần máy, người sửa ở giữa, tải về, 👍."""
    return [
        _ev(0, "session_start", "open", feature="", task=""),
        _ev(2, "feature_open", "open"),
        _ev(6, "upload_done", "prepare"),
        _ev(9, "job_queued", "run", job_id="j1"),
        _ev(11, "job_started", "run", job_id="j1"),
        _ev(41, "job_done", "run", lat=30000, job_id="j1", phase="asr_translate"),
        _ev(120, "edit_translation", "review"),
        _ev(200, "job_queued", "run", job_id="j2"),
        _ev(202, "job_started", "run", job_id="j2"),
        _ev(260, "job_done", "run", lat=59000, job_id="j2", phase="tts"),
        _ev(275, "output_download", "output", value="mp4", count=2,
            success_type="download"),
        _ev(277, "thumbs_feedback", "output", value="up"),
        _ev(280, "feature_complete", "done", success_type="download"),
    ]


# ── Chuẩn hoá dòng log ───────────────────────────────────────────

def test_normalize_drops_zero_latency():
    """latency_ms chỉ dành cho dòng MÁY chạy: 0/rỗng phải thành None.

    Nhét 0 vào là kéo tụt trung bình tốc độ máy (spec §2 "quy tắc vàng").
    """
    assert store.normalize({"event_name": "x", "user_id": "a", "latency_ms": 0})["latency_ms"] is None
    assert store.normalize({"event_name": "x", "user_id": "a", "latency_ms": ""})["latency_ms"] is None
    assert store.normalize({"event_name": "x", "user_id": "a", "latency_ms": -5})["latency_ms"] is None
    assert store.normalize({"event_name": "x", "user_id": "a", "latency_ms": 1200})["latency_ms"] == 1200


def test_normalize_parses_time_and_props():
    r = store.normalize({
        "event_name": "feature_open", "user_id": "  AN  ",
        "event_time": "2026-08-24T09:25:09.300+07:00",
        "properties": {"entry": "sidebar"},
    })
    assert r["user_id"] == "an"                       # chuẩn hoá để gom đúng người
    assert abs(r["event_ts"] - 1787538309.3) < 1.0     # 09:25:09.300 +07:00
    assert '"entry"' in r["properties_json"]
    # thời gian sai định dạng -> lùi về giờ máy chủ, không làm vỡ luồng
    bad = store.normalize({"event_name": "x", "user_id": "a", "event_time": "hôm qua"})
    assert bad["event_ts"] > 0


# ── Kho lưu ──────────────────────────────────────────────────────

def test_insert_query_and_dedup():
    rows = _one_full_task()
    n = store.insert(rows)
    assert n == len(rows)
    assert store.insert(rows) == 0                    # cùng event_id -> không nhân đôi
    got = store.query(user_id="an")
    assert len(got) >= len(rows)
    assert all(g["user_id"] == "an" for g in got)
    # dòng thiếu user_id thì bỏ, không rác vào bảng
    assert store.insert([store.normalize({"event_name": "x"})]) == 0


def test_query_newest_first_respects_limit():
    """Cắt `limit` mà xếp tăng dần sẽ lấy nhầm N dòng CŨ NHẤT — màn log thô cần MỚI nhất."""
    old = store.query(limit=3)
    new = store.query(limit=3, newest_first=True)
    assert len(old) == len(new) == 3
    assert new[0]["server_ts"] >= old[-1]["server_ts"]
    assert [r["server_ts"] for r in new] == sorted([r["server_ts"] for r in new], reverse=True)


def test_purge_old_keeps_recent():
    before = store.stats()["rows"]
    assert store.purge_old(days=3650) == 0            # chưa có gì cũ tới vậy
    assert store.stats()["rows"] == before


# ── ETL: dựng task & chỉ số ──────────────────────────────────────

def test_task_metrics_from_one_session():
    evs = _one_full_task()
    tasks = etl.build_tasks(evs)
    assert len(tasks) == 1
    t = tasks[0]

    assert t["has_output"] and t["success_type"] == "download"
    # A1 = từ lúc mở tool tới ĐẦU RA đầu tiên (275 - 2 giây)
    assert abs(t["duration_ms"] - 273000) < 50
    # B4 = tổng thời gian máy chạy thật, KHÔNG gồm chờ hàng đợi
    assert t["machine_ms"] == 89000
    assert abs(t["wait_ms"] - 4000) < 50               # (11-9) + (202-200)
    assert t["attempts"] == 2                          # A4: hai lần job_done
    assert t["produced"] == 2                          # B1: count=2
    assert t["thumbs_up"] == 1 and t["thumbs_down"] == 0


def test_step_durations_stop_at_done():
    """Bấm 👍 sau khi xong mang step=output; dòng thời gian không được hiện thêm bước sau 'done'."""
    steps = [d["step"] for d in etl.step_durations(etl.build_tasks(_one_full_task())[0])]
    assert steps[-1] == "done"
    assert steps.count("done") == 1


def test_abandon_and_completion_rate():
    """A3 = task có đầu ra / tổng task; bỏ dở ghi rõ bước cuối chạm tới."""
    evs = _one_full_task() + [
        _ev(300, "feature_open", "open", feature="shuffle", task="task_2"),
        _ev(340, "upload_done", "prepare", feature="shuffle", task="task_2"),
        _ev(900, "feature_abandon", "prepare", feature="shuffle", task="task_2",
            last_step="prepare", reason="qua_nhieu_clip"),
    ]
    s = etl.summarize(evs)
    assert s["overall"]["tasks"] == 2
    assert s["overall"]["tasks_done"] == 1
    assert abs(s["overall"]["completion_rate"] - 0.5) < 1e-9
    sh = next(f for f in s["features"] if f["feature"] == "shuffle")
    assert sh["abandon_by_step"] == {"prepare": 1}
    assert sh["label"] == "Xào video"


def test_forward_counts_as_success():
    """Chuyển tiếp sang tool khác cũng là ĐẦU RA hợp lệ (spec §6.2)."""
    evs = [
        _ev(0, "feature_open", "open", feature="down_list", task="t_fw"),
        _ev(5, "output_forward", "output", feature="down_list", task="t_fw",
            value="cut", success_type="forward", to="cut", count=3),
    ]
    t = etl.build_tasks(evs)[0]
    assert t["has_output"] and t["success_type"] == "forward"
    prods = etl.summarize(evs)["products"]
    assert prods[0]["kind"].startswith("→")            # đọc ra "chuyển sang", không lẫn đuôi file
    assert prods[0]["count"] == 3


def test_job_speed_split_by_phase():
    """B2 phải tách tốc độ theo TỪNG BƯỚC máy, không gộp một cục."""
    jobs = etl.summarize(_one_full_task())["jobs"]
    phases = {j["phase"]: j["median_ms"] for j in jobs}
    assert phases["asr_translate"] == 30000
    assert phases["tts"] == 59000


def test_contribution_score_ranks_tools():
    """Điểm đóng góp = 40% nhu cầu + 20% ít bấm chạy lại + 40% sản phẩm đầu ra.

    Tool A: 2 việc, 4 sản phẩm, mỗi việc bấm 1 lần  -> phải đứng trên
    Tool B: 1 việc, 1 sản phẩm, bấm 2 lần/việc.
    """
    evs = []
    for i in (1, 2):                       # cut: 2 việc, mỗi việc 1 lần bấm, 2 sản phẩm
        evs += [
            _ev(i * 10, "feature_open", "open", feature="cut", task=f"c{i}"),
            _ev(i * 10 + 2, "job_done", "run", feature="cut", task=f"c{i}", lat=1000),
            _ev(i * 10 + 4, "output_download", "output", feature="cut", task=f"c{i}",
                value="mp4", count=2),
        ]
    evs += [                               # shuffle: 1 việc, 2 lần bấm, 1 sản phẩm
        _ev(100, "feature_open", "open", feature="shuffle", task="s1"),
        _ev(102, "job_done", "run", feature="shuffle", task="s1", lat=1000),
        _ev(104, "regenerate", "run", feature="shuffle", task="s1"),
        _ev(106, "output_download", "output", feature="shuffle", task="s1",
            value="mp4", count=1),
    ]
    feats = {f["feature"]: f for f in etl.summarize(evs)["features"]}
    cut, sh = feats["cut"], feats["shuffle"]
    # cut cao nhất ở cả 3 thành phần -> tròn 100 điểm
    assert cut["score_demand"] == 100.0 and cut["score_output"] == 100.0
    assert cut["score_rerun"] == 100.0 and cut["score_total"] == 100.0
    # shuffle: nhu cầu 1/2, sản phẩm 1/4, bấm gấp đôi -> 0.4*50 + 0.2*50 + 0.4*25
    assert sh["score_demand"] == 50.0
    assert sh["score_output"] == 25.0
    assert sh["score_rerun"] == 50.0
    assert abs(sh["score_total"] - 40.0) < 0.05
    assert cut["score_total"] > sh["score_total"]


def test_contribution_rerun_zero_when_never_ran():
    """Tool mở ra rồi bỏ (chưa bấm chạy lần nào) KHÔNG được 100đ "ít chạy lại".

    Kẹp sàn 1.0 như bản đầu làm cả bảng cùng 100 -> thành phần 20% mất tác dụng.
    Chưa chạy = chưa đóng góp -> 0 điểm.
    """
    evs = [
        _ev(0, "feature_open", "open", feature="cut", task="c1"),
        _ev(2, "job_done", "run", feature="cut", task="c1", lat=1000),
        _ev(4, "output_download", "output", feature="cut", task="c1", value="mp4", count=1),
        _ev(50, "feature_open", "open", feature="text_video", task="v1"),   # mở rồi bỏ
    ]
    feats = {f["feature"]: f for f in etl.summarize(evs)["features"]}
    assert feats["cut"]["score_rerun"] == 100.0
    assert feats["text_video"]["attempts_avg"] == 0
    assert feats["text_video"]["score_rerun"] == 0.0
    assert feats["cut"]["score_total"] > feats["text_video"]["score_total"]


def test_contribution_score_handles_single_tool():
    """Chỉ có 1 tool -> không vỡ, tự là tool cao nhất (100 điểm)."""
    f = etl.summarize(_one_full_task())["features"]
    assert len(f) == 1 and f[0]["score_total"] == 100.0


def test_passive_pages_are_not_tasks():
    """page_view (trang chủ/hướng dẫn) không có task_id -> không tính vào A3."""
    evs = _one_full_task() + [
        _ev(400, "page_view", "open", feature="home", task=""),
        _ev(410, "page_view", "open", feature="guide", task=""),
    ]
    s = etl.summarize(evs)
    assert s["overall"]["tasks"] == 1
    assert all(f["feature"] not in ("home", "guide") for f in s["features"])


# ── ETL: phiên làm việc ──────────────────────────────────────────

def test_session_splits_on_long_silence():
    """Im lặng quá ngưỡng -> tách thành hai phiên (spec §6: session_end là mốc mềm)."""
    gap = (settings.SESSION_IDLE_MINUTES + 10) * 60
    evs = [
        _ev(0, "session_start", "open", feature="", task="", sess="s"),
        _ev(60, "feature_open", "open", task="t1", sess="s"),
        _ev(60 + gap, "feature_open", "open", task="t2", sess="s"),
        _ev(120 + gap, "upload_done", "prepare", task="t2", sess="s"),
    ]
    sessions = etl.build_sessions(evs)
    assert len(sessions) == 2
    # thời lượng DÙNG THẬT không được cộng cả quãng im lặng
    assert sum(s["active_ms"] for s in sessions) < gap * 1000
    assert abs(sessions[0]["active_ms"] - 60000) < 50


def test_returning_user_needs_two_days():
    """A5 = một user xuất hiện ở >= 2 ngày khác nhau."""
    day = 86400
    evs = [
        _ev(0, "feature_open", "open", task="t1", user="an"),
        _ev(-2 * day, "feature_open", "open", task="t0", user="an", sess="s0"),
        _ev(0, "feature_open", "open", task="t2", user="binh", sess="sb"),
    ]
    users = {u["user_id"]: u for u in etl.summarize(evs)["users"]}
    assert users["an"]["returning"] is True
    assert users["binh"]["returning"] is False


def test_user_detail_lists_sessions_and_tasks():
    d = etl.user_detail(_one_full_task(), "an")
    assert d["user_id"] == "an" and len(d["sessions"]) == 1
    task = d["sessions"][0]["tasks"][0]
    assert task["label"] == "Dịch / đổi giọng"
    assert task["has_output"] and task["machine_ms"] == 89000
    assert [j["latency_ms"] for j in task["jobs"]] == [30000, 59000]


# ── Dịch log sang tiếng Việt + xuất CSV ──────────────────────────

def test_labels_translate_every_column():
    """Mỗi cột log thô phải ra được một câu tiếng Việt, không lọt khoá máy."""
    from app.services.analytics import labels as lb
    assert lb.feature_label("dub") == "Dịch / đổi giọng"
    assert lb.step_label("run") == "3. Máy chạy"
    assert lb.event_label("feature_abandon") == "Bỏ dở"
    assert lb.value_label("up").startswith("Hài lòng")
    assert lb.value_label("mp4") == "Video MP4"
    assert lb.duration_label(53043) == "53,0 giây"
    assert lb.duration_label(None) == ""       # thao tác NGƯỜI -> để trống, không in "0"


def test_labels_keep_unknown_keys():
    """Tool/thuộc tính mới thêm mà quên cập nhật bảng thì giữ nguyên khoá gốc,
    tuyệt đối không được nuốt mất dữ liệu."""
    from app.services.analytics import labels as lb
    assert lb.feature_label("tool_moi") == "tool_moi"
    assert lb.event_label("su_kien_la") == "su_kien_la"
    assert "thuoc_tinh_la: 7" in lb.describe_props({"thuoc_tinh_la": 7})


def test_describe_props_unpacks_json_blob():
    from app.services.analytics import labels as lb
    txt = lb.describe_props({"job_id": "f09c", "kind": "download", "attempt": 1,
                             "phase": "asr_translate", "dur_ms": 63290})
    # thuộc tính quan trọng lên trước, mili-giây in ra kiểu người đọc
    assert txt.startswith("Lần bấm chạy thứ: 1")
    assert "Bước máy: Bóc lời & dịch" in txt
    assert "Loại job: Tải video" in txt
    assert "Kéo dài: 1 phút 3 giây" in txt


def test_export_csv_is_readable_and_raw_keeps_schema():
    """CSV mặc định phải đọc được; `?raw=1` giữ nguyên 11 cột schema v2."""
    import csv as _csv
    import io as _io
    from app.api import analytics as api

    store.insert(_one_full_task())
    api.settings.VIDEO_ADMIN_TOKEN = "tok"

    def _read(**kw):
        # StreamingResponse bọc iterator đồng bộ thành async generator -> phải
        # rút bằng vòng lặp asyncio chứ không list() thẳng được.
        import asyncio
        resp = api.admin_export_csv(token="tok", days=2, **kw)

        async def drain():
            return [c async for c in resp.body_iterator]

        text = "".join(c.decode("utf-8") if isinstance(c, bytes) else c
                       for c in asyncio.run(drain()))
        return list(_csv.reader(_io.StringIO(text.lstrip(""))))

    rows = _read()
    head = rows[0]
    assert head[:8] == ["STT", "Ngày", "Giờ", "Người dùng", "Mã người dùng",
                        "Tool", "Bước", "Hành động"]
    body = [r for r in rows[1:] if r]
    assert any(r[5] == "Dịch / đổi giọng" and r[7] == "Job chạy xong"
               and r[9] == "30,0 giây" for r in body)
    assert any(r[7] == "Đánh giá kết quả" and r[8].startswith("Hài lòng") for r in body)

    raw = _read(raw=1)
    assert raw[0] == ["event_id", "event_time", "user_id", "session_id", "task_id",
                      "feature", "step", "event_name", "event_value", "latency_ms",
                      "properties_json"]


# ── Chỉ số mức độ bám app (nhóm "Tình trạng người dùng") ─────────

def test_adoption_counts_ignore_lookers():
    """`used_users` chỉ đếm người có thao tác THẬT; hàng ngày/hàng tuần tính
    theo mốc mới nhất TRONG KHOẢNG LỌC, không phải theo giờ chạy báo cáo."""
    day = 86400
    evs = [
        _ev(0, "feature_open", "open", user="an", task="t1"),        # làm thật
        _ev(-3 * day, "run_click", "run", user="binh", task="t2", sess="sb"),
        _ev(-10 * day, "page_view", "open", user="cuong", task="", sess="sc"),
    ]
    o = etl.summarize(evs)["overall"]
    assert o["used_users"] == 2                # cuong chỉ ngó trang -> không tính
    assert o["daily_users"] == 1               # trong 24h cuối: an
    assert o["weekly_users"] == 2              # trong 7 ngày cuối: an + binh
    assert o["session_max_ms"] >= 0


def test_job_rows_carry_attempt_and_vietnamese_phase():
    evs = _one_full_task()
    jobs = {j["phase"]: j for j in etl.summarize(evs)["jobs"]}
    assert jobs["tts"]["phase_label"] == "Đọc thành tiếng"
    assert jobs["asr_translate"]["runs"] == 1
    assert jobs["asr_translate"]["attempt_avg"] == 1.0