# FILE: _St/Shared/test_shared_storage.py | ROLE: Shared SQLite 저장·조회 회귀 테스트
from __future__ import annotations

import json
import re
import sqlite3
import sys
from pathlib import Path

import pytest

SHARED_DIR = Path(__file__).resolve().parent
if str(SHARED_DIR.parent) not in sys.path:
    sys.path.insert(0, str(SHARED_DIR.parent))

from Shared.storage.sort_storage import (
    DEFAULT_STORAGE_CODE,
    SCHEMA_VERSION,
    SOURCE_MODES,
    build_instagram_media_path,
    find_instagram_media_file,
    connect_database,
    canonical_instagram_post_url,
    create_download_batch,
    create_download_job,
    default_database_path,
    initialize_database,
    list_database_files,
    open_database_file,
    create_database_file,
    save_database_as,
    record_media_saved,
    record_downloaded_file,
    upsert_observation,
    list_accounts,
    list_account_posts,
    get_post_stats,
    soft_delete_account,
    soft_delete_post,
    list_deleted_accounts,
    normalize_collection_username,
    list_collection_accounts,
    replace_collection_accounts,
)

REQUIRED_TABLES = {
    "schema_migrations",
    "collector_profile",
    "device_profile",
    "storage_roots",
    "social_accounts",
    "collection_profiles",
    "account_collection_rules",
    "posts",
    "post_stat_latest",
    "post_stat_history",
    "media_assets",
    "transcripts",
    "hook_segments",
    "categories",
    "post_category_assignments",
    "tags",
    "post_tag_assignments",
    "post_notes",
    "review_templates",
    "review_items",
    "post_reviews",
    "review_scores",
    "post_comments",
    "aggregate_definitions",
    "aggregate_accounts",
    "p2p_inventory",
    "sync_queue",
    "point_ledger_cache",
    "download_batches",
    "download_jobs",
    "media_saved_events",
    "post_stat_10m",
    "deletion_events",
    "collection_account_queue",
}


def make_db(tmp_path: Path) -> tuple[Path, Path]:
    st_root = tmp_path / "bundle" / "_St"
    st_root.mkdir(parents=True)
    db_path = st_root / "Shared" / "data" / "sort_local.sqlite3"
    initialize_database(db_path, st_root)
    return st_root, db_path


def insert_post(db_path: Path, suffix: str = "A") -> tuple[int, int]:
    conn = connect_database(db_path)
    try:
        account_id = conn.execute(
            "INSERT INTO social_accounts(username) VALUES (?)",
            (f"account_{suffix}",),
        ).lastrowid
        post_id = conn.execute(
            "INSERT INTO posts(account_id, shortcode, post_url) VALUES (?, ?, ?)",
            (
                account_id,
                f"SHORT{suffix}",
                f"https://www.instagram.com/reel/SHORT{suffix}/",
            ),
        ).lastrowid
        conn.commit()
        return int(account_id), int(post_id)
    finally:
        conn.close()


def test_sqlite_initialization_and_fixed_stdown(tmp_path: Path) -> None:
    st_root, db_path = make_db(tmp_path)
    result = initialize_database(db_path, st_root)
    expected = tmp_path / "bundle" / "_StDown" / "instagram"
    assert expected.is_dir() and result["storage_root"] == str(expected.resolve())
    assert result["schema_version"] == SCHEMA_VERSION == 7
    assert result["journal_mode"] == "wal" and result["foreign_keys"] == 1
    conn = connect_database(db_path)
    try:
        tables = {
            row[0]
            for row in conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'"
            )
        }
        assert REQUIRED_TABLES.issubset(tables)
        root = conn.execute(
            "SELECT root_code, root_path, is_default FROM storage_roots"
        ).fetchone()
        assert root["root_code"] == DEFAULT_STORAGE_CODE
        assert root["root_path"] == str(expected.resolve()) and root["is_default"] == 1
        versions = {
            row[0] for row in conn.execute("SELECT version FROM schema_migrations")
        }
        assert {1, 2, 3, 4, 5, 6}.issubset(versions)
    finally:
        conn.close()



