Paddle Ocr

# 此目录用于存储转换历史和归档文件
# 每次转换会创建一个子目录，包含：
# - input.*: 原始输入文件
# - response.json: API 响应
# - result.md: 转换后的 Markdown 文件

# ===== PaddleOCR API 官方配置 =====
# 从 https://www.paddleocr.com 对应模型 API 页面复制
PADDLEOCR_DOC_PARSING_API_URL=https://your-endpoint.example.com/layout-parsing
PADDLEOCR_ACCESS_TOKEN=your_access_token_here

# ===== 识别选项 =====
PADDLEOCR_DOC_ORIENTATION=false
PADDLEOCR_DOC_UNWARP=false
PADDLEOCR_CHART_RECOG=false

# ===== 稳定性与大文件策略 =====
PADDLEOCR_DOC_PARSING_TIMEOUT=600
PADDLEOCR_BATCH_PAGES=40
PADDLEOCR_MAX_BASE64_MB=20

# ===== 日志 =====
PADDLEOCR_LOG_LEVEL=medium

MIT License

Copyright (c) 2025 杨卫薪律师（微信ywxlaw）

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

ObjC.import("Foundation");

function shellQuote(value) {
  return "'" + String(value).replace(/'/g, "'\\''") + "'";
}

function getScriptPath() {
  const args = $.NSProcessInfo.processInfo.arguments;
  if (args.count < 4) {
    throw new Error("无法定位当前脚本路径");
  }
  return ObjC.unwrap(args.objectAtIndex(3));
}

function run(argv) {
  const app = Application.currentApplication();
  app.includeStandardAdditions = true;
  const sh = (cmd) => app.doShellScript(cmd);

  try {
    const scriptPath = getScriptPath();
    const scriptDir = sh(`/usr/bin/dirname ${shellQuote(scriptPath)}`).trim();
    const pythonScript = `${scriptDir}/convert.py`;
    const skillRoot = sh(`/usr/bin/dirname ${shellQuote(scriptDir)}`).trim();
    const uvPath = sh(`/bin/zsh -lc 'command -v uv || true'`).trim();
    const runner = uvPath ? `${shellQuote(uvPath)} run` : "/usr/bin/python3";
    const argList = (argv || []).map(shellQuote).join(" ");
    const command =
      `cd ${shellQuote(skillRoot)} && ${runner} ${shellQuote(pythonScript)}` +
      (argList ? ` ${argList}` : "");

    const output = sh(command);
    console.log(output);
    return output;
  } catch (error) {
    const message =
      "PaddleOCR 转换失败\n" +
      "===============================================\n" +
      `${error.message}\n\n` +
      "建议：\n" +
      "1. 先运行 `uv run scripts/smoke_test.py --skip-api-test`\n" +
      "2. 检查 `config/.env` 是否已配置\n" +
      "3. 直接运行 `uv run scripts/convert.py \"文件路径\"` 查看详细日志";
    console.log(message);
    return message;
  }
}

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.9"
# dependencies = [
#   "httpx>=0.27.0",
#   "pypdfium2>=4.30.0",
# ]
# ///

"""面向法律 PDF 的高层转换入口：产出 Markdown，并写入 archive。"""

from __future__ import annotations

import argparse
import base64
import hashlib
import json
import shutil
import sys
import tempfile
import urllib.request
from datetime import datetime
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parent))

from lib import (  # noqa: E402
    SUPPORTED_LOCAL_SUFFIXES,
    extract_markdown_and_images,
    get_runtime_config,
    get_skill_root,
    is_pdf_file,
    parse_document,
    sanitize_name,
)
from split_pdf import (  # noqa: E402
    extract_pages_to_pdf,
    format_pages_compact,
    get_pdf_page_count,
    parse_pages_spec,
    split_pdf_by_batch_size,
)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="将法律 PDF / 图片转换为 Markdown，并保留 archive 归档"
    )
    parser.add_argument("input", help="本地 PDF 或图片路径")
    parser.add_argument(
        "--output",
        help="输出 Markdown 路径；如果不是 .md，则按目录处理",
    )
    parser.add_argument(
        "--pages",
        help='只处理指定页码，例如 "1-20" 或 "1-5,8,10-12"（仅 PDF）',
    )
    parser.add_argument(
        "--archive-name",
        help="自定义 archive 目录名后缀，默认使用输入文件名",
    )
    parser.add_argument(
        "--no-archive",
        action="store_true",
        help="不写入 archive，仅输出 Markdown",
    )
    return parser.parse_args()


def sha256_of_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def estimate_base64_mb(path: Path) -> float:
    return (path.stat().st_size * 4 / 3) / 1024 / 1024


