Your Name
358 lines
14 KiB
Python
358 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from datetime import datetime, timezone
|
|
from email.utils import parsedate_to_datetime
|
|
from typing import Any
|
|
from urllib.parse import parse_qs, urljoin, urlparse
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from bs4 import BeautifulSoup, Tag
|
|
|
|
|
|
class ConfigError(ValueError):
|
|
pass
|
|
|
|
|
|
def normalize_config(value: dict[str, Any]) -> dict[str, Any]:
|
|
if not isinstance(value, dict) or not value:
|
|
raise ConfigError("config is required and cannot be empty")
|
|
|
|
if "discovery" not in value:
|
|
follow_links = value.get("follow_links", False)
|
|
config = {
|
|
"version": 1,
|
|
"discovery": {"type": "rss", "limit": value.get("max_items", 20)},
|
|
"detail": {
|
|
"enabled": follow_links,
|
|
"always": follow_links,
|
|
"root_selector": value.get("content_selector", ""),
|
|
"fields": {
|
|
"text": {
|
|
"selector": value.get("text_selector") or ":root",
|
|
"extract": "text",
|
|
"required": True,
|
|
}
|
|
},
|
|
"media": [
|
|
{"type": "photo", "selector": "img", "attributes": ["src", "data-src"]},
|
|
{"type": "video", "selector": "iframe, video, source", "attributes": ["src", "data-src"]},
|
|
],
|
|
},
|
|
"access": {"type": value.get("access", "auto")},
|
|
"retry": {"attempts": 1, "delay_seconds": 0},
|
|
"min_text_length": value.get("min_text_length", 0),
|
|
}
|
|
else:
|
|
config = dict(value)
|
|
|
|
if config.get("version", 1) != 1:
|
|
raise ConfigError("only config.version=1 is supported")
|
|
discovery = config.get("discovery")
|
|
if not isinstance(discovery, dict) or discovery.get("type") not in {"rss", "html"}:
|
|
raise ConfigError("discovery.type must be rss or html")
|
|
try:
|
|
limit = int(discovery.get("limit", 20))
|
|
except (TypeError, ValueError) as exc:
|
|
raise ConfigError("discovery.limit must be an integer") from exc
|
|
if not 1 <= limit <= 100:
|
|
raise ConfigError("discovery.limit must be between 1 and 100")
|
|
discovery["limit"] = limit
|
|
if discovery["type"] == "html" and not str(discovery.get("item_selector") or "").strip():
|
|
raise ConfigError("discovery.item_selector is required for html discovery")
|
|
|
|
access = config.get("access", {"type": "auto"})
|
|
if isinstance(access, str):
|
|
access = {"type": access}
|
|
if not isinstance(access, dict) or access.get("type", "auto") not in {"auto", "http", "browser", "cloudflare"}:
|
|
raise ConfigError("access.type must be auto, http, browser or cloudflare")
|
|
config["access"] = access
|
|
|
|
detail = config.get("detail") or {"enabled": False}
|
|
if not isinstance(detail, dict):
|
|
raise ConfigError("detail must be an object")
|
|
detail["enabled"] = bool(detail.get("enabled", False))
|
|
detail["always"] = bool(detail.get("always", detail["enabled"]))
|
|
if detail["enabled"]:
|
|
if not str(detail.get("root_selector") or "").strip():
|
|
raise ConfigError("detail.root_selector is required when detail is enabled")
|
|
if not isinstance(detail.get("fields") or {}, dict):
|
|
raise ConfigError("detail.fields must be an object")
|
|
if not isinstance(detail.get("media") or [], list):
|
|
raise ConfigError("detail.media must be an array")
|
|
if detail.get("transport", "browser") not in {"http", "fetch", "browser"}:
|
|
raise ConfigError("detail.transport must be http, fetch or browser")
|
|
detail["transport"] = detail.get("transport", "browser")
|
|
config["detail"] = detail
|
|
|
|
retry = config.get("retry") or {}
|
|
try:
|
|
attempts = int(retry.get("attempts", 2))
|
|
delay = float(retry.get("delay_seconds", 2))
|
|
timeout = float(retry.get("timeout_seconds", 90))
|
|
except (TypeError, ValueError) as exc:
|
|
raise ConfigError("retry values must be numeric") from exc
|
|
if not 1 <= attempts <= 5 or not 0 <= delay <= 60 or not 10 <= timeout <= 300:
|
|
raise ConfigError("retry attempts must be 1..5, delay_seconds 0..60 and timeout_seconds 10..300")
|
|
config["retry"] = {"attempts": attempts, "delay_seconds": delay, "timeout_seconds": timeout}
|
|
|
|
try:
|
|
min_length = int(config.get("min_text_length", 0))
|
|
except (TypeError, ValueError) as exc:
|
|
raise ConfigError("min_text_length must be an integer") from exc
|
|
if not 0 <= min_length <= 100_000:
|
|
raise ConfigError("min_text_length must be between 0 and 100000")
|
|
config["min_text_length"] = min_length
|
|
return config
|
|
|
|
|
|
def nested_value(value: Any, path: str) -> Any:
|
|
current = value
|
|
for part in path.split("."):
|
|
if isinstance(current, dict):
|
|
current = current.get(part)
|
|
else:
|
|
return None
|
|
return current
|
|
|
|
|
|
def json_path_values(value: Any, path: str) -> list[Any]:
|
|
parts = path.split(".")
|
|
found: list[Any] = []
|
|
|
|
def visit(node: Any, index: int) -> None:
|
|
if index == len(parts):
|
|
found.append(node)
|
|
return
|
|
if isinstance(node, list):
|
|
for entry in node:
|
|
visit(entry, index)
|
|
elif isinstance(node, dict):
|
|
if parts[index] in node:
|
|
visit(node[parts[index]], index + 1)
|
|
for entry in node.values():
|
|
if isinstance(entry, (dict, list)):
|
|
visit(entry, index)
|
|
|
|
visit(value, 0)
|
|
return found
|
|
|
|
|
|
def candidate_elements(root: Tag | BeautifulSoup, candidate: dict[str, Any]) -> list[Tag]:
|
|
selectors = candidate.get("selectors") or candidate.get("selector") or []
|
|
if isinstance(selectors, str):
|
|
selectors = [selectors]
|
|
elements: list[Tag] = []
|
|
for selector in selectors:
|
|
if selector == ":root":
|
|
elements.append(root)
|
|
elif str(selector).strip():
|
|
elements.extend(root.select(str(selector)))
|
|
return elements
|
|
|
|
|
|
def element_value(element: Tag, candidate: dict[str, Any]) -> Any:
|
|
mode = str(candidate.get("extract") or "text")
|
|
if mode == "text":
|
|
return element.get_text("\n", strip=True)
|
|
if mode == "html":
|
|
return element.decode_contents()
|
|
if mode == "attr":
|
|
attributes = candidate.get("attributes") or candidate.get("attribute") or []
|
|
if isinstance(attributes, str):
|
|
attributes = [attributes]
|
|
for attribute in attributes:
|
|
value = element.get(str(attribute))
|
|
if value:
|
|
return value
|
|
return None
|
|
if mode == "json":
|
|
try:
|
|
payload = json.loads(element.string or element.get_text("", strip=True))
|
|
except (TypeError, json.JSONDecodeError):
|
|
return None
|
|
values = json_path_values(payload, str(candidate.get("path") or ""))
|
|
return next((entry for entry in values if entry is not None and entry != ""), None)
|
|
raise ConfigError(f"unsupported extract mode: {mode}")
|
|
|
|
|
|
def parse_date(value: Any, candidate: dict[str, Any]) -> str | None:
|
|
if not value:
|
|
return None
|
|
raw = str(value).strip()
|
|
try:
|
|
parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
try:
|
|
parsed = parsedate_to_datetime(raw)
|
|
except (TypeError, ValueError, OverflowError):
|
|
parsed = None
|
|
if parsed is None:
|
|
formats = candidate.get("formats") or candidate.get("date_format") or []
|
|
if isinstance(formats, str):
|
|
formats = [formats]
|
|
for date_format in formats:
|
|
try:
|
|
parsed = datetime.strptime(raw, str(date_format))
|
|
break
|
|
except ValueError:
|
|
continue
|
|
if parsed is None:
|
|
return None
|
|
if parsed.tzinfo is None:
|
|
try:
|
|
parsed = parsed.replace(tzinfo=ZoneInfo(str(candidate.get("timezone") or "UTC")))
|
|
except Exception:
|
|
parsed = parsed.replace(tzinfo=timezone.utc)
|
|
return parsed.astimezone(timezone.utc).isoformat()
|
|
|
|
|
|
def apply_regex(value: Any, candidate: dict[str, Any]) -> Any:
|
|
pattern = candidate.get("regex")
|
|
if not pattern or value is None:
|
|
return value
|
|
match = re.search(str(pattern), str(value), flags=re.DOTALL)
|
|
if not match:
|
|
return None
|
|
group = candidate.get("group", 1 if match.lastindex else 0)
|
|
try:
|
|
return match.group(group)
|
|
except (IndexError, KeyError):
|
|
return None
|
|
|
|
|
|
def extract_field(
|
|
root: Tag | BeautifulSoup | None,
|
|
rule: dict[str, Any],
|
|
source: dict[str, Any] | None = None,
|
|
*,
|
|
field_name: str = "",
|
|
) -> Any:
|
|
candidates = rule.get("candidates") or [rule]
|
|
for candidate in candidates:
|
|
if not isinstance(candidate, dict):
|
|
continue
|
|
values: list[Any] = []
|
|
if candidate.get("source"):
|
|
values.append(nested_value(source or {}, str(candidate["source"])))
|
|
elif root is not None:
|
|
values.extend(element_value(element, candidate) for element in candidate_elements(root, candidate))
|
|
for value in values:
|
|
value = apply_regex(value, candidate)
|
|
if value is None or (isinstance(value, str) and not value.strip()):
|
|
continue
|
|
if field_name == "published_at" or candidate.get("type") == "date":
|
|
value = parse_date(value, candidate)
|
|
if not value:
|
|
continue
|
|
if isinstance(value, str):
|
|
value = value.strip()
|
|
if len(str(value)) < int(rule.get("min_length", 0)):
|
|
continue
|
|
return value
|
|
return None
|
|
|
|
|
|
def extract_fields(
|
|
root: Tag | BeautifulSoup | None,
|
|
fields: dict[str, Any],
|
|
source: dict[str, Any] | None = None,
|
|
) -> tuple[dict[str, Any], list[dict[str, str]]]:
|
|
values: dict[str, Any] = {}
|
|
errors: list[dict[str, str]] = []
|
|
for name, rule in fields.items():
|
|
if not isinstance(rule, dict):
|
|
errors.append({"field": name, "error": "field rule must be an object"})
|
|
continue
|
|
try:
|
|
value = extract_field(root, rule, source, field_name=name)
|
|
except Exception as exc:
|
|
errors.append({"field": name, "error": str(exc)})
|
|
continue
|
|
if value is None and rule.get("required"):
|
|
errors.append({"field": name, "error": "required field not found"})
|
|
elif value is not None:
|
|
values[name] = value
|
|
return values, errors
|
|
|
|
|
|
def youtube_url(value: str, base_url: str = "") -> str | None:
|
|
url = urljoin(base_url, str(value or "").strip())
|
|
parsed = urlparse(url)
|
|
host = (parsed.hostname or "").lower().removeprefix("www.").removeprefix("m.")
|
|
video_id = ""
|
|
if host == "youtu.be":
|
|
video_id = parsed.path.strip("/").split("/", 1)[0]
|
|
elif host in {"youtube.com", "youtube-nocookie.com"}:
|
|
if parsed.path == "/watch":
|
|
video_id = (parse_qs(parsed.query).get("v") or [""])[0]
|
|
elif parsed.path.startswith(("/embed/", "/shorts/", "/live/")):
|
|
video_id = parsed.path.strip("/").split("/", 1)[1]
|
|
if not re.fullmatch(r"[A-Za-z0-9_-]{6,20}", video_id):
|
|
return None
|
|
return f"https://www.youtube.com/watch?v={video_id}"
|
|
|
|
|
|
def srcset_urls(value: str) -> list[str]:
|
|
return [entry.strip().split()[0] for entry in value.split(",") if entry.strip()]
|
|
|
|
|
|
def extract_media(root: Tag | BeautifulSoup, specs: list[dict[str, Any]], base_url: str) -> list[dict[str, str]]:
|
|
media: list[dict[str, str]] = []
|
|
for spec in specs:
|
|
if not isinstance(spec, dict):
|
|
continue
|
|
media_type = str(spec.get("type") or "photo")
|
|
attributes = spec.get("attributes") or spec.get("attribute") or ["src"]
|
|
if isinstance(attributes, str):
|
|
attributes = [attributes]
|
|
for element in candidate_elements(root, spec):
|
|
raw_values: list[str] = []
|
|
for attribute in attributes:
|
|
raw = element.get(str(attribute))
|
|
if not raw:
|
|
continue
|
|
raw_values.extend(srcset_urls(str(raw)) if attribute == "srcset" else [str(raw)])
|
|
break
|
|
for raw in raw_values:
|
|
url = urljoin(base_url, raw.strip())
|
|
if not url.startswith(("http://", "https://")):
|
|
continue
|
|
item_type = media_type
|
|
provider = ""
|
|
if media_type == "video":
|
|
normalized = youtube_url(url)
|
|
if normalized:
|
|
url, provider = normalized, "youtube"
|
|
else:
|
|
host = (urlparse(url).hostname or "").lower()
|
|
provider = "twitch" if "twitch.tv" in host else host
|
|
if element.name in {"iframe", "a"} or provider == "twitch":
|
|
item_type = "external_video"
|
|
item = {"type": item_type, "url": url}
|
|
if provider:
|
|
item["provider"] = provider
|
|
media.append(item)
|
|
return list({(item["type"], item["url"]): item for item in media}.values())
|
|
|
|
|
|
def detail_from_html(
|
|
html: str,
|
|
url: str,
|
|
config: dict[str, Any],
|
|
discovery_values: dict[str, Any] | None = None,
|
|
) -> tuple[dict[str, Any], list[dict[str, str]]]:
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
detail = config["detail"]
|
|
root = soup.select_one(str(detail["root_selector"]))
|
|
if root is None:
|
|
return {}, [{"field": "detail", "error": f"root selector not found: {detail['root_selector']}"}]
|
|
for selector in detail.get("remove_selectors") or []:
|
|
for element in root.select(str(selector)):
|
|
element.decompose()
|
|
values, errors = extract_fields(root, detail.get("fields") or {}, discovery_values)
|
|
values["media"] = extract_media(root, detail.get("media") or [], url)
|
|
values["html"] = str(root)
|
|
return values, errors
|