def test_migrate_existing_v1_database_to_v7(tmp_path: Path) -> None:
    st_root = tmp_path / "bundle" / "_St"
    st_root.mkdir(parents=True)
    db_path = st_root / "Shared" / "data" / "sort_local.sqlite3"
    db_path.parent.mkdir(parents=True)
    conn = sqlite3.connect(db_path)
    try:
        schema_v1 = (SHARED_DIR / "storage" / "schema_sqlite_v1.sql").read_text(encoding="utf-8")
        conn.executescript(schema_v1)
        conn.commit()
        assert conn.execute("PRAGMA user_version").fetchone()[0] == 1
    finally:
        conn.close()

    result = initialize_database(db_path, st_root)
    assert result["schema_version"] == 7
    conn = connect_database(db_path)
    try:
        tables = {
            row[0]
            for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
        }
        assert {"download_batches", "download_jobs", "media_saved_events", "post_stat_10m", "deletion_events", "collection_account_queue"}.issubset(tables)
        for table_name in ("post_stat_latest", "post_stat_history", "post_stat_10m"):
            columns = {row["name"] for row in conn.execute(f"PRAGMA table_info({table_name})")}
            assert {"sort_by", "sort_value", "source_kind"}.issubset(columns)
    finally:
        conn.close()

def test_sqlite_idempotent_and_post_unique(tmp_path: Path) -> None:
    st_root, db_path = make_db(tmp_path)
    initialize_database(db_path, st_root)
    conn = connect_database(db_path)
    try:
        assert conn.execute("SELECT COUNT(*) FROM storage_roots").fetchone()[0] == 1
        conn.execute(
            "INSERT INTO posts(platform, shortcode, post_url) VALUES('instagram','ABC123','https://www.instagram.com/p/ABC123/')"
        )
        with pytest.raises(sqlite3.IntegrityError):
            conn.execute(
                "INSERT INTO posts(platform, shortcode, post_url) VALUES('instagram','ABC123','https://www.instagram.com/reel/ABC123/')"
            )
    finally:
        conn.close()


@pytest.mark.parametrize("source_mode", sorted(SOURCE_MODES))
def test_unified_media_saved_flow_for_every_source_mode(
    tmp_path: Path, source_mode: str
) -> None:
    _, db_path = make_db(tmp_path)
    account_id, post_id = insert_post(db_path, source_mode[:3])
    batch_id = create_download_batch(
        db_path,
        source_mode=source_mode,
        account_id=account_id,
        sort_by="views" if source_mode == "TOP_SORTED" else "",
        filter_days=7 if source_mode in {"TOP_SORTED", "BULK_FILTERED"} else None,
        requested_count=1,
    )
    job_id = create_download_job(
        db_path,
        batch_id=batch_id,
        post_id=post_id,
        source_mode=source_mode,
        asset_kind="VIDEO",
        rank_no=1 if source_mode == "TOP_SORTED" else None,
        source_url="https://cdn.example/video.mp4",
        expected_file_name="video.mp4",
    )
    event_uuid = f"event-{source_mode}"
    result = record_media_saved(
        db_path,
        post_id=post_id,
        source_mode=source_mode,
        asset_kind="VIDEO",
        relative_path=f"account_{source_mode[:3]}/video.mp4",
        file_name="video.mp4",
        file_size=12345,
        sha256="a" * 64,
        source_url="https://cdn.example/video.mp4",
        download_job_id=job_id,
        event_uuid=event_uuid,
        rank_no=1 if source_mode == "TOP_SORTED" else None,
    )
    duplicate = record_media_saved(
        db_path,
        post_id=post_id,
        source_mode=source_mode,
        asset_kind="VIDEO",
        relative_path=f"account_{source_mode[:3]}/video.mp4",
        file_name="video.mp4",
        file_size=12345,
        sha256="a" * 64,
        event_uuid=event_uuid,
    )
    assert result["idempotent"] is False
    assert duplicate["idempotent"] is True and duplicate["event_id"] == result["event_id"]

    conn = connect_database(db_path)
    try:
        event = conn.execute(
            "SELECT * FROM media_saved_events WHERE id = ?", (result["event_id"],)
        ).fetchone()
        asset = conn.execute(
            "SELECT * FROM media_assets WHERE id = ?", (result["asset_id"],)
        ).fetchone()
        job = conn.execute("SELECT * FROM download_jobs WHERE id = ?", (job_id,)).fetchone()
        post = conn.execute("SELECT media_state FROM posts WHERE id = ?", (post_id,)).fetchone()
        sync = conn.execute(
            "SELECT * FROM sync_queue WHERE entity_type = 'MEDIA_SAVED'"
        ).fetchall()
        assert event["source_mode"] == source_mode
        assert event["folder_name"] == f"account_{source_mode[:3]}"
        assert asset["local_status"] == "AVAILABLE" and asset["sha256"] == "a" * 64
        assert asset["folder_name"] == f"account_{source_mode[:3]}"
        assert asset["file_name"] == "video.mp4"
        batch = conn.execute("SELECT * FROM download_batches WHERE id = ?", (batch_id,)).fetchone()
        assert job["job_status"] == "COMPLETE"
        assert batch["batch_status"] == "COMPLETE"
        assert asset["storage_root_id"] is not None
        assert post["media_state"] == "MEDIA_AVAILABLE"
        assert len(sync) == 1
        sync_payload = json.loads(sync[0]["payload_json"])
        assert sync_payload["source_mode"] == source_mode
        assert "relative_path" not in sync_payload and "local_path" not in sync_payload
        assert "folder_name" not in sync_payload
    finally:
        conn.close()


