#!/usr/bin/env python3
"""Exercise synchronous video deletion and the live recycle-bin workflow."""

from __future__ import annotations

import argparse
import hashlib
import json
import time
from pathlib import Path
from typing import Any

import httpx


def call(client: httpx.Client, method: str, path: str, **kwargs: Any) -> httpx.Response:
    response = client.request(method, path, **kwargs)
    if response.is_error:
        raise RuntimeError(f"{method} {path} -> {response.status_code}: {response.text[:1000]}")
    return response


def digest(path: Path) -> str:
    value = hashlib.sha256()
    with path.open("rb") as handle:
        while block := handle.read(1024 * 1024):
            value.update(block)
    return value.hexdigest()


def videos(client: httpx.Client) -> list[dict[str, Any]]:
    return call(client, "GET", "/api/v1/videos?limit=500").json()


def find_video(client: httpx.Client, source_id: str, filename: str) -> dict[str, Any] | None:
    return next(
        (
            item
            for item in videos(client)
            if item.get("source_id") == source_id
            and item.get("source_key", "").rsplit("/", 1)[-1] == filename
        ),
        None,
    )


def upload(
    client: httpx.Client,
    fixture: Path,
    source_id: str,
    filename: str,
    title: str,
) -> dict[str, Any]:
    visible = find_video(client, source_id, filename)
    if visible:
        return visible
    existing = call(client, "GET", "/api/v1/uploads?limit=500").json()
    task = next(
        (
            item
            for item in existing
            if item.get("filename") == filename
            and item.get("status") in {"receiving", "queued", "transferring", "indexing"}
        ),
        None,
    )
    if task is None:
        task = call(
            client,
            "POST",
            "/api/v1/uploads",
            json={
                "source_id": source_id,
                "relative_path": "ingest",
                "filename": filename,
                "title": title,
                "size_bytes": fixture.stat().st_size,
                "sha256": digest(fixture),
                "conflict": "skip",
            },
        ).json()
    upload_id = task["id"]
    if task.get("status") != "completed":
        chunk_size = int(task["chunk_size"])
        received = {int(value) for value in task.get("received_chunks", task.get("received", []))}
        with fixture.open("rb") as handle:
            for index in range(int(task["total_chunks"])):
                block = handle.read(chunk_size)
                if index in received:
                    continue
                call(
                    client,
                    "PUT",
                    f"/api/v1/uploads/{upload_id}/chunks/{index}",
                    content=block,
                    headers={"X-Chunk-SHA256": hashlib.sha256(block).hexdigest()},
                )
        call(client, "POST", f"/api/v1/uploads/{upload_id}/complete")
    deadline = time.monotonic() + 180
    while time.monotonic() < deadline:
        tasks = call(client, "GET", "/api/v1/uploads?limit=500").json()
        task = next(item for item in tasks if item["id"] == upload_id)
        if task["status"] == "completed":
            break
        if task["status"] in {"failed", "cancelled"}:
            raise AssertionError(f"upload failed: {task.get('message') or task.get('error')}")
        time.sleep(1)
    else:
        raise TimeoutError(f"upload did not complete: {filename}")
    deadline = time.monotonic() + 60
    while time.monotonic() < deadline:
        video = find_video(client, source_id, filename)
        if video:
            return video
        time.sleep(1)
    raise TimeoutError(f"completed upload did not appear in /videos: {filename}")


def purge_source_key(client: httpx.Client, source_id: str, key: str) -> None:
    try:
        result = call(
            client,
            "POST",
            "/api/v1/files/trash",
            json={"source_id": source_id, "keys": [key]},
        ).json()
    except RuntimeError as exc:
        if "404" in str(exc):
            return
        raise
    for trash_id in result.get("ids", []):
        call(client, "DELETE", f"/api/v1/trash/{trash_id}")


