diff options
Diffstat (limited to 'scripts/course-daily-update.py')
| -rw-r--r-- | scripts/course-daily-update.py | 138 |
1 files changed, 138 insertions, 0 deletions
diff --git a/scripts/course-daily-update.py b/scripts/course-daily-update.py new file mode 100644 index 0000000..4f509d6 --- /dev/null +++ b/scripts/course-daily-update.py @@ -0,0 +1,138 @@ +#!/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): + try: + r = requests.post(OLLAMA, json={'model': MODEL, 'stream': False, 'temperature': 0.4, 'max_tokens': n, + 'messages': [{'role': 'system', 'content': system}, {'role': 'user', 'content': user}]}, timeout=180) + if not r.ok: return None + return (r.json().get('choices') or [{}])[0].get('message', {}).get('content') + except Exception: + 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 <cmd>` 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) |