def resolve_output_markdown_path(input_path: Path, output_arg: str | None) -> Path:
    if not output_arg:
        return input_path.with_suffix(".md").resolve()

    output_path = Path(output_arg).expanduser()
    if output_path.suffix.lower() == ".md":
        return output_path.resolve()

    if output_arg.endswith("/") or output_path.is_dir() or not output_path.suffix:
        return (output_path / f"{input_path.stem}.md").resolve()

    return output_path.resolve()


def resolve_images_dir(markdown_path: Path) -> Path:
    return markdown_path.with_name(f"{markdown_path.stem}_images")


def write_markdown(markdown_path: Path, content: str) -> None:
    markdown_path.parent.mkdir(parents=True, exist_ok=True)
    markdown_path.write_text(content, encoding="utf-8")


def decode_base64_image(raw_data: str) -> bytes:
    payload = raw_data.strip()
    if payload.startswith("data:") and "," in payload:
        payload = payload.split(",", 1)[1]
    return base64.b64decode(payload)


def save_images(
    batch_outputs: list[dict[str, Any]],
    images_dir: Path,
) -> list[dict[str, str]]:
    saved: list[dict[str, str]] = []
    if images_dir.exists():
        shutil.rmtree(images_dir)
    if not any(batch["images"] for batch in batch_outputs):
        return saved

    images_dir.mkdir(parents=True, exist_ok=True)
    used_names: set[str] = set()

    for batch in batch_outputs:
        batch_label = sanitize_name(batch["label"])
        for index, (source_path, image_data) in enumerate(
            sorted(batch["images"].items()),
            start=1,
        ):
            suffix = Path(source_path).suffix.lower() or ".png"
            filename = f"{batch_label}_{index:03d}{suffix}"
            while filename in used_names:
                filename = f"{batch_label}_{index:03d}_{len(used_names):03d}{suffix}"
            used_names.add(filename)

            target_path = images_dir / filename
            if str(image_data).startswith(("http://", "https://")):
                urllib.request.urlretrieve(str(image_data), target_path)
            else:
                target_path.write_bytes(decode_base64_image(str(image_data)))

            saved.append(
                {
                    "batch": batch["label"],
                    "source": str(source_path),
                    "path": str(target_path),
                    "filename": filename,
                }
            )

    return saved


def build_archive_paths(archive_root: Path, archive_name: str) -> dict[str, Path]:
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    slug = sanitize_name(archive_name)
    base = archive_root / f"{timestamp}_{slug}"
    return {
        "root": base,
        "input": base / "input",
        "output": base / "output",
        "images": base / "output" / "images",
        "batches": base / "batches",
        "metadata": base / "metadata.json",
        "result_md": base / "output" / "result.md",
        "result_json": base / "output" / "result.json",
    }


