from __future__ import annotations

import argparse
import hashlib
import json
import re
import sqlite3
import urllib.parse
import uuid
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from typing import Any, Mapping

SCHEMA_VERSION = 7
SCHEMA_FILES = (
    Path(__file__).with_name("schema_sqlite_v1.sql"),
    Path(__file__).with_name("schema_sqlite_v2.sql"),
    Path(__file__).with_name("schema_sqlite_v3.sql"),
    Path(__file__).with_name("schema_sqlite_v4.sql"),
    Path(__file__).with_name("schema_sqlite_v5.sql"),
    Path(__file__).with_name("schema_sqlite_v6.sql"),
    Path(__file__).with_name("schema_sqlite_v7.sql"),
)
DEFAULT_DB_FILENAME = "sort_local.sqlite3"
DEFAULT_DB_RELATIVE = Path("Shared") / "data" / DEFAULT_DB_FILENAME
DATABASE_SELECTION_RELATIVE = Path("Shared") / "data" / "sort_database_selection.json"
_DATABASE_FILENAME_RE = re.compile(r"^[0-9A-Za-z가-힣._-]+$")
DEFAULT_STORAGE_CODE = "STDOWN_INSTAGRAM"
SOURCE_MODES = frozenset(
    {
        "MANUAL_SINGLE",
        "BULK_ALL",
        "BULK_FILTERED",
        "TOP_SORTED",
        "TRANSCRIPT_REQUEST",
    }
)


def derive_stdown_root(st_root: str | Path) -> Path:
    """Return the fixed sibling _StDown directory for the supplied _St root."""
    root = Path(st_root).expanduser().resolve()
    if root.name != "_St":
        raise ValueError(f"st_root must point to the _St directory: {root}")
    return root.parent / "_StDown"


def database_directory(st_root: str | Path) -> Path:
    return Path(st_root).expanduser().resolve() / "Shared" / "data"


def database_selection_path(st_root: str | Path) -> Path:
    return Path(st_root).expanduser().resolve() / DATABASE_SELECTION_RELATIVE


def normalize_database_filename(value: object) -> str:
    raw = str(value or "").strip()
    if not raw:
        raise ValueError("database filename required")
    if "/" in raw or "\\" in raw or Path(raw).name != raw:
        raise ValueError("database filename must not contain a path")
    name = raw
    if not _DATABASE_FILENAME_RE.fullmatch(name):
        raise ValueError("database filename contains unsupported characters")
    if not name.lower().endswith(".sqlite3"):
        name += ".sqlite3"
    if name in {".", ".."} or name.startswith("."):
        raise ValueError("database filename is invalid")
    return name


def read_database_selection(st_root: str | Path) -> str:
    path = database_selection_path(st_root)
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
        return normalize_database_filename(payload.get("filename") or DEFAULT_DB_FILENAME)
    except FileNotFoundError:
        return DEFAULT_DB_FILENAME
    except Exception:
        return DEFAULT_DB_FILENAME


