#!/usr/bin/env python3 """AcroForm, crop, metadata and embedded-image operations using existing pypdf.""" from __future__ import annotations import json import math import sys from pathlib import Path from _pdf_common import (SkillArgumentParser, input_pdf, output_pdf, output_directory, new_temp_pdf, publish_temp_file, parse_page_spec, run_cli) def load_data(inline, filename): if bool(inline) == bool(filename): raise ValueError('必须且只能提供 --data 或 --data-file') if filename and Path(filename).stat().st_size > 2 * 1024 * 1024: raise ValueError('JSON 上限为 2 MiB') raw = Path(filename).read_bytes() if filename else inline.encode('utf-8') if len(raw) > 2 * 1024 * 1024: raise ValueError('JSON 上限为 2 MiB') data = json.loads(raw) if not isinstance(data, dict): raise ValueError('JSON 必须是对象') return data def inherited_value(field, key, default=None): """Read an inheritable field attribute without following malformed cycles.""" seen = set() while field is not None: field = field.get_object() if id(field) in seen: raise ValueError('PDF 表单字段的父级引用存在循环') seen.add(id(field)) if key in field: return field[key] field = field.get('/Parent') return default def field_info(reader): result = {} for name, field in (reader.get_fields() or {}).items(): # get_fields() is a summary: /AP and /MaxLen live on the full object. original = field.indirect_reference.get_object() ft = str(inherited_value(original, '/FT', '')) flags = int(inherited_value(original, '/Ff', 0)) kind = {'/Tx':'text', '/Ch':'choice', '/Sig':'signature'}.get(ft, 'unknown') if ft == '/Btn': kind = 'pushbutton' if flags & (1 << 16) else ('radio' if flags & (1 << 15) else 'checkbox') widgets = [original, *[kid.get_object() for kid in original.get('/Kids', [])]] states = set() if kind in {'checkbox', 'radio'}: for widget in widgets: appearance = widget.get('/AP') normal = appearance.get_object().get('/N') if appearance is not None else None if normal is not None: states.update(str(key) for key in normal.get_object().keys()) if kind == 'radio' and flags & (1 << 14): states.discard('/Off') options = [{'value':str(option[0]), 'label':str(option[1])} if isinstance(option, list) else {'value':str(option), 'label':str(option)} for option in inherited_value(original, '/Opt', [])] value = inherited_value(original, '/V') result[name] = {'id':name, 'type':kind, 'read_only':bool(flags & 1), 'flags':flags, 'current_value':[str(v) for v in value] if isinstance(value, list) else (str(value) if value is not None else None), 'states':sorted(states), 'options':options, 'max_length':inherited_value(original, '/MaxLen')} return result def validated_values(infos, data): from pypdf.generic import NameObject if not data or len(data) > 500: raise ValueError('填表数据需为 1–500 个字段') values = {} for name, value in data.items(): if name not in infos: raise ValueError(f'表单字段不存在：{name}') field = infos[name] if field['read_only']: raise ValueError(f'字段为只读：{name}') kind = field['type'] if kind in {'signature','pushbutton','unknown'}: raise ValueError(f'不支持填写 {kind} 字段：{name}') if kind in {'checkbox','radio'}: states = field['states'] if type(value) is bool and kind == 'checkbox': choices = [s for s in states if s != '/Off'] if value and len(choices) != 1: raise ValueError(f'{name} 的选中状态不唯一，请使用具体状态名') value = choices[0] if value else '/Off' elif isinstance(value, str): value = '/' + value.lstrip('/') else: raise ValueError(f'{name} 需布尔值或有效状态名') if value not in states: raise ValueError(f'{name} 状态无效；可选：{states}') values[name] = NameObject(value) elif kind == 'choice': choices = {entry['value'] for entry in field['options']} selected = value if isinstance(value, list) else [value] if isinstance(value, list) and not field['flags'] & (1 << 21): raise ValueError(f'{name} 不支持多选') editable = bool(field['flags'] & (1 << 18)) if not all(isinstance(item, str) and (editable or item in choices) for item in selected): raise ValueError(f'{name} 选项无效；可选：{sorted(choices)}') values[name] = value else: if not isinstance(value, str) or (field['max_length'] and len(value) > field['max_length']): raise ValueError(f'{name} 需文本，且不得超过字段长度上限') values[name] = value return values def execute(args): from pypdf import PdfReader, PdfWriter from pypdf.generic import RectangleObject source = input_pdf(args.input) with source.open('rb') as handle: reader = PdfReader(handle) if reader.is_encrypted: raise ValueError('PDF 已加密，请提供已解密副本') if args.operation == 'form-info': fields = list(field_info(reader).values()) if args.offset < 0 or not 1 <= args.limit <= 200: raise ValueError('offset 不能小于 0，limit 需为 1–200') stop = min(len(fields), args.offset + args.limit) acroform = reader.root_object.get('/AcroForm') return {'field_count':len(fields), 'fields':fields[args.offset:stop], 'has_more':stop < len(fields), 'next_offset':stop if stop < len(fields) else None, 'has_xfa':bool(acroform and '/XFA' in acroform.get_object())} if args.operation == 'extract-images': return extract_images(reader, args) output = output_pdf(args.output, args.overwrite) if output == source: raise ValueError('输出文件不能覆盖源 PDF') writer = PdfWriter(clone_from=reader) temporary = new_temp_pdf(output) extra = {} values = {} try: if args.operation == 'form-fill': acroform = reader.root_object.get('/AcroForm') if acroform and '/XFA' in acroform.get_object(): raise ValueError('XFA 表单不属于 AcroForm 固定接口，不能声称填写成功') values = validated_values(field_info(reader), load_data(args.data, args.data_file)) writer.update_page_form_field_values(None, values, auto_regenerate=False) extra = {'fields_filled':list(values), 'requires_visual_review':True} elif args.operation == 'metadata': data = load_data(args.data, args.data_file) allowed = {'Title','Author','Subject','Keywords','Creator','Producer'} if set(data) - allowed or not all(isinstance(v,str) and len(v) <= 4096 for v in data.values()): raise ValueError('元数据仅支持 Title/Author/Subject/Keywords/Creator/Producer 文本字段') writer.add_metadata({'/'+key:value for key,value in data.items()}) extra = {'updated_keys':list(data), 'xmp_preserved':'/Metadata' in reader.root_object} elif args.operation == 'crop': box = [float(v) for v in args.box.split(',')] if len(box) != 4 or not all(math.isfinite(v) for v in box) or box[0] >= box[2] or box[1] >= box[3]: raise ValueError('box 需为左,下,右,上四个有限坐标，单位 pt') pages = parse_page_spec(args.pages, len(writer.pages)) for number in pages: media = writer.pages[number-1].mediabox if box[0] < media.left or box[1] < media.bottom or box[2] > media.right or box[3] > media.top: raise ValueError(f'裁剪框超出第 {number} 页 MediaBox') writer.pages[number-1].cropbox = RectangleObject((box[0], box[1], box[2], box[3])) extra = {'cropped_pages':pages, 'box':box, 'warning':'裁剪只改变可见范围，不删除隐藏内容，不能用于脱敏。'} with temporary.open('wb') as stream: writer.write(stream) check = PdfReader(temporary) if len(check.pages) != len(reader.pages): raise ValueError('编辑后页数异常') if args.operation == 'form-fill': actual = field_info(check) for name, value in values.items(): wanted = [str(v) for v in value] if isinstance(value, list) else str(value) if name not in actual or actual[name]['current_value'] != wanted: raise ValueError(f'字段回读校验失败：{name}') publish_temp_file(temporary, output, args.overwrite) finally: writer.close() temporary.unlink(missing_ok=True) return {'path':str(output), 'source':str(source), 'page_count':len(reader.pages), 'operation':args.operation, **extra} def extract_images(reader, args): if not 1 <= args.max_images <= 50 or args.start_image < 0: raise ValueError('max-images 需为 1–50，start-image 不能小于 0') pages = parse_page_spec(args.pages, len(reader.pages)) if len(pages) > 4: raise ValueError('一次最多处理 4 页，请指定 pages') destination = output_directory(args.output_dir) result, total_bytes = [], 0 for page_index, number in enumerate(pages): images = reader.pages[number-1].images start = args.start_image if page_index == 0 else 0 for index in range(start, len(images)): if len(result) >= args.max_images: return {'images':result, 'has_more':True, 'next_page':number, 'next_image':index, 'remaining_pages':pages[page_index:], 'note':'嵌入图片不是整页截图；页面外观请用 render_pdf.py。'} item = images[index] if item.image.width * item.image.height > 20_000_000: raise ValueError('嵌入图像超过 2000 万像素，请改为限制 DPI 的页面渲染') total_bytes += len(item.data) if total_bytes > 25 * 1024 * 1024: raise ValueError('本批嵌入图片超过 25 MiB，请减少页数或图片数') extension = Path(item.name).suffix.lower() if extension not in {'.png','.jpg','.jpeg','.jp2','.tif','.tiff'}: raise ValueError(f'不支持的嵌入图片编码：{extension}，请用页面渲染') output = destination / f'page-{number:04d}-image-{index:03d}{extension}' temporary = new_temp_pdf(output) try: temporary.write_bytes(item.data) publish_temp_file(temporary, output, args.overwrite) finally: temporary.unlink(missing_ok=True) result.append({'page':number, 'image_index':index, 'path':str(output)}) return {'images':result, 'has_more':False, 'note':'嵌入图片不是整页截图；页面外观请用 render_pdf.py。'} def main(argv=None): parser = SkillArgumentParser(description='PDF 表单、裁剪、元数据和嵌入图片固定接口') commands = parser.add_subparsers(dest='operation', required=True) for name in ('form-info','form-fill','metadata','crop','extract-images'): command = commands.add_parser(name) command.add_argument('--input', required=True) if name == 'form-info': command.add_argument('--offset', type=int, default=0) command.add_argument('--limit', type=int, default=50) else: command.add_argument('--overwrite', action='store_true') command.add_argument('--output-dir' if name == 'extract-images' else '--output', required=True) if name in {'form-fill','metadata'}: command.add_argument('--data') command.add_argument('--data-file') if name in {'crop','extract-images'}: command.add_argument('--pages') if name == 'crop': command.add_argument('--box', required=True) if name == 'extract-images': command.add_argument('--start-image', type=int, default=0) command.add_argument('--max-images', type=int, default=20) return run_cli(lambda: execute(parser.parse_args(sys.argv[1:] if argv is None else argv))) if __name__ == '__main__': raise SystemExit(main())