#!/usr/bin/env python3 # -*- coding: utf-8 -*- """对比页右侧可编辑后台：静态服务 + /api/save 自动保存到 v7-live.md 启动: python3 compare_edit.py （端口 8897） """ import http.server, socketserver, os, re, json, shutil, datetime, threading, uuid from urllib.parse import urlparse, parse_qs BASE = os.path.dirname(os.path.abspath(__file__)) LIVE = '/root/media-hub/export/v7-live.md' ANN = '/root/media-hub/export/compare-annotations.json' HTMLS = '/root/media-hub/export/v7-live-html.json' BAKDIR = '/root/media-hub/export/.compare-backups' TOKEN = open(os.path.join(BASE, '.compare-token'), encoding='utf-8').read().strip() PORT = 8897 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): 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('批注是空的') rgs = [] for rg in (data.get('ranges') or [])[:80]: try: rgs.append({ 'sec': int(rg.get('sec')), 'para': int(rg.get('para')), 'start': max(0, int(rg.get('start'))), 'end': max(0, int(rg.get('end'))), 't': str(rg.get('t') or '')[:2000], }) except Exception: pass items.append({ 'id': uuid.uuid4().hex[:12], 'sec': int(data.get('sec')), 'para': int(data.get('para')), 'ranges': rgs, '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 # 节头只认「【1. 标题】」这种带编号的，正文里的【助理】单独成行也不会被误认成新节 HEADS_RE = re.compile(r'^【\s*\d+\s*[.．、]\s*.*】$') # 正文里出现这种行 = 用户不小心造出了节头，拒绝保存 FAKE_HEAD_RE = re.compile(r'^【\s*\d+\s*[.．、]\s*.*】$') def norm(text): """把浏览器 innerText 归一成 markdown：去 \r、每行 strip、段间一个空行""" text = text.replace('\r', '').replace('\u00a0', ' ') out = [] for ln in text.split('\n'): ln = ln.strip() if ln: out.append(ln) elif out and out[-1] != '': out.append('') while out and out[-1] == '': out.pop() return '\n'.join(out) def htmls_load(): try: with open(HTMLS, encoding='utf-8') as f: d = json.load(f) return d if isinstance(d, dict) else {} except Exception: return {} def htmls_put(sec, arr): if not isinstance(arr, list) or not arr: return d = htmls_load() d[str(sec)] = [str(x)[:20000] for x in arr] tmp = HTMLS + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump(d, f, ensure_ascii=False) os.replace(tmp, HTMLS) def read_live(): return open(LIVE, encoding='utf-8').read() def save_sec(sec, text, html=None): with LOCK: raw = read_live() lines = raw.split('\n') heads = [k for k, l in enumerate(lines) if HEADS_RE.match(l.strip())] if not 1 <= sec <= len(heads): raise ValueError('节号越界') body = norm(text) if len(re.findall(r'[\u4e00-\u9fff]', body)) < 10: raise ValueError('这一节汉字太少，怕你误删，没存') for ln in body.split('\n'): if FAKE_HEAD_RE.match(ln.strip()): raise ValueError('这一行「%s」长得像节标题，存进去会把整篇拆坏，没存' % ln.strip()[:20]) start = heads[sec - 1] end = heads[sec] if sec < len(heads) else len(lines) new_lines = lines[:start + 1] + [''] + body.split('\n') + [''] + lines[end:] out = '\n'.join(new_lines) ts = datetime.datetime.now().strftime('%Y%m%d-%H%M%S') shutil.copy2(LIVE, os.path.join(BAKDIR, f'v7live-{ts}.md')) olds = sorted(os.listdir(BAKDIR)) for f in olds[:-300]: try: os.remove(os.path.join(BAKDIR, f)) except OSError: pass tmp = LIVE + '.tmp' open(tmp, 'w', encoding='utf-8').write(out) os.replace(tmp, LIVE) htmls_put(sec, html) return len(re.findall(r'[\u4e00-\u9fff]', body)), ts def all_sections(): raw = read_live() lines = raw.split('\n') heads = [k for k, l in enumerate(lines) if HEADS_RE.match(l.strip())] res = {} for i, st in enumerate(heads, 1): en = heads[i] if i < len(heads) else len(lines) res[i] = norm('\n'.join(lines[st + 1:en])) return res 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 _json(self, obj, code=200): body = json.dumps(obj, ensure_ascii=False).encode() self.send_response(code) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Content-Length', str(len(body))) self.send_header('Cache-Control', 'no-store') self.end_headers() self.wfile.write(body) def do_GET(self): p = self.path.split('?')[0] if p == '/api/ping': return self._json({'ok': True, 'mtime': os.path.getmtime(LIVE), 'sections': len(all_sections())}) if p == '/api/sections': return self._json({'ok': True, 'mtime': os.path.getmtime(LIVE), 'secs': all_sections(), 'htmls': htmls_load()}) if p == '/api/ann': q = parse_qs(urlparse(self.path).query) if (q.get('k') or [''])[0] != TOKEN: return self._json({'ok': False, 'err': 'token'}) return self._json({'ok': True, 'items': ann_load()}) return super().do_GET() def do_POST(self): p = self.path.split('?')[0] if p not in ('/api/save', '/api/save_all', '/api/ann'): return self.send_error(404) 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 p == '/api/ann': return self._json({'ok': True, 'items': ann_apply(data)}) if p == '/api/save': han, ts = save_sec(int(data['sec']), str(data['text']), data.get('html')) return self._json({'ok': True, 'chars': han, 'ts': ts, 'mtime': os.path.getmtime(LIVE)}) secs = data.get('secs') or {} done = [] for k, v in secs.items(): han, ts = save_sec(int(k), str(v), (data.get('htmls') or {}).get(k)) done.append({'sec': int(k), 'chars': han}) return self._json({'ok': True, 'saved': done, 'mtime': os.path.getmtime(LIVE)}) except Exception as e: return self._json({'ok': False, 'err': str(e)}) socketserver.TCPServer.allow_reuse_address = True with socketserver.ThreadingTCPServer(('0.0.0.0', PORT), H) as httpd: httpd.daemon_threads = True print('compare edit server on', PORT) httpd.serve_forever()