#!/usr/bin/env python3 # -*- coding: utf-8 -*- """可编辑提词器后台：静态服务 + /api/save 自动保存到 draft-latest.md 并重出全部产物 启动: python3 edit_server.py """ import speech_rate as SR import http.server, socketserver, os, re, json, shutil, subprocess, datetime, threading, uuid from urllib.parse import urlparse, parse_qs BASE = os.path.dirname(os.path.abspath(__file__)) DRAFT = '/root/media-hub/export/draft-latest.md' BAKDIR = '/root/media-hub/export/.edit-backups' ANN = '/root/media-hub/export/annotations.json' TOKEN = open(os.path.join(BASE, '.edit-token'), encoding='utf-8').read().strip() PORT = 8896 os.makedirs(BAKDIR, exist_ok=True) LOCK = threading.Lock() ANN_LOCK = threading.Lock() def ann_load(): try: with open(ANN, encoding='utf-8') as f: d = json.load(f) return d if isinstance(d, list) else [] except Exception: return [] def ann_write(items): tmp = ANN + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump(items, f, ensure_ascii=False, indent=1) os.replace(tmp, ANN) def ann_apply(data): """add / upd / del 一条批注，返回最新全量列表""" act = data.get('action') with ANN_LOCK: items = ann_load() if act == 'add': note = str(data.get('note') or '').strip() if not note: raise ValueError('批注是空的') items.append({ 'id': uuid.uuid4().hex[:12], 'sec': int(data.get('sec')), 'para': int(data.get('para')), 'note': note, 'text': str(data.get('text') or '')[:300], 'created': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), }) elif act == 'upd': note = str(data.get('note') or '').strip() if not note: raise ValueError('批注是空的') for it in items: if it.get('id') == data.get('id'): it['note'] = note it['updated'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') break else: raise ValueError('找不到这条批注') elif act == 'del': before = len(items) items = [it for it in items if it.get('id') != data.get('id')] if len(items) == before: raise ValueError('找不到这条批注') else: raise ValueError('未知动作') ann_write(items) return items def split_sections(raw): parts = re.split(r'^【(.+?)】\s*$', raw, flags=re.M) return parts[0], [[h, b] for h, b in zip(parts[1::2], parts[2::2])] def regen(): for s in ['sync_prompter.py', 'gen_mobile_txt.py']: try: subprocess.run(['python3', os.path.join(BASE, s)], cwd=BASE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=120) except Exception as e: print('regen fail', s, e) def save_para(sec, idx, text): with LOCK: raw = open(DRAFT, encoding='utf-8').read() lines = raw.split('\n') heads = [k for k, l in enumerate(lines) if re.match(r'^【.*】$', l.strip())] if not 1 <= sec <= len(heads): raise ValueError('节号越界') start = heads[sec - 1] end = heads[sec] if sec < len(heads) else len(lines) body = lines[start + 1:end] pos = [k for k, l in enumerate(body) if l.strip()] if not 0 <= idx < len(pos): raise ValueError('段落越界') text = ' '.join(text.replace('\r', '').split('\n')).strip() if not text: raise ValueError('不能存成空段') body[pos[idx]] = text lines[start + 1:end] = body out = '\n'.join(lines) # 顺手更新文件头统计行（与 gen_prompter 同口径：逐节取整后累加） ls = out.split('\n') hs = [k for k, l in enumerate(ls) if re.match(r'^【.*】$', l.strip())] tot_han = tot_sec = 0 for k, st in enumerate(hs): en = hs[k + 1] if k + 1 < len(hs) else len(ls) n = len(re.findall(r'[\u4e00-\u9fff]', '\n'.join(ls[st + 1:en]))) tot_han += n tot_sec += SR.sec_of(k + 1, n) stat = (f'> {len(hs)} 节 · {tot_han} 口播汉字 · 纯念 {tot_sec // 60}:{tot_sec % 60:02d}' f' · 加停顿成片约 {round(tot_sec * 1.19 / 60)} 分') out = re.sub(r'^> 14 节 · .*$', stat, out, count=1, flags=re.M) ts = datetime.datetime.now().strftime('%Y%m%d-%H%M%S') shutil.copy2(DRAFT, os.path.join(BAKDIR, f'draft-{ts}.md')) olds = sorted(os.listdir(BAKDIR)) for f in olds[:-200]: try: os.remove(os.path.join(BAKDIR, f)) except OSError: pass open(DRAFT, 'w', encoding='utf-8').write(out) threading.Thread(target=regen, daemon=True).start() return len(text) class H(http.server.SimpleHTTPRequestHandler): def __init__(self, *a, **kw): super().__init__(*a, directory=BASE, **kw) def guess_type(self, path): t = super().guess_type(path) if path.endswith(('.md', '.txt', '.json', '.html', '.js', '.csv')): base = t.split(';')[0] if base.startswith('application/octet-stream') or base in ('text/markdown', 'text/plain'): base = 'text/plain' if path.endswith(('.md', '.txt', '.csv')) else base return base + '; charset=utf-8' return t def log_message(self, *a): pass def do_GET(self): if self.path.startswith('/api/ping'): self.send_response(200) self.send_header('Content-Type', 'application/json; charset=utf-8') self.end_headers() self.wfile.write(json.dumps({'ok': True, 'mtime': os.path.getmtime(DRAFT)}).encode()) return if self.path.startswith('/api/ann'): q = parse_qs(urlparse(self.path).query) if (q.get('k') or [''])[0] != TOKEN: res = {'ok': False, 'err': 'token'} else: res = {'ok': True, 'items': ann_load()} body = json.dumps(res, ensure_ascii=False).encode() self.send_response(200) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Content-Length', str(len(body))) self.end_headers() self.wfile.write(body) return return super().do_GET() def do_POST(self): if not (self.path.startswith('/api/save') or self.path.startswith('/api/ann')): self.send_error(404) return try: n = int(self.headers.get('Content-Length') or 0) data = json.loads(self.rfile.read(n).decode('utf-8')) if data.get('k') != TOKEN: raise ValueError('token') if self.path.startswith('/api/ann'): res = {'ok': True, 'items': ann_apply(data)} else: cnt = save_para(int(data['sec']), int(data['para']), str(data['text'])) res = {'ok': True, 'chars': cnt} except Exception as e: res = {'ok': False, 'err': str(e)} body = json.dumps(res, ensure_ascii=False).encode() self.send_response(200) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Content-Length', str(len(body))) self.end_headers() self.wfile.write(body) socketserver.TCPServer.allow_reuse_address = True with socketserver.ThreadingTCPServer(('0.0.0.0', PORT), H) as httpd: print('edit server on', PORT) httpd.serve_forever()