#!/usr/bin/env python3
"""
Daily GitHub Release Checker
Runs as cron job, checks 11 projects for new releases.
Writes to releases-check.json baseline and produces report on stdout.
"""
import json, urllib.request, urllib.error, time, re, sys
from datetime import datetime, timezone, timedelta
BASELINE_PATH = '/opt/data/releases-check.json'
def load_baseline():
with open(BASELINE_PATH) as f:
return json.load(f)
def save_baseline(data):
with open(BASELINE_PATH, 'w') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
def fetch_json(url, retries=2):
for attempt in range(retries):
try:
req = urllib.request.Request(url, headers={
'User-Agent': 'Hermes-Cron-Release-Checker/1.0',
'Accept': 'application/json'
})
with urllib.request.urlopen(req, timeout=30) as resp:
data = resp.read().decode()
if not data.strip(): return None
return json.loads(data)
except urllib.error.HTTPError as e:
if e.code == 404: return {'_error': '404', '_status': 404}
if e.code == 403: return {'_error': 'rate_limit', '_status': 403}
return {'_error': str(e), '_status': e.code}
except (urllib.error.URLError, json.JSONDecodeError, TimeoutError) as e:
if attempt < retries - 1: time.sleep(2)
except Exception as e: return {'_error': str(e)}
return None
def fetch_text(url, retries=2):
for attempt in range(retries):
try:
req = urllib.request.Request(url, headers={'User-Agent': 'Hermes-Cron-Release-Checker/1.0'})
with urllib.request.urlopen(req, timeout=30) as resp:
return resp.read().decode()
except Exception as e:
if attempt < retries - 1: time.sleep(2)
else: return None
return None
def scrape_github_release(repo):
url = f'https://github.com/{repo}/releases'
html = fetch_text(url)
if not html: return None, None
m = re.search(r'/releases/tag/([^"\']+)', html)
if m:
tag = m.group(1)
m2 = re.search(r'<relative-time[^>]*datetime="([^"]+)"', html)
date = m2.group(1)[:10] if m2 else None
return tag, date
return None, None
def compare_versions(old_tag, new_tag, name=''):
if old_tag == new_tag: return False
if name == 'cline':
old = old_tag.replace('cli-', 'v', 1) if old_tag.startswith('cli-') else old_tag
new = new_tag.replace('cli-', 'v', 1) if new_tag.startswith('cli-') else new_tag
else: old, new = old_tag, new_tag
def extract_nums(tag): return [int(x) for x in re.findall(r'\d+', tag)]
old_nums = extract_nums(old)
new_nums = extract_nums(new)
if not old_nums or not new_nums: return True
max_len = max(len(old_nums), len(new_nums))
old_padded = old_nums + [0] * (max_len - len(old_nums))
new_padded = new_nums + [0] * (max_len - len(new_nums))
if new_padded < old_padded: return 'rollback'
return True
projects = [
('opencode', 'anomalyco/opencode'),
('oh-my-openagent', 'code-yeongyu/oh-my-openagent'),
('Hermes Agent', 'NousResearch/hermes-agent'),
('openclaw', 'openclaw/openclaw'),
('cline', 'cline/cline'),
('pinchtab', 'pinchtab/pinchtab'),
('llama.cpp', 'ggerganov/llama.cpp'),
('one-api', 'songquanpeng/one-api'),
('stdd', 'leonai42/stdd'),
('nvitop', 'XuehaiPan/nvitop'),
('open-webui', 'open-webui/open-webui'),
]
baseline = load_baseline()
now = datetime.now(timezone(timedelta(hours=8)))
today_str = now.strftime('%Y-%m-%d')
total_checks = baseline.get('total_checks', 0)
total_updates = baseline.get('total_updates', 0)
updates = []; errors = []; anomalies = []; new_projects = 0
start_time = time.time()
for name, repo in projects:
if name not in baseline:
baseline[name] = {'version': '', 'date': '', 'exists': True, 'repo_path': repo, 'history': []}
entry = baseline[name]
old_version = entry.get('version', '')
is_new_project = not old_version
history = entry.get('history', [])
if name == 'stdd':
stdd_repo = '/opt/data/home/.stdd'
import subprocess
try:
subprocess.run(['git', 'fetch', '--tags', 'origin'], cwd=stdd_repo, capture_output=True, timeout=30)
result = subprocess.run(['git', 'tag', '--sort=-creatordate', '--list', 'v[0-9]*'], cwd=stdd_repo, capture_output=True, text=True, timeout=10)
tags = [t.strip() for t in result.stdout.strip().split('\n') if t.strip()]
if not tags: errors.append(f'{name}: no version tags found'); continue
latest_tag = tags[0]
date_result = subprocess.run(['git', 'log', '-1', '--format=%ai', latest_tag], cwd=stdd_repo, capture_output=True, text=True, timeout=10)
tag_date_str = date_result.stdout.strip()
latest_date = tag_date_str[:10] if tag_date_str else today_str
except subprocess.CalledProcessError as e: errors.append(f'{name}: git error: {e}'); continue
except FileNotFoundError: errors.append(f'{name}: git not found'); continue
elif name == 'cline':
api_url = f'https://api.github.com/repos/{repo}/releases?per_page=15'
data = fetch_json(api_url)
if data is None: errors.append(f'{name}: no response'); continue
if isinstance(data, dict) and '_error' in data:
err = data['_error']
if err == '404': errors.append(f'{name}: 404 (no releases)'); entry['exists'] = False; continue
elif err == 'rate_limit':
tag, date = scrape_github_release(repo)
if tag: latest_tag, latest_date = tag, date or today_str
else: errors.append(f'{name}: rate limited and HTML scrape failed'); continue
else: errors.append(f'{name}: {err}'); continue
else:
found = None
for r in data:
tag = r.get('tag_name', '')
if tag and re.match(r'^v\d', tag): found = r; break
if found:
latest_tag = found['tag_name']
published_at = found.get('published_at') or found.get('created_at', '')
latest_date = published_at[:10] if published_at else today_str
else: errors.append(f'{name}: no main release found'); continue
else:
api_url = f'https://api.github.com/repos/{repo}/releases/latest'
data = fetch_json(api_url)
if data is None: errors.append(f'{name}: no response'); continue
if isinstance(data, dict) and '_error' in data:
err = data['_error']
if err == '404': errors.append(f'{name}: 404 (no releases)'); entry['exists'] = False; continue
elif err == 'rate_limit':
tag, date = scrape_github_release(repo)
if tag: latest_tag, latest_date = tag, date or today_str
else: errors.append(f'{name}: rate limited and HTML scrape failed'); continue
else: errors.append(f'{name}: {err}'); continue
latest_tag = data.get('tag_name', '')
if not latest_tag: errors.append(f'{name}: empty tag_name'); continue
published_at = data.get('published_at') or data.get('created_at', '')
latest_date = published_at[:10] if published_at else today_str
result = compare_versions(old_version, latest_tag, name)
if is_new_project:
entry['version'] = latest_tag; entry['date'] = latest_date; entry['exists'] = True
entry['history'] = [{'version': latest_tag, 'date': latest_date}]
new_projects += 1
elif result == 'rollback':
anomalies.append(f'{name}: {old_version} -> {latest_tag} (rollback)')
elif result is True:
entry['version'] = latest_tag; entry['date'] = latest_date; entry['exists'] = True
if not history or history[-1].get('version') != latest_tag:
history.append({'version': latest_tag, 'date': latest_date})
entry['history'] = history
updates.append((name, old_version, latest_tag, latest_date))
total_updates += 1
total_checks += 1
baseline['last_checked'] = now.strftime('%Y-%m-%d %H:%M CST') if baseline.get('last_checked', '').split()[0] != today_str else baseline['last_checked']
baseline['total_checks'] = total_checks
baseline['total_updates'] = total_updates
save_baseline(baseline)
elapsed = time.time() - start_time
print(f"ELAPSED={elapsed:.0f}")
print(f"UPDATES={','.join(f'{n}:{o}->{v}' for n,o,v,d in updates) if updates else ''}")
print(f"ERRORS={','.join(errors) if errors else ''}")
print(f"ANOMALIES={','.join(anomalies) if anomalies else ''}")
print(f"NEW_PROJECTS={new_projects}")
print(f"TOTAL_CHECKS={total_checks}")
print(f"TOTAL_UPDATES={total_updates}")
print(f"UPDATE_COUNT={len(updates)}")
print("==STATUS==")
for name, repo in projects:
entry = baseline.get(name, {})
v = entry.get('version', '-'); d = entry.get('date', '-'); e = entry.get('exists', True)
if not e: print(f" {name}: 404 (no releases)")
elif any(name == e.split(':')[0].strip() for e in errors): print(f" {name}: ERROR")
else: print(f" {name}: {v} ({d})")
if updates:
print("==UPDATES==")
for n, o, v, d in updates: print(f" {n}: {o} -> {v} ({d})")
if errors:
print("==ERRORS==")
for e in errors: print(f" {e}")
if anomalies:
print("==ANOMALIES==")
for a in anomalies: print(f" {a}")