Jeena
f55f3d8e91
The monolithic check_updates.py has been split into the update_monitor/ package with separate modules for fetching, extraction, comparison, classification, and notification. This makes the codebase easier to navigate and test. The flat services.yaml is replaced by per-service directories under services/<name>/config.yaml, co-locating config with update scripts and AI review examples. When --auto-update is enabled, services with an auto-update.sh script get their release notes classified by a local Ollama model (using few-shot examples from review-examples/) to decide whether the update can be applied automatically or needs manual admin intervention. Results are reported in a single consolidated Matrix message. Immich is the first service with auto-update configured, using 7 real release note examples to train the classifier.
77 lines
2.2 KiB
Python
77 lines
2.2 KiB
Python
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
|