import json import re from typing import Any, Dict, Optional from .models import ExtractRule def extract_jsonpath(payload: Any, path: str) -> Optional[str]: if not path.startswith("$."): return None current = payload for part in path[2:].split("."): if isinstance(current, list): if not part.isdigit(): return None index = int(part) if index >= len(current): return None current = current[index] continue if not isinstance(current, dict): return None if part not in current: return None current = current[part] return None if current is None else str(current) def extract_version( body: str, rule: ExtractRule, headers: Optional[Dict[str, str]] = None, ) -> Optional[str]: if rule.type == "jsonpath": try: return extract_jsonpath(json.loads(body), rule.value) except json.JSONDecodeError: return None if rule.type == "jsonpath_join": try: payload = json.loads(body) except json.JSONDecodeError: return None parts = [] for path in rule.value.split(","): value = extract_jsonpath(payload, path.strip()) if value is None: return None parts.append(value) return ".".join(parts) if rule.type == "header": return headers.get(rule.value.lower()) if headers else None if rule.type == "regex": match = re.search(rule.value, body) if not match: return None return match.group(1) if match.lastindex else match.group(0) if rule.type == "text": return rule.value if rule.value in body else None return None def build_upstream_fallback(url: Optional[str]) -> Optional[Dict[str, Any]]: if not url: return None for host in ("api.github.com/repos/", "codeberg.org/api/v1/repos/"): if host in url and url.endswith("/releases/latest"): return { "url": url.replace("/releases/latest", "/tags"), "extract": ExtractRule(type="jsonpath", value="$.0.name"), } return None