def test_media_saved_rejects_path_escape(tmp_path: Path) -> None:
    _, db_path = make_db(tmp_path)
    _, post_id = insert_post(db_path, "PATH")
    with pytest.raises(ValueError, match="storage root"):
        record_media_saved(
            db_path,
            post_id=post_id,
            source_mode="MANUAL_SINGLE",
            asset_kind="IMAGE",
            relative_path="../outside.jpg",
            file_name="outside.jpg",
            file_size=1,
        )


def test_mysql_additive_schema() -> None:
    v45 = (
        SHARED_DIR / "storage" / "mysql" / "migration_v45_from_yellow_v44.sql"
    ).read_text(encoding="utf-8")
    v46 = (
        SHARED_DIR / "storage" / "mysql" / "migration_v46_media_saved_flow.sql"
    ).read_text(encoding="utf-8")
    combined = v45 + "\n" + v46
    upper = combined.upper()
    assert "DROP TABLE" not in upper and "TRUNCATE TABLE" not in upper
    assert "ALTER TABLE `IG_MEDIA`" not in upper
    assert "ALTER TABLE `IG_MEDIA_USER_DOWNLOAD`" not in upper
    created = {
        match.group(1)
        for match in re.finditer(
            r"CREATE\s+TABLE\s+IF\s+NOT\s+EXISTS\s+`([^`]+)`", combined, re.I
        )
    }
    assert {
        "collector_members",
        "collector_devices",
        "collector_groups",
        "collector_posts",
        "collector_post_stat_latest",
        "collector_asset_inventory",
        "collector_transcripts",
        "collector_hook_segments",
        "collector_categories",
        "collector_reviews",
        "collector_comments",
        "collector_aggregates",
        "collector_tasks",
        "collector_sync_events",
        "collector_point_ledger",
        "collector_download_batches",
        "collector_download_jobs",
        "collector_media_saved_events",
    }.issubset(created)
    assert "`collector_code` varchar(8) NOT NULL" in v45
    assert "`token_hash` char(64)" in v45
    assert "local_path" not in v46.lower()
    assert "relative_path" not in v46.lower()
    assert "source_mode" in v46 and "client_event_id" in v46

def test_record_downloaded_file_creates_context_for_actual_download_root(tmp_path: Path) -> None:
    st_root = tmp_path / "_St"
    st_root.mkdir()
    media_dir = tmp_path / "BrowserDownloads" / "campaign" / "account_a"
    media_dir.mkdir(parents=True)
    media_file = media_dir / "account_a_upload_unknown_No없음_CODE123.mp4"
    media_file.write_bytes(b"video-data")

    result = record_downloaded_file(
        st_root=st_root,
        file_path=media_file,
        source_mode="BULK_ALL",
        shortcode="CODE123",
        owner_username="account_a",
        asset_kind="VIDEO",
        source_url="https://cdn.example/video.mp4",
        post_url="https://www.instagram.com/reel/CODE123/",
        md5="m" * 32,
        sha256="s" * 64,
        file_size=media_file.stat().st_size,
        account_path="account_a",
        batch_key="batch-all-1",
        requested_count=1,
        event_uuid="event-all-1",
        stats={"views": 100, "likes": 10, "comments": 2, "reposts": 1, "captured_at": "2026-07-19T00:00:00"},
    )

    db_path = default_database_path(st_root)
    with connect_database(db_path) as conn:
        event = conn.execute("SELECT * FROM media_saved_events WHERE id = ?", (result["event_id"],)).fetchone()
        post = conn.execute("SELECT * FROM posts WHERE id = ?", (result["post_id"],)).fetchone()
        batch = conn.execute("SELECT * FROM download_batches WHERE id = ?", (result["batch_id"],)).fetchone()
        job = conn.execute("SELECT * FROM download_jobs WHERE id = ?", (result["job_id"],)).fetchone()
        stat = conn.execute("SELECT * FROM post_stat_latest WHERE post_id = ?", (result["post_id"],)).fetchone()
        root = conn.execute("SELECT * FROM storage_roots WHERE id = ?", (result["storage_root_id"],)).fetchone()
        assert event["relative_path"] == "account_a/" + media_file.name
        assert Path(root["root_path"]) == media_dir.parent
        assert post["media_state"] == "MEDIA_AVAILABLE"
        assert batch["source_mode"] == "BULK_ALL"
        assert batch["batch_status"] == "COMPLETE"
        assert job["job_status"] == "COMPLETE"
        assert stat["views"] == 100
        payload = json.loads(conn.execute("SELECT payload_json FROM sync_queue WHERE entity_type = 'MEDIA_SAVED'").fetchone()[0])
        assert "relative_path" not in payload
        assert "storage_root_path" not in payload


