#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ AI 漫剧 · 无限画布 —— SQLite 本地数据服务器 ============================================ 用法: python3 aicomic-server.py # 默认端口 8765，数据库 aicomic.db python3 aicomic-server.py 9000 # 指定端口 AICOMIC_DB=/path/x.db python3 aicomic-server.py # 指定数据库路径 与 manju pipeline 合并后的新接口（画布 → 一键成片）: POST /api/export-shots 把画布卡片转为 manju/script/shots.json POST /api/build 后台启动 manju/build.py 生成成片 GET /api/build-status 查询生成进度与日志 GET /api/output 播放/下载成片 MP4 启动后浏览器打开: http://127.0.0.1:8765/ 数据全部存在 SQLite 数据库文件（默认: aicomic.db）中，可随时拷贝备份。 """ import json import os import sqlite3 import subprocess import sys import threading import time import webbrowser from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlparse BASE_DIR = os.path.dirname(os.path.abspath(__file__)) _ARGS = [a for a in sys.argv[1:] if a != '--no-browser'] AUTO_OPEN = '--no-browser' not in sys.argv PORT = int(_ARGS[0]) if _ARGS and _ARGS[0].isdigit() else 8765 DB_PATH = os.environ.get('AICOMIC_DB') or (_ARGS[1] if len(_ARGS) > 1 else os.path.join(BASE_DIR, 'aicomic.db')) HTML_PATH = os.path.join(BASE_DIR, 'aicomic-canvas.html') MAX_SNAPSHOTS = 20 def db(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row conn.execute(''' CREATE TABLE IF NOT EXISTS canvas_state ( id INTEGER PRIMARY KEY CHECK (id = 1), cards TEXT NOT NULL, view TEXT NOT NULL, updated_at REAL NOT NULL ) ''') conn.execute(''' CREATE TABLE IF NOT EXISTS snapshots ( id INTEGER PRIMARY KEY AUTOINCREMENT, cards TEXT NOT NULL, view TEXT NOT NULL, created_at REAL NOT NULL ) ''') return conn def load_state(): conn = db() row = conn.execute('SELECT cards, view, updated_at FROM canvas_state WHERE id=1').fetchone() conn.close() if not row: return None return {'cards': json.loads(row['cards']), 'view': json.loads(row['view']), 'updated_at': row['updated_at']} def save_state(cards, view): conn = db() now = time.time() conn.execute(''' INSERT INTO canvas_state (id, cards, view, updated_at) VALUES (1, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET cards=excluded.cards, view=excluded.view, updated_at=excluded.updated_at ''', (json.dumps(cards, ensure_ascii=False), json.dumps(view, ensure_ascii=False), now)) conn.execute('INSERT INTO snapshots (cards, view, created_at) VALUES (?, ?, ?)', (json.dumps(cards, ensure_ascii=False), json.dumps(view, ensure_ascii=False), now)) conn.execute('DELETE FROM snapshots WHERE id NOT IN (SELECT id FROM snapshots ORDER BY id DESC LIMIT ?)', (MAX_SNAPSHOTS,)) conn.commit() conn.close() def list_snapshots(): conn = db() rows = conn.execute('SELECT id, created_at FROM snapshots ORDER BY id DESC LIMIT ?', (MAX_SNAPSHOTS,)).fetchall() conn.close() return [{'id': r['id'], 'created_at': r['created_at']} for r in rows] def restore_snapshot(sid): conn = db() row = conn.execute('SELECT cards, view FROM snapshots WHERE id=?', (sid,)).fetchone() conn.close() if not row: return None return {'cards': json.loads(row['cards']), 'view': json.loads(row['view'])} # ================= manju pipeline 桥接（画布 → 一键成片） ================= MANJU_DIR = os.path.join(BASE_DIR, "manju") VENV_PY = os.path.join(BASE_DIR, ".venv", "bin", "python") BUILD_LOG = os.path.join(BASE_DIR, "aicomic-build.log") _BUILD = {"proc": None, "started_at": None, "args": []} def _output_name(): try: cfg = json.load(open(os.path.join(MANJU_DIR, "config.json"), encoding="utf-8")) return "manju_ep%02d.mp4" % int(cfg.get("episode", 1)) except Exception: return "manju_ep01.mp4" def _build_running(): p = _BUILD["proc"] return p is not None and p.poll() is None def start_build(flags): """后台启动 manju/build.py；同一时间只允许一个任务""" if _build_running(): return False, "已有生成任务进行中，请稍候" flags = flags or {} args = [VENV_PY, "build.py"] if flags.get("skip_images"): args.append("--skip-images") if flags.get("skip_tts"): args.append("--skip-tts") if flags.get("skip_clips"): args.append("--skip-clips") if flags.get("use_ai_clips"): args.append("--use-ai-clips") if flags.get("force"): args.append("--force") try: logf = open(BUILD_LOG, "w", encoding="utf-8") except OSError as e: return False, "无法写入日志: %s" % e try: proc = subprocess.Popen(args, cwd=MANJU_DIR, stdout=logf, stderr=subprocess.STDOUT) except Exception as e: logf.close() return False, "启动 pipeline 失败: %s" % e _BUILD.update(proc=proc, started_at=time.time(), args=args) return True, "已启动" def build_status(): p = _BUILD["proc"] running = _build_running() done = (p is not None) and not running code = p.poll() if done else None tail = "" try: with open(BUILD_LOG, "r", encoding="utf-8", errors="replace") as f: lines = f.readlines() tail = "".join(lines[-40:]) except OSError: tail = "" output = None out_path = os.path.join(MANJU_DIR, "output", _output_name()) if os.path.exists(out_path): output = "/api/output" return { "running": running, "done": done, "code": code, "started_at": _BUILD["started_at"], "log": tail, "output": output, "args": _BUILD["args"], } class Handler(BaseHTTPRequestHandler): protocol_version = 'HTTP/1.1' def _cors(self): self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type') def _json(self, code, obj): body = json.dumps(obj, ensure_ascii=False).encode('utf-8') self.send_response(code) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Content-Length', str(len(body))) self._cors() self.end_headers() self.wfile.write(body) def _read_json(self): length = int(self.headers.get('Content-Length', 0) or 0) if length <= 0: return None try: return json.loads(self.rfile.read(length).decode('utf-8')) except Exception: return None def do_OPTIONS(self): self.send_response(204) self._cors() self.end_headers() def do_GET(self): path = urlparse(self.path).path if path in ('/', '/index.html', '/aicomic-canvas.html'): try: with open(HTML_PATH, 'rb') as f: body = f.read() self.send_response(200) self.send_header('Content-Type', 'text/html; charset=utf-8') self.send_header('Content-Length', str(len(body))) self._cors() self.end_headers() self.wfile.write(body) return except FileNotFoundError: self._json(404, {'error': 'aicomic-canvas.html not found, run server from its directory'}) return if path == '/api/state': st = load_state() if st is None: self._json(200, {'cards': None, 'view': None, 'updated_at': None}) else: self._json(200, st) return if path == '/api/snapshots': self._json(200, list_snapshots()) return if path == '/api/build-status': self._json(200, build_status()) return if path == '/api/output': out_path = os.path.join(MANJU_DIR, 'output', _output_name()) if not os.path.exists(out_path): self._json(404, {'error': '成片不存在，请先在画布上「一键成片」'}) return try: with open(out_path, 'rb') as f: body = f.read() self.send_response(200) self.send_header('Content-Type', 'video/mp4') self.send_header('Content-Length', str(len(body))) self.send_header('Content-Disposition', 'inline; filename="' + _output_name() + '"') self._cors() self.end_headers() self.wfile.write(body) except Exception as e: self._json(500, {'error': '读取成片失败: %s' % e}) return self._json(404, {'error': 'not found: ' + path}) def do_POST(self): path = urlparse(self.path).path if path == '/api/state': data = self._read_json() if data is None or not isinstance(data.get('cards'), list): self._json(400, {'error': 'body must be {"cards": [...], "view": {...}}'}) return save_state(data['cards'], data.get('view', {})) self._json(200, {'ok': True, 'saved_at': time.time()}) return if path == '/api/import-novel': data = self._read_json() if data is None or not data.get('text'): self._json(400, {'error': 'body must be {"text": "...", "target_chars": 1500, "use_llm": true}'}) return try: import aicomic_import res = aicomic_import.run_import( data['text'], target_chars=int(data.get('target_chars', 1500) or 1500), use_llm=bool(data.get('use_llm', True))) self._json(200, res) except Exception as e: self._json(500, {'error': '导入失败: %s' % str(e)}) return if path == '/api/image-prompt': data = self._read_json() or {} try: import aicomic_import prompt = aicomic_import.optimize_image_prompt( scene=data.get('scene', ''), detail=data.get('prompt', ''), chars=data.get('chars') or [], move=data.get('move', ''), style=data.get('style', ''), style_pack=data.get('style_pack', 'cel'), quality=data.get('quality', 'high')) self._json(200, {'ok': True, 'prompt': prompt}) except Exception as e: self._json(200, {'ok': False, 'error': str(e), 'prompt': ''}) return if path.startswith('/api/snapshots/') and path.endswith('/restore'): sid = path.split('/')[3] try: sid = int(sid) except ValueError: self._json(400, {'error': 'bad snapshot id'}) return st = restore_snapshot(sid) if st is None: self._json(404, {'error': 'snapshot not found'}) else: self._json(200, st) return if path == '/api/export-shots': data = self._read_json() if data is None or not isinstance(data.get('cards'), list): self._json(400, {'error': 'body must be {"cards": [...]}'}) return try: import aicomic_bridge shots, removed = aicomic_bridge.export_shots(data['cards']) self._json(200, {'ok': True, 'count': len(shots), 'removed': removed, 'file': aicomic_bridge.SHOTS_PATH}) except Exception as e: self._json(500, {'error': '导出分镜失败: %s' % e}) return if path == '/api/build': data = self._read_json() or {} ok, msg = start_build(data) self._json(200 if ok else 409, {'ok': ok, 'message': msg}) return self._json(404, {'error': 'not found: ' + path}) def log_message(self, fmt, *args): sys.stderr.write('[%s] %s\n' % (time.strftime('%H:%M:%S'), fmt % args)) def main(): print('=' * 56) print(' AI 漫剧 · 无限画布 SQLite 服务器（已合并 manju pipeline）') print(' URL : http://127.0.0.1:%d/' % PORT) print(' DB : %s' % DB_PATH) print(' 画布「一键成片」→ manju pipeline → /api/output') print(' 停止: Ctrl+C') print('=' * 56) if not os.path.exists(HTML_PATH): print('警告: 未找到 %s ，请把本脚本和 aicomic-canvas.html 放在同一目录' % HTML_PATH) server = ThreadingHTTPServer(('127.0.0.1', PORT), Handler) if AUTO_OPEN: try: threading.Timer(1.0, lambda: webbrowser.open('http://127.0.0.1:%d/' % PORT)).start() except Exception: pass try: server.serve_forever() except KeyboardInterrupt: print('\n已停止') if __name__ == '__main__': main()