def search_ids(client: httpx.Client, text: str) -> list[str]:
    result = call(
        client,
        "POST",
        "/api/v1/search",
        json={"text": text, "recognition_types": ["metadata"]},
    ).json()
    return [item["video_id"] for item in result.get("items", [])]


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--base-url", required=True)
    auth = parser.add_mutually_exclusive_group(required=True)
    auth.add_argument("--token-file", type=Path)
    auth.add_argument("--password")
    parser.add_argument("--run-dir", type=Path, required=True)
    args = parser.parse_args()

    state = json.loads((args.run_dir / "state.json").read_text())
    headers = {}
    if args.token_file is not None:
        headers["Authorization"] = f"Bearer {args.token_file.read_text().strip()}"
    client = httpx.Client(
        base_url=args.base_url.rstrip("/"),
        headers=headers,
        timeout=httpx.Timeout(60, connect=10),
    )
    if args.password is not None:
        login = call(
            client,
            "POST",
            "/api/v1/auth/login",
            json={"password": args.password, "remember_device": False},
        ).json()
        csrf = str(login.get("csrf_token") or "")
        if not csrf:
            raise AssertionError("login response omitted CSRF token")
        client.headers["X-CSRF-Token"] = csrf
    source_id = state["source_id"]
    run_id = state["run_id"]
    cases = [
        {
            "mode": "trash",
            "fixture": args.run_dir / "e2e-positive.mp4",
            "filename": f"{run_id}-delete-trash.mp4",
            "title": "删除回收验收 481",
            "key": f"ingest/{run_id}-delete-trash.mp4",
        },
        {
            "mode": "tombstone",
            "fixture": args.run_dir / "e2e-negative.mp4",
            "filename": f"{run_id}-delete-only.mp4",
            "title": "仅移出验收 592",
            "key": f"ingest/{run_id}-delete-only.mp4",
        },
    ]
    report: dict[str, Any] = {}
    try:
        trash_case = cases[0]
        first = upload(
            client,
            trash_case["fixture"],
            source_id,
            trash_case["filename"],
            trash_case["title"],
        )
        deleted = call(client, "DELETE", f"/api/v1/videos/{first['id']}?delete_source=true").json()
        if not deleted.get("source_deleted") or not deleted.get("trash_id"):
            raise AssertionError(f"source deletion result mismatch: {deleted}")
        if find_video(client, source_id, trash_case["filename"]):
            raise AssertionError("source-deleted video remained visible")
        if first["id"] in search_ids(client, "删除回收验收 481"):
            raise AssertionError("source-deleted video remained in metadata search")
        trash_rows = call(client, "GET", f"/api/v1/trash?source_id={source_id}").json()
        trash_item = next((item for item in trash_rows if item["id"] == deleted["trash_id"]), None)
        if not trash_item:
            raise AssertionError("deleted source was not listed in recycle bin")
        restored = call(client, "POST", f"/api/v1/trash/{trash_item['id']}/restore", json={}).json()
        deadline = time.monotonic() + 90
        restored_video = None
        while time.monotonic() < deadline:
            restored_video = find_video(client, source_id, trash_case["filename"])
            if restored_video:
                break
            time.sleep(1)
        if not restored_video:
            raise AssertionError("restored source did not return to /videos")
        removed_again = call(
            client,
            "DELETE",
            f"/api/v1/videos/{restored_video['id']}?delete_source=true",
        ).json()
        call(client, "DELETE", f"/api/v1/trash/{removed_again['trash_id']}")
        report["trash"] = {
            "deleted": deleted,
            "restored_key": restored["key"],
            "restored_video_id": restored_video["id"],
            "purged": True,
        }

        tombstone_case = cases[1]
        second = upload(
            client,
            tombstone_case["fixture"],
            source_id,
            tombstone_case["filename"],
            tombstone_case["title"],
        )
        removed = call(client, "DELETE", f"/api/v1/videos/{second['id']}?delete_source=false").json()
        if removed.get("source_deleted") or removed.get("trash_id"):
            raise AssertionError(f"library-only deletion result mismatch: {removed}")
        scan = call(client, "POST", f"/api/v1/sources/{source_id}/scan", json={}).json()
        time.sleep(5)
        if find_video(client, source_id, tombstone_case["filename"]):
            raise AssertionError("tombstoned video reappeared after source scan")
        if second["id"] in search_ids(client, "仅移出验收 592"):
            raise AssertionError("tombstoned video remained in metadata search")
        purge_source_key(client, source_id, tombstone_case["key"])
        report["tombstone"] = {
            "deleted": removed,
            "scan_job_id": scan.get("job_id"),
            "remained_hidden_after_scan": True,
            "source_cleaned": True,
        }
    finally:
        for case in cases:
            video = find_video(client, source_id, case["filename"])
            if video:
                try:
                    deletion = call(
                        client,
                        "DELETE",
                        f"/api/v1/videos/{video['id']}?delete_source=true",
                    ).json()
                    if deletion.get("trash_id"):
                        call(client, "DELETE", f"/api/v1/trash/{deletion['trash_id']}")
                except Exception:
                    pass
            else:
                try:
                    purge_source_key(client, source_id, case["key"])
                except Exception:
                    pass
    (args.run_dir / "delete-report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2) + "\n"
    )
    print(json.dumps(report, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()