def test_record_downloaded_file_is_idempotent_across_ws_retry(tmp_path: Path) -> None:
    st_root = tmp_path / "_St"
    st_root.mkdir()
    media_dir = tmp_path / "downloads" / "top" / "account_b"
    media_dir.mkdir(parents=True)
    media_file = media_dir / "account_b_upload_unknown_No001_CODE999.jpg"
    media_file.write_bytes(b"image-data")
    kwargs = dict(
        st_root=st_root,
        file_path=media_file,
        source_mode="TOP_SORTED",
        shortcode="CODE999",
        owner_username="account_b",
        asset_kind="IMAGE",
        rank_no=1,
        account_path="account_b",
        batch_key="top-batch-1",
        event_uuid="top-event-1",
        file_size=media_file.stat().st_size,
    )
    first = record_downloaded_file(**kwargs)
    second = record_downloaded_file(**kwargs)
    assert first["event_id"] == second["event_id"]
    assert second["idempotent"] is True
    with connect_database(default_database_path(st_root)) as conn:
        assert conn.execute("SELECT COUNT(*) FROM media_saved_events").fetchone()[0] == 1
        assert conn.execute("SELECT COUNT(*) FROM sync_queue WHERE entity_type = 'MEDIA_SAVED'").fetchone()[0] == 1


def test_download_batch_completes_only_after_requested_files(tmp_path: Path) -> None:
    st_root = tmp_path / "_St"
    st_root.mkdir()
    media_dir = tmp_path / "downloads" / "bulk" / "account_c"
    media_dir.mkdir(parents=True)
    first_file = media_dir / "account_c_upload_unknown_No없음_AAA.jpg"
    second_file = media_dir / "account_c_upload_unknown_No없음_BBB.jpg"
    first_file.write_bytes(b"one")
    second_file.write_bytes(b"two")

    first = record_downloaded_file(
        st_root=st_root,
        file_path=first_file,
        source_mode="BULK_FILTERED",
        shortcode="AAA",
        owner_username="account_c",
        asset_kind="IMAGE",
        account_path="account_c",
        batch_key="filtered-batch-two",
        requested_count=2,
        event_uuid="filtered-event-one",
        file_size=3,
    )
    db_path = default_database_path(st_root)
    with connect_database(db_path) as conn:
        status = conn.execute("SELECT batch_status FROM download_batches WHERE id = ?", (first["batch_id"],)).fetchone()[0]
        assert status == "RUNNING"

    second = record_downloaded_file(
        st_root=st_root,
        file_path=second_file,
        source_mode="BULK_FILTERED",
        shortcode="BBB",
        owner_username="account_c",
        asset_kind="IMAGE",
        account_path="account_c",
        batch_key="filtered-batch-two",
        requested_count=2,
        event_uuid="filtered-event-two",
        file_size=3,
    )
    assert second["batch_id"] == first["batch_id"]
    with connect_database(db_path) as conn:
        status = conn.execute("SELECT batch_status FROM download_batches WHERE id = ?", (first["batch_id"],)).fetchone()[0]
        assert status == "COMPLETE"


