#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
拍书取字 · 后端服务
手机拍照 / 相册选图 → 上传 → 文本提取（本地 RapidOCR 为主，DeepSeek VL 可选增强）
照片与原页文本自动存档到 data/books/<书名>/（同时可在下载中心 8899 看到）。
启动：bash start.sh 停止：bash stop.sh
默认端口 8901，监听 0.0.0.0（局域网手机访问 http://<本机IP>:8901）
"""
from __future__ import annotations
import base64
import collections
import io
import json
import mimetypes
import os
import re
import shutil
import threading
import time
import datetime as _dt
from pathlib import Path
from flask import Flask, jsonify, request, send_file, send_from_directory
# ---------------------------------------------------------------------------
# 常量与配置
# ---------------------------------------------------------------------------
BASE = Path(__file__).resolve().parent
STATIC_DIR = BASE / "static"
BOOKS_DIR = BASE / "data" / "books"
TRASH_DIR = BASE / "data" / "trash" # 回收站：删除先移到此处，可恢复
DEEPSEEK_KEY_FILE = BASE / "deepseek_key.txt"
DEEPSEEK_ENDPOINT = os.environ.get(
"DEEPSEEK_ENDPOINT", "https://api.deepseek.com/v1/chat/completions"
)
DEEPSEEK_MODEL = os.environ.get("DEEPSEEK_MODEL", "deepseek-v4-flash-vision-exp")
HOST = os.environ.get("OCR_HOST", "0.0.0.0")
PORT = int(os.environ.get("OCR_PORT", "8901"))
ALLOWED_EXT = {"jpg", "jpeg", "png", "webp", "gif", "bmp", "heic", "heif", "tif", "tiff"}
PHOTO_EXT = ".jpg"
# OCR 提示词（VL 引擎用）：按阅读顺序整页转写正文
OCR_PROMPT = (
"你是古籍/实体书页面转写助手。这是拍摄的一页印刷实体书（简体中文为主）。\n"
"请把这一页的正文内容完整、逐字转写为纯文本：\n"
"1) 只输出正文文字本身，按自上而下、从左到右的阅读顺序，不要输出任何解释或格式标记；\n"
"2) 忽略页眉、页脚、页码、注释边栏、水印、手指、阴影等无关内容；\n"
"3) 段落之间保留一个空行，标点符号（中文句号逗号等）按原样保留；\n"
"4) 若包含英文单词或数字请原样保留，不要翻译。\n"
)
_ocr_lock = threading.Lock()
_rapid_engine = None
_vl_ready_cache: bool | None = None
app = Flask(__name__, static_folder=None)
# 单次上传限制：最多 100 张、每张最大 10MB；请求体上限按 100×10MB + 表单开销估算（≈1.02GB）
MAX_FILES_PER_UPLOAD = int(os.environ.get("OCR_MAX_FILES", "100")) # 单次张数上限
MAX_FILE_BYTES = int(os.environ.get("OCR_MAX_FILE_MB", "10")) * 1024 * 1024 # 单张大小上限
app.config["MAX_CONTENT_LENGTH"] = int(os.environ.get(
"OCR_MAX_UPLOAD_MB", str(MAX_FILES_PER_UPLOAD * MAX_FILE_BYTES // 1048576 + 16))) * 1024 * 1024
# ---------------------------------------------------------------------------
# 工具函数
# ---------------------------------------------------------------------------
def _sanitize(name: str, fallback: str, max_len: int = 48) -> str:
name = (name or "").strip()
name = re.sub(r"[\\/:*?\"<>|\s]+", "_", name)
name = re.sub(r"[^\w\u4e00-\u9fff.-]", "", name)
name = name.strip("._")
if not name:
return fallback
return name[:max_len]
def _now_tag() -> str:
return _dt.datetime.now().strftime("%Y%m%d_%H%M%S")
def _load_image(raw: bytes, max_side: int = 2600):
"""解码 → 按 EXIF 摆正 → 等比缩放到 max_side（默认 2600，足够 OCR 又控内存）。"""
from PIL import Image, ImageOps
img = Image.open(io.BytesIO(raw))
img = ImageOps.exif_transpose(img)
if img.mode not in ("RGB", "L"):
img = img.convert("RGB")
w, h = img.size
scale = min(1.0, max_side / max(w, h))
if scale < 1.0:
img = img.resize((int(w * scale), int(h * scale)), Image.LANCZOS)
return img
def _image_bytes(img, fmt: str = "JPEG", quality: int = 92) -> bytes:
buf = io.BytesIO()
img.save(buf, format=fmt, quality=quality)
return buf.getvalue()
def _jpeg_bytes_for_vl(img, max_side: int = 1568) -> bytes:
"""VL 上传前缩到较小尺寸并转 JPEG（控制带宽与 API 耗时）。"""
from PIL import Image
w, h = img.size
scale = min(1.0, max_side / max(w, h))
if scale < 1.0:
img = img.resize((int(w * scale), int(h * scale)), Image.LANCZOS)
if img.mode != "RGB":
img = img.convert("RGB")
return _image_bytes(img, "JPEG", 88)
# ---------------------------------------------------------------------------
# 本地 RapidOCR 引擎（懒加载）
# ---------------------------------------------------------------------------
def get_local_engine():
global _rapid_engine
if _rapid_engine is None:
from rapidocr_onnxruntime import RapidOCR
_rapid_engine = RapidOCR()
return _rapid_engine
def _normalize_result(res) -> list:
"""rapidocr 返回结构跨版本有差异，这里做防御性归一：
结果 → [ {box:[左上,右上,右下,左下], text:str, score:float}, ... ] 按阅读顺序排序。"""
if res is None:
return []
if isinstance(res, tuple):
res = res[0]
if isinstance(res, dict):
res = res.get("result") or res.get("res") or []
rows = []
for item in res or []:
try:
if isinstance(item, dict):
box = item.get("box") or item.get("boxes") or item.get("dt_box")
text = item.get("text") or item.get("rec_text") or ""
score = item.get("score") or item.get("rec_score") or 0.0
else:
box, text, score = item[0], item[1], item[2]
if not text:
continue
rows.append({"box": box, "text": str(text), "score": float(score or 0)})
except Exception:
continue
# 按 y 分桶（容忍轻微倾斜），桶内按 x 排 —— 得到大致阅读顺序
def cy(r):
b = r["box"]
return (b[0][1] + b[3][1]) / 2.0 if b and len(b) == 4 else 0.0
def cx(r):
b = r["box"]
return (b[0][0] + b[1][0]) / 2.0 if b and len(b) == 4 else 0.0
rows.sort(key=lambda r: (cy(r) // 36, cx(r)))
return rows
def _assemble_text(rows: list, img_h: int) -> str:
"""按检测框把文本行组装成段落：行间垂直间距大 → 空行分段。"""
if not rows:
return ""
lines: list[tuple[float, float, str]] = []
for r in rows:
b = r["box"]
top = min(p[1] for p in b) if b else 0
bot = max(p[1] for p in b) if b else img_h
lines.append((top, bot, r["text"]))
parts = [lines[0][2]]
prev_bot = lines[0][1]
for top, bot, text in lines[1:]:
if top - prev_bot > 0.8 * (bot - top) + 4: # 明显空隙 → 段落换行
parts.append("\n")
parts.append(text)
prev_bot = max(prev_bot, bot)
return "\n".join(parts).strip("\n") + "\n"
def ocr_local(img) -> tuple[str, float, dict]:
"""整图输入返回 (文本, 耗时秒, 诊断信息)。"""
t0 = time.time()
import numpy as np
rgb = np.asarray(img.convert("RGB"))
bgr = rgb[:, :, ::-1].copy()
with _ocr_lock:
result = get_local_engine()(bgr)
rows = _normalize_result(result)
text = _assemble_text(rows, bgr.shape[0])
det_n = len(rows)
chars = sum(len(r["text"]) for r in rows)
return text, time.time() - t0, {"lines": det_n, "chars": chars}
# ---------------------------------------------------------------------------
# DeepSeek VL 增强引擎
# ---------------------------------------------------------------------------
def get_deepseek_key() -> str | None:
try:
return DEEPSEEK_KEY_FILE.read_text(encoding="utf-8").strip() or None
except Exception:
return None
def vl_ready() -> bool:
global _vl_ready_cache
if _vl_ready_cache is None:
_vl_ready_cache = bool(get_deepseek_key())
return _vl_ready_cache
def ocr_vl(img) -> tuple[str, float, dict]:
"""调用 DeepSeek 视觉模型转写整页。"""
import requests
key = get_deepseek_key()
if not key:
raise RuntimeError("DeepSeek API key 未配置（缺 deepseek_key.txt）")
t0 = time.time()
jpg = _jpeg_bytes_for_vl(img)
b64 = base64.b64encode(jpg).decode("ascii")
payload = {
"model": DEEPSEEK_MODEL,
"messages": [
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{b64}"},
},
{"type": "text", "text": OCR_PROMPT},
],
}
],
"temperature": 0.2,
"max_tokens": 4096,
"stream": False,
}
headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json"}
resp = requests.post(DEEPSEEK_ENDPOINT, json=payload, headers=headers, timeout=180)
if resp.status_code != 200:
raise RuntimeError(f"DeepSeek API {resp.status_code}: {resp.text[:200]}")
data = resp.json()
text = data["choices"][0]["message"]["content"]
usage = data.get("usage", {})
return (text or "").strip() + "\n", time.time() - t0, {
"usage": {k: usage.get(k) for k in ("prompt_tokens", "completion_tokens")},
"model": data.get("model", DEEPSEEK_MODEL),
}
# ---------------------------------------------------------------------------
# 结构化 VL（整书批量用）：识别 页码/页眉/页脚/正文
# ---------------------------------------------------------------------------
OCR_PROMPT_VL_STRUCT = (
"你是整书扫描排版助手。这是拍摄的一页实体书。请分四部分输出：\n"
"1【页码】：这一页印刷的页码（阿拉伯数字或中文数字或罗马数字；若该页没有明确页码写“无”，"
"但务必优先从页脚/页边识别真实页码，它是整书排序的关键）；\n"
"2【页眉】：页面顶部重复出现的书眉（如书名/章节名；没有写“无”）；\n"
"3【页脚】：页面底部重复出现的文字（真实页码不要重复放在这里，单独在第1部分；其余底部文字写这里，没有写“无”）；\n"
"4【正文】：完整、逐字转写正文，按自上而下阅读顺序，含标题与小节名，段落之间保留空行，"
"保留中文标点，英文/数字原样输出，不要漏句，不要添加解释。\n"
"★表格规则（务必执行）：页面中出现的表格（含“续表”“表 x-x”以及只有横竖线、"
"行列对齐排列的表格，如项目章程表、组织结构表、会议记录表、职责分工表、进度/预算表），"
"必须用 **Markdown 管道表格** 逐行转写，不要拍平成用“/”或空格分隔的多行纯文本：\n"
" ① 首行写表头：| 列1 | 列2 | … |（表头缺失或跨页续表时，用该表上一页的表头；确实无表头写 | 列1 | 列2 | … |）；\n"
" ② 第二行写分隔：| --- | --- | … |（列数与表头一致）；\n"
" ③ 之后每行一条数据，列数保持一致；合并单元格把该值在其覆盖的每个格子里重复填写；\n"
" ④ 单元格为空写空（两个管道符之间留空），表格里的文字一个都不能丢，也不要改写成句子；\n"
" ⑤ 表格前后若有小标题（如“续表”“表 2-3 …”）按普通正文行照写；一个页面有多个表格就输出多个 Markdown 表格。\n"
" ⑥ 单元格内不要使用
或换行，同一格内多条内容用“；”分隔，一条数据占一行；\n"
" ⑦ 图表/组织结构里的上下级关系按“上层 → 下层”拆成一条条记录，不要输出完全相同的重复行；\n"
" ⑧ 表格若印有真实表头（表头行写着“项目目标”“成功标准”等），照抄真实表头，不要写“列1/列2”。\n"
"严格按下述标记逐段输出，标记各占一行，不要输出其它任何内容：\n"
"【页码】<页码或“无”>\n"
"【页眉】<内容或“无”>\n"
"【页脚】<内容或“无”>\n"
"【正文开始】\n<正文>\n【正文结束】\n"
)
_CN_DIGIT = {'零': 0, '一': 1, '二': 2, '两': 2, '三': 3, '四': 4, '五': 5, '六': 6,
'七': 7, '八': 8, '九': 9}
def _cn_num(s: str):
"""极简中文数字（1~9999）转 int；失败返回 None。"""
s = (s or '').strip()
if not s or any(ch not in _CN_DIGIT and ch not in '十百千零' for ch in s):
return None
total, section = 0, 0
for ch in s:
if ch == '千':
section *= 1000
elif ch == '百':
section *= 100
elif ch == '十':
section = (section or 1) * 10
elif ch == '零':
continue
else:
section += _CN_DIGIT[ch]
return total + section or None
def _page_num_int(raw) -> int | None:
if raw is None:
return None
s = re.sub(r"[^\d0-9一二三四五六七八九两十百千]", "", str(raw))
if not s:
return None
if s.isdigit():
return int(s)
return _cn_num(s)
def _parse_vl_output(raw: str) -> dict:
"""解析四段标记输出 → {page_raw, page_num, header, footer, body, parse_ok}"""
meta = {"page_raw": "", "header": "", "footer": "", "body": "", "parse_ok": False}
if not raw:
return meta
sec = raw.split("【正文开始】")
if len(sec) >= 2 and "【正文结束】" in sec[1]:
meta["body"] = sec[1].split("【正文结束】")[0].strip("\n")
meta["parse_ok"] = True
pre = sec[0]
cur, buf = None, []
for ln in (pre or "").splitlines():
m = re.match(r"^【(页码|页眉|页脚)】\s*(.*)$", ln.strip())
if m:
if cur:
meta[cur] = "\n".join(buf).strip()
cur, buf = {"页码": "page_raw", "页眉": "header", "页脚": "footer"}[m.group(1)], [m.group(2)]
elif cur is not None:
buf.append(ln.strip())
if cur:
meta[cur] = "\n".join(buf).strip()
for k in ("page_raw", "header", "footer"):
meta[k] = (meta[k] or "").replace("无", "").strip() or "无"
else:
# 兜底：无标记时把整段当正文，同时尝试抓数字页眉脚
meta["body"] = raw.strip()
meta["parse_ok"] = False
meta["page_num"] = _page_num_int(None if meta["page_raw"] == "无" else meta["page_raw"])
return meta
def ocr_vl_struct(img) -> tuple[dict, float, dict]:
"""结构化 VL 整页识别。返回 (meta, 耗时, diag)。"""
import requests
key = get_deepseek_key()
if not key:
raise RuntimeError("DeepSeek API key 未配置（缺 deepseek_key.txt）")
t0 = time.time()
b64 = base64.b64encode(_jpeg_bytes_for_vl(img)).decode("ascii")
payload = {
"model": DEEPSEEK_MODEL,
"messages": [{"role": "user", "content": [
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}},
{"type": "text", "text": OCR_PROMPT_VL_STRUCT},
]}],
"temperature": 0.1,
"max_tokens": 6000,
"stream": False,
}
headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json"}
resp = requests.post(DEEPSEEK_ENDPOINT, json=payload, headers=headers, timeout=240)
if resp.status_code != 200:
raise RuntimeError(f"DeepSeek API {resp.status_code}: {resp.text[:200]}")
data = resp.json()
raw = data["choices"][0]["message"]["content"] or ""
meta = _parse_vl_output(raw)
usage = data.get("usage", {})
meta["model"] = data.get("model", DEEPSEEK_MODEL)
diag = {"usage": {k: usage.get(k) for k in ("prompt_tokens", "completion_tokens")}}
return meta, time.time() - t0, diag
def run_vl_structured_page(book_dir: Path, stem: str) -> str:
"""对已存档页跑结构化 VL，写 .vl.json，返回正文。
结构化结果为空（模型未按格式输出等）时自动回退为整页普通 VL 转写，避免空页。"""
img = _load_image((book_dir / f"{stem}.jpg").read_bytes())
meta, _sec, _diag = ocr_vl_struct(img)
meta.setdefault("stem", stem)
meta["at"] = _dt.datetime.now().isoformat(timespec="seconds")
if not (meta.get("body") or "").strip():
# 回退：普通整页 VL；仍失败则抛出（由调用方写 err.json，页面可重试，避免“空成功”）
try:
text, _s2, _d2 = ocr_vl(img)
except Exception as e: # noqa: BLE001
raise RuntimeError(f"结构化 VL 为空且普通 VL 调用失败: {e}") from e
if not (text or "").strip():
raise RuntimeError("VL 未返回文字（结构化与普通两路皆空），请重试或重拍本页")
meta = {"page_raw": "无", "header": "无", "footer": "无", "body": (text or "").strip(),
"page_num": None, "parse_ok": False, "fallback": "plain-vl",
"model": meta.get("model"), "stem": stem,
"at": _dt.datetime.now().isoformat(timespec="seconds")}
(book_dir / f"{stem}.vl.json").write_text(json.dumps(meta, ensure_ascii=False), encoding="utf-8")
return meta.get("body") or ""
def _read_vl_meta(book_dir: Path, stem: str) -> dict | None:
try:
m = json.loads((book_dir / f"{stem}.vl.json").read_text(encoding="utf-8"))
return m if isinstance(m, dict) else None
except Exception: # noqa: BLE001
return None
# ---------------------------------------------------------------------------
# 服务器端识别队列：照片先上传存档（pending），OCR 由后台队列逐条处理
# ---------------------------------------------------------------------------
_OCR_QUEUE: collections.deque = collections.deque()
_OCR_QUEUED: set = set() # (book, stem)
_OCR_ACTIVE: set = set() # (book, stem)
_OCR_WORKER = None
_OCR_WORKER_LOCK = threading.Lock()
def _meta_engine(book_dir: Path, stem: str, default: str = "local") -> str:
try:
e = json.loads((book_dir / f"{stem}.meta.json").read_text(encoding="utf-8")).get("engine", default)
return e if e in ("local", "vl", "vl2") else default
except Exception: # noqa: BLE001
return default
def _save_page_text(book_dir: Path, stem: str, text: str, *, force: bool = False):
"""保存页面正文前的安全闸：空结果、或比现有正文骤减（<30%）时**拒绝覆盖**并写 err.json。
背景：VL 偶发返回空正文，旧实现无条件覆盖，会把已识别好的正文清空。
返回 (是否写入, 说明)。force=True 时跳过闸门（人工修正/强制重跑用）。
"""
txt = book_dir / f"{stem}.txt"
new_len = len((text or "").strip())
old_len = len(txt.read_text(encoding="utf-8").strip()) if txt.is_file() else 0
if not force and (new_len == 0 or (old_len >= 200 and new_len < old_len * 0.3)):
(book_dir / f"{stem}.err.json").write_text(
json.dumps({"error": f"识别结果异常：新结果 {new_len} 字，原正文 {old_len} 字，已保留原正文（请重试）",
"at": _dt.datetime.now().isoformat(timespec="seconds")}, ensure_ascii=False),
encoding="utf-8")
return False, f"识别结果异常（新 {new_len} 字 / 原 {old_len} 字），已保留原正文，请重试"
txt.write_text(text, encoding="utf-8")
return True, None
def _run_page_ocr(book_dir: Path, stem: str, engine: str):
"""对已存档的一页执行 OCR 并写盘（txt 或 err.json）。
engine: local=本地 | vl=旧式整页 VL | vl2=结构化 VL（页码/页眉/页脚/正文）"""
jpg = book_dir / f"{stem}.jpg"
txt = book_dir / f"{stem}.txt"
err = book_dir / f"{stem}.err.json"
try:
if not jpg.is_file():
raise RuntimeError("服务器上照片不存在")
img = _load_image(jpg.read_bytes())
if engine == "local":
text, _sec, _diag = ocr_local(img)
(book_dir / f"{stem}.vl.json").unlink(missing_ok=True) # 本地结果替代后旧 VL 元数据失效
elif engine == "vl2":
text = run_vl_structured_page(book_dir, stem) # 内部写 .vl.json
else:
if not vl_ready():
engine = "local"
text, _sec, _diag = ocr_local(img)
else:
text, _sec, _diag = ocr_vl(img)
ok_save, why = _save_page_text(book_dir, stem, text)
if not ok_save:
raise RuntimeError(why)
err.unlink(missing_ok=True)
except Exception as e: # noqa: BLE001
try:
err.write_text(json.dumps({"error": str(e)[:500], "at": _dt.datetime.now().isoformat(timespec="seconds")},
ensure_ascii=False), encoding="utf-8")
except Exception: # noqa: BLE001
pass
def _schedule_ocr(book_name: str, stem: str):
key = (book_name, stem)
if key in _OCR_QUEUED or key in _OCR_ACTIVE:
return
_OCR_QUEUED.add(key)
_OCR_QUEUE.append(key)
_ensure_ocr_worker()
def _ensure_ocr_worker():
global _OCR_WORKER
with _OCR_WORKER_LOCK:
if _OCR_WORKER is None or not _OCR_WORKER.is_alive():
_OCR_WORKER = threading.Thread(target=_ocr_worker_loop, daemon=True, name="ocr-worker")
_OCR_WORKER.start()
def _ocr_worker_loop():
while True:
try:
book, stem = _OCR_QUEUE.popleft()
except IndexError:
time.sleep(0.4)
continue
key = (book, stem)
_OCR_QUEUED.discard(key)
_OCR_ACTIVE.add(key)
try:
book_dir = BOOKS_DIR / book
if book_dir.is_dir():
_run_page_ocr(book_dir, stem, _meta_engine(book_dir, stem))
finally:
_OCR_ACTIVE.discard(key)
def _page_status(rec, book_name: str):
"""由磁盘 + 队列状态推导页状态：queued/running/error/done/pending"""
key = (book_name, rec["stem"])
if key in _OCR_ACTIVE:
return "running", None
if key in _OCR_QUEUED:
return "queued", None
if rec["err"].is_file():
try:
msg = json.loads(rec["err"].read_text(encoding="utf-8")).get("error", "识别失败")
except Exception: # noqa: BLE001
msg = "识别失败"
return "error", msg
if rec["txt"].is_file():
return "done", None
return "pending", None
@app.get("/")
def index():
return send_from_directory(STATIC_DIR, "index.html")
@app.get("/api/health")
def health():
return jsonify(
{
"ok": True,
"service": "paishu-quzi",
"local_engine": "rapidocr",
"vl_enabled": vl_ready(),
"vl_model": DEEPSEEK_MODEL if vl_ready() else None,
}
)
def _make_stem(book_dir: Path, seq: int) -> str:
base = f"第{seq:03d}页_{int(time.time() * 1000)}"
s, n = base, 1
while (book_dir / f"{s}.jpg").exists() or (book_dir / f"{s}.txt").exists():
s = f"{base}_{n}"
n += 1
return s
@app.errorhandler(413)
def _handle_413(e): # noqa: ARG001
"""请求体超限时返回 JSON，前端据此提示"分批上传"。"""
mb = app.config["MAX_CONTENT_LENGTH"] / 1048576
return jsonify(ok=False,
error="本次上传超过服务器单次上限（约 %.0fMB）。请分批上传：每批 ≤%d 张、每张 ≤%dMB。"
% (mb, MAX_FILES_PER_UPLOAD, MAX_FILE_BYTES // 1048576),
limit_mb=round(mb)), 413
@app.post("/api/photo")
def api_photo():
"""照片即时上传存档（不识别，识别进后台队列逐条处理）。
multipart：files[]（可多张）+ image(单张兼容) + book_name + engine=local|vl
返回 {ok, book, engine, items:[{seq, stem, status}], failed:[{name,error}]}
"""
files = request.files.getlist("files")
if not files and "image" in request.files:
files = [request.files["image"]]
files = [f for f in files if f and f.filename]
if not files:
return jsonify(ok=False, error="缺少图片文件（字段 files[] 或 image）"), 400
if len(files) > MAX_FILES_PER_UPLOAD:
return jsonify(ok=False,
error="单次最多上传 %d 张（本次 %d 张），请分批上传"
% (MAX_FILES_PER_UPLOAD, len(files)),
max_files=MAX_FILES_PER_UPLOAD), 400
engine = (request.form.get("engine") or "local").strip().lower()
if engine not in ("local", "vl"):
return jsonify(ok=False, error="engine 只支持 local / vl"), 400
if engine == "vl" and not vl_ready():
engine = "local"
book_name = _sanitize(request.form.get("book_name"), f"未命名书_{_dt.datetime.now():%Y%m%d}")
book_dir = BOOKS_DIR / book_name
book_dir.mkdir(parents=True, exist_ok=True)
max_seq = max((r["seq"] for r in _page_records(book_dir)), default=0)
items, failed = [], []
for idx, f in enumerate(files, start=1):
name = f.filename
try:
raw = f.read()
if len(raw) < 100:
raise RuntimeError("文件为空")
if len(raw) > MAX_FILE_BYTES:
raise RuntimeError("单张超过 %dMB 上限（本张 %.1fMB）"
% (MAX_FILE_BYTES // 1048576, len(raw) / 1048576))
# 即时解码/EXIF 摆正/压缩并落盘（照片上传成功 = 已上服务器）
img = _load_image(raw)
seq = max_seq + idx
stem = _make_stem(book_dir, seq)
(book_dir / f"{stem}.jpg").write_bytes(_image_bytes(img))
meta_engine = "vl2" if engine == "vl" else "local" # VL 一律走结构化（页码/页眉/页脚/正文）
(book_dir / f"{stem}.meta.json").write_text(
json.dumps({"engine": meta_engine}, ensure_ascii=False), encoding="utf-8")
items.append({"seq": seq, "stem": stem, "status": "queued"})
except Exception as e: # noqa: BLE001
failed.append({"name": name, "error": str(e)[:200]})
if items:
for it in items:
_schedule_ocr(book_name, it["stem"])
return jsonify(ok=bool(items), book=book_name, engine=engine,
items=items, failed=failed,
n=len(items), n_failed=len(failed))
@app.post("/api/batch/process")
def api_batch_process():
"""把某书（或全部书）里“已上传但还没识别/失败”的页重新排入识别队列。
body: {book_name?: str, retry_errors?: bool}"""
data = request.get_json(silent=True) or {}
retry_errors = bool(data.get("retry_errors"))
targets: list[Path]
if data.get("book_name"):
b, d = _book_dir_or_400(data.get("book_name"))
if not b:
return jsonify(ok=False, error=d), 400
if not d.is_dir():
return jsonify(ok=False, error=f"项目《{b}》不存在"), 404
targets = [d]
else:
targets = [d for d in BOOKS_DIR.iterdir() if d.is_dir()]
scheduled = []
for d in targets:
for rec in _page_records(d):
st, _msg = _page_status(rec, d.name)
if st == "done":
continue
if st == "error" and not retry_errors:
continue
_schedule_ocr(d.name, rec["stem"])
scheduled.append({"book": d.name, "stem": rec["stem"], "status": "queued"})
return jsonify(ok=True, scheduled=scheduled, n=len(scheduled))
@app.post("/api/batch/vlall")
def api_batch_vlall():
"""整书 VL 批量：把项目页面全部/待处理页标记为结构化 VL（vl2）并排入识别队列。
body: {book_name, scope:'all'|'todo', retry_errors?:bool}
'all' 会用 VL 重跑所有页（含已本地识别的页）；'todo' 只处理 待识别/失败 页。"""
data = request.get_json(silent=True) or {}
b, d = _book_dir_or_400(data.get("book_name"))
if not b:
return jsonify(ok=False, error=d), 400
if not d.is_dir():
return jsonify(ok=False, error=f"项目《{b}》不存在"), 404
if not vl_ready():
return jsonify(ok=False, error="DeepSeek VL 未启用（未配置 key）"), 400
scope = (data.get("scope") or "all").strip().lower()
if scope not in ("all", "todo"):
return jsonify(ok=False, error="scope 只支持 all / todo"), 400
retry_errors = bool(data.get("retry_errors"))
stem_whitelist = {str(s) for s in (data.get("stems") or [])}
scheduled, running = [], []
for rec in _page_records(d):
st, _ = _page_status(rec, b)
if stem_whitelist and rec["stem"] not in stem_whitelist:
continue
if st in ("queued", "running"):
running.append(rec["stem"])
continue
if scope == "todo" and st == "done":
continue
if st == "error" and not retry_errors:
continue
(d / f"{rec['stem']}.meta.json").write_text(
json.dumps({"engine": "vl2"}, ensure_ascii=False), encoding="utf-8")
_schedule_ocr(b, rec["stem"])
scheduled.append(rec["stem"])
return jsonify(ok=True, book=b, scope=scope, n=len(scheduled),
running=running, model=DEEPSEEK_MODEL)
@app.post("/api/ocr")
def api_ocr():
upload = request.files.get("image")
if upload is None or not upload.filename:
return jsonify(ok=False, error="缺少图片文件（字段名 image）"), 400
ext = upload.filename.rsplit(".", 1)[-1].lower() if "." in upload.filename else ""
if ext not in ALLOWED_EXT:
return jsonify(ok=False, error=f"不支持的图片格式: .{ext}"), 400
raw = upload.read()
if len(raw) < 100:
return jsonify(ok=False, error="图片内容为空"), 400
engine = (request.form.get("engine") or "local").strip().lower()
if engine not in ("local", "vl"):
return jsonify(ok=False, error="engine 参数只支持 local / vl"), 400
if engine == "vl" and not vl_ready():
return jsonify(ok=False, error="DeepSeek VL 未启用：服务器未配置 API key"), 400
book_name = _sanitize(
request.form.get("book_name"),
f"未命名书_{_dt.datetime.now():%Y%m%d}",
)
try:
seq = max(1, int(request.form.get("seq") or 1))
except ValueError:
seq = 1
try:
img = _load_image(raw)
except Exception as e: # noqa: BLE001
return jsonify(ok=False, error=f"图片解码失败: {e}"), 400
# ---- 存档目录：data/books/<书名>/ ----
book_dir = BOOKS_DIR / book_name
book_dir.mkdir(parents=True, exist_ok=True)
stem = f"第{seq:03d}页_{_now_tag()}"
try:
if engine == "vl":
text, sec, diag = ocr_vl(img)
else:
text, sec, diag = ocr_local(img)
except Exception as e: # noqa: BLE001
return jsonify(ok=False, error=f"识别失败({engine}): {e}"), 500
# ---- 落盘：照片 + 文本 ----
photo_path = book_dir / f"{stem}{PHOTO_EXT}"
photo_path.write_bytes(_image_bytes(img))
txt_path = book_dir / f"{stem}.txt"
txt_path.write_text(text, encoding="utf-8")
rel = f"{photo_path.relative_to(BOOKS_DIR)}" # <书名>/第xxx页.jpg（相对 data/books）
rel_txt = f"{photo_path.relative_to(BOOKS_DIR).with_suffix('.txt')}"
return jsonify(
ok=True,
engine=engine,
text=text,
chars=len(text.strip()),
sec=round(sec, 2),
diag=diag,
book=book_name,
seq=seq,
stem=stem, # 页码文件主名（第xxx页_时间戳），供整书合成引用
photo=f"/f/{rel}",
thumb=f"/f/{rel}?w=480",
text_file=f"/txt/{rel_txt}",
saved_at=_dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
)
@app.get("/pc")
def index_pc():
"""PC 端页面（拍摄 + VL 修正）。摄像头需经 HTTPS 访问。"""
return send_from_directory(STATIC_DIR, "index_pc.html")
@app.post("/api/page/reocr")
def api_page_reocr():
"""对服务器已存档的一页重跑 OCR。
body(json): {book_name, stem, engine: local|vl, save?: bool}
engine=vl 走**结构化 VL**（识别页码/页眉/页脚并写 .vl.json，正文为 txt）；
save=true 时结果覆盖该页存档 txt（供整书合成使用）。
"""
data = request.get_json(silent=True) or {}
book_name = _sanitize(data.get("book_name"), "")
stem = str(data.get("stem") or "").strip()
if not book_name:
return jsonify(ok=False, error="缺少 book_name"), 400
if not re.fullmatch(r"[\w\u4e00-\u9fff.-]+", stem) or stem.startswith("."):
return jsonify(ok=False, error="非法 stem"), 400
engine = (data.get("engine") or "local").strip().lower()
if engine not in ("local", "vl"):
return jsonify(ok=False, error="engine 只支持 local / vl"), 400
if engine == "vl" and not vl_ready():
return jsonify(ok=False, error="DeepSeek VL 未启用：服务器未配置 API key"), 400
book_dir = BOOKS_DIR / book_name
jpg_path = (book_dir / f"{stem}.jpg").resolve()
if not str(jpg_path).startswith(str(BOOKS_DIR.resolve())) or not jpg_path.is_file():
return jsonify(ok=False, error="服务器上没有该页照片，请重新拍摄/上传本页"), 404
extra: dict = {}
try:
img = _load_image(jpg_path.read_bytes())
if engine == "vl":
meta, sec, diag = ocr_vl_struct(img) # 结构化：页码/页眉/页脚/正文
text = meta.get("body") or ""
extra = {"page_num": meta.get("page_num"), "page_raw": meta.get("page_raw"),
"header": meta.get("header"), "footer": meta.get("footer")}
(book_dir / f"{stem}.vl.json").write_text(
json.dumps({**meta, "stem": stem, "at": _dt.datetime.now().isoformat(timespec="seconds")},
ensure_ascii=False), encoding="utf-8")
else:
text, sec, diag = ocr_local(img)
except Exception as e: # noqa: BLE001
return jsonify(ok=False, error=f"识别失败({engine}): {e}"), 500
saved = False
if data.get("save"):
force = bool(data.get("force"))
ok_save, why = _save_page_text(book_dir, stem, text, force=force)
if not ok_save:
if engine == "vl":
(book_dir / f"{stem}.vl.json").unlink(missing_ok=True) # 空结果不留下坏元数据
return jsonify(ok=False, error=why, chars=len(text.strip()), engine=engine,
book=book_name, stem=stem, saved=False), 422
if engine == "local":
(book_dir / f"{stem}.vl.json").unlink(missing_ok=True)
saved = True
return jsonify(
ok=True,
engine=engine,
text=text,
chars=len(text.strip()),
sec=round(sec, 2),
diag=diag,
book=book_name,
stem=stem,
saved=saved,
**extra,
)
@app.post("/api/page/text")
def api_page_text():
"""手动修正：覆盖存档的一页文本。body: {book_name, stem, text}"""
data = request.get_json(silent=True) or {}
book_name = _sanitize(data.get("book_name"), "")
stem = str(data.get("stem") or "").strip()
if not book_name:
return jsonify(ok=False, error="缺少 book_name"), 400
if not re.fullmatch(r"[\w\u4e00-\u9fff.-]+", stem) or stem.startswith("."):
return jsonify(ok=False, error="非法 stem"), 400
book_dir = BOOKS_DIR / book_name
txt_path = (book_dir / f"{stem}.txt").resolve()
if not str(txt_path).startswith(str(BOOKS_DIR.resolve())) or not txt_path.is_file():
return jsonify(ok=False, error="存档页不存在（请先识别该页）"), 404
text = str(data.get("text") or "")
txt_path.write_text(text, encoding="utf-8")
return jsonify(ok=True, book=book_name, stem=stem, chars=len(text.strip()))
# ---------------------------------------------------------------------------
# 项目管理（浏览 / 修正 / 整书合成 / 重命名 / 删除，数据即 data/books/<书名>/）
# ---------------------------------------------------------------------------
_PAGE_RE = re.compile(r"^第(\d+)页_(.+)$") # 第001页_<时间戳[+序号]>，后缀任意
def _seq_of_stem(stem: str) -> int:
m = re.match(r"^第(\d+)页", stem)
return int(m.group(1)) if m else 0
def _page_records(book_dir: Path):
"""按 (页码, 文件名) 排序返回该书所有页记录（jpg 即存在=照片已上传服务器）。
只认“第NNN页_…”开头的文件名，避免把书目录里的成品文件(如 第1章_…_纯净全文.txt)误当页。"""
stems = set()
for p in book_dir.iterdir():
if p.suffix in (".jpg", ".txt") and re.match(r"^第\d+页_", p.name):
stems.add(p.stem)
recs = []
for s in stems:
recs.append({
"stem": s,
"seq": _seq_of_stem(s),
"jpg": book_dir / f"{s}.jpg",
"txt": book_dir / f"{s}.txt",
"err": book_dir / f"{s}.err.json",
})
recs.sort(key=lambda r: (r["seq"], r["stem"]))
return recs
def _status_of(rec) -> tuple[str, str | None]:
if rec["err"].is_file():
try:
msg = json.loads(rec["err"].read_text(encoding="utf-8")).get("error", "识别失败")
except Exception: # noqa: BLE001
msg = "识别失败"
return "error", msg
if rec["txt"].is_file():
return "done", None
return "pending", None
def _book_dir_or_400(name: str):
"""校验书名并返回 (sanitized_name, dir)；非法返回 (None, errstr)。"""
b = _sanitize(name or "", "")
if not b:
return None, "缺少有效的 book_name"
d = (BOOKS_DIR / b).resolve()
if not str(d).startswith(str(BOOKS_DIR.resolve()) + os.sep):
return None, "路径越界"
return b, d
def _list_pages(book_dir: Path):
"""返回按 (页码, 时间戳) 排序的页记录：{stem, seq, txt, jpg}"""
rows = []
for txt in book_dir.glob("第*页_*.txt"):
m = _PAGE_RE.match(txt.stem)
if not m:
continue
jpg = txt.with_suffix(".jpg")
rows.append({"stem": txt.stem, "seq": int(m.group(1)),
"txt": txt, "jpg": jpg if jpg.is_file() else None})
rows.sort(key=lambda r: (r["seq"], r["txt"].name))
return rows
@app.get("/api/books")
def api_books_list():
"""列出服务器上全部项目（= data/books/ 下目录）及统计（含待识别/失败页）。"""
out = []
for d in sorted(BOOKS_DIR.iterdir(), key=lambda p: p.stat().st_mtime if p.is_dir() else 0, reverse=True):
if not d.is_dir() or d.name.startswith("."):
continue
recs = _page_records(d)
chars = 0
size = 0
updated = 0
done = todo = errors = 0
for r in recs:
st, _ = _page_status(r, d.name)
if st == "done":
done += 1
elif st == "error":
errors += 1
else:
todo += 1
for f in (r["jpg"], r["txt"], r["err"]):
if f and f.is_file():
try:
size += f.stat().st_size
updated = max(updated, f.stat().st_mtime)
except OSError:
pass
if r["txt"].is_file():
try:
chars += len(r["txt"].read_text(encoding="utf-8", errors="ignore").strip())
except OSError:
pass
out.append({
"name": d.name,
"pages": len(recs),
"done": done,
"todo": todo,
"errors": errors,
"chars": chars,
"size": size,
"updated_at": _dt.datetime.fromtimestamp(updated).strftime("%Y-%m-%d %H:%M:%S") if updated else "-",
"first": f"/f/{d.name}/{recs[0]['stem']}.jpg?w=480" if recs and recs[0]["jpg"].is_file() else None,
})
TRASH_DIR.mkdir(parents=True, exist_ok=True)
n_trash = sum(1 for p in TRASH_DIR.iterdir() if p.is_dir())
return jsonify(ok=True, books=out, trash=n_trash)
@app.get("/api/books//pages")
def api_book_pages(book: str):
"""列出某项目全部页（含状态：done/pending/queued/running/error）；?text=1 附每页全文。"""
b, d = _book_dir_or_400(book)
if not b:
return jsonify(ok=False, error=d), 400
if not d.is_dir():
return jsonify(ok=False, error=f"项目《{b}》不存在"), 404
with_text = request.args.get("text") == "1"
out = []
for rec in _page_records(d):
st, err_msg = _page_status(rec, b)
vl = _read_vl_meta(d, rec["stem"])
item = {
"stem": rec["stem"], "seq": rec["seq"], "status": st, "err": err_msg,
"thumb": f"/f/{b}/{rec['stem']}.jpg?w=480" if rec["jpg"].is_file() else None,
"photo": f"/f/{b}/{rec['stem']}.jpg" if rec["jpg"].is_file() else None,
"time": _dt.datetime.fromtimestamp(
max(p.stat().st_mtime for p in (rec["jpg"], rec["txt"], rec["err"]) if p.is_file())
).strftime("%Y-%m-%d %H:%M:%S"),
}
if vl:
item["vl"] = True
item["page_num"] = vl.get("page_num")
item["page_raw"] = (vl.get("page_raw") or "")[:24]
item["header"] = (vl.get("header") or "")[:80]
item["footer"] = (vl.get("footer") or "")[:80]
if rec["txt"].is_file():
item["txt_url"] = f"/txt/{b}/{rec['stem']}.txt"
if with_text:
if rec["txt"].is_file():
try:
item["text"] = rec["txt"].read_text(encoding="utf-8", errors="ignore")
item["chars"] = len(item["text"].strip())
except OSError:
item["text"], item["chars"] = "", 0
else:
item["text"], item["chars"] = "", 0
out.append(item)
return jsonify(ok=True, book=b, pages=out)
@app.post("/api/books/delete")
def api_book_delete():
"""删除整个项目：先移入回收站（data/trash，可恢复），不直接删。body: {book_name}"""
data = request.get_json(silent=True) or {}
b, d = _book_dir_or_400(data.get("book_name"))
if not b:
return jsonify(ok=False, error=d), 400
if not d.is_dir():
return jsonify(ok=False, error=f"项目《{b}》不存在"), 404
TRASH_DIR.mkdir(parents=True, exist_ok=True)
entry = TRASH_DIR / f"{int(time.time() * 1000)}_book"
os.rename(d, entry) # 同文件系统改名即"移动"
(entry / "meta.json").write_text(
json.dumps({"kind": "book", "book": b}, ensure_ascii=False), encoding="utf-8"
)
return jsonify(ok=True, moved_to_trash=True, trash=entry.name, book=b)
@app.post("/api/books/rename")
def api_book_rename():
"""重命名项目目录。body: {book_name, new_name}"""
data = request.get_json(silent=True) or {}
b, d = _book_dir_or_400(data.get("book_name"))
if not b:
return jsonify(ok=False, error=d), 400
new_name, _ = _book_dir_or_400(data.get("new_name"))
if not new_name:
return jsonify(ok=False, error="缺少有效的新名称"), 400
if new_name == b:
return jsonify(ok=False, error="新旧名称相同"), 400
if not d.is_dir():
return jsonify(ok=False, error=f"项目《{b}》不存在"), 404
nd = BOOKS_DIR / new_name
if nd.exists():
return jsonify(ok=False, error=f"已存在同名项目《{new_name}》"), 409
os.rename(d, nd)
return jsonify(ok=True, old=b, new=new_name)
@app.post("/api/page/delete")
def api_page_delete():
"""删除某一页：先移入回收站（可恢复），不直接删。body: {book_name, stem}"""
data = request.get_json(silent=True) or {}
b, d = _book_dir_or_400(data.get("book_name"))
if not b:
return jsonify(ok=False, error=d), 400
stem = str(data.get("stem") or "").strip()
if not re.fullmatch(r"[\w\u4e00-\u9fff.-]+", stem) or stem.startswith("."):
return jsonify(ok=False, error="非法 stem"), 400
if not d.is_dir():
return jsonify(ok=False, error=f"项目《{b}》不存在"), 404
TRASH_DIR.mkdir(parents=True, exist_ok=True)
entry = TRASH_DIR / f"{int(time.time() * 1000)}_page"
entry.mkdir()
moved = []
for suffix in (".txt", ".jpg"):
f = (d / f"{stem}{suffix}").resolve()
if str(f).startswith(str(d.resolve()) + os.sep) and f.is_file():
os.rename(f, entry / f"{stem}{suffix}")
moved.append(f"{stem}{suffix}")
(entry / "meta.json").write_text(
json.dumps({"kind": "page", "book": b, "stem": stem}, ensure_ascii=False),
encoding="utf-8",
)
return jsonify(ok=True, book=b, stem=stem, removed=moved, trash=entry.name)
# ---------------------------------------------------------------------------
# 回收站（data/trash）：列表 / 恢复 / 永久删除
# ---------------------------------------------------------------------------
@app.get("/api/trash")
def api_trash_list():
TRASH_DIR.mkdir(parents=True, exist_ok=True)
out = []
for d in sorted(TRASH_DIR.iterdir(), key=lambda p: p.stat().st_mtime, reverse=True):
if not d.is_dir():
continue
try:
meta = json.loads((d / "meta.json").read_text(encoding="utf-8"))
except Exception: # noqa: BLE001
continue
files = [f.name for f in d.iterdir() if f.is_file() and f.name != "meta.json"]
size = sum(f.stat().st_size for f in d.iterdir() if f.is_file())
out.append({
"entry": d.name,
"kind": meta.get("kind"),
"book": meta.get("book"),
"stem": meta.get("stem"),
"files": len(files),
"size": size,
"time": _dt.datetime.fromtimestamp(d.stat().st_mtime).strftime("%Y-%m-%d %H:%M:%S"),
})
return jsonify(ok=True, entries=out)
@app.post("/api/trash/restore")
def api_trash_restore():
"""从回收站恢复。body: {entry}（列表里的 entry 名）"""
data = request.get_json(silent=True) or {}
name = str(data.get("entry") or "").strip()
entry = (TRASH_DIR / name).resolve()
if not name or not str(entry).startswith(str(TRASH_DIR.resolve()) + os.sep) or not entry.is_dir():
return jsonify(ok=False, error="回收站条目不存在"), 404
try:
meta = json.loads((entry / "meta.json").read_text(encoding="utf-8"))
except Exception: # noqa: BLE001
return jsonify(ok=False, error="条目信息损坏"), 400
kind, book = meta.get("kind"), meta.get("book")
if kind == "book":
target = BOOKS_DIR / book
if target.exists():
target = BOOKS_DIR / f"{book}_已恢复{int(time.time())}"
os.rename(entry, target)
(target / "meta.json").unlink(missing_ok=True)
return jsonify(ok=True, kind=kind, book=book, to=target.name)
if kind == "page":
stem = meta.get("stem")
book_dir = BOOKS_DIR / book
book_dir.mkdir(parents=True, exist_ok=True)
restored, conflict = [], []
for f in entry.iterdir():
if f.name == "meta.json" or not f.is_file():
continue
tgt = book_dir / f.name
if tgt.exists():
tgt = book_dir / f"{f.stem}_恢复{f.suffix}"
conflict.append(tgt.name)
os.rename(f, tgt)
restored.append(tgt.name)
(entry / "meta.json").unlink(missing_ok=True)
entry.rmdir()
return jsonify(ok=True, kind=kind, book=book, stem=stem, restored=restored,
conflict=conflict)
return jsonify(ok=False, error="未知条目类型"), 400
@app.post("/api/trash/purge")
def api_trash_purge():
"""永久删除回收站条目（真正不可恢复）。body: {entry}"""
data = request.get_json(silent=True) or {}
name = str(data.get("entry") or "").strip()
entry = (TRASH_DIR / name).resolve()
if not name or not str(entry).startswith(str(TRASH_DIR.resolve()) + os.sep) or not entry.is_dir():
return jsonify(ok=False, error="回收站条目不存在"), 404
shutil.rmtree(entry)
return jsonify(ok=True, purged=name)
@app.get("/f/")
def photo_file(filepath: str):
"""读取存档照片；?w=N 时实时缩放做缩略图（LAN 场景够用）。"""
p = (BOOKS_DIR / filepath).resolve()
if not str(p).startswith(str(BOOKS_DIR.resolve())):
return jsonify(ok=False, error="路径越界"), 400
if not p.is_file():
return jsonify(ok=False, error="文件不存在"), 404
want_w = request.args.get("w", type=int)
if want_w:
from PIL import Image
try:
im = Image.open(p)
im = ImageOps_rotate_thumb(im, want_w)
return send_file(io.BytesIO(_image_bytes(im)), mimetype="image/jpeg")
except Exception:
pass
return send_file(p, mimetype=mimetypes.guess_type(p.name)[0] or "application/octet-stream")
def ImageOps_rotate_thumb(im, want_w):
from PIL import Image, ImageOps
im = ImageOps.exif_transpose(im)
w, h = im.size
if w > want_w:
im = im.resize((want_w, int(h * want_w / w)), Image.LANCZOS)
return im
# ---------------------------------------------------------------------------
# 导出全套成品（服务器统一生成）：正文标注版 / 纯净全文 / 含页眉页脚txt / 含页眉页脚MD
# ---------------------------------------------------------------------------
def _rows_for_book(book_dir: Path, book_name: str) -> dict:
"""folder 级取行并排序（与 /api/batch/export 的 folder 逻辑一致）：
章首页置前 → 印刷页码排序（无页码页插值）→ 相似重复页去重。"""
from difflib import SequenceMatcher as _SM
rows = []
skipped = []
for rec in _page_records(book_dir):
if rec["txt"].is_file():
rows.append({"stem": rec["stem"], "seq": rec["seq"],
"path": rec["txt"], "vl": _read_vl_meta(book_dir, rec["stem"])})
else:
st, em = _page_status(rec, book_name)
skipped.append({"seq": rec["seq"], "stem": rec["stem"],
"reason": em or ("待识别" if st in ("pending", "queued", "running") else f"状态:{st}")})
for i, r in enumerate(rows):
r["i"] = i
try:
r["body"] = r["path"].read_text(encoding="utf-8").strip()
except OSError:
r["body"] = ""
r["front"] = bool(not (r["vl"] or {}).get("page_num")
and re.match(r"^第[0-9一二三四五六七八九十百两]+章", r["body"]))
def _dedupe(lst):
kept, dropped = [], []
for r in lst:
dup = bool(r["body"]) and any(
_SM(None, r["body"], k["body"]).ratio() > 0.9 for k in kept if k["body"])
(dropped if dup else kept).append(r)
return kept, dropped
front_rows = sorted([r for r in rows if r["front"]], key=lambda r: (r["seq"], r["i"]))
front_kept, front_dropped = _dedupe(front_rows)
rest_rows = [r for r in rows if not r["front"]]
for i, r in enumerate(rest_rows):
r["i"] = i
any_num = any((r["vl"] or {}).get("page_num") is not None for r in rest_rows)
if any_num:
seq_sorted = sorted(rest_rows, key=lambda r: (r["seq"], r["i"]))
nums = {r["stem"]: (r["vl"] or {}).get("page_num") for r in rest_rows}
def _est(idx):
if nums[seq_sorted[idx]["stem"]] is not None:
return float(nums[seq_sorted[idx]["stem"]])
below = above = None
for j in range(idx - 1, -1, -1):
if nums[seq_sorted[j]["stem"]] is not None:
below = (j, float(nums[seq_sorted[j]["stem"]]))
break
for k in range(idx + 1, len(seq_sorted)):
if nums[seq_sorted[k]["stem"]] is not None:
above = (k, float(nums[seq_sorted[k]["stem"]]))
break
if below and above:
(j, v1), (k, v2) = below, above
return v1 + (v2 - v1) * (idx - j) / (k - j)
if above:
return above[1] - (above[0] - idx)
if below:
return below[1] + (idx - below[0])
return 1e9 + idx
est = {seq_sorted[idx]["stem"]: _est(idx) for idx in range(len(seq_sorted))}
rest_rows.sort(key=lambda r: (est[r["stem"]], r["i"]))
ordered_by = "page_num"
else:
rest_rows.sort(key=lambda r: (r["seq"], r["i"]))
ordered_by = "seq"
rest_kept, rest_dropped = _dedupe(rest_rows)
order = front_kept + rest_kept
return {"rows": order, "skipped": skipped,
"dropped": front_dropped + rest_dropped, "ordered_by": ordered_by}
def _row_label(r: dict) -> str:
label = f"第 {r['seq']} 页"
raw = (r["vl"] or {}).get("page_raw")
if raw not in (None, "", "无"):
label += f"（原书页码 {raw}）"
elif r["front"]:
label += "（章首页·无页码）"
return label
def _row_content(r: dict, include_side: bool) -> str:
if include_side and r["vl"]:
seg = []
if r["vl"].get("header") not in (None, "", "无"):
seg.append(f"【页眉】{r['vl']['header']}")
if r["body"]:
seg.append(r["body"])
if r["vl"].get("footer") not in (None, "", "无"):
seg.append(f"【页脚】{r['vl']['footer']}")
return "\n".join(seg)
return r["body"]
def _unique_path(book_dir: Path, prefix: str, suffix: str = ".txt") -> Path:
now = _dt.datetime.now().strftime("%Y%m%d_%H%M%S")
p, n = book_dir / f"{prefix}_{now}{suffix}", 1
while p.exists():
n += 1
p = book_dir / f"{prefix}_{now}_{n}{suffix}"
return p
def _paras(text: str) -> list[str]:
ps = []
for chunk in (text or "").split("\n\n"):
chunk = chunk.strip("\n")
if chunk.strip():
ps.append(chunk)
return ps
@app.post("/api/batch/exportall")
def api_batch_exportall():
"""一键导出该项目全套成品（正文标注版 / 纯净全文 / 含页眉页脚 txt / 含页眉页脚 md）。
body: {book_name} → 四份文件写入 data/books/<书>/ 并返回下载链接。"""
data = request.get_json(silent=True) or {}
b, d = _book_dir_or_400(data.get("book_name"))
if not b:
return jsonify(ok=False, error=d), 400
if not d.is_dir():
return jsonify(ok=False, error=f"项目《{b}》不存在"), 404
res = _rows_for_book(d, b)
rows, skipped, dropped = res["rows"], res["skipped"], res["dropped"]
if not rows:
return jsonify(ok=False, error="没有可导出的已识别页"), 404
ts = _dt.datetime.now()
title_line = b.replace("_", " ").strip()
files: list[dict] = []
# 1) 正文标注版（每页带分隔与页码标注）
parts = []
for r in rows:
body = _row_content(r, include_side=False)
parts.append(f"────────── {_row_label(r)} ──────────\n{body if body else '（本页未能识别出文字）'}")
head = [f"《{b}》整书文本（带页码标注）",
f"生成时间：{ts:%Y-%m-%d %H:%M:%S} 共 {len(parts)} 页 {sum(len(p) for p in parts)} 字",
f"排序依据：{'VL 印刷页码（章首页置前）' if res['ordered_by'] == 'page_num' else '拍摄顺序'} 正文（剔除页眉页脚页码）"]
if dropped:
head.append(f"（自动合并 {len(dropped)} 页重复拍摄）")
if skipped:
head.append(f"（{len(skipped)} 页未识别/失败未并入）")
text1 = "\n".join(head) + "\n\n" + "\n\n".join(parts) + "\n"
p1 = _unique_path(d, "全书_正文标注")
p1.write_text(text1, encoding="utf-8")
files.append({"tag": "正文标注版", "name": p1.name,
"url": f"/txt/{p1.relative_to(BOOKS_DIR)}", "chars": len(text1.strip())})
# 2) 纯净全文（无任何装饰行，页间空行）
clean_paras = []
for r in rows:
clean_paras.append("\n\n".join(_paras(r["body"])))
text2 = "\n\n".join(clean_paras).strip() + "\n"
p2 = _unique_path(d, "全书_纯净全文")
p2.write_text(text2, encoding="utf-8")
files.append({"tag": "纯净全文", "name": p2.name,
"url": f"/txt/{p2.relative_to(BOOKS_DIR)}", "chars": len(text2.replace("\n", ""))})
# 3) 含页眉页脚 txt
parts3 = []
for r in rows:
c = _row_content(r, include_side=True)
parts3.append(f"────────── {_row_label(r)} ──────────\n{c if c else '（本页未能识别出文字）'}")
text3 = "\n".join(head) + "\n\n" + "\n\n".join(parts3) + "\n"
p3 = _unique_path(d, "全书_含页眉页脚")
p3.write_text(text3, encoding="utf-8")
files.append({"tag": "含页眉页脚(txt)", "name": p3.name,
"url": f"/txt/{p3.relative_to(BOOKS_DIR)}", "chars": len(text3.strip())})
# 4) 含页眉页脚 Markdown（带格式）
md = [f"# {title_line}", "",
f"> 📄 **版本**：含页眉页脚（页眉/页脚以引用样式标注） · **整理**：拍书取字（VL 结构化识别，按印刷页码自动排序）",
f"> 🗓 生成时间：{ts:%Y-%m-%d %H:%M:%S} · 共 **{len(parts3)} 页**", "", "---", ""]
if skipped:
md.append(f"> ⚠️ {len(skipped)} 页未识别/失败未并入")
md.append("")
for r in rows:
md.append(f"## {_row_label(r)}\n")
if r["vl"]:
if r["vl"].get("header") not in (None, "", "无"):
md.append(f"> 🏷 **页眉**：{r['vl']['header']}\n")
for para in _paras(r["body"]):
md.append(para + "\n")
if r["vl"]:
if r["vl"].get("footer") not in (None, "", "无"):
md.append(f"> 🔖 **页脚**：{r['vl']['footer']}\n")
md.append("---\n")
text4 = "\n".join(md).rstrip() + "\n"
p4 = _unique_path(d, "全书_含页眉页脚", suffix=".md")
p4.write_text(text4, encoding="utf-8")
files.append({"tag": "含页眉页脚(md)", "name": p4.name,
"url": f"/txt/{p4.relative_to(BOOKS_DIR)}", "chars": len(text4.replace("\n", ""))})
return jsonify(ok=True, book=b, ordered_by=res["ordered_by"],
dropped=[x["stem"] for x in dropped], skipped=skipped, files=files)
@app.post("/api/batch/export")
def api_batch_export():
"""一次性合成整书文本（自动排序 + 章首页置前 + 重复页去重）。
body(json): {book_name, scope:"session"|"folder", stems?:[页stem], include_side?:0|1}
排序：VL 识别出的印刷页码为主；无页码的“章首页/书前页”（正文以 第N章 开头）自动放最前；
其余无页码页按相邻有页码页位置插值；正文高度相似(>0.9)的重复拍摄页只保留 1 份。
"""
data = request.get_json(silent=True) or {}
book_name = _sanitize(data.get("book_name"), f"未命名书_{_dt.datetime.now():%Y%m%d}")
scope = (data.get("scope") or "session").strip().lower()
if scope not in ("session", "folder"):
return jsonify(ok=False, error="scope 只支持 session / folder"), 400
book_dir = BOOKS_DIR / book_name
if not book_dir.is_dir():
return jsonify(ok=False, error=f"服务器上还没有《{book_name}》的存档"), 404
# 收集行：有 txt 的页（可并） + 待识别/失败说明
rows: list[dict] = []
missing: list[str] = []
skipped: list[dict] = []
if scope == "folder":
for rec in _page_records(book_dir):
if rec["txt"].is_file():
rows.append({"stem": rec["stem"], "seq": rec["seq"],
"path": rec["txt"], "vl": _read_vl_meta(book_dir, rec["stem"])})
else:
st, em = _page_status(rec, book_name)
skipped.append({"seq": rec["seq"], "stem": rec["stem"],
"reason": em or ("待识别" if st in ("pending", "queued", "running") else f"状态:{st}")})
else:
for s in data.get("stems") or []:
s = str(s).strip()
if not re.fullmatch(r"[\w\u4e00-\u9fff.-]+", s) or s.startswith("."):
return jsonify(ok=False, error=f"非法的页文件名: {s}"), 400
txt_path = (book_dir / f"{s}.txt").resolve()
if not str(txt_path).startswith(str(book_dir.resolve())) or not txt_path.is_file():
missing.append(s)
continue
rows.append({"stem": s, "seq": _seq_of_stem(s), "path": txt_path,
"vl": _read_vl_meta(book_dir, s)})
if not rows:
return jsonify(ok=False, error="没有可合成的页文本"), 404
# 读取正文并标记“章首页/书前页”（无印刷页码、正文以 第N章 开头）
from difflib import SequenceMatcher as _SM
for i, r in enumerate(rows):
r["i"] = i
try:
r["body"] = r["path"].read_text(encoding="utf-8").strip()
except OSError:
r["body"] = ""
r["front"] = bool(not (r["vl"] or {}).get("page_num")
and re.match(r"^第[0-9一二三四五六七八九十百两]+章", r["body"]))
def _dedupe(lst):
kept, dropped = [], []
for r in lst:
dup = bool(r["body"]) and any(
_SM(None, r["body"], k["body"]).ratio() > 0.9 for k in kept if k["body"])
(dropped if dup else kept).append(r)
return kept, dropped
front_rows = [r for r in rows if r["front"]]
rest_rows = [r for r in rows if not r["front"]]
front_rows.sort(key=lambda r: (r["seq"], r["i"]))
front_kept, front_dropped = _dedupe(front_rows)
# 其余页：优先按 VL 页码，无页码页按相邻有页码页插值
for i, r in enumerate(rest_rows):
r["i"] = i
any_num = any((r["vl"] or {}).get("page_num") is not None for r in rest_rows)
if any_num:
seq_sorted = sorted(rest_rows, key=lambda r: (r["seq"], r["i"]))
nums = {r["stem"]: (r["vl"] or {}).get("page_num") for r in rest_rows}
def est_num(idx: int):
if nums[seq_sorted[idx]["stem"]] is not None:
return float(nums[seq_sorted[idx]["stem"]])
below = above = None
for j in range(idx - 1, -1, -1):
if nums[seq_sorted[j]["stem"]] is not None:
below = (j, float(nums[seq_sorted[j]["stem"]]))
break
for k in range(idx + 1, len(seq_sorted)):
if nums[seq_sorted[k]["stem"]] is not None:
above = (k, float(nums[seq_sorted[k]["stem"]]))
break
if below and above:
(j, v1), (k, v2) = below, above
return v1 + (v2 - v1) * (idx - j) / (k - j)
if above:
return above[1] - (above[0] - idx)
if below:
return below[1] + (idx - below[0])
return 1e9 + idx
est = {seq_sorted[idx]["stem"]: est_num(idx) for idx in range(len(seq_sorted))}
rest_rows.sort(key=lambda r: (est[r["stem"]], r["i"]))
ordered_by = "page_num"
else:
rest_rows.sort(key=lambda r: (r["seq"], r["i"]))
ordered_by = "seq"
rest_kept, rest_dropped = _dedupe(rest_rows)
order = front_kept + rest_kept
dup_dropped = front_dropped + rest_dropped
for r in order:
r["i"] = r["i"] # keep stable
# 更新 i 仅用于保持原始相对顺序字段
for n, r in enumerate(order):
r["order"] = n
include_side = bool(int(data.get("include_side", 0)))
parts: list[str] = []
total_chars = 0
for r in order:
body = r["body"]
label = f"第 {r['seq']} 页"
if r["vl"] and r["vl"].get("page_raw") not in (None, "", "无"):
label += f"（原书页码 {r['vl']['page_raw']}）"
elif r["front"]:
label += "（章首页·无页码）"
if include_side and r["vl"]:
seg = []
if r["vl"].get("header") not in (None, "", "无"):
seg.append(f"【页眉】{r['vl']['header']}")
if body:
seg.append(body)
if r["vl"].get("footer") not in (None, "", "无"):
seg.append(f"【页脚】{r['vl']['footer']}")
content = "\n".join(seg)
else:
content = body
total_chars += len(content)
parts.append(f"────────── {label} ──────────\n{content if content else '（本页未能识别出文字）'}")
now = _dt.datetime.now()
head_lines = [
f"《{book_name}》整书文本",
f"生成时间：{now:%Y-%m-%d %H:%M:%S} 共 {len(parts)} 页 {total_chars} 字",
f"排序依据：{'VL 识别出的印刷页码（章首页置前）' if ordered_by == 'page_num' else '拍摄顺序'} " + ("含页眉页脚" if include_side else "正文（已剔除页眉页脚页码）"),
]
if missing:
head_lines.append(f"（另有 {len(missing)} 页存档缺失，未并入）")
if skipped:
head_lines.append(f"（另有 {len(skipped)} 页未识别/识别失败，未并入——VL 整书处理后重新合成即可）")
if dup_dropped:
head_lines.append(f"（自动合并 {len(dup_dropped)} 页重复拍摄：{', '.join(d['stem'] for d in dup_dropped[:5])}{'…' if len(dup_dropped) > 5 else ''}）")
content = "\n".join(head_lines) + "\n\n" + "\n\n".join(parts) + "\n"
# 输出文件名（同秒多次合成自动加序号，避免覆盖）
base_name = f"全书文本_{now:%Y%m%d_%H%M%S}"
out_name, n = base_name + ".txt", 1
while (book_dir / out_name).exists():
n += 1
out_name = f"{base_name}_{n}.txt"
out_path = book_dir / out_name
out_path.write_text(content, encoding="utf-8")
rel = f"{out_path.relative_to(BOOKS_DIR)}"
preview = content[:1600] + ("\n…（余下内容见文件）" if len(content) > 1600 else "")
vl_pages = sum(1 for r in order if r["vl"] and r["vl"].get("page_num") is not None)
return jsonify(
ok=True,
scope=scope,
book=book_name,
pages=len(parts),
chars=total_chars,
file=out_name,
url=f"/txt/{rel}",
missing=missing,
skipped=skipped if scope == "folder" else None,
ordered_by=ordered_by,
include_side=include_side,
vl_pages=vl_pages,
duplicates_dropped=[r["stem"] for r in dup_dropped],
preview=preview,
)
@app.get("/txt/")
def text_file(filepath: str):
p = (BOOKS_DIR / filepath).resolve()
if not str(p).startswith(str(BOOKS_DIR.resolve())):
return jsonify(ok=False, error="路径越界"), 400
if not p.is_file():
return jsonify(ok=False, error="文件不存在"), 404
return send_file(p, mimetype="text/plain", as_attachment=True)
if __name__ == "__main__":
BOOKS_DIR.mkdir(parents=True, exist_ok=True)
cert, key = BASE / "cert.pem", BASE / "key.pem"
ssl_ctx = None
if os.environ.get("OCR_SSL") == "1":
if cert.is_file() and key.is_file():
ssl_ctx = (str(cert), str(key))
else:
print("[警告] OCR_SSL=1 但缺少 cert.pem/key.pem，请先运行 bash gen_cert.sh")
print(f"[拍书取字] 本地 OCR 引擎: rapidocr-onnxruntime | VL 增强: {'启用' if vl_ready() else '未配置'}")
print(f"[拍书取字] 存档目录: {BOOKS_DIR}")
print(f"[拍书取字] 服务地址: http{'s' if ssl_ctx else ''}://{HOST}:{PORT}/ （手机/PC 连同一局域网访问本机 IP）")
app.run(host=HOST, port=PORT, threaded=True, ssl_context=ssl_ctx)