David Freitag
ea7d420975
- Implemented a new beets plugin to import and manage music videos, supporting various video formats and providing metadata from IMVDB and MusicBrainz for autotagging. - Added installation instructions and configuration options in README.md. - Created IMVDBApi client for interacting with the IMVDB API. - Defined typing for various API responses and utility functions for mapping MusicBrainz data to beets TrackInfo. - Included desktop entry and JSON info files for a video titled "[SAST] The Evolution of Search-based Software Testing". - Added utility functions for handling artist credits and related metadata. - Introduced a grabbed.txt file for tracking video sources.
95 lines
3.3 KiB
Python
95 lines
3.3 KiB
Python
from typing import Any
|
|
from lapidary.runtime import (
|
|
Body,
|
|
ClientBase,
|
|
Query,
|
|
Response,
|
|
Responses,
|
|
UnexpectedResponse,
|
|
get,
|
|
)
|
|
from typing import Annotated, Any, Literal, Sequence, cast, Self
|
|
from httpx_auth import HeaderApiKey
|
|
|
|
from beetsplug.beets_music_videos.types import SearchResponseType, VideoType
|
|
from beets import logging
|
|
|
|
log = logging.getLogger("beets")
|
|
|
|
|
|
class IMVDBApi(ClientBase): # type: ignore
|
|
"""Lapidary-based IMVDB client. Use api_key in security; do not hardcode."""
|
|
|
|
def __init__(
|
|
self,
|
|
api_key: str,
|
|
base_url: str = "https://imvdb.com/api/v1",
|
|
**kwargs: Any,
|
|
) -> None: # type: ignore
|
|
# Some APIs return 500/502 for non-browser User-Agent; send a browser-like one.
|
|
headers = dict(kwargs.pop("headers", {}))
|
|
headers.setdefault(
|
|
"User-Agent",
|
|
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
|
)
|
|
super().__init__( # type: ignore
|
|
base_url=base_url,
|
|
security=[{"api_key": [api_key]}],
|
|
headers=headers,
|
|
**kwargs,
|
|
)
|
|
# Send API key as header IMVDB-APP-KEY on every request.
|
|
self.lapidary_authenticate(
|
|
api_key=HeaderApiKey(api_key, header_name="IMVDB-APP-KEY"),
|
|
)
|
|
|
|
@get("/search/videos") # type: ignore[misc]
|
|
def video_search(
|
|
self: Self,
|
|
q: Annotated[str, Query()],
|
|
) -> Annotated[
|
|
SearchResponseType,
|
|
Responses(
|
|
responses={
|
|
"2xx": Response(body=Body({"application/json": SearchResponseType})),
|
|
}
|
|
),
|
|
]:
|
|
"""Search IMVDB; returns response with .results (list of VideoType).
|
|
|
|
Implemented at runtime by Lapidary; this stub is never executed.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
# def sync_video_search(self, query: str) -> list[VideoType]:
|
|
# """Sync wrapper used by the plugin. Runs video_search and returns .results."""
|
|
# import asyncio
|
|
# # Reuse a single event loop to avoid "Event loop is closed" on subsequent calls.
|
|
# try:
|
|
# loop = asyncio.get_event_loop()
|
|
# if loop.is_closed():
|
|
# loop = asyncio.new_event_loop()
|
|
# asyncio.set_event_loop(loop)
|
|
# except RuntimeError:
|
|
# loop = asyncio.new_event_loop()
|
|
# asyncio.set_event_loop(loop)
|
|
# try:
|
|
# resp = loop.run_until_complete(self.video_search(q=query))
|
|
# return resp["results"]
|
|
# except UnexpectedResponse as e:
|
|
# # Lapidary raises UnexpectedResponse when body validation fails
|
|
# # (e.g. API shape differs from SearchResponseType). For 200, parse
|
|
# # body and return results.
|
|
# if e.response.status_code == 200:
|
|
# try:
|
|
# body = e.response.json()
|
|
# if isinstance(body, dict):
|
|
# return cast(list[VideoType], body["results"])
|
|
# except Exception:
|
|
# pass
|
|
# log.debug("imvdb search failed for %r: %s", query[:50], e)
|
|
# return []
|
|
# except Exception as e:
|
|
# log.debug("imvdb search failed for %r: %s", query[:50], e)
|
|
# return []
|