def test_observation_10m_repository_and_soft_delete(tmp_path: Path) -> None:
    _, db_path = make_db(tmp_path)
    first = upsert_observation(
        db_path, username="account_a", shortcode="POST1", views=100, likes=10, comments=2,
        observed_at="2026-07-19T14:02:37Z",
    )
    same_bucket = upsert_observation(
        db_path, username="account_a", shortcode="POST1", views=130, likes=12, comments=3,
        observed_at="2026-07-19T14:09:59Z",
    )
    next_bucket = upsert_observation(
        db_path, username="account_a", shortcode="POST1", views=190, likes=18, comments=5,
        observed_at="2026-07-19T14:12:01Z",
    )
    assert first["bucket_at"] == same_bucket["bucket_at"] == "2026-07-19T14:00:00Z"
    assert next_bucket["bucket_at"] == "2026-07-19T14:10:00Z"
    stats = get_post_stats(db_path, first["post_id"])
    assert len(stats) == 2
    assert stats[0]["views"] == 190 and stats[1]["views"] == 130
    assert stats[1]["sample_count"] == 2

    accounts = list_accounts(db_path)
    assert accounts["total"] == 1 and accounts["items"][0]["username"] == "account_a"
    posts = list_account_posts(db_path, account_id=first["account_id"], sort_by="views_delta")
    assert posts["total"] == 1
    assert posts["items"][0]["views_delta"] == 60
    assert posts["items"][0]["views_per_hour"] == 360

    deleted_post = soft_delete_post(db_path, first["post_id"])
    assert deleted_post["deleted"] is True
    assert list_account_posts(db_path, account_id=first["account_id"])["total"] == 0

    upsert_observation(
        db_path, username="account_b", shortcode="POST2", views=20,
        observed_at="2026-07-19T15:01:00Z",
    )
    account_b = next(item for item in list_accounts(db_path)["items"] if item["username"] == "account_b")
    deleted_account = soft_delete_account(db_path, account_b["id"])
    assert deleted_account["username"] == "account_b"
    ignored = upsert_observation(
        db_path, username="account_b", shortcode="POST3", views=999,
        observed_at="2026-07-19T15:12:00Z",
    )
    assert ignored["ignored_deleted"] is True and ignored["deleted_entity"] == "ACCOUNT"
    assert all(item["username"] != "account_b" for item in list_accounts(db_path)["items"])
    assert any(item["username"] == "account_b" for item in list_deleted_accounts(db_path))
    conn = connect_database(db_path)
    try:
        rule = conn.execute("SELECT active, deleted_at FROM social_accounts WHERE id=?", (account_b["id"],)).fetchone()
        assert rule["active"] == 0 and rule["deleted_at"]
        sync_ops = conn.execute("SELECT entity_type, operation FROM sync_queue WHERE operation='DELETE'").fetchall()
        assert {(row["entity_type"], row["operation"]) for row in sync_ops} >= {("POST", "DELETE"), ("ACCOUNT", "DELETE")}
    finally:
        conn.close()



def test_observation_skips_unchanged_metrics_across_new_bucket(tmp_path: Path) -> None:
    _, db_path = make_db(tmp_path)
    first = upsert_observation(
        db_path, username="unchanged_account", shortcode="UNCHANGED1",
        views=500, views_available=1, likes=50, comments=5, reposts=2,
        sort_by="views", sort_value=500, source_kind="LOCAL",
        observed_at="2026-07-25T12:02:00Z",
    )
    duplicate = upsert_observation(
        db_path, username="unchanged_account", shortcode="UNCHANGED1",
        views=500, views_available=1, likes=50, comments=5, reposts=2,
        sort_by="likes", sort_value=50, source_kind="LOCAL",
        observed_at="2026-07-25T12:12:00Z",
    )
    assert first["saved"] is True
    assert duplicate["saved"] is False and duplicate["skipped_unchanged"] is True
    assert duplicate["skipped_all_writes"] is True
    with connect_database(db_path) as conn:
        history_count = conn.execute("SELECT COUNT(*) FROM post_stat_history WHERE post_id=?", (first["post_id"],)).fetchone()[0]
        bucket_count = conn.execute("SELECT COUNT(*) FROM post_stat_10m WHERE post_id=?", (first["post_id"],)).fetchone()[0]
        latest = conn.execute("SELECT views, views_available, likes, comments, reposts FROM post_stat_latest WHERE post_id=?", (first["post_id"],)).fetchone()
        post_observed = conn.execute("SELECT last_observed_at FROM posts WHERE id=?", (first["post_id"],)).fetchone()[0]
        assert history_count == 1
        assert bucket_count == 1
        assert tuple(latest) == (500, 1, 50, 5, 2)
        assert post_observed == "2026-07-25T12:02:00Z"

    missing_views = upsert_observation(
        db_path, username="unchanged_account", shortcode="UNCHANGED1",
        views=0, views_available=0, likes=51, comments=5, reposts=2,
        sort_by="likes", sort_value=51, source_kind="LOCAL",
        observed_at="2026-07-25T12:22:00Z",
    )
    assert missing_views["saved"] is True
    with connect_database(db_path) as conn:
        latest = conn.execute("SELECT views, views_available, likes FROM post_stat_latest WHERE post_id=?", (first["post_id"],)).fetchone()
        assert tuple(latest) == (500, 1, 51)


