117 lines
4.7 KiB
Python
117 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Remove narrowly prefixed records created by live acceptance runs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
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 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("--prefix", required=True)
|
|
parser.add_argument("--keep", action="append", default=[])
|
|
parser.add_argument("--report", type=Path)
|
|
args = parser.parse_args()
|
|
if len(args.prefix) < 12 or not args.prefix.startswith("imagefind-e2e-"):
|
|
raise ValueError("cleanup prefix must be a narrowly scoped imagefind-e2e-* value")
|
|
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(120, 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
|
|
keep = set(args.keep)
|
|
report: dict[str, Any] = {
|
|
"prefix": args.prefix,
|
|
"videos": [],
|
|
"collections": [],
|
|
"tags": [],
|
|
"tag_groups": [],
|
|
"people_hidden": [],
|
|
}
|
|
|
|
rows = call(client, "GET", "/api/v1/videos?limit=500").json()
|
|
for video in rows:
|
|
filename = str(video.get("source_key") or "").rsplit("/", 1)[-1]
|
|
if args.prefix not in str(video.get("source_key") or "") or filename in keep:
|
|
continue
|
|
deleted = call(client, "DELETE", f"/api/v1/videos/{video['id']}?delete_source=true").json()
|
|
if trash_id := deleted.get("trash_id"):
|
|
call(client, "DELETE", f"/api/v1/trash/{trash_id}")
|
|
report["videos"].append({"id": video["id"], "filename": filename, **deleted})
|
|
|
|
for collection in call(client, "GET", "/api/v1/collections").json():
|
|
if not str(collection.get("name") or "").startswith(args.prefix):
|
|
continue
|
|
detail = call(client, "GET", f"/api/v1/collections/{collection['id']}").json()
|
|
if detail.get("video_count"):
|
|
continue
|
|
result = call(client, "DELETE", f"/api/v1/collections/{collection['id']}").json()
|
|
report["collections"].append(result)
|
|
|
|
for group in call(client, "GET", "/api/v1/tag-groups").json():
|
|
if not str(group.get("name") or "").startswith(args.prefix):
|
|
continue
|
|
tags = call(client, "GET", f"/api/v1/tags?group_id={group['id']}").json()
|
|
if any(int(tag.get("video_count") or 0) for tag in tags):
|
|
continue
|
|
for tag in tags:
|
|
call(client, "DELETE", f"/api/v1/tags/{tag['id']}")
|
|
report["tags"].append(tag["id"])
|
|
call(client, "DELETE", f"/api/v1/tag-groups/{group['id']}")
|
|
report["tag_groups"].append(group["id"])
|
|
|
|
for source in call(client, "GET", "/api/v1/sources").json():
|
|
for item in call(client, "GET", f"/api/v1/trash?source_id={source['id']}").json():
|
|
if args.prefix in str(item.get("display_name") or ""):
|
|
call(client, "DELETE", f"/api/v1/trash/{item['id']}")
|
|
|
|
# Named people are deliberately preserved when their last face is removed.
|
|
# There is no destructive people endpoint, so hide narrowly prefixed test
|
|
# identities after their fixture videos have been deleted.
|
|
for person in call(client, "GET", "/api/v1/people").json():
|
|
name = str(person.get("name") or "")
|
|
if not name.startswith(args.prefix) or person.get("hidden"):
|
|
continue
|
|
call(client, "PATCH", f"/api/v1/people/{person['id']}", json={"name": name, "hidden": True})
|
|
report["people_hidden"].append(person["id"])
|
|
|
|
encoded = json.dumps(report, ensure_ascii=False, indent=2) + "\n"
|
|
if args.report:
|
|
args.report.parent.mkdir(parents=True, exist_ok=True)
|
|
args.report.write_text(encoded)
|
|
print(encoded, end="")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|