1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
#!/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 <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)
|