def test_observation_v7_preserves_sort_mode_source_and_views_availability(tmp_path: Path) -> None:
    _, db_path = make_db(tmp_path)
    first = upsert_observation(
        db_path,
        username="stats_account",
        shortcode="STATS1",
        views=100,
        views_available=1,
        likes=10,
        comments=2,
        reposts=1,
        sort_by="likes",
        sort_value=10,
        source_kind="LOCAL",
        observed_at="2026-07-25T11:02:00Z",
    )
    second = upsert_observation(
        db_path,
        username="stats_account",
        shortcode="STATS1",
        views=150,
        views_available=1,
        likes=18,
        comments=4,
        reposts=2,
        sort_by="likes",
        sort_value=18,
        source_kind="SERVER",
        observed_at="2026-07-25T11:08:00Z",
    )
    assert first["bucket_at"] == second["bucket_at"] == "2026-07-25T11:00:00Z"
    with connect_database(db_path) as conn:
        latest = conn.execute("SELECT * FROM post_stat_latest WHERE post_id=?", (first["post_id"],)).fetchone()
        tenm = conn.execute("SELECT * FROM post_stat_10m WHERE post_id=?", (first["post_id"],)).fetchone()
        assert latest["views_available"] == 1
        assert latest["sort_by"] == "likes" and latest["sort_value"] == 18 and latest["source_kind"] == "SERVER"
        assert tenm["sample_count"] == 2
        assert tenm["views_available"] == 1
        assert tenm["sort_by"] == "likes" and tenm["sort_value"] == 18 and tenm["source_kind"] == "SERVER"


def test_canonical_instagram_post_url_rejects_scope_and_profile_urls() -> None:
    code = "Da0WTq-OosH"
    assert canonical_instagram_post_url(code, "https://www.instagram.com/reel/Da0WTq-OosH/?utm_source=x") == "https://www.instagram.com/reel/Da0WTq-OosH/"
    assert canonical_instagram_post_url(code, "https://www.instagram.com/reels/Da0WTq-OosH/") == "https://www.instagram.com/reel/Da0WTq-OosH/"
    assert canonical_instagram_post_url(code, "https://www.instagram.com/explore/") == "https://www.instagram.com/p/Da0WTq-OosH/"
    assert canonical_instagram_post_url(code, "https://www.instagram.com/account_name/") == "https://www.instagram.com/p/Da0WTq-OosH/"
    assert canonical_instagram_post_url(code, "https://cdn.example/video.mp4") == "https://www.instagram.com/p/Da0WTq-OosH/"
    assert canonical_instagram_post_url(code, "https://www.instagram.com/explore/", fallback_kind="reel") == "https://www.instagram.com/reel/Da0WTq-OosH/"


def test_download_and_analyze_reads_never_expose_scope_url_as_post_url(tmp_path: Path) -> None:
    st_root = tmp_path / "_St"
    st_root.mkdir()
    media_dir = tmp_path / "downloads" / "account_scope"
    media_dir.mkdir(parents=True)
    media_file = media_dir / "account_scope_upload_unknown_No없음_Da0WTq-OosH.jpg"
    media_file.write_bytes(b"image")

    result = record_downloaded_file(
        st_root=st_root,
        file_path=media_file,
        source_mode="MANUAL_SINGLE",
        shortcode="Da0WTq-OosH",
        owner_username="account_scope",
        asset_kind="IMAGE",
        post_url="https://www.instagram.com/explore/",
        account_path="account_scope",
        event_uuid="scope-url-event",
        file_size=media_file.stat().st_size,
    )
    db_path = default_database_path(st_root)
    with connect_database(db_path) as conn:
        stored = conn.execute("SELECT post_url FROM posts WHERE id=?", (result["post_id"],)).fetchone()[0]
        account_id = conn.execute("SELECT account_id FROM posts WHERE id=?", (result["post_id"],)).fetchone()[0]
        assert stored == "https://www.instagram.com/p/Da0WTq-OosH/"
        conn.execute("UPDATE posts SET post_url='https://www.instagram.com/explore/' WHERE id=?", (result["post_id"],))
        conn.commit()
    listed = list_account_posts(db_path, account_id=account_id)["items"][0]
    assert listed["post_url"] == "https://www.instagram.com/p/Da0WTq-OosH/"