def write_database_selection(st_root: str | Path, filename: object) -> Path:
    root = Path(st_root).expanduser().resolve()
    name = normalize_database_filename(filename)
    path = database_selection_path(root)
    path.parent.mkdir(parents=True, exist_ok=True)
    temp = path.with_suffix(path.suffix + ".tmp")
    temp.write_text(json.dumps({"filename": name}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    temp.replace(path)
    return database_directory(root) / name


def default_database_path(st_root: str | Path) -> Path:
    root = Path(st_root).expanduser().resolve()
    return database_directory(root) / read_database_selection(root)


def list_database_files(st_root: str | Path) -> dict[str, Any]:
    root = Path(st_root).expanduser().resolve()
    directory = database_directory(root)
    directory.mkdir(parents=True, exist_ok=True)
    active = read_database_selection(root)
    files = sorted((path.name for path in directory.glob("*.sqlite3") if path.is_file()), key=str.casefold)
    if DEFAULT_DB_FILENAME not in files and not files:
        files = [DEFAULT_DB_FILENAME]
    return {"active_filename": active, "active_path": str((directory / active).resolve()), "files": files}


def open_database_file(st_root: str | Path, filename: object) -> dict[str, Any]:
    root = Path(st_root).expanduser().resolve()
    name = normalize_database_filename(filename)
    target = database_directory(root) / name
    if not target.is_file():
        raise ValueError(f"database file not found: {name}")
    initialize_database(target, root)
    write_database_selection(root, name)
    return {**list_database_files(root), "db_path": str(target.resolve()), "filename": name}


def create_database_file(st_root: str | Path, filename: object) -> dict[str, Any]:
    root = Path(st_root).expanduser().resolve()
    name = normalize_database_filename(filename)
    target = database_directory(root) / name
    if target.exists():
        raise ValueError(f"database file already exists: {name}")
    initialize_database(target, root)
    write_database_selection(root, name)
    return {**list_database_files(root), "db_path": str(target.resolve()), "filename": name}


def save_database_as(st_root: str | Path, source_db_path: str | Path, filename: object) -> dict[str, Any]:
    root = Path(st_root).expanduser().resolve()
    name = normalize_database_filename(filename)
    source = Path(source_db_path).expanduser().resolve()
    target = database_directory(root) / name
    if target.exists():
        raise ValueError(f"database file already exists: {name}")
    initialize_database(source, root)
    target.parent.mkdir(parents=True, exist_ok=True)
    src = connect_database(source)
    dst = sqlite3.connect(target, timeout=10.0)
    try:
        src.backup(dst)
        dst.commit()
    except Exception:
        dst.close()
        src.close()
        try:
            target.unlink(missing_ok=True)
        except Exception:
            pass
        raise
    else:
        dst.close()
        src.close()
    initialize_database(target, root)
    write_database_selection(root, name)
    return {**list_database_files(root), "db_path": str(target.resolve()), "filename": name}


def connect_database(db_path: str | Path) -> sqlite3.Connection:
    path = Path(db_path).expanduser().resolve()
    path.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(path, timeout=10.0)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    conn.execute("PRAGMA busy_timeout = 10000")
    conn.execute("PRAGMA journal_mode = WAL")
    conn.execute("PRAGMA synchronous = NORMAL")
    return conn


def _table_columns(conn: sqlite3.Connection, table_name: str) -> set[str]:
    return {str(row["name"]) for row in conn.execute(f"PRAGMA table_info({table_name})")}


def _ensure_column(conn: sqlite3.Connection, table_name: str, column_name: str, definition: str) -> None:
    if column_name not in _table_columns(conn, table_name):
        conn.execute(f"ALTER TABLE {table_name} ADD COLUMN {column_name} {definition}")


def _apply_schema(conn: sqlite3.Connection) -> None:
    for schema_file in SCHEMA_FILES:
        conn.executescript(schema_file.read_text(encoding="utf-8"))
    _ensure_column(conn, "social_accounts", "deleted_at", "TEXT")
    _ensure_column(conn, "posts", "deleted_at", "TEXT")
    _ensure_column(conn, "media_assets", "folder_name", "TEXT NOT NULL DEFAULT ''")
    _ensure_column(conn, "media_saved_events", "folder_name", "TEXT NOT NULL DEFAULT ''")
    for table_name in ("post_stat_latest", "post_stat_history", "post_stat_10m"):
        _ensure_column(conn, table_name, "sort_by", "TEXT NOT NULL DEFAULT ''")
        _ensure_column(conn, table_name, "sort_value", "INTEGER NOT NULL DEFAULT 0")
        _ensure_column(conn, table_name, "source_kind", "TEXT NOT NULL DEFAULT 'LOCAL'")
        _ensure_column(conn, table_name, "views_available", "INTEGER NOT NULL DEFAULT 0")
    migrations = (
        (1, "initial_storage_foundation"),
        (2, "unified_media_saved_flow"),
        (3, "analyze8780_repository_and_10m_stats"),
        (4, "collection_account_queue_sqlite_source"),
        (5, "original_filename_and_account_folder_display"),
        (6, "observation_sort_mode_and_hourly_central_sync"),
        (7, "views_availability_and_central_hour_history"),
    )
    conn.executemany(
        "INSERT OR IGNORE INTO schema_migrations(version, name) VALUES (?, ?)",
        migrations,
    )


def initialize_database(
    db_path: str | Path,
    st_root: str | Path,
    *,
    create_storage_dirs: bool = True,
) -> dict[str, Any]:
    stdown_root = derive_stdown_root(st_root)
    instagram_root = stdown_root / "instagram"
    if create_storage_dirs:
        instagram_root.mkdir(parents=True, exist_ok=True)

    conn = connect_database(db_path)
    try:
        _apply_schema(conn)
        conn.execute(
            """
            INSERT INTO storage_roots(root_code, root_path, is_default)
            VALUES (?, ?, 1)
            ON CONFLICT(root_code) DO UPDATE SET
                root_path = excluded.root_path,
                is_default = 1,
                updated_at = CURRENT_TIMESTAMP
            """,
            (DEFAULT_STORAGE_CODE, str(instagram_root)),
        )
        conn.execute("INSERT OR IGNORE INTO collector_profile(id) VALUES (1)")
        conn.execute("INSERT OR IGNORE INTO device_profile(id) VALUES (1)")
        conn.commit()

        table_count = conn.execute(
            "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'"
        ).fetchone()[0]
        journal_mode = conn.execute("PRAGMA journal_mode").fetchone()[0]
        foreign_keys = conn.execute("PRAGMA foreign_keys").fetchone()[0]
        user_version = conn.execute("PRAGMA user_version").fetchone()[0]
        return {
            "db_path": str(Path(db_path).expanduser().resolve()),
            "storage_root": str(instagram_root),
            "schema_version": int(user_version),
            "table_count": int(table_count),
            "journal_mode": str(journal_mode).lower(),
            "foreign_keys": int(foreign_keys),
        }
    finally:
        conn.close()


def normalize_source_mode(source_mode: str) -> str:
    normalized = str(source_mode or "").strip().upper()
    if normalized not in SOURCE_MODES:
        allowed = ", ".join(sorted(SOURCE_MODES))
        raise ValueError(f"unsupported source_mode={source_mode!r}; allowed={allowed}")
    return normalized


def canonical_instagram_post_url(shortcode: str, *candidate_urls: str, fallback_kind: str = "p") -> str:
    """Return a real Instagram post permalink, never a feed/profile scope URL."""
    code = str(shortcode or "").strip()
    if not code:
        return ""
    canonical_code = urllib.parse.quote(code, safe="-_")
    kind_map = {"p": "p", "reel": "reel", "reels": "reel", "tv": "tv"}
    for candidate in candidate_urls:
        raw = str(candidate or "").strip()
        if not raw:
            continue
        if raw.startswith("/"):
            raw = "https://www.instagram.com" + raw
        try:
            parsed = urllib.parse.urlparse(raw)
        except Exception:
            continue
        host = str(parsed.hostname or "").lower()
        if host != "instagram.com" and not host.endswith(".instagram.com"):
            continue
        parts = [urllib.parse.unquote(part) for part in str(parsed.path or "").split("/") if part]
        if len(parts) < 2:
            continue
        kind = kind_map.get(parts[0].lower())
        if not kind or parts[1] != code:
            continue
        return f"https://www.instagram.com/{kind}/{canonical_code}/"
    fallback = kind_map.get(str(fallback_kind or "p").strip().lower(), "p")
    return f"https://www.instagram.com/{fallback}/{canonical_code}/"


def normalize_relative_path(relative_path: str | Path) -> str:
    raw = str(relative_path or "").replace("\\", "/").strip()
    if not raw:
        raise ValueError("relative_path is required")
    path = PurePosixPath(raw)
    if path.is_absolute() or ".." in path.parts:
        raise ValueError(f"relative_path must stay below the configured storage root: {raw}")
    return path.as_posix()


def create_download_batch(
    db_path: str | Path,
    *,
    source_mode: str,
    batch_key: str | None = None,
    account_id: int | None = None,
    sort_by: str = "",
    filter_days: int | None = None,
    requested_count: int = 0,
    metadata: Mapping[str, Any] | None = None,
) -> int:
    mode = normalize_source_mode(source_mode)
    key = str(batch_key or uuid.uuid4())
    conn = connect_database(db_path)
    try:
        cursor = conn.execute(
            """
            INSERT INTO download_batches(
                batch_key, source_mode, account_id, sort_by, filter_days,
                requested_count, metadata_json
            ) VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            (
                key,
                mode,
                account_id,
                str(sort_by or ""),
                filter_days,
                max(0, int(requested_count)),
                json.dumps(dict(metadata or {}), ensure_ascii=False, sort_keys=True),
            ),
        )
        conn.commit()
        return int(cursor.lastrowid)
    finally:
        conn.close()


def create_download_job(
    db_path: str | Path,
    *,
    post_id: int,
    source_mode: str,
    asset_kind: str,
    media_index: int = 0,
    batch_id: int | None = None,
    rank_no: int | None = None,
    source_url: str = "",
    expected_file_name: str = "",
) -> int:
    mode = normalize_source_mode(source_mode)
    kind = str(asset_kind or "").strip().upper()
    if not kind:
        raise ValueError("asset_kind is required")
    conn = connect_database(db_path)
    try:
        cursor = conn.execute(
            """
            INSERT INTO download_jobs(
                batch_id, post_id, source_mode, asset_kind, media_index,
                rank_no, source_url, expected_file_name
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                batch_id,
                int(post_id),
                mode,
                kind,
                max(0, int(media_index)),
                rank_no,
                str(source_url or ""),
                str(expected_file_name or ""),
            ),
        )
        conn.commit()
        return int(cursor.lastrowid)
    finally:
        conn.close()


def record_media_saved(
    db_path: str | Path,
    *,
    post_id: int,
    source_mode: str,
    asset_kind: str,
    relative_path: str | Path,
    file_name: str,
    folder_name: str = "",
    file_size: int,
    media_index: int = 0,
    rank_no: int | None = None,
    md5: str = "",
    sha256: str = "",
    source_url: str = "",
    storage_root_id: int | None = None,
    download_job_id: int | None = None,
    event_uuid: str | None = None,
    completed_at: str | None = None,
    payload: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    """Record the common MEDIA_SAVED result for every download entry path.

    The server sync payload intentionally excludes the local relative path. The
    local event retains only a path relative to the configured storage root.
    """
    mode = normalize_source_mode(source_mode)
    kind = str(asset_kind or "").strip().upper()
    if not kind:
        raise ValueError("asset_kind is required")
    safe_relative_path = normalize_relative_path(relative_path)
    safe_file_name = str(file_name or "").strip()
    if not safe_file_name:
        raise ValueError("file_name is required")
    relative_parent = PurePosixPath(safe_relative_path).parent
    derived_folder_name = "" if str(relative_parent) in ("", ".") else relative_parent.name
    safe_folder_name = str(folder_name or derived_folder_name or "instagram").strip()
    size = max(0, int(file_size))
    index = max(0, int(media_index))
    event_key = str(event_uuid or uuid.uuid4())
    completed_value = completed_at or None

    conn = connect_database(db_path)
    try:
        conn.execute("BEGIN IMMEDIATE")
        existing = conn.execute(
            "SELECT id, asset_id FROM media_saved_events WHERE event_uuid = ?",
            (event_key,),
        ).fetchone()
        if existing:
            conn.commit()
            return {
                "event_id": int(existing["id"]),
                "asset_id": int(existing["asset_id"]),
                "event_uuid": event_key,
                "source_mode": mode,
                "idempotent": True,
            }

        job_row = None
        if download_job_id is not None:
            job_row = conn.execute(
                """
                SELECT id, batch_id, post_id, source_mode, asset_kind, media_index
                FROM download_jobs WHERE id = ?
                """,
                (int(download_job_id),),
            ).fetchone()
            if not job_row:
                raise ValueError(f"download_job_id not found: {download_job_id}")
            expected = (int(post_id), mode, kind, index)
            actual = (
                int(job_row["post_id"]),
                str(job_row["source_mode"]),
                str(job_row["asset_kind"]),
                int(job_row["media_index"]),
            )
            if actual != expected:
                raise ValueError(
                    f"download job mismatch: expected={expected!r} actual={actual!r}"
                )

        post = conn.execute(
            "SELECT id, account_id, shortcode FROM posts WHERE id = ?",
            (int(post_id),),
        ).fetchone()
        if not post:
            raise ValueError(f"post_id not found: {post_id}")

        if storage_root_id is None:
            default_root = conn.execute(
                "SELECT id FROM storage_roots WHERE is_default = 1 ORDER BY id LIMIT 1"
            ).fetchone()
            storage_root_id = int(default_root["id"]) if default_root else None

        conn.execute(
            """
            INSERT INTO media_assets(
                post_id, asset_kind, media_index, source_url, storage_root_id,
                relative_path, file_name, folder_name, file_size, md5, sha256,
                local_status, verified_at
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'AVAILABLE', COALESCE(?, CURRENT_TIMESTAMP))
            ON CONFLICT(post_id, asset_kind, media_index) DO UPDATE SET
                source_url = CASE WHEN excluded.source_url <> '' THEN excluded.source_url ELSE media_assets.source_url END,
                storage_root_id = COALESCE(excluded.storage_root_id, media_assets.storage_root_id),
                relative_path = excluded.relative_path,
                file_name = excluded.file_name,
                folder_name = excluded.folder_name,
                file_size = excluded.file_size,
                md5 = excluded.md5,
                sha256 = excluded.sha256,
                local_status = 'AVAILABLE',
                verified_at = excluded.verified_at,
                updated_at = CURRENT_TIMESTAMP
            """,
            (
                int(post_id),
                kind,
                index,
                str(source_url or ""),
                storage_root_id,
                safe_relative_path,
                safe_file_name,
                safe_folder_name,
                size,
                str(md5 or ""),
                str(sha256 or ""),
                completed_value,
            ),
        )
        asset = conn.execute(
            "SELECT id FROM media_assets WHERE post_id = ? AND asset_kind = ? AND media_index = ?",
            (int(post_id), kind, index),
        ).fetchone()
        asset_id = int(asset["id"])

        local_payload = dict(payload or {})
        event_cursor = conn.execute(
            """
            INSERT INTO media_saved_events(
                event_uuid, download_job_id, post_id, asset_id, account_id,
                shortcode, source_mode, asset_kind, media_index, rank_no,
                relative_path, file_name, folder_name, file_size, md5, sha256,
                completed_at, payload_json
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, COALESCE(?, CURRENT_TIMESTAMP), ?)
            """,
            (
                event_key,
                download_job_id,
                int(post_id),
                asset_id,
                post["account_id"],
                str(post["shortcode"]),
                mode,
                kind,
                index,
                rank_no,
                safe_relative_path,
                safe_file_name,
                safe_folder_name,
                size,
                str(md5 or ""),
                str(sha256 or ""),
                completed_value,
                json.dumps(local_payload, ensure_ascii=False, sort_keys=True),
            ),
        )
        event_id = int(event_cursor.lastrowid)

        if download_job_id is not None:
            conn.execute(
                """
                UPDATE download_jobs
                SET job_status = 'COMPLETE', completed_at = COALESCE(?, CURRENT_TIMESTAMP),
                    last_error = '', updated_at = CURRENT_TIMESTAMP
                WHERE id = ?
                """,
                (completed_value, int(download_job_id)),
            )
            batch_id = job_row["batch_id"] if job_row else None
            if batch_id is not None:
                batch_progress = conn.execute(
                    """
                    SELECT b.requested_count,
                           COUNT(j.id) AS total_jobs,
                           SUM(CASE WHEN j.job_status <> 'COMPLETE' THEN 1 ELSE 0 END) AS remaining_jobs
                    FROM download_batches b
                    LEFT JOIN download_jobs j ON j.batch_id = b.id
                    WHERE b.id = ?
                    GROUP BY b.id, b.requested_count
                    """,
                    (int(batch_id),),
                ).fetchone()
                requested = int(batch_progress["requested_count"] or 0) if batch_progress else 0
                total_jobs = int(batch_progress["total_jobs"] or 0) if batch_progress else 0
                remaining = int(batch_progress["remaining_jobs"] or 0) if batch_progress else 0
                if remaining == 0 and (requested <= 0 or total_jobs >= requested):
                    conn.execute(
                        """
                        UPDATE download_batches
                        SET batch_status = 'COMPLETE',
                            completed_at = COALESCE(?, CURRENT_TIMESTAMP),
                            updated_at = CURRENT_TIMESTAMP
                        WHERE id = ?
                        """,
                        (completed_value, int(batch_id)),
                    )
                else:
                    conn.execute(
                        """
                        UPDATE download_batches
                        SET batch_status = 'RUNNING', completed_at = NULL,
                            updated_at = CURRENT_TIMESTAMP
                        WHERE id = ?
                        """,
                        (int(batch_id),),
                    )

        conn.execute(
            "UPDATE posts SET media_state = 'MEDIA_AVAILABLE', updated_at = CURRENT_TIMESTAMP WHERE id = ?",
            (int(post_id),),
        )
        conn.execute(
            """
            INSERT INTO p2p_inventory(asset_id, available, updated_at)
            VALUES (?, 1, CURRENT_TIMESTAMP)
            ON CONFLICT(asset_id) DO UPDATE SET available = 1, updated_at = CURRENT_TIMESTAMP
            """,
            (asset_id,),
        )

        server_payload = {
            "client_event_id": event_key,
            "post_id": int(post_id),
            "shortcode": str(post["shortcode"]),
            "source_mode": mode,
            "asset_kind": kind,
            "media_index": index,
            "rank_no": rank_no,
            "file_name": safe_file_name,
            "file_size": size,
            "md5": str(md5 or ""),
            "sha256": str(sha256 or ""),
            "completed_at": completed_value,
        }
        conn.execute(
            """
            INSERT INTO sync_queue(entity_type, entity_local_id, operation, payload_json)
            VALUES ('MEDIA_SAVED', ?, 'UPSERT', ?)
            """,
            (event_id, json.dumps(server_payload, ensure_ascii=False, sort_keys=True)),
        )
        conn.commit()
        return {
            "event_id": event_id,
            "asset_id": asset_id,
            "event_uuid": event_key,
            "source_mode": mode,
            "folder_name": safe_folder_name,
            "file_name": safe_file_name,
            "idempotent": False,
        }
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

def ensure_storage_root(
    db_path: str | Path,
    *,
    root_path: str | Path,
    root_code: str | None = None,
    is_default: bool = False,
) -> int:
    resolved = Path(root_path).expanduser().resolve()
    code = str(root_code or "").strip()
    if not code:
        digest = hashlib.sha256(str(resolved).encode("utf-8")).hexdigest()[:12].upper()
        code = f"LOCAL_DOWNLOAD_{digest}"
    conn = connect_database(db_path)
    try:
        if is_default:
            conn.execute("UPDATE storage_roots SET is_default = 0 WHERE is_default = 1")
        conn.execute(
            """
            INSERT INTO storage_roots(root_code, root_path, is_default)
            VALUES (?, ?, ?)
            ON CONFLICT(root_code) DO UPDATE SET
                root_path = excluded.root_path,
                is_default = excluded.is_default,
                updated_at = CURRENT_TIMESTAMP
            """,
            (code, str(resolved), 1 if is_default else 0),
        )
        row = conn.execute("SELECT id FROM storage_roots WHERE root_code = ?", (code,)).fetchone()
        conn.commit()
        return int(row["id"])
    finally:
        conn.close()


def _is_relative_to(path: Path, root: Path) -> bool:
    try:
        path.relative_to(root)
        return True
    except ValueError:
        return False


def resolve_download_storage(
    db_path: str | Path,
    *,
    file_path: str | Path,
    account_path: str = "",
    account_folder_merge: bool = False,
) -> dict[str, Any]:
    file = Path(file_path).expanduser().resolve()
    if not file.is_file():
        raise FileNotFoundError(str(file))

    conn = connect_database(db_path)
    try:
        default = conn.execute(
            "SELECT id, root_path FROM storage_roots WHERE is_default = 1 ORDER BY id LIMIT 1"
        ).fetchone()
    finally:
        conn.close()

    if default:
        default_root = Path(str(default["root_path"])).expanduser().resolve()
        if _is_relative_to(file, default_root):
            return {
                "storage_root_id": int(default["id"]),
                "storage_root_path": str(default_root),
                "relative_path": file.relative_to(default_root).as_posix(),
                "folder_name": file.parent.name or "instagram",
                "storage_layout": "STDOWN_DEFAULT",
            }

    parent = file.parent
    root = parent
    account_parts = [part for part in str(account_path or "").replace("\\", "/").split("/") if part]
    if not account_folder_merge and account_parts:
        probe = parent
        matched = True
        for expected in reversed(account_parts):
            if probe.name.casefold() != expected.casefold():
                matched = False
                break
            probe = probe.parent
        if matched:
            root = probe

    root_id = ensure_storage_root(db_path, root_path=root)
    return {
        "storage_root_id": root_id,
        "storage_root_path": str(root),
        "relative_path": file.relative_to(root).as_posix(),
        "folder_name": file.parent.name or "instagram",
        "storage_layout": "LOCAL_DOWNLOAD_ROOT",
    }


def upsert_download_post(
    db_path: str | Path,
    *,
    shortcode: str,
    owner_username: str,
    post_url: str = "",
    caption: str = "",
    thumbnail_url: str = "",
    video_url: str = "",
    published_at: str | None = None,
    stats: Mapping[str, Any] | None = None,
    fallback_kind: str = "",
) -> dict[str, int]:
    code = str(shortcode or "").strip()
    owner = str(owner_username or "").strip().lstrip("@")
    if not code:
        raise ValueError("shortcode is required")
    if not owner:
        raise ValueError("owner_username is required")
    kind_hint = str(fallback_kind or "").strip().lower()
    if kind_hint not in {"reel", "p"}:
        kind_hint = "reel" if str(video_url or "").strip() else "p"
    url = canonical_instagram_post_url(code, post_url, fallback_kind=kind_hint)
    conn = connect_database(db_path)
    try:
        conn.execute("BEGIN IMMEDIATE")
        conn.execute(
            """
            INSERT INTO social_accounts(platform, username, profile_url)
            VALUES ('instagram', ?, ?)
            ON CONFLICT(platform, username) DO UPDATE SET
                profile_url = CASE WHEN excluded.profile_url <> '' THEN excluded.profile_url ELSE social_accounts.profile_url END,
                updated_at = CURRENT_TIMESTAMP
            """,
            (owner, f"https://www.instagram.com/{owner}/"),
        )
        account = conn.execute(
            "SELECT id FROM social_accounts WHERE platform = 'instagram' AND username = ?",
            (owner,),
        ).fetchone()
        account_id = int(account["id"])
        conn.execute(
            """
            INSERT INTO posts(
                account_id, platform, shortcode, post_url, caption, published_at,
                thumbnail_url, video_url, source_url_observed_at
            ) VALUES (?, 'instagram', ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
            ON CONFLICT(platform, shortcode) DO UPDATE SET
                account_id = excluded.account_id,
                post_url = CASE WHEN excluded.post_url <> '' THEN excluded.post_url ELSE posts.post_url END,
                caption = CASE WHEN excluded.caption <> '' THEN excluded.caption ELSE posts.caption END,
                published_at = COALESCE(excluded.published_at, posts.published_at),
                thumbnail_url = CASE WHEN excluded.thumbnail_url <> '' THEN excluded.thumbnail_url ELSE posts.thumbnail_url END,
                video_url = CASE WHEN excluded.video_url <> '' THEN excluded.video_url ELSE posts.video_url END,
                source_url_observed_at = CURRENT_TIMESTAMP,
                last_observed_at = CURRENT_TIMESTAMP,
                updated_at = CURRENT_TIMESTAMP
            """,
            (
                account_id,
                code,
                url,
                str(caption or ""),
                published_at,
                str(thumbnail_url or ""),
                str(video_url or ""),
            ),
        )
        post = conn.execute(
            "SELECT id FROM posts WHERE platform = 'instagram' AND shortcode = ?",
            (code,),
        ).fetchone()
        post_id = int(post["id"])

        values = dict(stats or {})
        if values:
            collected_at = str(values.get("captured_at") or values.get("collected_at") or "").strip() or None
            metric_values = (
                int(values.get("views") or 0),
                int(values.get("likes") or 0),
                int(values.get("comments") or 0),
                int(values.get("reposts") or 0),
            )
            conn.execute(
                """
                INSERT INTO post_stat_latest(post_id, views, likes, comments, reposts, collected_at)
                VALUES (?, ?, ?, ?, ?, COALESCE(?, CURRENT_TIMESTAMP))
                ON CONFLICT(post_id) DO UPDATE SET
                    views = excluded.views,
                    likes = excluded.likes,
                    comments = excluded.comments,
                    reposts = excluded.reposts,
                    collected_at = excluded.collected_at
                """,
                (post_id, *metric_values, collected_at),
            )
            conn.execute(
                """
                INSERT OR IGNORE INTO post_stat_history(
                    post_id, views, likes, comments, reposts, collected_at
                ) VALUES (?, ?, ?, ?, ?, COALESCE(?, CURRENT_TIMESTAMP))
                """,
                (post_id, *metric_values, collected_at),
            )
        conn.commit()
        return {"account_id": account_id, "post_id": post_id}
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def ensure_download_batch(
    db_path: str | Path,
    *,
    batch_key: str,
    source_mode: str,
    account_id: int | None = None,
    sort_by: str = "",
    filter_days: int | None = None,
    requested_count: int = 0,
    metadata: Mapping[str, Any] | None = None,
) -> int:
    key = str(batch_key or "").strip()
    if not key:
        raise ValueError("batch_key is required")
    mode = normalize_source_mode(source_mode)
    conn = connect_database(db_path)
    try:
        conn.execute(
            """
            INSERT INTO download_batches(
                batch_key, source_mode, account_id, sort_by, filter_days,
                requested_count, batch_status, metadata_json
            ) VALUES (?, ?, ?, ?, ?, ?, 'RUNNING', ?)
            ON CONFLICT(batch_key) DO UPDATE SET
                account_id = COALESCE(excluded.account_id, download_batches.account_id),
                sort_by = CASE WHEN excluded.sort_by <> '' THEN excluded.sort_by ELSE download_batches.sort_by END,
                filter_days = COALESCE(excluded.filter_days, download_batches.filter_days),
                requested_count = MAX(download_batches.requested_count, excluded.requested_count),
                batch_status = CASE WHEN download_batches.batch_status = 'COMPLETE' THEN 'COMPLETE' ELSE 'RUNNING' END,
                metadata_json = excluded.metadata_json,
                updated_at = CURRENT_TIMESTAMP
            """,
            (
                key,
                mode,
                account_id,
                str(sort_by or ""),
                filter_days,
                max(0, int(requested_count)),
                json.dumps(dict(metadata or {}), ensure_ascii=False, sort_keys=True),
            ),
        )
        row = conn.execute("SELECT id FROM download_batches WHERE batch_key = ?", (key,)).fetchone()
        conn.commit()
        return int(row["id"])
    finally:
        conn.close()


def ensure_download_job(
    db_path: str | Path,
    *,
    batch_id: int,
    post_id: int,
    source_mode: str,
    asset_kind: str,
    media_index: int = 0,
    rank_no: int | None = None,
    source_url: str = "",
    expected_file_name: str = "",
) -> int:
    mode = normalize_source_mode(source_mode)
    kind = str(asset_kind or "").strip().upper()
    if not kind:
        raise ValueError("asset_kind is required")
    conn = connect_database(db_path)
    try:
        conn.execute(
            """
            INSERT INTO download_jobs(
                batch_id, post_id, source_mode, asset_kind, media_index,
                rank_no, source_url, expected_file_name, job_status
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'DOWNLOADING')
            ON CONFLICT(batch_id, post_id, asset_kind, media_index) DO UPDATE SET
                rank_no = COALESCE(excluded.rank_no, download_jobs.rank_no),
                source_url = CASE WHEN excluded.source_url <> '' THEN excluded.source_url ELSE download_jobs.source_url END,
                expected_file_name = CASE WHEN excluded.expected_file_name <> '' THEN excluded.expected_file_name ELSE download_jobs.expected_file_name END,
                job_status = CASE WHEN download_jobs.job_status = 'COMPLETE' THEN 'COMPLETE' ELSE 'DOWNLOADING' END,
                last_error = '',
                updated_at = CURRENT_TIMESTAMP
            """,
            (
                int(batch_id),
                int(post_id),
                mode,
                kind,
                max(0, int(media_index)),
                rank_no,
                str(source_url or ""),
                str(expected_file_name or ""),
            ),
        )
        row = conn.execute(
            """
            SELECT id FROM download_jobs
            WHERE batch_id = ? AND post_id = ? AND asset_kind = ? AND media_index = ?
            """,
            (int(batch_id), int(post_id), kind, max(0, int(media_index))),
        ).fetchone()
        conn.commit()
        return int(row["id"])
    finally:
        conn.close()


def record_downloaded_file(
    *,
    st_root: str | Path,
    file_path: str | Path,
    source_mode: str,
    shortcode: str,
    owner_username: str,
    asset_kind: str,
    media_index: int = 0,
    rank_no: int | None = None,
    source_url: str = "",
    post_url: str = "",
    caption: str = "",
    thumbnail_url: str = "",
    video_url: str = "",
    published_at: str | None = None,
    md5: str = "",
    sha256: str = "",
    file_size: int = 0,
    account_path: str = "",
    account_folder_merge: bool = False,
    batch_key: str = "",
    requested_count: int = 0,
    sort_by: str = "",
    filter_days: int | None = None,
    event_uuid: str = "",
    stats: Mapping[str, Any] | None = None,
    payload: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    root = Path(st_root).expanduser().resolve()
    db_path = default_database_path(root)
    initialize_database(db_path, root)
    mode = normalize_source_mode(source_mode)
    event_key = str(event_uuid or uuid.uuid4())
    context = upsert_download_post(
        db_path,
        shortcode=shortcode,
        owner_username=owner_username,
        post_url=post_url,
        caption=caption,
        thumbnail_url=thumbnail_url,
        video_url=video_url,
        published_at=published_at,
        stats=stats,
        fallback_kind="reel" if str(asset_kind or "").strip().upper() == "VIDEO" else "p",
    )
    storage = resolve_download_storage(
        db_path,
        file_path=file_path,
        account_path=account_path,
        account_folder_merge=account_folder_merge,
    )
    effective_batch_key = str(batch_key or f"single-{event_key}")
    batch_id = ensure_download_batch(
        db_path,
        batch_key=effective_batch_key,
        source_mode=mode,
        account_id=context["account_id"],
        sort_by=sort_by,
        filter_days=filter_days,
        requested_count=max(1, int(requested_count or 0)),
        metadata={
            "account_path": str(account_path or ""),
            "storage_layout": storage["storage_layout"],
        },
    )
    job_id = ensure_download_job(
        db_path,
        batch_id=batch_id,
        post_id=context["post_id"],
        source_mode=mode,
        asset_kind=asset_kind,
        media_index=media_index,
        rank_no=rank_no,
        source_url=source_url,
        expected_file_name=Path(file_path).name,
    )
    local_payload = dict(payload or {})
    local_payload.update(
        {
            "batch_key": effective_batch_key,
            "storage_layout": storage["storage_layout"],
        }
    )
    try:
        result = record_media_saved(
            db_path,
            post_id=context["post_id"],
            source_mode=mode,
            asset_kind=asset_kind,
            relative_path=storage["relative_path"],
            file_name=Path(file_path).name,
            folder_name=storage["folder_name"],
            file_size=file_size,
            media_index=media_index,
            rank_no=rank_no,
            md5=md5,
            sha256=sha256,
            source_url=source_url,
            storage_root_id=storage["storage_root_id"],
            download_job_id=job_id,
            event_uuid=event_key,
            payload=local_payload,
        )
    except Exception as exc:
        conn = connect_database(db_path)
        try:
            conn.execute(
                """
                UPDATE download_jobs
                SET job_status = 'FAILED', last_error = ?, updated_at = CURRENT_TIMESTAMP
                WHERE id = ?
                """,
                (str(exc), job_id),
            )
            conn.commit()
        finally:
            conn.close()
        raise
    return {
        **result,
        "db_path": str(db_path),
        "account_id": context["account_id"],
        "post_id": context["post_id"],
        "batch_id": batch_id,
        "job_id": job_id,
        "relative_path": storage["relative_path"],
        "folder_name": storage["folder_name"],
        "file_name": Path(file_path).name,
        "storage_root_id": storage["storage_root_id"],
        "storage_layout": storage["storage_layout"],
    }



def _parse_observed_at(value: str | datetime | None) -> datetime:
    if isinstance(value, datetime):
        dt = value
    else:
        raw = str(value or "").strip()
        if not raw:
            dt = datetime.now(timezone.utc)
        else:
            if raw.endswith("Z"):
                raw = raw[:-1] + "+00:00"
            dt = datetime.fromisoformat(raw)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt.astimezone(timezone.utc)


def ten_minute_bucket(value: str | datetime | None = None) -> str:
    dt = _parse_observed_at(value)
    minute = (dt.minute // 10) * 10
    return dt.replace(minute=minute, second=0, microsecond=0).isoformat().replace("+00:00", "Z")


def _iso_utc(value: str | datetime | None = None) -> str:
    return _parse_observed_at(value).isoformat().replace("+00:00", "Z")


def upsert_observation(
    db_path: str | Path,
    *,
    username: str,
    shortcode: str,
    views: int = 0,
    views_available: int = 0,
    likes: int = 0,
    comments: int = 0,
    reposts: int = 0,
    observed_at: str | datetime | None = None,
    platform_account_id: str = "",
    display_name: str = "",
    profile_url: str = "",
    post_url: str = "",
    caption: str = "",
    published_at: str | None = None,
    thumbnail_url: str = "",
    video_url: str = "",
    sort_by: str = "",
    sort_value: int = 0,
    source_kind: str = "LOCAL",
) -> dict[str, Any]:
    owner = str(username or "").strip().lstrip("@")
    code = str(shortcode or "").strip()
    if not owner:
        raise ValueError("username is required")
    if not code:
        raise ValueError("shortcode is required")
    observed = _iso_utc(observed_at)
    bucket = ten_minute_bucket(observed_at)
    metrics = tuple(max(0, int(v or 0)) for v in (views, likes, comments, reposts))
    views_available_flag = 1 if int(views_available or 0) else 0
    sort_field = str(sort_by or "").strip()
    sort_metric_value = max(0, int(sort_value or 0))
    observation_source = str(source_kind or "LOCAL").strip().upper() or "LOCAL"
    conn = connect_database(db_path)
    try:
        conn.execute("BEGIN IMMEDIATE")
        deleted_account = conn.execute(
            "SELECT 1 FROM deletion_events WHERE entity_type='ACCOUNT' AND lower(entity_key)=lower(?) LIMIT 1",
            (owner,),
        ).fetchone()
        deleted_post = conn.execute(
            "SELECT 1 FROM deletion_events WHERE entity_type='POST' AND entity_key=? LIMIT 1",
            (code,),
        ).fetchone()
        if deleted_account or deleted_post:
            conn.commit()
            return {
                "account_id": 0,
                "post_id": 0,
                "bucket_at": bucket,
                "observed_at": observed,
                "ignored_deleted": True,
                "deleted_entity": "ACCOUNT" if deleted_account else "POST",
            }
        existing_observation = conn.execute(
            """
            SELECT
                p.id AS post_id, p.account_id, p.post_url, p.caption, p.published_at,
                p.thumbnail_url, p.video_url, p.deleted_at AS post_deleted_at,
                a.username, a.platform_account_id, a.display_name, a.profile_url,
                a.active, a.deleted_at AS account_deleted_at,
                s.views, s.views_available, s.likes, s.comments, s.reposts
            FROM posts p
            JOIN social_accounts a ON a.id=p.account_id
            LEFT JOIN post_stat_latest s ON s.post_id=p.id
            WHERE p.platform='instagram' AND p.shortcode=?
            LIMIT 1
            """,
            (code,),
        ).fetchone()
        if existing_observation is not None and existing_observation["views"] is not None:
            effective_views = metrics[0]
            effective_views_available = views_available_flag
            if not views_available_flag and int(existing_observation["views_available"] or 0):
                effective_views = max(0, int(existing_observation["views"] or 0))
                effective_views_available = 1
            unchanged_metrics = (
                int(existing_observation["views"] or 0),
                int(existing_observation["views_available"] or 0),
                int(existing_observation["likes"] or 0),
                int(existing_observation["comments"] or 0),
                int(existing_observation["reposts"] or 0),
            ) == (
                effective_views, effective_views_available, metrics[1], metrics[2], metrics[3]
            )
            incoming_post_url = str(post_url or "").strip()
            resolved_incoming_post_url = (
                canonical_instagram_post_url(
                    code, incoming_post_url,
                    fallback_kind="reel" if str(video_url or "").strip() else "p",
                )
                if incoming_post_url else ""
            )
            metadata_changed = any((
                str(existing_observation["username"] or "").strip().lower() != owner.lower(),
                int(existing_observation["active"] or 0) != 1,
                existing_observation["account_deleted_at"] is not None,
                existing_observation["post_deleted_at"] is not None,
                bool(str(platform_account_id or "").strip()) and str(existing_observation["platform_account_id"] or "") != str(platform_account_id or ""),
                bool(str(display_name or "").strip()) and str(existing_observation["display_name"] or "") != str(display_name or ""),
                bool(str(profile_url or "").strip()) and str(existing_observation["profile_url"] or "") != str(profile_url or ""),
                bool(resolved_incoming_post_url) and canonical_instagram_post_url(code, existing_observation["post_url"], fallback_kind="reel" if str(existing_observation["video_url"] or "").strip() else "p") != resolved_incoming_post_url,
                bool(str(caption or "")) and str(existing_observation["caption"] or "") != str(caption or ""),
                published_at is not None and str(existing_observation["published_at"] or "") != str(published_at or ""),
                bool(str(thumbnail_url or "").strip()) and str(existing_observation["thumbnail_url"] or "") != str(thumbnail_url or ""),
                bool(str(video_url or "").strip()) and str(existing_observation["video_url"] or "") != str(video_url or ""),
            ))
            if unchanged_metrics and not metadata_changed:
                conn.commit()
                return {
                    "account_id": int(existing_observation["account_id"]),
                    "post_id": int(existing_observation["post_id"]),
                    "bucket_at": bucket,
                    "observed_at": observed,
                    "views_available": effective_views_available,
                    "sort_by": sort_field,
                    "sort_value": sort_metric_value,
                    "source_kind": observation_source,
                    "saved": False,
                    "skipped_unchanged": True,
                    "skipped_all_writes": True,
                }
        conn.execute(
            """
            INSERT INTO social_accounts(
                platform, platform_account_id, username, display_name, profile_url, active, deleted_at
            ) VALUES ('instagram', ?, ?, ?, ?, 1, NULL)
            ON CONFLICT(platform, username) DO UPDATE SET
                platform_account_id = CASE WHEN excluded.platform_account_id <> '' THEN excluded.platform_account_id ELSE social_accounts.platform_account_id END,
                display_name = CASE WHEN excluded.display_name <> '' THEN excluded.display_name ELSE social_accounts.display_name END,
                profile_url = CASE WHEN excluded.profile_url <> '' THEN excluded.profile_url ELSE social_accounts.profile_url END,
                active = 1,
                deleted_at = NULL,
                updated_at = CURRENT_TIMESTAMP
            """,
            (str(platform_account_id or ""), owner, str(display_name or ""), str(profile_url or "") or f"https://www.instagram.com/{owner}/"),
        )
        account_id = int(conn.execute(
            "SELECT id FROM social_accounts WHERE platform='instagram' AND username=?", (owner,)
        ).fetchone()["id"])
        existing_post = conn.execute(
            "SELECT post_url, video_url FROM posts WHERE platform='instagram' AND shortcode=?",
            (code,),
        ).fetchone()
        if existing_post and not str(post_url or "").strip() and not str(video_url or "").strip():
            resolved_post_url = canonical_instagram_post_url(
                code,
                existing_post["post_url"],
                fallback_kind="reel" if str(existing_post["video_url"] or "").strip() else "p",
            )
        else:
            resolved_post_url = canonical_instagram_post_url(
                code,
                post_url,
                fallback_kind="reel" if str(video_url or "").strip() else "p",
            )
        conn.execute(
            """
            INSERT INTO posts(
                account_id, platform, shortcode, post_url, caption, published_at,
                thumbnail_url, video_url, source_url_observed_at, last_observed_at, deleted_at
            ) VALUES (?, 'instagram', ?, ?, ?, ?, ?, ?, ?, ?, NULL)
            ON CONFLICT(platform, shortcode) DO UPDATE SET
                account_id = excluded.account_id,
                post_url = CASE WHEN excluded.post_url <> '' THEN excluded.post_url ELSE posts.post_url END,
                caption = CASE WHEN excluded.caption <> '' THEN excluded.caption ELSE posts.caption END,
                published_at = COALESCE(excluded.published_at, posts.published_at),
                thumbnail_url = CASE WHEN excluded.thumbnail_url <> '' THEN excluded.thumbnail_url ELSE posts.thumbnail_url END,
                video_url = CASE WHEN excluded.video_url <> '' THEN excluded.video_url ELSE posts.video_url END,
                source_url_observed_at = excluded.source_url_observed_at,
                last_observed_at = excluded.last_observed_at,
                deleted_at = NULL,
                updated_at = CURRENT_TIMESTAMP
            """,
            (
                account_id, code, resolved_post_url,
                str(caption or ""), published_at, str(thumbnail_url or ""), str(video_url or ""),
                observed, observed,
            ),
        )
        post_id = int(conn.execute(
            "SELECT id FROM posts WHERE platform='instagram' AND shortcode=?", (code,)
        ).fetchone()["id"])
        previous_stat = conn.execute(
            """
            SELECT views, views_available, likes, comments, reposts
            FROM post_stat_latest
            WHERE post_id=?
            """,
            (post_id,),
        ).fetchone()
        effective_views = metrics[0]
        effective_views_available = views_available_flag
        if previous_stat is not None and not views_available_flag and int(previous_stat["views_available"] or 0):
            effective_views = max(0, int(previous_stat["views"] or 0))
            effective_views_available = 1
        effective_metrics = (effective_views, metrics[1], metrics[2], metrics[3])
        unchanged = previous_stat is not None and (
            int(previous_stat["views"] or 0),
            int(previous_stat["views_available"] or 0),
            int(previous_stat["likes"] or 0),
            int(previous_stat["comments"] or 0),
            int(previous_stat["reposts"] or 0),
        ) == (
            effective_metrics[0],
            effective_views_available,
            effective_metrics[1],
            effective_metrics[2],
            effective_metrics[3],
        )
        if unchanged:
            conn.execute(
                "DELETE FROM deletion_events WHERE entity_type='ACCOUNT' AND entity_local_id=?", (account_id,)
            )
            conn.execute(
                "DELETE FROM deletion_events WHERE entity_type='POST' AND entity_local_id=?", (post_id,)
            )
            conn.commit()
            return {
                "account_id": account_id,
                "post_id": post_id,
                "bucket_at": bucket,
                "observed_at": observed,
                "views_available": effective_views_available,
                "sort_by": sort_field,
                "sort_value": sort_metric_value,
                "source_kind": observation_source,
                "saved": False,
                "skipped_unchanged": True,
            }
        conn.execute(
            """
            INSERT INTO post_stat_latest(post_id, views, views_available, likes, comments, reposts, collected_at, sort_by, sort_value, source_kind)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(post_id) DO UPDATE SET
                views=excluded.views, views_available=excluded.views_available, likes=excluded.likes, comments=excluded.comments,
                reposts=excluded.reposts, collected_at=excluded.collected_at,
                sort_by=excluded.sort_by, sort_value=excluded.sort_value, source_kind=excluded.source_kind
            """,
            (post_id, effective_metrics[0], effective_views_available, effective_metrics[1], effective_metrics[2], effective_metrics[3], observed, sort_field, sort_metric_value, observation_source),
        )
        conn.execute(
            """
            INSERT INTO post_stat_history(post_id, views, views_available, likes, comments, reposts, collected_at, sort_by, sort_value, source_kind)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(post_id, collected_at) DO UPDATE SET
                views=excluded.views, views_available=excluded.views_available, likes=excluded.likes, comments=excluded.comments, reposts=excluded.reposts,
                sort_by=excluded.sort_by, sort_value=excluded.sort_value, source_kind=excluded.source_kind
            """,
            (post_id, effective_metrics[0], effective_views_available, effective_metrics[1], effective_metrics[2], effective_metrics[3], observed, sort_field, sort_metric_value, observation_source),
        )
        conn.execute(
            """
            INSERT INTO post_stat_10m(
                post_id, bucket_at, observed_at, views, views_available, likes, comments, reposts, sample_count, sort_by, sort_value, source_kind
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?, ?)
            ON CONFLICT(post_id, bucket_at) DO UPDATE SET
                observed_at=excluded.observed_at, views=excluded.views, views_available=excluded.views_available, likes=excluded.likes,
                comments=excluded.comments, reposts=excluded.reposts, sort_by=excluded.sort_by,
                sort_value=excluded.sort_value, source_kind=excluded.source_kind,
                sample_count=post_stat_10m.sample_count + 1, updated_at=CURRENT_TIMESTAMP
            """,
            (post_id, bucket, observed, effective_metrics[0], effective_views_available, effective_metrics[1], effective_metrics[2], effective_metrics[3], sort_field, sort_metric_value, observation_source),
        )
        conn.execute(
            "DELETE FROM deletion_events WHERE entity_type='ACCOUNT' AND entity_local_id=?", (account_id,)
        )
        conn.execute(
            "DELETE FROM deletion_events WHERE entity_type='POST' AND entity_local_id=?", (post_id,)
        )
        conn.commit()
        return {"account_id": account_id, "post_id": post_id, "bucket_at": bucket, "observed_at": observed, "views_available": effective_views_available, "sort_by": sort_field, "sort_value": sort_metric_value, "source_kind": observation_source, "saved": True, "skipped_unchanged": False}
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def list_accounts(
    db_path: str | Path, *, page: int = 1, page_size: int = 50, include_deleted: bool = False
) -> dict[str, Any]:
    page = max(1, int(page or 1))
    page_size = max(1, min(200, int(page_size or 50)))
    where = "1=1" if include_deleted else "a.active=1 AND a.deleted_at IS NULL"
    conn = connect_database(db_path)
    try:
        total = int(conn.execute(f"SELECT COUNT(*) AS n FROM social_accounts a WHERE {where}").fetchone()["n"])
        rows = conn.execute(
            f"""
            SELECT a.id, a.username, a.display_name, a.profile_url, a.active, a.deleted_at,
                   COUNT(p.id) AS post_count,
                   COALESCE(MAX(s.views), 0) AS max_views,
                   COALESCE(SUM(s.views), 0) AS total_views,
                   MAX(COALESCE(s.collected_at, p.last_observed_at, a.updated_at)) AS last_observed_at
            FROM social_accounts a
            LEFT JOIN posts p ON p.account_id=a.id AND p.deleted_at IS NULL
            LEFT JOIN post_stat_latest s ON s.post_id=p.id
            WHERE {where}
            GROUP BY a.id
            ORDER BY COALESCE(last_observed_at, a.updated_at) DESC, a.username COLLATE NOCASE
            LIMIT ? OFFSET ?
            """,
            (page_size, (page - 1) * page_size),
        ).fetchall()
        return {"items": [dict(row) for row in rows], "page": page, "page_size": page_size, "total": total}
    finally:
        conn.close()


def list_account_posts(
    db_path: str | Path, *, account_id: int, page: int = 1, page_size: int = 50, sort_by: str = "views"
) -> dict[str, Any]:
    page = max(1, int(page or 1))
    page_size = max(1, min(200, int(page_size or 50)))
    sort_map = {
        "views": "views DESC", "likes": "likes DESC", "comments": "comments DESC",
        "reposts": "reposts DESC", "views_delta": "views_delta DESC",
        "views_per_hour": "views_per_hour DESC", "published_at": "p.published_at DESC",
    }
    order = sort_map.get(str(sort_by or "views"), sort_map["views"])
    conn = connect_database(db_path)
    try:
        total = int(conn.execute(
            "SELECT COUNT(*) AS n FROM posts WHERE account_id=? AND deleted_at IS NULL", (int(account_id),)
        ).fetchone()["n"])
        rows = conn.execute(
            f"""
            WITH ranked AS (
                SELECT t.*, ROW_NUMBER() OVER(PARTITION BY t.post_id ORDER BY t.bucket_at DESC) AS rn
                FROM post_stat_10m t
            ), recent AS (
                SELECT post_id,
                       MAX(CASE WHEN rn=1 THEN views END) AS last_views,
                       MAX(CASE WHEN rn=2 THEN views END) AS prev_views,
                       MAX(CASE WHEN rn=1 THEN bucket_at END) AS bucket_at
                FROM ranked WHERE rn <= 2 GROUP BY post_id
            )
            SELECT p.id, p.shortcode, p.post_url, p.caption, p.published_at, p.thumbnail_url,
                   p.video_url, p.media_state, p.last_observed_at,
                   COALESCE(s.views,0) AS views, COALESCE(s.views_available,0) AS views_available, COALESCE(s.likes,0) AS likes,
                   COALESCE(s.comments,0) AS comments, COALESCE(s.reposts,0) AS reposts,
                   COALESCE(recent.last_views, s.views, 0)-COALESCE(recent.prev_views, recent.last_views, s.views, 0) AS views_delta,
                   (COALESCE(recent.last_views, s.views, 0)-COALESCE(recent.prev_views, recent.last_views, s.views, 0))*6 AS views_per_hour,
                   CASE WHEN COALESCE(s.views,0)>0 THEN ROUND(COALESCE(s.likes,0)*100.0/s.views,4) ELSE 0 END AS like_rate,
                   CASE WHEN COALESCE(s.views,0)>0 THEN ROUND(COALESCE(s.comments,0)*100.0/s.views,4) ELSE 0 END AS comment_rate,
                   recent.bucket_at AS latest_bucket_at,
                   (SELECT ma.folder_name FROM media_assets ma WHERE ma.post_id=p.id AND ma.local_status='AVAILABLE' ORDER BY ma.media_index, ma.id LIMIT 1) AS folder_name,
                   (SELECT ma.file_name FROM media_assets ma WHERE ma.post_id=p.id AND ma.local_status='AVAILABLE' ORDER BY ma.media_index, ma.id LIMIT 1) AS file_name,
                   (SELECT ma.relative_path FROM media_assets ma WHERE ma.post_id=p.id AND ma.local_status='AVAILABLE' ORDER BY ma.media_index, ma.id LIMIT 1) AS relative_path,
                   EXISTS(
                       SELECT 1 FROM media_assets ma
                       WHERE ma.post_id=p.id AND ma.local_status='AVAILABLE'
                         AND UPPER(ma.asset_kind) IN ('IMAGE','JPG','JPEG')
                   ) AS has_jpg,
                   EXISTS(
                       SELECT 1 FROM media_assets ma
                       WHERE ma.post_id=p.id AND ma.local_status='AVAILABLE'
                         AND UPPER(ma.asset_kind) IN ('VIDEO','MP4')
                   ) AS has_mp4
            FROM posts p
            LEFT JOIN post_stat_latest s ON s.post_id=p.id
            LEFT JOIN recent ON recent.post_id=p.id
            WHERE p.account_id=? AND p.deleted_at IS NULL
            ORDER BY {order}, p.id DESC
            LIMIT ? OFFSET ?
            """,
            (int(account_id), page_size, (page - 1) * page_size),
        ).fetchall()
        items = [dict(row) for row in rows]
        for item in items:
            item["post_url"] = canonical_instagram_post_url(item.get("shortcode"), item.get("post_url"), fallback_kind="reel" if str(item.get("video_url") or "").strip() else "p")
        return {"items": items, "page": page, "page_size": page_size, "total": total, "sort_by": sort_by}
    finally:
        conn.close()


def get_post_detail(db_path: str | Path, post_id: int) -> dict[str, Any] | None:
    conn = connect_database(db_path)
    try:
        row = conn.execute(
            """
            SELECT p.*, a.username, a.display_name,
                   COALESCE(s.views,0) AS views, COALESCE(s.views_available,0) AS views_available, COALESCE(s.likes,0) AS likes,
                   COALESCE(s.comments,0) AS comments, COALESCE(s.reposts,0) AS reposts,
                   s.collected_at
            FROM posts p
            LEFT JOIN social_accounts a ON a.id=p.account_id
            LEFT JOIN post_stat_latest s ON s.post_id=p.id
            WHERE p.id=? AND p.deleted_at IS NULL
            """,
            (int(post_id),),
        ).fetchone()
        if not row:
            return None
        item = dict(row)
        item["post_url"] = canonical_instagram_post_url(item.get("shortcode"), item.get("post_url"), fallback_kind="reel" if str(item.get("video_url") or "").strip() else "p")
        assets = conn.execute(
            """
            SELECT asset_kind, media_index, folder_name, file_name, relative_path, file_size, md5, sha256, local_status
            FROM media_assets WHERE post_id=? ORDER BY media_index, id
            """,
            (int(post_id),),
        ).fetchall()
        item["media_assets"] = [dict(asset) for asset in assets]
        return item
    finally:
        conn.close()


def get_post_stats(db_path: str | Path, post_id: int, limit: int = 144) -> list[dict[str, Any]]:
    conn = connect_database(db_path)
    try:
        rows = conn.execute(
            """
            SELECT bucket_at, observed_at, views, views_available, likes, comments, reposts, sample_count
            FROM post_stat_10m WHERE post_id=? ORDER BY bucket_at DESC LIMIT ?
            """,
            (int(post_id), max(1, min(1008, int(limit or 144)))),
        ).fetchall()
        return [dict(row) for row in rows]
    finally:
        conn.close()



_COLLECTION_RESERVED_USERNAMES = frozenset({
    "", "p", "reel", "reels", "explore", "explorer", "stories", "accounts",
    "direct", "graphql", "api", "oauth", "developer", "about", "privacy",
    "terms", "legal", "emails", "challenge", "search", "home", "instagram",
})


def normalize_collection_username(value: object) -> str:
    """Return a canonical Instagram username for the collection queue.

    The queue stores usernames only. A legacy profile URL is accepted only for
    migration; feed/detail scopes such as /explore/, /reels/, /p/ and /reel/
    are rejected and never become collection accounts.
    """
    raw = str(value or "").strip().replace("🗑️", "", 1).strip().lstrip("@")
    if not raw:
        return ""
    candidate = raw
    if "://" in raw or raw.startswith("www.") or raw.startswith("instagram.com/"):
        url = raw if "://" in raw else "https://" + raw
        try:
            parsed = urllib.parse.urlparse(url)
        except Exception:
            return ""
        host = str(parsed.hostname or "").lower().removeprefix("www.")
        if host != "instagram.com":
            return ""
        parts = [urllib.parse.unquote(part) for part in str(parsed.path or "").split("/") if part]
        if not parts:
            return ""
        first = str(parts[0] or "").strip()
        if first.lower() in _COLLECTION_RESERVED_USERNAMES:
            return ""
        # Only a profile root or its own reels tab is a legacy account input.
        if len(parts) > 1 and str(parts[1] or "").lower() not in {"reels"}:
            return ""
        candidate = first
    elif "/" in raw:
        return ""
    candidate = candidate.strip().lstrip("@").lower()
    if candidate in _COLLECTION_RESERVED_USERNAMES:
        return ""
    if not candidate or len(candidate) > 30:
        return ""
    if any(ch not in "abcdefghijklmnopqrstuvwxyz0123456789._" for ch in candidate):
        return ""
    return candidate


def list_collection_accounts(db_path: str | Path, *, include_deleted: bool = True) -> list[dict[str, Any]]:
    conn = connect_database(db_path)
    try:
        where = "" if include_deleted else "WHERE enabled=1 AND is_deleted=0"
        rows = conn.execute(
            f"""
            SELECT id, username, position, enabled, is_deleted, source_type, updated_at
            FROM collection_account_queue
            {where}
            ORDER BY position ASC, id ASC
            """
        ).fetchall()
        return [
            {
                **dict(row),
                "enabled": bool(row["enabled"]),
                "is_deleted": bool(row["is_deleted"]),
                "account_url": f"https://www.instagram.com/{row['username']}/",
            }
            for row in rows
        ]
    finally:
        conn.close()


def replace_collection_accounts(db_path: str | Path, items: list[object]) -> dict[str, Any]:
    """Replace the SIDE collection queue with a username-only ordered list."""
    normalized_by_username: dict[str, dict[str, Any]] = {}
    for raw_item in list(items or []):
        item = raw_item if isinstance(raw_item, Mapping) else {"username": raw_item}
        username = normalize_collection_username(
            item.get("username") or item.get("account") or item.get("account_url") or item.get("url")
        )
        if not username:
            continue
        deleted = 1 if bool(item.get("is_deleted") or item.get("deleted")) else 0
        enabled = 0 if deleted else (1 if bool(item.get("enabled", True)) else 0)
        previous = normalized_by_username.get(username)
        if previous is not None:
            previous["is_deleted"] = 1 if previous["is_deleted"] or deleted else 0
            previous["enabled"] = 0 if previous["is_deleted"] else (1 if previous["enabled"] and enabled else 0)
            continue
        normalized_by_username[username] = {
            "username": username,
            "position": 0,
            "enabled": enabled,
            "is_deleted": deleted,
        }

    # Preserve the exact user-visible order received from Side. Sorting is an
    # explicit Side action; a plain save must never reorder the queue.
    normalized = list(normalized_by_username.values())
    seen = set(normalized_by_username)
    for position, item in enumerate(normalized, start=1):
        item["position"] = position

    conn = connect_database(db_path)
    try:
        conn.execute("BEGIN IMMEDIATE")
        deleted_names = {
            str(row["entity_key"] or "").strip().lower()
            for row in conn.execute(
                "SELECT entity_key FROM deletion_events WHERE entity_type='ACCOUNT'"
            ).fetchall()
        }
        for item in normalized:
            if item["username"] in deleted_names:
                item["is_deleted"] = 1
                item["enabled"] = 0
            conn.execute(
                """
                INSERT INTO collection_account_queue(username, position, enabled, is_deleted, source_type)
                VALUES (?, ?, ?, ?, 'SIDE')
                ON CONFLICT(username) DO UPDATE SET
                    position=excluded.position,
                    enabled=excluded.enabled,
                    is_deleted=excluded.is_deleted,
                    source_type='SIDE',
                    updated_at=CURRENT_TIMESTAMP
                """,
                (item["username"], item["position"], item["enabled"], item["is_deleted"]),
            )
        if seen:
            placeholders = ",".join("?" for _ in seen)
            conn.execute(
                f"DELETE FROM collection_account_queue WHERE source_type='SIDE' AND username NOT IN ({placeholders})",
                tuple(sorted(seen)),
            )
        else:
            conn.execute("DELETE FROM collection_account_queue WHERE source_type='SIDE'")
        payload = {
            "items": [
                {
                    "username": item["username"],
                    "position": item["position"],
                    "enabled": bool(item["enabled"]),
                    "is_deleted": bool(item["is_deleted"]),
                }
                for item in normalized
            ]
        }
        conn.execute(
            "INSERT INTO sync_queue(entity_type, operation, payload_json) VALUES('COLLECTION_ACCOUNT_QUEUE', 'REPLACE', ?)",
            (json.dumps(payload, ensure_ascii=False, sort_keys=True),),
        )
        conn.commit()
        return {"items": list_collection_accounts(db_path), "total": len(normalized)}
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def soft_delete_account(db_path: str | Path, account_id: int, reason: str = "user_delete") -> dict[str, Any]:
    conn = connect_database(db_path)
    try:
        conn.execute("BEGIN IMMEDIATE")
        row = conn.execute("SELECT id, username FROM social_accounts WHERE id=?", (int(account_id),)).fetchone()
        if not row:
            raise ValueError(f"account_id not found: {account_id}")
        conn.execute("UPDATE social_accounts SET active=0, deleted_at=CURRENT_TIMESTAMP, updated_at=CURRENT_TIMESTAMP WHERE id=?", (int(account_id),))
        conn.execute("UPDATE account_collection_rules SET enabled=0, updated_at=CURRENT_TIMESTAMP WHERE account_id=?", (int(account_id),))
        conn.execute("UPDATE collection_account_queue SET enabled=0, is_deleted=1, updated_at=CURRENT_TIMESTAMP WHERE username=? COLLATE NOCASE", (str(row["username"]),))
        conn.execute("UPDATE posts SET deleted_at=COALESCE(deleted_at,CURRENT_TIMESTAMP), updated_at=CURRENT_TIMESTAMP WHERE account_id=?", (int(account_id),))
        conn.execute(
            """INSERT INTO deletion_events(entity_type, entity_local_id, entity_key, reason)
               VALUES('ACCOUNT', ?, ?, ?)
               ON CONFLICT(entity_type, entity_local_id) DO UPDATE SET
                   entity_key=excluded.entity_key, reason=excluded.reason, created_at=CURRENT_TIMESTAMP, synced_at=NULL""",
            (int(account_id), str(row["username"]), str(reason or "user_delete")),
        )
        payload = json.dumps({"account_id": int(account_id), "username": str(row["username"]), "deleted": True}, ensure_ascii=False, sort_keys=True)
        conn.execute("INSERT INTO sync_queue(entity_type, entity_local_id, operation, payload_json) VALUES('ACCOUNT', ?, 'DELETE', ?)", (int(account_id), payload))
        conn.commit()
        return {"account_id": int(account_id), "username": str(row["username"]), "deleted": True}
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def soft_delete_post(db_path: str | Path, post_id: int, reason: str = "user_delete") -> dict[str, Any]:
    conn = connect_database(db_path)
    try:
        conn.execute("BEGIN IMMEDIATE")
        row = conn.execute("SELECT id, shortcode FROM posts WHERE id=?", (int(post_id),)).fetchone()
        if not row:
            raise ValueError(f"post_id not found: {post_id}")
        conn.execute("UPDATE posts SET deleted_at=CURRENT_TIMESTAMP, updated_at=CURRENT_TIMESTAMP WHERE id=?", (int(post_id),))
        conn.execute(
            """INSERT INTO deletion_events(entity_type, entity_local_id, entity_key, reason)
               VALUES('POST', ?, ?, ?)
               ON CONFLICT(entity_type, entity_local_id) DO UPDATE SET
                   entity_key=excluded.entity_key, reason=excluded.reason, created_at=CURRENT_TIMESTAMP, synced_at=NULL""",
            (int(post_id), str(row["shortcode"]), str(reason or "user_delete")),
        )
        payload = json.dumps({"post_id": int(post_id), "shortcode": str(row["shortcode"]), "deleted": True}, ensure_ascii=False, sort_keys=True)
        conn.execute("INSERT INTO sync_queue(entity_type, entity_local_id, operation, payload_json) VALUES('POST', ?, 'DELETE', ?)", (int(post_id), payload))
        conn.commit()
        return {"post_id": int(post_id), "shortcode": str(row["shortcode"]), "deleted": True}
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def list_deleted_accounts(db_path: str | Path) -> list[dict[str, Any]]:
    conn = connect_database(db_path)
    try:
        rows = conn.execute(
            """SELECT entity_local_id AS account_id, entity_key AS username, created_at
               FROM deletion_events WHERE entity_type='ACCOUNT' ORDER BY created_at DESC"""
        ).fetchall()
        return [dict(row) for row in rows]
    finally:
        conn.close()

def main() -> int:
    parser = argparse.ArgumentParser(description="Initialize sort local SQLite storage.")
    parser.add_argument("--st-root", required=True, help="Absolute path to the _St directory")
    parser.add_argument("--db", help="SQLite path. Defaults to _St/Shared/data/sort_local.sqlite3")
    parser.add_argument("--no-create-storage-dirs", action="store_true")
    args = parser.parse_args()

    st_root = Path(args.st_root)
    db_path = Path(args.db) if args.db else default_database_path(st_root)
    try:
        result = initialize_database(
            db_path,
            st_root,
            create_storage_dirs=not args.no_create_storage_dirs,
        )
    except (OSError, sqlite3.Error, ValueError) as exc:
        print(json.dumps({"overall": "FAIL", "error": str(exc)}, ensure_ascii=False))
        return 1

    print(json.dumps({"overall": "OK", **result}, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())