def copy_into_archive(
    archive_paths: dict[str, Path],
    input_path: Path,
    output_md_path: Path,
    images_dir: Path,
    result_json: dict[str, Any],
    batch_outputs: list[dict[str, Any]],
    metadata: dict[str, Any],
) -> Path:
    archive_paths["input"].mkdir(parents=True, exist_ok=True)
    archive_paths["output"].mkdir(parents=True, exist_ok=True)
    archive_paths["batches"].mkdir(parents=True, exist_ok=True)

    shutil.copy2(input_path, archive_paths["input"] / input_path.name)
    shutil.copy2(output_md_path, archive_paths["result_md"])

    archive_paths["result_json"].write_text(
        json.dumps(result_json, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    archive_paths["metadata"].write_text(
        json.dumps(metadata, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    if images_dir.exists():
        shutil.copytree(images_dir, archive_paths["images"], dirs_exist_ok=True)

    for index, batch in enumerate(batch_outputs, start=1):
        label = sanitize_name(batch["label"])
        batch_json_path = archive_paths["batches"] / f"batch_{index:03d}_{label}.json"
        batch_json_path.write_text(
            json.dumps(batch["envelope"], ensure_ascii=False, indent=2),
            encoding="utf-8",
        )

    return archive_paths["root"]


def build_batch_output(
    label: str,
    input_path: Path,
) -> dict[str, Any]:
    envelope = parse_document(file_path=str(input_path))
    if not envelope.get("ok"):
        error = envelope.get("error") or {}
        raise RuntimeError(f"{label} 处理失败：{error.get('message', '未知错误')}")

    text, images = extract_markdown_and_images(envelope["result"])
    if not text.strip():
        raise RuntimeError(f"{label} OCR 完成，但未提取到有效文本")

    return {
        "label": label,
        "input_path": str(input_path),
        "envelope": envelope,
        "text": text,
        "images": images,
    }


def prepare_pdf_batches(
    input_path: Path,
    pages_spec: str | None,
    temp_dir: Path,
    batch_pages: int,
    max_base64_mb: float,
) -> tuple[list[tuple[str, Path]], dict[str, Any]]:
    total_pages = get_pdf_page_count(input_path)
    selected_pages = (
        parse_pages_spec(pages_spec, total_pages)
        if pages_spec
        else list(range(total_pages))
    )
    selected_label = format_pages_compact(selected_pages)
    processed_page_count = len(selected_pages)

    estimated_subset_base64_mb = estimate_base64_mb(input_path)
    if processed_page_count and total_pages:
        estimated_subset_base64_mb *= processed_page_count / total_pages

    needs_batch = (
        processed_page_count > batch_pages or estimated_subset_base64_mb > max_base64_mb
    )

    if needs_batch:
        batch_specs = split_pdf_by_batch_size(
            input_path=input_path,
            output_dir=temp_dir / "pdf-batches",
            batch_size=batch_pages,
            page_indices=selected_pages,
        )
        batches = [(spec["label"], spec["path"]) for spec in batch_specs]
    else:
        if pages_spec:
            single_path = temp_dir / f"{input_path.stem}_selected.pdf"
            extract_pages_to_pdf(input_path, single_path, selected_pages)
            batches = [(selected_label, single_path)]
        else:
            batches = [("all-pages", input_path)]

    info = {
        "total_pages": total_pages,
        "processed_pages": processed_page_count,
        "selected_pages": selected_label,
        "needs_batch": needs_batch,
        "batch_pages": batch_pages,
        "estimated_base64_mb": round(estimated_subset_base64_mb, 2),
    }
    return batches, info


def process_input(
    input_path: Path,
    pages_spec: str | None,
    batch_pages: int,
    max_base64_mb: float,
    temp_dir: Path,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    if is_pdf_file(input_path):
        batch_inputs, pdf_info = prepare_pdf_batches(
            input_path=input_path,
            pages_spec=pages_spec,
            temp_dir=temp_dir,
            batch_pages=batch_pages,
            max_base64_mb=max_base64_mb,
        )
    else:
        if pages_spec:
            raise ValueError("--pages 仅适用于 PDF 文件")
        batch_inputs = [(input_path.stem, input_path)]
        pdf_info = {
            "total_pages": None,
            "processed_pages": 1,
            "selected_pages": None,
            "needs_batch": False,
            "batch_pages": None,
            "estimated_base64_mb": round(estimate_base64_mb(input_path), 2),
        }

    outputs = [build_batch_output(label, path) for label, path in batch_inputs]
    return outputs, pdf_info


def main() -> int:
    args = parse_args()
    runtime = get_runtime_config()
    input_path = Path(args.input).expanduser().resolve()

    if not input_path.exists():
        print(f"错误：文件不存在：{input_path}", file=sys.stderr)
        return 1
    if input_path.suffix.lower() not in SUPPORTED_LOCAL_SUFFIXES:
        print(
            f"错误：不支持的文件类型：{input_path.suffix.lower()}",
            file=sys.stderr,
        )
        return 1

    output_md_path = resolve_output_markdown_path(input_path, args.output)
    output_images_dir = resolve_images_dir(output_md_path)
    archive_root = get_skill_root() / "archive"
    archive_name = args.archive_name or input_path.stem

    temp_dir = Path(tempfile.mkdtemp(prefix="paddleocr_convert_"))
    archive_path: Path | None = None

    try:
        batch_outputs, pdf_info = process_input(
            input_path=input_path,
            pages_spec=args.pages,
            batch_pages=runtime["batch_pages"],
            max_base64_mb=runtime["max_base64_mb"],
            temp_dir=temp_dir,
        )

        merged_text = "\n\n".join(batch["text"].strip() for batch in batch_outputs).strip()
        write_markdown(output_md_path, merged_text)
        saved_images = save_images(batch_outputs, output_images_dir)

        result_json = {
            "ok": True,
            "source": {
                "path": str(input_path),
                "name": input_path.name,
                "sha256": sha256_of_file(input_path),
                "kind": "pdf" if is_pdf_file(input_path) else "image",
            },
            "processing": {
                "mode": "batched" if len(batch_outputs) > 1 else "single",
                "batch_count": len(batch_outputs),
                **pdf_info,
            },
            "text": merged_text,
            "images": [
                {
                    "batch": image["batch"],
                    "source": image["source"],
                    "filename": image["filename"],
                    "path": image["path"],
                }
                for image in saved_images
            ],
            "batches": [
                {
                    "index": index,
                    "label": batch["label"],
                    "input_path": batch["input_path"],
                    "text_length": len(batch["text"]),
                    "image_count": len(batch["images"]),
                }
                for index, batch in enumerate(batch_outputs, start=1)
            ],
        }

        metadata = {
            "created_at": datetime.now().isoformat(timespec="seconds"),
            "provider": "PaddleOCR Document Parsing API",
            "purpose": "法律 PDF / 图片 OCR，主产出为 Markdown，保留 archive 追溯链",
            "config": {
                "timeout_seconds": runtime["timeout_seconds"],
                "doc_orientation": runtime["doc_orientation"],
                "doc_unwarp": runtime["doc_unwarp"],
                "chart_recognition": runtime["chart_recognition"],
                "batch_pages": runtime["batch_pages"],
                "max_base64_mb": runtime["max_base64_mb"],
            },
            "output_markdown": str(output_md_path),
            "image_output_dir": str(output_images_dir) if saved_images else None,
        }

        if not args.no_archive:
            archive_paths = build_archive_paths(archive_root, archive_name)
            archive_path = copy_into_archive(
                archive_paths=archive_paths,
                input_path=input_path,
                output_md_path=output_md_path,
                images_dir=output_images_dir,
                result_json=result_json,
                batch_outputs=batch_outputs,
                metadata=metadata,
            )

        print("转换完成")
        print(f"Markdown: {output_md_path}")
        if saved_images:
            print(f"图片目录: {output_images_dir}")
        if archive_path:
            print(f"Archive: {archive_path}")
        print(
            "模式: "
            + ("自动分批" if len(batch_outputs) > 1 else "单次请求")
        )
        return 0
    except Exception as error:  # noqa: BLE001
        print(f"转换失败：{error}", file=sys.stderr)
        return 1
    finally:
        shutil.rmtree(temp_dir, ignore_errors=True)


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.9"
# dependencies = [
#   "httpx>=0.27.0",
# ]
# ///

"""底层接口调用脚本：输出稳定 JSON envelope。"""

from __future__ import annotations

import argparse
import json
import sys
import tempfile
import uuid
from datetime import datetime
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

from lib import parse_document  # noqa: E402


def default_output_path() -> Path:
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
    short_id = uuid.uuid4().hex[:8]
    return (
        Path(tempfile.gettempdir())
        / "paddleocr"
        / "legal-doc-parsing"
        / f"result_{timestamp}_{short_id}.json"
    )


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="调用 PaddleOCR 接口并输出稳定 JSON envelope"
    )
    source_group = parser.add_mutually_exclusive_group(required=True)
    source_group.add_argument("--file-path", help="本地 PDF / 图片路径")
    source_group.add_argument("--file-url", help="远程 PDF / 图片 URL")
    parser.add_argument(
        "--file-type",
        type=int,
        choices=[0, 1],
        help="显式指定文件类型：0=PDF，1=图片",
    )
    parser.add_argument("--pretty", action="store_true", help="格式化 JSON 输出")
    output_group = parser.add_mutually_exclusive_group()
    output_group.add_argument("--stdout", action="store_true", help="直接打印 JSON")
    output_group.add_argument("--output", help="写入指定 JSON 文件")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    envelope = parse_document(
        file_path=args.file_path,
        file_url=args.file_url,
        file_type=args.file_type,
    )
    json_text = json.dumps(
        envelope,
        ensure_ascii=False,
        indent=2 if args.pretty else None,
    )

    if args.stdout:
        print(json_text)
    else:
        output_path = Path(args.output).expanduser().resolve() if args.output else default_output_path()
        output_path.parent.mkdir(parents=True, exist_ok=True)
        output_path.write_text(json_text, encoding="utf-8")
        print(f"Result saved to: {output_path}", file=sys.stderr)

    return 0 if envelope.get("ok") else 1


if __name__ == "__main__":
    raise SystemExit(main())

from __future__ import annotations

import base64
import math
import os
import re
from pathlib import Path
from typing import Any
from urllib.parse import unquote, urlparse

import httpx

DEFAULT_TIMEOUT_SECONDS = 600
DEFAULT_BATCH_PAGES = 40
DEFAULT_MAX_BASE64_MB = 20.0

SUPPORTED_IMAGE_SUFFIXES = (
    ".png",
    ".jpg",
    ".jpeg",
    ".bmp",
    ".tiff",
    ".tif",
    ".webp",
)
SUPPORTED_LOCAL_SUFFIXES = (".pdf",) + SUPPORTED_IMAGE_SUFFIXES


def get_skill_root() -> Path:
    return Path(__file__).resolve().parent.parent


def get_config_path() -> Path:
    return get_skill_root() / "config" / ".env"


def read_env_file(path: Path) -> dict[str, str]:
    if not path.exists():
        return {}

    env: dict[str, str] = {}
    for raw_line in path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        env[key.strip()] = value.strip().strip('"').strip("'")
    return env


def parse_bool(value: str | bool | None, default: bool = False) -> bool:
    if value is None:
        return default
    if isinstance(value, bool):
        return value
    normalized = str(value).strip().lower()
    if not normalized:
        return default
    return normalized in {"1", "true", "yes", "on"}


def parse_positive_int(value: str | None, default: int) -> int:
    if not value or not str(value).strip():
        return default
    try:
        parsed = int(str(value).strip())
    except ValueError as error:
        raise ValueError(f"整数配置无效：{value}") from error
    if parsed <= 0:
        raise ValueError(f"整数配置必须大于 0：{value}")
    return parsed


def parse_positive_float(value: str | None, default: float) -> float:
    if not value or not str(value).strip():
        return default
    try:
        parsed = float(str(value).strip())
    except ValueError as error:
        raise ValueError(f"数字配置无效：{value}") from error
    if not math.isfinite(parsed) or parsed <= 0:
        raise ValueError(f"数字配置必须大于 0：{value}")
    return parsed


def sanitize_name(value: str) -> str:
    sanitized = re.sub(r"[^\w\u4e00-\u9fff.-]+", "_", value, flags=re.UNICODE)
    sanitized = sanitized.strip("._")
    return sanitized or "document"


def first_non_empty(env: dict[str, str], *keys: str) -> str:
    for key in keys:
        value = env.get(key, "").strip()
        if value:
            return value
    return ""


def normalize_api_url(api_url: str) -> str:
    url = api_url.strip()
    if not url:
        raise ValueError("未配置 PaddleOCR API 地址")
    if "://" not in url:
        url = f"https://{url}"

    parsed = urlparse(url)
    host = (parsed.hostname or "").lower()
    if parsed.scheme not in {"https", "http"}:
        raise ValueError("PaddleOCR API 地址必须以 https:// 或 http:// 开头")
    if parsed.scheme == "http" and host not in {"127.0.0.1", "localhost"}:
        raise ValueError("仅允许 localhost / 127.0.0.1 使用 http://")
    if not parsed.path.rstrip("/").endswith("/layout-parsing"):
        raise ValueError(
            "PaddleOCR API 地址必须是完整的 layout-parsing 端点，例如 "
            "https://your-endpoint/layout-parsing"
        )
    return url


def get_runtime_config() -> dict[str, Any]:
    file_env = read_env_file(get_config_path())
    merged_env = {**file_env, **os.environ}

    api_url = first_non_empty(merged_env, "PADDLEOCR_DOC_PARSING_API_URL")
    access_token = first_non_empty(merged_env, "PADDLEOCR_ACCESS_TOKEN")

    if not api_url:
        raise ValueError(
            "未配置 PADDLEOCR_DOC_PARSING_API_URL。请先编辑 paddle-ocr/config/.env。"
        )
    if not access_token:
        raise ValueError(
            "未配置 PADDLEOCR_ACCESS_TOKEN。请先编辑 paddle-ocr/config/.env。"
        )

    return {
        "api_url": normalize_api_url(api_url),
        "access_token": access_token,
        "doc_orientation": parse_bool(
            first_non_empty(merged_env, "PADDLEOCR_DOC_ORIENTATION"),
            default=False,
        ),
        "doc_unwarp": parse_bool(
            first_non_empty(merged_env, "PADDLEOCR_DOC_UNWARP"),
            default=False,
        ),
        "chart_recognition": parse_bool(
            first_non_empty(merged_env, "PADDLEOCR_CHART_RECOG"),
            default=False,
        ),
        "timeout_seconds": parse_positive_float(
            first_non_empty(merged_env, "PADDLEOCR_DOC_PARSING_TIMEOUT"),
            default=DEFAULT_TIMEOUT_SECONDS,
        ),
        "batch_pages": parse_positive_int(
            first_non_empty(merged_env, "PADDLEOCR_BATCH_PAGES"),
            default=DEFAULT_BATCH_PAGES,
        ),
        "max_base64_mb": parse_positive_float(
            first_non_empty(merged_env, "PADDLEOCR_MAX_BASE64_MB"),
            default=DEFAULT_MAX_BASE64_MB,
        ),
        "log_level": first_non_empty(merged_env, "PADDLEOCR_LOG_LEVEL") or "medium",
    }


def detect_file_type(path_or_url: str) -> int:
    normalized = path_or_url.lower()
    if normalized.startswith(("http://", "https://")):
        normalized = unquote(urlparse(normalized).path)

    if normalized.endswith(".pdf"):
        return 0
    if normalized.endswith(SUPPORTED_IMAGE_SUFFIXES):
        return 1
    raise ValueError(f"不支持的文件类型：{path_or_url}")


def is_pdf_file(path: Path) -> bool:
    return path.suffix.lower() == ".pdf"


def load_file_as_base64(file_path: str) -> str:
    path = Path(file_path)
    if not path.exists():
        raise FileNotFoundError(f"文件不存在：{file_path}")
    if not path.is_file():
        raise ValueError(f"不是普通文件：{file_path}")
    if path.stat().st_size == 0:
        raise ValueError(f"文件为空：{file_path}")
    return base64.b64encode(path.read_bytes()).decode("utf-8")


def _error(code: str, message: str) -> dict[str, Any]:
    return {
        "ok": False,
        "text": "",
        "result": None,
        "error": {"code": code, "message": message},
    }


def make_request(
    *,
    api_url: str,
    access_token: str,
    payload: dict[str, Any],
    timeout_seconds: float,
) -> dict[str, Any]:
    headers = {
        "Authorization": f"token {access_token}",
        "Content-Type": "application/json",
        "Client-Platform": "private-legal-skill",
    }

    try:
        with httpx.Client(timeout=timeout_seconds) as client:
            response = client.post(api_url, json=payload, headers=headers)
    except httpx.TimeoutException as error:
        raise RuntimeError(f"请求超时：{timeout_seconds} 秒") from error
    except httpx.RequestError as error:
        raise RuntimeError(f"网络请求失败：{error}") from error

    if response.status_code != 200:
        detail = response.text[:500].strip() or "空响应"
        if response.status_code == 403:
            raise RuntimeError(f"鉴权失败（403）：{detail}")
        if response.status_code == 429:
            raise RuntimeError(f"配额或频率受限（429）：{detail}")
        raise RuntimeError(f"接口错误（{response.status_code}）：{detail}")

    try:
        result = response.json()
    except ValueError as error:
        raise RuntimeError(f"接口返回的不是合法 JSON：{response.text[:200]}") from error

    if not isinstance(result, dict):
        raise RuntimeError("接口返回结构异常：顶层不是对象")

    if result.get("errorCode", 0) != 0:
        raise RuntimeError(f"接口返回错误：{result.get('errorMsg', '未知错误')}")

    return result


def extract_markdown_and_images(provider_result: dict[str, Any]) -> tuple[str, dict[str, str]]:
    raw_result = provider_result.get("result")
    if not isinstance(raw_result, dict):
        raise ValueError("接口返回结构异常：缺少 result 对象")

    layout_results = raw_result.get("layoutParsingResults")
    if not isinstance(layout_results, list) or not layout_results:
        raise ValueError("接口未返回 layoutParsingResults")

    texts: list[str] = []
    images: dict[str, str] = {}

    for index, page_result in enumerate(layout_results):
        if not isinstance(page_result, dict):
            raise ValueError(f"第 {index + 1} 页结构异常")
        markdown = page_result.get("markdown")
        if not isinstance(markdown, dict):
            raise ValueError(f"第 {index + 1} 页缺少 markdown 字段")
        text = markdown.get("text")
        if isinstance(text, str) and text.strip():
            texts.append(text)
        page_images = markdown.get("images")
        if isinstance(page_images, dict):
            for key, value in page_images.items():
                images[str(key)] = str(value)

    return "\n\n".join(texts), images


def parse_document(
    *,
    file_path: str | None = None,
    file_url: str | None = None,
    file_type: int | None = None,
    visualize: bool = False,
) -> dict[str, Any]:
    if bool(file_path) == bool(file_url):
        return _error("INPUT_ERROR", "必须在 file_path 和 file_url 中二选一")

    try:
        runtime = get_runtime_config()
    except ValueError as error:
        return _error("CONFIG_ERROR", str(error))

    try:
        if file_path:
            resolved_file_type = file_type if file_type is not None else detect_file_type(file_path)
            payload = {
                "file": load_file_as_base64(file_path),
                "fileType": resolved_file_type,
            }
        else:
            assert file_url is not None
            payload = {
                "file": file_url.strip(),
            }
            try:
                resolved_file_type = file_type if file_type is not None else detect_file_type(file_url)
            except ValueError:
                resolved_file_type = None
            if resolved_file_type is not None:
                payload["fileType"] = resolved_file_type

        payload["useDocOrientationClassify"] = runtime["doc_orientation"]
        payload["useDocUnwarping"] = runtime["doc_unwarp"]
        payload["useChartRecognition"] = runtime["chart_recognition"]
        payload["visualize"] = visualize
    except (ValueError, OSError, MemoryError) as error:
        return _error("INPUT_ERROR", str(error))

    try:
        provider_result = make_request(
            api_url=runtime["api_url"],
            access_token=runtime["access_token"],
            payload=payload,
            timeout_seconds=runtime["timeout_seconds"],
        )
        text, _images = extract_markdown_and_images(provider_result)
    except (RuntimeError, ValueError) as error:
        return _error("API_ERROR", str(error))

    return {
        "ok": True,
        "text": text,
        "result": provider_result,
        "error": None,
    }

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.9"
# dependencies = [
#   "Pillow>=10.0.0",
# ]
# ///

"""扫描图片压缩工具，适合先压缩后做 OCR。"""

from __future__ import annotations

import argparse
import math
from pathlib import Path

from PIL import Image

DEFAULT_QUALITY = 85
DEFAULT_TARGET_MB = 20.0
SUPPORTED_SUFFIXES = (".png", ".jpg", ".jpeg", ".bmp", ".tiff", ".tif", ".webp")


def positive_int(value: str) -> int:
    parsed = int(value)
    if parsed < 1 or parsed > 100:
        raise argparse.ArgumentTypeError("quality 必须在 1-100 之间")
    return parsed


def positive_float(value: str) -> float:
    parsed = float(value)
    if not math.isfinite(parsed) or parsed <= 0:
        raise argparse.ArgumentTypeError("target-size 必须大于 0")
    return parsed


def optimize_image(input_path: Path, output_path: Path, quality: int, target_mb: float) -> None:
    image = Image.open(input_path)
    original_size = input_path.stat().st_size / 1024 / 1024
    print(f"原始大小：{original_size:.2f} MB")
    print(f"原始尺寸：{image.size[0]}x{image.size[1]}")

    is_jpeg_like = output_path.suffix.lower() in {".jpg", ".jpeg", ".webp"}
    if is_jpeg_like and image.mode in {"RGBA", "LA", "P"}:
        background = Image.new("RGB", image.size, (255, 255, 255))
        if image.mode == "P":
            image = image.convert("RGBA")
        background.paste(image, mask=image.split()[-1] if image.mode in {"RGBA", "LA"} else None)
        image = background

    save_kwargs = {"optimize": True}
    if is_jpeg_like:
        save_kwargs["quality"] = quality

    def save_current(current_image: Image.Image) -> float:
        output_path.parent.mkdir(parents=True, exist_ok=True)
        current_image.save(output_path, **save_kwargs)
        return output_path.stat().st_size / 1024 / 1024

    current_size = save_current(image)
    scale_factor = 0.9
    while current_size > target_mb and scale_factor >= 0.4:
        width = max(1, int(image.size[0] * scale_factor))
        height = max(1, int(image.size[1] * scale_factor))
        resized = image.resize((width, height), Image.Resampling.LANCZOS)
        current_size = save_current(resized)
        print(f"调整后尺寸：{width}x{height}，大小：{current_size:.2f} MB")
        scale_factor -= 0.1

    print(f"优化后大小：{current_size:.2f} MB")
    print(f"已输出：{output_path}")


def main() -> int:
    parser = argparse.ArgumentParser(description="压缩 OCR 输入图片")
    parser.add_argument("input", help="输入图片")
    parser.add_argument("output", help="输出图片")
    parser.add_argument("--quality", type=positive_int, default=DEFAULT_QUALITY)
    parser.add_argument("--target-size", type=positive_float, default=DEFAULT_TARGET_MB)
    args = parser.parse_args()

    input_path = Path(args.input).expanduser().resolve()
    output_path = Path(args.output).expanduser().resolve()
    if not input_path.exists():
        print(f"错误：文件不存在：{input_path}")
        return 1
    if input_path.suffix.lower() not in SUPPORTED_SUFFIXES:
        print(f"错误：不支持的图片格式：{input_path.suffix.lower()}")
        return 1

    optimize_image(input_path, output_path, args.quality, args.target_size)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.9"
# dependencies = [
#   "httpx>=0.27.0",
# ]
# ///

"""配置与连通性自检。"""

from __future__ import annotations

import argparse
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

from lib import get_runtime_config, parse_document  # noqa: E402

DEFAULT_TEST_URL = (
    "https://paddle-model-ecology.bj.bcebos.com/paddlex/imgs/demo_image/"
    "pp_structure_v3_demo.png"
)


def masked_token(token: str) -> str:
    if len(token) <= 12:
        return "***"
    return f"{token[:8]}...{token[-4:]}"


def main() -> int:
    parser = argparse.ArgumentParser(description="PaddleOCR skill 自检")
    parser.add_argument("--skip-api-test", action="store_true", help="只检查配置，不发请求")
    parser.add_argument("--test-url", help="覆盖默认测试文档 URL")
    args = parser.parse_args()

    print("=" * 60)
    print("PaddleOCR Skill Smoke Test")
    print("=" * 60)

    try:
        runtime = get_runtime_config()
    except ValueError as error:
        print(f"配置错误：{error}")
        print("请先编辑 paddle-ocr/config/.env")
        return 1

    print("配置检查通过")
    print(f"API URL: {runtime['api_url']}")
    print(f"Token: {masked_token(runtime['access_token'])}")
    print(f"Timeout: {runtime['timeout_seconds']} 秒")
    print(f"Batch pages: {runtime['batch_pages']}")

    if args.skip_api_test:
        print("已跳过 API 连通性测试")
        return 0

    test_url = args.test_url or DEFAULT_TEST_URL
    print(f"测试文档: {test_url}")
    result = parse_document(file_url=test_url)
    if not result.get("ok"):
        error = result.get("error") or {}
        print(f"API 测试失败：{error.get('message', '未知错误')}")
        return 1

    preview = result.get("text", "").replace("\n", " ")[:200]
    if preview:
        print(f"文本预览: {preview}...")
    print("API 测试通过")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.9"
# dependencies = [
#   "pypdfium2>=4.30.0",
# ]
# ///

"""PDF 页码提取与自动分批工具。"""

from __future__ import annotations

import argparse
from pathlib import Path

import pypdfium2 as pdfium


def get_pdf_page_count(input_path: Path) -> int:
    document = pdfium.PdfDocument(str(input_path))
    try:
        return len(document)
    finally:
        document.close()


def parse_pages_spec(pages_spec: str, total_pages: int) -> list[int]:
    if not pages_spec or not pages_spec.strip():
        raise ValueError("页码范围不能为空")

    selected: list[int] = []
    seen: set[int] = set()

    def add_page(page_number: int) -> None:
        if page_number < 1 or page_number > total_pages:
            raise ValueError(f"页码超出范围：{page_number}，有效范围是 1-{total_pages}")
        index = page_number - 1
        if index not in seen:
            seen.add(index)
            selected.append(index)

    for token in [part.strip() for part in pages_spec.split(",") if part.strip()]:
        if "-" in token:
            start_text, end_text = token.split("-", 1)
            if not start_text.isdigit() or not end_text.isdigit():
                raise ValueError(f"非法页码范围：{token}")
            start_page, end_page = int(start_text), int(end_text)
            if start_page > end_page:
                raise ValueError(f"起始页不能大于结束页：{token}")
            for page_number in range(start_page, end_page + 1):
                add_page(page_number)
        else:
            if not token.isdigit():
                raise ValueError(f"非法页码：{token}")
            add_page(int(token))

    if not selected:
        raise ValueError("没有解析出有效页码")
    return selected


def format_pages_compact(page_indices: list[int]) -> str:
    if not page_indices:
        return "none"

    pages = sorted(index + 1 for index in page_indices)
    ranges: list[str] = []
    start = prev = pages[0]

    for page in pages[1:]:
        if page == prev + 1:
            prev = page
            continue
        ranges.append(f"{start}-{prev}" if start != prev else str(start))
        start = prev = page

    ranges.append(f"{start}-{prev}" if start != prev else str(start))
    return ",".join(ranges)


def extract_pages_to_pdf(input_path: Path, output_path: Path, page_indices: list[int]) -> None:
    source_pdf = pdfium.PdfDocument(str(input_path))
    try:
        output_pdf = pdfium.PdfDocument.new()
        try:
            output_pdf.import_pages(source_pdf, page_indices)
            output_path.parent.mkdir(parents=True, exist_ok=True)
            output_pdf.save(str(output_path))
        finally:
            output_pdf.close()
    finally:
        source_pdf.close()


def split_pdf_by_batch_size(
    *,
    input_path: Path,
    output_dir: Path,
    batch_size: int,
    page_indices: list[int] | None = None,
) -> list[dict[str, Path | str]]:
    total_pages = get_pdf_page_count(input_path)
    selected_pages = page_indices or list(range(total_pages))
    if batch_size <= 0:
        raise ValueError("batch_size 必须大于 0")

    output_dir.mkdir(parents=True, exist_ok=True)
    batches: list[dict[str, Path | str]] = []

    for batch_index in range(0, len(selected_pages), batch_size):
        chunk = selected_pages[batch_index : batch_index + batch_size]
        label = format_pages_compact(chunk)
        filename = f"batch_{batch_index // batch_size + 1:03d}_{label.replace(',', '_')}.pdf"
        output_path = output_dir / filename
        extract_pages_to_pdf(input_path, output_path, chunk)
        batches.append({"label": label, "path": output_path})

    return batches


def main() -> int:
    parser = argparse.ArgumentParser(description="按页码范围提取 PDF")
    parser.add_argument("input_pdf", help="输入 PDF")
    parser.add_argument("output_pdf", help="输出 PDF")
    parser.add_argument("--pages", required=True, help='页码范围，例如 "1-5,8,10-12"')
    args = parser.parse_args()

    input_path = Path(args.input_pdf).expanduser().resolve()
    output_path = Path(args.output_pdf).expanduser().resolve()

    if not input_path.exists():
        print(f"错误：文件不存在：{input_path}")
        return 1
    if input_path.suffix.lower() != ".pdf":
        print(f"错误：输入必须是 PDF：{input_path}")
        return 1
    if output_path.suffix.lower() != ".pdf":
        print(f"错误：输出必须是 PDF：{output_path}")
        return 1

    total_pages = get_pdf_page_count(input_path)
    page_indices = parse_pages_spec(args.pages, total_pages)
    extract_pages_to_pdf(input_path, output_path, page_indices)

    print(f"已生成：{output_path}")
    print(f"原始总页数：{total_pages}")
    print(f"提取页码：{format_pages_compact(page_indices)}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())