def test_collection_account_queue_username_only_and_scope_rejection(tmp_path: Path) -> None:
    _, db_path = make_db(tmp_path)
    assert normalize_collection_username("@Double_King0.1") == "double_king0.1"
    assert normalize_collection_username("https://www.instagram.com/Alpha.User/") == "alpha.user"
    assert normalize_collection_username("https://www.instagram.com/Alpha.User/reels/") == "alpha.user"
    for bad in (
        "https://www.instagram.com/explore/",
        "https://www.instagram.com/reels/",
        "https://www.instagram.com/reel/CODE1/",
        "https://www.instagram.com/p/CODE1/",
    ):
        assert normalize_collection_username(bad) == ""

    saved = replace_collection_accounts(
        db_path,
        [
            {"username": "Double_King0.1", "position": 99},
            {"account_url": "https://www.instagram.com/alpha.user/"},
            {"username": "explore"},
            {"username": "double_king0.1"},
        ],
    )
    assert saved["total"] == 2
    items = list_collection_accounts(db_path)
    assert [item["username"] for item in items] == ["double_king0.1", "alpha.user"]
    assert [item["position"] for item in items] == [1, 2]
    assert items[0]["account_url"] == "https://www.instagram.com/double_king0.1/"
    with connect_database(db_path) as conn:
        sync = conn.execute(
            "SELECT operation, payload_json FROM sync_queue WHERE entity_type='COLLECTION_ACCOUNT_QUEUE' ORDER BY id DESC LIMIT 1"
        ).fetchone()
        assert sync["operation"] == "REPLACE"
        payload = json.loads(sync["payload_json"])
        assert [item["username"] for item in payload["items"]] == ["double_king0.1", "alpha.user"]


def test_duplicate_deleted_collection_entry_always_wins(tmp_path: Path) -> None:
    _, db_path = make_db(tmp_path)
    result = replace_collection_accounts(
        db_path,
        [
            {"username": "@beta", "enabled": True},
            {"account_url": "https://www.instagram.com/alpha/"},
            {"username": "beta", "is_deleted": True},
            {"username": "ALPHA"},
        ],
    )
    assert [item["username"] for item in result["items"]] == ["beta", "alpha"]
    beta = next(item for item in result["items"] if item["username"] == "beta")
    assert beta["is_deleted"] is True and beta["enabled"] is False


def test_deleted_account_stays_deleted_in_collection_queue(tmp_path: Path) -> None:
    _, db_path = make_db(tmp_path)
    result = upsert_observation(
        db_path,
        username="alpha",
        shortcode="DELETE1",
        views=1,
        observed_at="2026-07-19T09:00:00Z",
    )
    replace_collection_accounts(db_path, [{"username": "alpha"}, {"username": "beta"}])
    soft_delete_account(db_path, result["account_id"])
    items = {item["username"]: item for item in list_collection_accounts(db_path)}
    assert items["alpha"]["is_deleted"] is True and items["alpha"]["enabled"] is False
    replaced = replace_collection_accounts(
        db_path,
        [
            {"account_url": "https://www.instagram.com/alpha/", "enabled": True},
            {"username": "@beta"},
            {"username": "alpha"},
        ],
    )
    assert [item["username"] for item in replaced["items"]] == ["alpha", "beta"]
    alpha = next(item for item in replaced["items"] if item["username"] == "alpha")
    assert alpha["is_deleted"] is True and alpha["enabled"] is False


def test_stdown_original_filename_folder_is_stored_and_listed(tmp_path: Path) -> None:
    st_root = tmp_path / "bundle" / "_St"
    st_root.mkdir(parents=True)
    media_dir = tmp_path / "bundle" / "_StDown" / "instagram" / "account_original"
    media_dir.mkdir(parents=True)
    original_name = "17912345678901234_n.mp4"
    media_file = media_dir / original_name
    media_file.write_bytes(b"original-media")

    result = record_downloaded_file(
        st_root=st_root,
        file_path=media_file,
        source_mode="MANUAL_SINGLE",
        shortcode="ORIGINAL1",
        owner_username="account_original",
        asset_kind="VIDEO",
        account_path="account_original",
        event_uuid="original-file-event",
        file_size=media_file.stat().st_size,
    )
    assert result["relative_path"] == f"account_original/{original_name}"
    assert result["folder_name"] == "account_original"
    assert result["file_name"] == original_name
    listed = list_account_posts(
        default_database_path(st_root), account_id=result["account_id"]
    )["items"][0]
    assert listed["folder_name"] == "account_original"
    assert listed["file_name"] == original_name
    assert listed["relative_path"] == f"account_original/{original_name}"


