| 1 | """Openverse provider. |
| 2 | |
| 3 | Zero-config (no API key required). Indexes openly licensed images across |
| 4 | Wikimedia, Flickr, museums, and other sources. |
| 5 | |
| 6 | API docs: https://api.openverse.org/v1/ |
| 7 | """ |
| 8 | |
| 9 | from __future__ import annotations |
| 10 | |
| 11 | import sys |
| 12 | from pathlib import Path |
| 13 | from urllib.parse import quote, unquote, urlparse |
| 14 | |
| 15 | _SCRIPTS_DIR = Path(__file__).resolve().parents[1] |
| 16 | if str(_SCRIPTS_DIR) not in sys.path: |
| 17 | sys.path.insert(0, str(_SCRIPTS_DIR)) |
| 18 | |
| 19 | from console_encoding import configure_utf8_stdio # noqa: E402 |
| 20 | |
| 21 | configure_utf8_stdio() |
| 22 | |
| 23 | if __name__ == "__main__": |
| 24 | print(__doc__) |
| 25 | print("Use via: python3 skills/ppt-master/scripts/image_search.py ...") |
| 26 | raise SystemExit(0 if any(arg in {"-h", "--help", "help"} for arg in sys.argv[1:]) else 1) |
| 27 | |
| 28 | import requests |
| 29 | |
| 30 | from image_sources.provider_common import ( |
| 31 | AssetCandidate, |
| 32 | ImageSearchRequest, |
| 33 | USER_AGENT, |
| 34 | build_query_progression, |
| 35 | classify_license, |
| 36 | normalize_license_name, |
| 37 | ) |
| 38 | |
| 39 | |
| 40 | API_URL = "https://api.openverse.org/v1/images/" |
| 41 | DEFAULT_PAGE_SIZE = 20 |
| 42 | DEFAULT_TIMEOUT = 30 |
| 43 | |
| 44 | # Map our orientation vocabulary to Openverse's ``aspect_ratio`` parameter. |
| 45 | _ASPECT_MAP = {"landscape": "wide", "portrait": "tall", "square": "square"} |
| 46 | |
| 47 | # Openverse license param values. ``cc0,pdm`` covers our "no-attribution" tier; |
| 48 | # adding ``by,by-sa`` opens the "attribution-required" tier. |
| 49 | _LICENSE_PARAM = { |
| 50 | "no-attribution-only": "cc0,pdm", |
| 51 | "all": "by,by-sa,cc0,pdm", |
| 52 | } |
| 53 | |
| 54 | |
| 55 | def _preview_url(item: dict, download_url: str) -> str: |
| 56 | """Prefer Wikimedia's bounded preview over Openverse's fragile proxy.""" |
| 57 | source = str(item.get("source") or item.get("provider") or "").lower() |
| 58 | parsed = urlparse(download_url) |
| 59 | if source == "wikimedia" and parsed.netloc == "upload.wikimedia.org": |
| 60 | filename = unquote(parsed.path.rsplit("/", 1)[-1]) |
| 61 | if filename: |
| 62 | return ( |
| 63 | "https://commons.wikimedia.org/wiki/Special:Redirect/file/" |
| 64 | f"{quote(filename, safe='')}?width=1024" |
| 65 | ) |
| 66 | return (item.get("thumbnail") or "").strip() |
| 67 | |
| 68 | |
| 69 | def parse_results(payload: dict) -> list[AssetCandidate]: |
| 70 | """Translate an Openverse search payload into a list of candidates.""" |
| 71 | candidates: list[AssetCandidate] = [] |
| 72 | for item in payload.get("results", []) or []: |
| 73 | license_name = (item.get("license") or "").strip() |
| 74 | license_url = (item.get("license_url") or "").strip() |
| 75 | tier = classify_license(license_name, license_url, provider="openverse") |
| 76 | if not tier: |
| 77 | continue |
| 78 | |
| 79 | download_url = (item.get("url") or item.get("thumbnail") or "").strip() |
| 80 | if not download_url: |
| 81 | continue |
| 82 | |
| 83 | candidates.append( |
| 84 | AssetCandidate( |
| 85 | provider="openverse", |
| 86 | title=(item.get("title") or "").strip() or "Untitled", |
| 87 | asset_id=str(item.get("id") or ""), |
| 88 | source_page_url=( |
| 89 | item.get("foreign_landing_url") or item.get("detail_url") or "" |
| 90 | ).strip(), |
| 91 | license_name=normalize_license_name(license_name), |
| 92 | license_url=license_url, |
| 93 | license_tier=tier, |
| 94 | width=int(item.get("width") or 0), |
| 95 | height=int(item.get("height") or 0), |
| 96 | download_url=download_url, |
| 97 | preview_url=_preview_url(item, download_url), |
| 98 | author=(item.get("creator") or "").strip(), |
| 99 | raw=item, |
| 100 | ) |
| 101 | ) |
| 102 | return candidates |
| 103 | |
| 104 | |
| 105 | def search( |
| 106 | request: ImageSearchRequest, |
| 107 | *, |
| 108 | license_tier_filter: str = "no-attribution-only", |
| 109 | page_size: int = DEFAULT_PAGE_SIZE, |
| 110 | timeout: int = DEFAULT_TIMEOUT, |
| 111 | ) -> list[AssetCandidate]: |
| 112 | """Search Openverse for candidates matching ``request``. |
| 113 | |
| 114 | ``license_tier_filter`` is one of ``"no-attribution-only"`` or ``"all"``. |
| 115 | Returns the candidates from the first non-empty query in the |
| 116 | progression — caller is responsible for picking the best one via |
| 117 | ``score_candidate``. |
| 118 | """ |
| 119 | if license_tier_filter not in _LICENSE_PARAM: |
| 120 | raise ValueError(f"unsupported license_tier_filter: {license_tier_filter!r}") |
| 121 | |
| 122 | orientation = (request.orientation or "").strip().lower() |
| 123 | |
| 124 | for query in build_query_progression(request.query): |
| 125 | params: dict[str, str | int] = { |
| 126 | "q": query, |
| 127 | "page_size": page_size, |
| 128 | "license": _LICENSE_PARAM[license_tier_filter], |
| 129 | "size": "large", |
| 130 | } |
| 131 | if orientation in _ASPECT_MAP: |
| 132 | params["aspect_ratio"] = _ASPECT_MAP[orientation] |
| 133 | |
| 134 | response = requests.get( |
| 135 | API_URL, |
| 136 | params=params, |
| 137 | headers={"User-Agent": USER_AGENT, "Accept": "application/json"}, |
| 138 | timeout=timeout, |
| 139 | ) |
| 140 | response.raise_for_status() |
| 141 | candidates = parse_results(response.json()) |
| 142 | if candidates: |
| 143 | return candidates |
| 144 | |
| 145 | return [] |
| 146 |