返回 ppt-master
provider_common.py
根目录 / skills / ppt-master / scripts / image_sources / provider_common.py
1 """Shared primitives for web image providers.
2
3 This module is the single home for everything that all four providers
4 (Openverse / Wikimedia / Pexels / Pixabay) need:
5
6 - License tier classification (the central abstraction of this module)
7 - Search request / asset candidate dataclasses
8 - Query simplification for keyword-based image APIs
9 - Candidate scoring
10 - Attribution text builder
11 - Small helpers (orientation, json path, etc.)
12
13 Provider-specific code (API URLs, payload shape, parse_results) lives in
14 the corresponding provider_<name>.py module and only imports from here.
15 """
16
17 from __future__ import annotations
18
19 import sys
20 from pathlib import Path
21
22 _SCRIPTS_DIR = Path(__file__).resolve().parents[1]
23 if str(_SCRIPTS_DIR) not in sys.path:
24 sys.path.insert(0, str(_SCRIPTS_DIR))
25
26 from console_encoding import configure_utf8_stdio # noqa: E402
27
28 configure_utf8_stdio()
29
30 if __name__ == "__main__":
31 print(__doc__)
32 print("This is an internal helper module used by image_search.py and the four web image providers.")
33 raise SystemExit(0 if any(arg in {"-h", "--help", "help"} for arg in sys.argv[1:]) else 1)
34
35 import re
36 from dataclasses import dataclass, field, replace
37 from typing import Any, Optional
38
39
40 # ---------------------------------------------------------------------------
41 # Project-wide constants
42 # ---------------------------------------------------------------------------
43
44 USER_AGENT = "PPTMaster/1.0 (https://github.com/hugohe3/ppt-master)"
45
46
47 # ---------------------------------------------------------------------------
48 # License tier classification
49 # ---------------------------------------------------------------------------
50 #
51 # Every accepted candidate is classified into exactly one of two tiers:
52 #
53 # "no-attribution" -> No on-slide credit needed (CC0, PD, Pexels,
54 # Pixabay). Default search target.
55 # "attribution-required" -> CC BY / CC BY-SA. Executor must add an
56 # inline credit text element on the slide.
57 #
58 # Anything else (CC BY-NC, CC BY-ND, all-rights-reserved, unknown) returns
59 # None and the candidate is rejected outright.
60
61 LICENSE_TIER_NO_ATTRIBUTION = "no-attribution"
62 LICENSE_TIER_ATTRIBUTION_REQUIRED = "attribution-required"
63
64 # Tokens that mark a license as "no attribution required".
65 NO_ATTRIBUTION_TOKENS: tuple[str, ...] = (
66 "cc0",
67 "public domain",
68 "publicdomain",
69 "creativecommons.org/publicdomain/",
70 "pexels license",
71 "pixabay content license",
72 "pixabay license",
73 )
74
75 # Tokens that mark a license as "attribution required".
76 ATTRIBUTION_REQUIRED_TOKENS: tuple[str, ...] = (
77 "cc by",
78 "cc-by",
79 "by-sa",
80 "by sa",
81 "creativecommons.org/licenses/by/",
82 "creativecommons.org/licenses/by-sa/",
83 )
84
85 # Tokens that disqualify a candidate entirely.
86 REJECTED_TOKENS: tuple[str, ...] = (
87 "by-nc",
88 "by nc",
89 "noncommercial",
90 "non-commercial",
91 "by-nd",
92 "by nd",
93 "no derivatives",
94 "noderivatives",
95 "all rights reserved",
96 )
97
98
99 # Canonical display forms for license names. Different providers report
100 # the same license with different capitalization (Openverse: "cc0",
101 # Wikimedia: "Public domain"); the Executor renders these as on-slide
102 # text, so a normalized form prevents inconsistent credits.
103 _LICENSE_NAME_CANON: dict[str, str] = {
104 "cc0": "CC0",
105 "cc 0": "CC0",
106 "public domain": "Public Domain",
107 "publicdomain": "Public Domain",
108 "pdm": "Public Domain",
109 "pexels license": "Pexels License",
110 "pixabay content license": "Pixabay Content License",
111 "pixabay license": "Pixabay Content License",
112 }
113
114 # CC license short-name pattern used to canonicalize "cc by 4.0" → "CC BY 4.0".
115 _CC_PATTERN = re.compile(
116 r"^\s*cc[\s-]+(by(?:[\s-]+(?:sa|nc|nd))*)\s*([0-9.]*)\s*$",
117 re.IGNORECASE,
118 )
119
120
121 def normalize_license_name(name: str) -> str:
122 """Return a canonical display form for a license name.
123
124 Maps common aliases to a consistent capitalization so the on-slide
125 credit text written by the Executor is uniform across providers.
126 Unknown inputs are returned trimmed but otherwise unchanged.
127 """
128 if not name:
129 return ""
130 key = name.strip().lower()
131 if not key:
132 return ""
133
134 if key in _LICENSE_NAME_CANON:
135 return _LICENSE_NAME_CANON[key]
136
137 cc_match = _CC_PATTERN.match(key)
138 if cc_match:
139 suffix_raw, version = cc_match.group(1), cc_match.group(2)
140 suffix = suffix_raw.replace(" ", "-").upper()
141 return f"CC {suffix} {version}".strip()
142
143 return name.strip()
144
145
146 def classify_license(
147 license_name: str,
148 license_url: str = "",
149 provider: str = "",
150 ) -> Optional[str]:
151 """Classify a license string into one of the two tiers, or reject it.
152
153 Returns:
154 ``"no-attribution"`` / ``"attribution-required"`` / ``None``.
155
156 The provider hint lets us treat Pexels and Pixabay's own licenses as
157 ``no-attribution`` even when the upstream API only returns a short
158 label like ``"Pexels"``.
159 """
160 text = " ".join(
161 part.strip().lower()
162 for part in (license_name or "", license_url or "")
163 if part
164 )
165 provider_key = (provider or "").strip().lower()
166
167 if not text and not provider_key:
168 return None
169
170 if any(token in text for token in REJECTED_TOKENS):
171 return None
172
173 if any(token in text for token in NO_ATTRIBUTION_TOKENS):
174 return LICENSE_TIER_NO_ATTRIBUTION
175
176 # Provider-default fallback: pexels / pixabay items often arrive with a
177 # bare "Pexels" / "Pixabay" license string. Their site-wide license is
178 # "free for commercial use, no attribution required".
179 #
180 # Guard: require the license text to actually mention the provider name,
181 # so an empty / missing license field never silently passes as no-attribution.
182 if (
183 provider_key in {"pexels", "pixabay"}
184 and provider_key in text
185 and not any(token in text for token in ATTRIBUTION_REQUIRED_TOKENS)
186 ):
187 return LICENSE_TIER_NO_ATTRIBUTION
188
189 if any(token in text for token in ATTRIBUTION_REQUIRED_TOKENS):
190 return LICENSE_TIER_ATTRIBUTION_REQUIRED
191
192 return None # unknown license -> reject
193
194
195 # ---------------------------------------------------------------------------
196 # Dataclasses
197 # ---------------------------------------------------------------------------
198
199
200 @dataclass
201 class ImageSearchRequest:
202 """A single image search intent passed to a provider."""
203
204 query: str
205 purpose: str = ""
206 orientation: str = "" # "landscape" / "portrait" / "square" / ""
207 min_width: int = 0
208 min_height: int = 0
209 filename: str = ""
210 slide: str = ""
211 required_terms: tuple[str, ...] = ()
212 query_variants: tuple[str, ...] = ()
213
214
215 @dataclass
216 class AssetCandidate:
217 """One ranked candidate returned by a provider's parse_results."""
218
219 provider: str
220 title: str
221 asset_id: str = ""
222 source_page_url: str = ""
223 license_name: str = ""
224 license_url: str = ""
225 license_tier: str = "" # one of LICENSE_TIER_* constants
226 width: int = 0
227 height: int = 0
228 download_url: str = ""
229 preview_url: str = ""
230 discovery_query: str = ""
231 author: str = ""
232 raw: Any = field(default=None)
233
234
235 # ---------------------------------------------------------------------------
236 # Query simplification
237 # ---------------------------------------------------------------------------
238 #
239 # Web image APIs do keyword matching against image metadata, not semantic
240 # search. Long, descriptive queries with brand names, HEX codes, and
241 # composition notes return zero results. We progressively trim the query
242 # down to the most concrete nouns.
243
244 _NOISE_WORDS = frozenset({
245 # Brand / product names
246 "claude", "openai", "gpt", "gemini", "copilot", "chatgpt", "midjourney",
247 "stable", "diffusion", "dall-e", "cursor", "anthropic", "microsoft",
248 "google", "apple", "meta", "nvidia", "tesla",
249 # Generic filler
250 "using", "with", "from", "that", "this", "have", "been", "will",
251 "into", "more", "also", "very", "some", "than", "them", "other",
252 })
253
254 # Words that look generic but are actually useful when they ARE the
255 # subject of the deck (e.g. a deck about AI). We only drop them when
256 # there are still other concrete nouns left.
257 _SOFT_NOISE_WORDS = frozenset({
258 "ai", "code", "software", "system", "digital", "platform", "solution",
259 "application", "interface", "framework", "algorithm", "api", "sdk",
260 "assistant", "tool", "service", "technology", "tech", "program",
261 # Visual-quality / usage terms. These are helpful in the full provider
262 # query, but should not consume the 3-4 keyword fallback budget or
263 # dominate relevance scoring over the real subject.
264 "professional", "editorial", "commercial", "premium", "stock",
265 "photo", "photograph", "photography", "image", "picture", "visual",
266 "background", "hero", "cover", "banner", "wallpaper",
267 "high", "quality", "resolution", "sharp", "clean", "cinematic",
268 "dramatic", "lighting", "light", "modern", "natural", "visible",
269 })
270
271 _TOKEN_STRIP_CHARS = ".,;:!?\"'()[]{},。;:!?、"
272 _MATCH_SEPARATOR_RE = re.compile(r"""[\s\-_./:;,'"()[\]{}]+""")
273 _ASCII_MATCH_TOKEN_RE = re.compile(r"[a-z0-9]+")
274
275
276 def simplify_query(query: str, max_words: int = 4) -> str:
277 """Trim a verbose query into a short keyword phrase.
278
279 Strategy:
280 1. Strip HEX color codes and parenthetical asides.
281 2. Drop hard-noise words (brand names, generic filler).
282 3. Drop soft-noise words ONLY if concrete nouns remain.
283 4. If the result would be empty, return the original query
284 (fail-open: better an over-broad search than zero results).
285 5. Cap at ``max_words`` words.
286 """
287 cleaned = re.sub(r"#[0-9a-fA-F]{3,8}", "", query)
288 cleaned = re.sub(r"\([^)]*\)", "", cleaned)
289 words = [w.strip(_TOKEN_STRIP_CHARS) for w in cleaned.split()]
290 words = [w for w in words if len(w) > 2]
291
292 after_hard = [w for w in words if w.lower() not in _NOISE_WORDS]
293 after_soft = [w for w in after_hard if w.lower() not in _SOFT_NOISE_WORDS]
294
295 # Only drop soft-noise if there are still concrete nouns left.
296 filtered = after_soft if after_soft else after_hard
297
298 if not filtered:
299 # Everything got filtered. Fail open: return the original query.
300 return query.strip()
301
302 return " ".join(filtered[:max_words])
303
304
305 def build_query_progression(query: str) -> list[str]:
306 """Return a list of progressively simpler queries to try in order.
307
308 Stops as soon as one of them yields candidates upstream. Duplicates
309 are dropped while preserving order.
310 """
311 seen: set[str] = set()
312 out: list[str] = []
313 for candidate in (
314 query,
315 simplify_query(query, max_words=4),
316 simplify_query(query, max_words=3),
317 simplify_query(query, max_words=2),
318 simplify_query(query, max_words=1),
319 ):
320 candidate = candidate.strip()
321 if candidate and candidate not in seen:
322 seen.add(candidate)
323 out.append(candidate)
324 return out
325
326
327 # ---------------------------------------------------------------------------
328 # Scoring
329 # ---------------------------------------------------------------------------
330
331
332 def normalize_orientation(width: int, height: int) -> str:
333 if width <= 0 or height <= 0:
334 return "unknown"
335 if width > height:
336 return "landscape"
337 if height > width:
338 return "portrait"
339 return "square"
340
341
342 def _query_tokens(query: str) -> list[str]:
343 """Extract ASCII keyword tokens from a query for relevance scoring.
344
345 Uses the same noise-word filtering as ``simplify_query`` so the
346 relevance signal lines up with the keywords we actually search by.
347 Non-ASCII tokens (CJK etc.) are dropped — image metadata is mostly
348 English even on multi-language providers, so substring matching CJK
349 against an English title is unreliable. When this leaves no tokens,
350 ``compute_relevance`` falls back to neutral (1.0) and lets the other
351 score dimensions decide.
352 """
353 cleaned = re.sub(r"#[0-9a-fA-F]{3,8}", "", query.lower())
354 cleaned = re.sub(r"\([^)]*\)", "", cleaned)
355 words = [w.strip(_TOKEN_STRIP_CHARS) for w in cleaned.split()]
356 words = [w for w in words if len(w) > 2 and w.isascii()]
357 if not words:
358 return []
359 after_hard = [w for w in words if w not in _NOISE_WORDS]
360 after_soft = [w for w in after_hard if w not in _SOFT_NOISE_WORDS]
361 return after_soft if after_soft else after_hard
362
363
364 def _candidate_text(candidate: AssetCandidate) -> str:
365 """Concatenate the candidate's matchable metadata fields for scoring."""
366 return " ".join(
367 filter(
368 None,
369 (
370 candidate.title,
371 candidate.author,
372 candidate.source_page_url,
373 ),
374 )
375 ).lower()
376
377
378 def _candidate_match_tokens(candidate: AssetCandidate) -> set[str]:
379 """Return whole ASCII tokens from candidate metadata for relevance scoring."""
380 return set(_ASCII_MATCH_TOKEN_RE.findall(_candidate_text(candidate)))
381
382
383 def _normalize_match_text(text: str) -> str:
384 """Normalize metadata / required terms for conservative substring matching."""
385 lowered = (text or "").lower()
386 return _MATCH_SEPARATOR_RE.sub(" ", lowered).strip()
387
388
389 def _term_group_alternatives(term_group: str) -> list[str]:
390 """Split one required term group into alternatives.
391
392 ``"Jiefangbei|Liberation Monument"`` means either alternative satisfies
393 that required group. Different list items are ANDed by
394 ``missing_required_terms``.
395 """
396 return [
397 _normalize_match_text(part)
398 for part in str(term_group or "").split("|")
399 if _normalize_match_text(part)
400 ]
401
402
403 def missing_required_terms(
404 candidate: AssetCandidate,
405 required_terms: tuple[str, ...] | list[str] | None,
406 ) -> list[str]:
407 """Return required term groups not present in candidate metadata.
408
409 This is an entity-safety gate, not a fuzzy visual classifier. Use it for
410 exact subjects such as landmarks, people, companies, or products where a
411 visually nice but wrong image is worse than no image.
412 """
413 if not required_terms:
414 return []
415
416 text = _normalize_match_text(_candidate_text(candidate))
417 compact_text = text.replace(" ", "")
418 missing: list[str] = []
419 for group in required_terms:
420 alternatives = _term_group_alternatives(group)
421 if not alternatives:
422 continue
423 matched = any(
424 alt in text or alt.replace(" ", "") in compact_text
425 for alt in alternatives
426 )
427 if not matched:
428 missing.append(str(group))
429 return missing
430
431
432 def compute_relevance(candidate: AssetCandidate, query: str) -> float:
433 """Fraction of query tokens that match whole candidate metadata tokens.
434
435 Range ``[0.0, 1.0]``. Returns ``1.0`` (neutral) when the query has no
436 ASCII tokens to match — this lets non-English queries fall through
437 to license / size scoring without being unfairly rejected. Whole-token
438 matching prevents false positives such as ``office`` matching ``officer``.
439 """
440 tokens = _query_tokens(query)
441 if not tokens:
442 return 1.0
443 candidate_tokens = _candidate_match_tokens(candidate)
444 if not candidate_tokens:
445 return 0.0
446 hits = sum(1 for token in tokens if token in candidate_tokens)
447 return hits / len(tokens)
448
449
450 def score_candidate(candidate: AssetCandidate, request: ImageSearchRequest) -> float:
451 """Score a candidate against a request. Higher is better; -inf rejects.
452
453 Relevance dominates: a candidate whose metadata shares no query
454 tokens is rejected outright, so size / license / orientation cannot
455 rescue an irrelevant image from a permissive provider.
456 """
457 if not candidate.license_tier:
458 return float("-inf")
459 if (
460 candidate.license_tier == LICENSE_TIER_ATTRIBUTION_REQUIRED
461 and not candidate.author.strip()
462 ):
463 return float("-inf")
464
465 required_misses = missing_required_terms(candidate, request.required_terms)
466 if required_misses:
467 return float("-inf")
468
469 relevance = compute_relevance(candidate, request.query)
470 if relevance == 0.0 and not request.required_terms:
471 return float("-inf")
472
473 score = relevance * 10000.0
474 title_text = _normalize_match_text(candidate.title)
475 compact_title = title_text.replace(" ", "")
476 for group in request.required_terms or ():
477 alternatives = _term_group_alternatives(group)
478 if any(
479 alt in title_text or alt.replace(" ", "") in compact_title
480 for alt in alternatives
481 ):
482 score += 1500.0
483
484 # Penalize infrastructure/transit metadata if the user didn't explicitly ask for it.
485 # This prevents high-res subway station photos from outranking actual tourist landmarks.
486 text = _candidate_text(candidate)
487 query_lower = request.query.lower()
488 infra_terms = [
489 "station", "subway", "metro", "rail", "transit", "airport", "bus",
490 "地铁", "站", "轨道",
491 ]
492
493 if not any(t in query_lower for t in infra_terms):
494 if any(t in text for t in infra_terms):
495 score -= 5000.0
496
497 candidate_orientation = normalize_orientation(candidate.width, candidate.height)
498 requested = (request.orientation or "").strip().lower()
499 if requested:
500 if candidate_orientation == requested:
501 score += 1000.0
502 else:
503 score -= 250.0
504
505 if request.min_width and candidate.width < request.min_width:
506 score -= 500.0
507 if request.min_height and candidate.height < request.min_height:
508 score -= 500.0
509
510 if candidate.license_tier == LICENSE_TIER_NO_ATTRIBUTION:
511 score += 250.0
512
513 # Larger images score higher, but only as a tie-breaker; entity accuracy
514 # and metadata relevance must dominate pixel count.
515 pixel_score = max(candidate.width, 0) * max(candidate.height, 0) / 1000.0
516 score += min(pixel_score, 1500.0)
517 return score
518
519
520 def score_review_candidate(
521 candidate: AssetCandidate,
522 request: ImageSearchRequest,
523 ) -> float:
524 """Score a wider visual-review pool without weakening automatic selection.
525
526 Metadata-verified candidates retain priority. A near match may enter the
527 thumbnail sheet only when at most one required identity group is missing
528 and the candidate still has meaningful relevance to the query that found
529 it. This never authorizes best-only download; ``score_candidate`` remains
530 the automatic-selection gate.
531 """
532 strict_score = score_candidate(candidate, request)
533 if strict_score != float("-inf"):
534 return strict_score + 20000.0
535 if not request.required_terms:
536 return strict_score
537
538 missing = missing_required_terms(candidate, request.required_terms)
539 if len(missing) != 1:
540 return float("-inf")
541
542 discovery_request = replace(
543 request,
544 query=candidate.discovery_query or request.query,
545 required_terms=(),
546 )
547 relevance = compute_relevance(candidate, discovery_request.query)
548 if relevance < 0.5:
549 return float("-inf")
550 relaxed_score = score_candidate(candidate, discovery_request)
551 if relaxed_score == float("-inf"):
552 return relaxed_score
553 return relaxed_score - 5000.0
554
555
556 # ---------------------------------------------------------------------------
557 # Attribution text
558 # ---------------------------------------------------------------------------
559
560
561 PROVIDER_DISPLAY_NAMES: dict[str, str] = {
562 "openverse": "Openverse",
563 "wikimedia": "Wikimedia Commons",
564 "pexels": "Pexels",
565 "pixabay": "Pixabay",
566 }
567
568
569 def build_attribution_text(filename: str, candidate: AssetCandidate) -> str:
570 """Render the canonical attribution string for the manifest.
571
572 Format:
573 ``filename — "title" by author, via Provider, license: name (url)``
574
575 Empty fields are gracefully omitted. The text is intended for use by
576 the Executor when generating in-SVG credit elements; it is not meant
577 to be machine-parsed downstream.
578 """
579 provider_name = PROVIDER_DISPLAY_NAMES.get(
580 candidate.provider, candidate.provider or "unknown"
581 )
582
583 parts: list[str] = [filename or candidate.download_url or "image"]
584 middle: list[str] = []
585 if candidate.title:
586 middle.append(f'"{candidate.title}"')
587 if candidate.author:
588 middle.append(f"by {candidate.author}")
589 middle.append(f"via {provider_name}")
590 parts.append(" ".join(middle))
591
592 license_part = candidate.license_name or candidate.license_url
593 if license_part:
594 if candidate.license_url and candidate.license_name:
595 license_part = f"{candidate.license_name} ({candidate.license_url})"
596 parts.append(f"license: {license_part}")
597
598 return " — ".join(parts)
599
600
601 # ---------------------------------------------------------------------------
602 # Small helpers
603 # ---------------------------------------------------------------------------
604
605
606 def ensure_json_parent(path: str | Path) -> Path:
607 """Make sure the parent directory of ``path`` exists; return as Path."""
608 p = Path(path)
609 p.parent.mkdir(parents=True, exist_ok=True)
610 return p
611
611 lines PYTHON