def test_mysql_v47_hourly_statistics_migration_is_non_destructive() -> None:
    sql = (SHARED_DIR / "storage" / "mysql" / "migration_v47_media_stat_hour.sql").read_text(encoding="utf-8")
    upper = sql.upper()
    for forbidden in ("DROP TABLE", "TRUNCATE TABLE", "DELETE FROM"):
        assert forbidden not in upper
    for token in (
        "ADD COLUMN `views_available`",
        "ADD COLUMN `stat_hour_bucket_utc`",
        "CREATE TABLE IF NOT EXISTS `ig_media_stat_hour`",
        "UNIQUE KEY `uq_collector_shortcode_hour`",
        "`hour_bucket_utc` datetime NOT NULL",
        "`sample_count` int(10) unsigned NOT NULL DEFAULT 1",
    ):
        assert token in sql, token


def test_sqlite_v7_preserves_views_availability(tmp_path: Path) -> None:
    st_root, db_path = make_db(tmp_path)
    result = upsert_observation(
        db_path, username="private_views", shortcode="PV1", views=0, views_available=0,
        likes=120, comments=30, reposts=7, observed_at="2026-07-26T01:07:00Z",
    )
    upsert_observation(
        db_path, username="public_views", shortcode="PUB1", views=4321, views_available=1,
        likes=120, comments=30, reposts=7, observed_at="2026-07-26T01:08:00Z",
    )
    with connect_database(db_path) as conn:
        assert int(conn.execute("PRAGMA user_version").fetchone()[0]) == 7
        for table in ("post_stat_latest", "post_stat_history", "post_stat_10m"):
            assert "views_available" in {row["name"] for row in conn.execute(f"PRAGMA table_info({table})")}
        private = conn.execute("SELECT views, views_available FROM post_stat_latest WHERE post_id=?", (result["post_id"],)).fetchone()
        public = conn.execute("""SELECT s.views, s.views_available FROM post_stat_latest s JOIN posts p ON p.id=s.post_id WHERE p.shortcode='PUB1'""").fetchone()
        assert tuple(private) == (0, 0)
        assert tuple(public) == (4321, 1)
    listed = list_account_posts(db_path, account_id=result["account_id"])["items"][0]
    assert listed["views_available"] == 0
    assert get_post_stats(db_path, result["post_id"])[0]["views_available"] == 0


def test_database_filename_selection_new_open_and_save_as(tmp_path: Path) -> None:
    st_root = tmp_path / "bundle" / "_St"
    st_root.mkdir(parents=True)
    default_path = default_database_path(st_root)
    initialize_database(default_path, st_root)
    replace_collection_accounts(default_path, [{"username": "alpha"}, {"username": "beta"}])

    copied = save_database_as(st_root, default_path, "archive")
    assert copied["filename"] == "archive.sqlite3"
    assert default_database_path(st_root).name == "archive.sqlite3"
    assert [item["username"] for item in list_collection_accounts(default_database_path(st_root))] == ["alpha", "beta"]

    created = create_database_file(st_root, "fresh.sqlite3")
    assert created["filename"] == "fresh.sqlite3"
    assert default_database_path(st_root).name == "fresh.sqlite3"
    assert list_collection_accounts(default_database_path(st_root)) == []

    opened = open_database_file(st_root, "archive.sqlite3")
    assert opened["filename"] == "archive.sqlite3"
    assert [item["username"] for item in list_collection_accounts(default_database_path(st_root))] == ["alpha", "beta"]
    info = list_database_files(st_root)
    assert info["active_filename"] == "archive.sqlite3"
    assert {"sort_local.sqlite3", "archive.sqlite3", "fresh.sqlite3"}.issubset(set(info["files"]))

    with pytest.raises(ValueError):
        create_database_file(st_root, "archive.sqlite3")
    with pytest.raises(ValueError):
        open_database_file(st_root, "../outside.sqlite3")
    with pytest.raises(ValueError):
        create_database_file(st_root, "../new-outside.sqlite3")


def test_shortcode_media_path_contract_uses_only_shortcode_actual_extension(tmp_path: Path) -> None:
    st_root, _ = make_db(tmp_path)
    target = build_instagram_media_path(st_root, "ameliauusa", "ABC123", ".webp")
    assert target == tmp_path / "bundle" / "_StDown" / "instagram" / "ameliauusa" / "ABC123.webp"
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_bytes(b"webp")
    assert find_instagram_media_file(st_root, "ameliauusa", "ABC123") == target
    target.unlink()
    legacy = target.with_name("ABC123_thumbnail.jpg")
    legacy.write_bytes(b"jpg")
    assert find_instagram_media_file(st_root, "ameliauusa", "ABC123") is None


def test_cache_and_auto_seen_source_modes_are_supported() -> None:
    assert {"CACHE8701_CDP", "AUTO_SEEN_JPG"}.issubset(SOURCE_MODES)
