#!/usr/bin/env python3 """Daily newOS4 course expansion. Grounded only: every command must exist in atlas.json.""" import json, os, re, sys, subprocess from datetime import datetime, timezone import requests ROOT = '/var/www/webx11.duckdns.org/spaces/newos4' COURSE = os.path.join(ROOT, 'course') ATLAS, FEED, FAQ = (os.path.join(COURSE, n) for n in ('atlas.json', 'feed.json', 'faq.json')) STATE = os.path.join(COURSE, '_state.json'); CHANGELOG = os.path.join(COURSE, 'changelog.jsonl') PENDING = os.path.join(COURSE, 'pending') SEARCH = 'https://webx11.duckdns.org/api/search' OLLAMA = 'http://127.0.0.1:11434/v1/chat/completions' MODEL = 'deepseek-v4.1-flash:cloud' TOPICS = ['terminal UX best practices command design', 'browser virtual filesystem design', 'module lazy loading cache busting web apps', 'Ollama local LLM tool calling best practices', 'WebDAV client sync reliability', 'developer onboarding interactive tutorial design', 'AI agent tool safety confirmation patterns', 'cron automation in single page apps', 'privacy preserving product analytics', 'prompt engineering grounded answers no hallucination', 'web terminal accessibility keyboard UX', 'self-hosted AI assistant cost control', 'spaced repetition micro learning programming', 'secure API key handling in browser apps'] SYS_DRAFT = ('You write ONE micro-lesson for the newOS4 terminal course. Return STRICT JSON only (no fences):\n' '{"title":"...","minutes":4,"blocks":[{"type":"text|code|cmd|tip|warn","text":"..."}],' '"exercise":{"instruction":"...","accept":["cmd ..."],"check":{"type":"manual"},"hint":"..."},"sources":["https://..."]}\n' 'Rules: 3-5 blocks; at least one cmd block; every command MUST appear verbatim in the ALLOWED list; ' 'only state facts present in the research snippets; sources must be URLs from the snippets.') def load(p, d): try: with open(p, encoding='utf-8') as f: return json.load(f) except Exception: return d def save(p, obj): tmp = p + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump(obj, f, ensure_ascii=False, indent=1) os.replace(tmp, p); os.chmod(p, 0o644) def parse_json(txt): txt = (txt or '').strip() if txt.startswith('`' * 3): txt = txt.split('\n', 1)[1] if '\n' in txt else '' txt = txt.strip() if txt.endswith('`' * 3): txt = txt[:-3].strip() m = re.search(r'\{[\s\S]*\}', txt) try: return json.loads(m.group(0) if m else txt) except Exception: return None def llm(system, user, n=900): for model in (MODEL, 'gemma4:31b-cloud'): try: r = requests.post(OLLAMA, json={'model': model, 'stream': False, 'temperature': 0.4, 'max_tokens': max(n, 1400), 'think': False, 'messages': [{'role': 'system', 'content': system}, {'role': 'user', 'content': user}]}, timeout=180) if not r.ok: continue msg = (r.json().get('choices') or [{}])[0].get('message', {}) content = (msg.get('content') or '').strip() if content: return content except Exception: continue return None def research(q): try: r = requests.get(SEARCH, params={'q': q}, timeout=25) if not r.ok: return [] return [{'title': (x.get('title') or '')[:160], 'url': x['url'], 'snippet': (x.get('content') or '')[:600]} for x in (r.json().get('results') or [])[:4] if x.get('url')] except Exception: return [] def validate_item(it, known, urls): if not isinstance(it, dict): return False, 'not object' if not (10 <= len(it.get('title') or '') <= 90): return False, 'title' blocks = it.get('blocks') or [] if not (3 <= len(blocks) <= 6): return False, 'block count' if any(b.get('type') not in ('text', 'code', 'cmd', 'tip', 'warn', 'table') or not (b.get('text') or '').strip() for b in blocks): return False, 'block' if not any(b.get('type') == 'cmd' for b in blocks): return False, 'no cmd' ex = it.get('exercise') or {} if not ex.get('instruction') or not ex.get('accept'): return False, 'exercise' for a in ex['accept']: if a.split(' ')[0] not in known: return False, 'unknown command ' + a.split(' ')[0] if not any(str(s) in urls for s in (it.get('sources') or []) if str(s).startswith('http')): return False, 'no source' return True, '' def main(): dry = '--dry-run' in sys.argv now = datetime.now(timezone.utc); day = now.strftime('%Y-%m-%d') state = load(STATE, {'seen_commands': [], 'topic_index': 0, 'last_run': None}) if not dry: subprocess.run([sys.executable, os.path.join(ROOT, 'scripts', 'gen-course-atlas.py')], check=True) atlas = load(ATLAS, {}); known = set(atlas.get('commands', {})) seen = set(state.get('seen_commands') or known) new = sorted(known - seen) if state.get('seen_commands') else [] feed = load(FEED, {'version': '', 'items': []}); added = []; rejected = [] if new: feed['items'].insert(0, {'id': 'new-' + day, 'date': day, 'title': f'{len(new)} new command(s) in newOS4', 'minutes': 2, 'generated': False, 'sources': ['atlas.json'], 'blocks': [{'type': 'text', 'text': 'Run `help ` for details.'}, {'type': 'cmd', 'text': '\n'.join(new[:12])}]}) added.append('new-' + day) if not any(i.get('id') == 'daily-' + day for i in feed['items']): topic = TOPICS[state.get('topic_index', 0) % len(TOPICS)] results = research(topic) if results: snippets = '\n'.join(f"- {r['title']} | {r['url']}\n {r['snippet']}" for r in results) draft = parse_json(llm(SYS_DRAFT, f'TOPIC: {topic}\n\nSNIPPETS:\n{snippets}\n\nALLOWED COMMANDS:\n' + ', '.join(sorted(known)))) ok, why = validate_item(draft, known, {r['url'] for r in results}) if draft else (False, 'no JSON') if ok: draft.update({'id': 'daily-' + day, 'date': day, 'generated': True}) feed['items'].insert(0, draft); added.append('daily-' + day) else: os.makedirs(PENDING, exist_ok=True) save(os.path.join(PENDING, f'daily-{day}.json'), {'draft': draft, 'why': why, 'topic': topic, 'results': results}) rejected.append('daily: ' + why) else: rejected.append('daily: no research results') state['topic_index'] = state.get('topic_index', 0) + 1 feed['items'] = feed['items'][:60]; feed['version'] = day if not dry: save(FEED, feed) if not dry: subprocess.run([sys.executable, os.path.join(ROOT, 'scripts', 'course-insights.py')], check=True) insights = load(os.path.join(COURSE, 'insights.json'), {}) faq = load(FAQ, {'items': []}); have = {f.get('q') for f in faq['items']}; faq_n = 0 for q in (insights.get('top_questions') or [])[:5]: if q['q'] in have: continue draft = parse_json(llm('Answer a repeated newOS4 question. Return STRICT JSON only: {"a":"<=400 chars","commands":["cmd"],"sources":["atlas.json#cmd"]}. Only use ALLOWED commands.', f'QUESTION: {q["q"]}\nALLOWED COMMANDS:\n' + ', '.join(sorted(known)), 500)) if draft and draft.get('a') and all(c in known for c in draft.get('commands', [])): faq['items'].append({'q': q['q'], 'a': draft['a'][:400], 'commands': draft['commands'], 'sources': draft.get('sources') or ['insights.json'], 'added': day}) have.add(q['q']); faq_n += 1 else: rejected.append('faq: ' + q['q'][:60]) if not dry: save(FAQ, faq) state.update({'seen_commands': sorted(known), 'last_run': now.strftime('%Y-%m-%dT%H:%M:%SZ')}) if not dry: save(STATE, state) with open(CHANGELOG, 'a', encoding='utf-8') as f: f.write(json.dumps({'date': day, 'added': added, 'faq': faq_n, 'rejected': rejected, 'atlas': f'{len(known)} commands'}, ensure_ascii=False) + '\n') print('newOS4 Course daily — ' + day) print('atlas: ' + str(len(known)) + ' commands | new: ' + str(len(new))) print('daily lesson: ' + next((i['title'] for i in feed['items'] if i.get('id') == 'daily-' + day), 'none added')) print('faq added: ' + str(faq_n) + ' | feed: ' + str(len(feed['items'])) + '/60') print('rejected: ' + ('; '.join(rejected) if rejected else 'none')) print('needs OK: none') if __name__ == '__main__': sys.exit(main() or 0)