feat(ai): add code interpreter for AI data analysis (#16559)

## Summary

- Add code interpreter tool that enables AI to execute Python code for
data analysis, CSV processing, and chart generation
- Support for both local (development) and E2B (sandboxed production)
execution drivers
- Real-time streaming of stdout/stderr and generated files
- Frontend components for displaying code execution results with
expandable sections

## Code Quality Improvements

- Extract `getMimeType` to shared utility to reduce code duplication
between drivers
- Fix security issue: escape single quotes/backslashes in E2B driver env
variable injection
- Add `buildExecutionState` helper to reduce duplicated state object
construction
- Add `DEFAULT_CODE_INTERPRETER_TIMEOUT_MS` constant for consistency
- Fix lingui linting warning and TypeScript theme errors in frontend

## Test Plan

- [ ] Test code interpreter with local driver in development
- [ ] Test code interpreter with E2B driver in production environment
- [ ] Verify streaming output displays correctly in chat UI
- [ ] Verify generated files (charts, CSVs) are uploaded and
downloadable
- [ ] Test file upload flow (CSV, Excel) triggers code interpreter

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> Updates generated i18n catalogs for Polish and pseudo-English, adding
strings for code execution/output (code interpreter) and various UI
messages, with minor text adjustments.
> 
> - **Localization**:
> - **Generated catalogs**: Refresh `locales/generated/pl-PL.ts` and
`locales/generated/pseudo-en.ts`.
> - Add strings for code execution/output (e.g., code, copy code/output,
running/waiting states, download files, generated files, Python code
execution).
> - Include new UI texts (errors, prompts, menus) and minor text
corrections.
>   - No changes to `pt-BR`; other files unchanged functionally.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
befc13d02c. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->
This commit is contained in:
Félix Malfait
2025-12-15 16:11:24 +01:00
committed by GitHub
parent 4281a71f40
commit 2e104c8e76
151 changed files with 10361 additions and 449 deletions
@@ -0,0 +1,53 @@
import { NodeEnvironment } from 'src/engine/core-modules/twenty-config/interfaces/node-environment.interface';
import { type TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
import {
CodeInterpreterDriverType,
type CodeInterpreterModuleOptions,
} from './code-interpreter.interface';
export const codeInterpreterModuleFactory = async (
twentyConfigService: TwentyConfigService,
): Promise<CodeInterpreterModuleOptions> => {
const driverType = twentyConfigService.get('CODE_INTERPRETER_TYPE');
const timeoutMs = twentyConfigService.get('CODE_INTERPRETER_TIMEOUT_MS');
switch (driverType) {
case CodeInterpreterDriverType.LOCAL: {
const nodeEnv = twentyConfigService.get('NODE_ENV');
if (nodeEnv === NodeEnvironment.PRODUCTION) {
throw new Error(
'LOCAL code interpreter driver is not allowed in production. Use E2B driver instead by setting CODE_INTERPRETER_TYPE=E2B and providing E2B_API_KEY.',
);
}
return {
type: CodeInterpreterDriverType.LOCAL,
options: { timeoutMs },
};
}
case CodeInterpreterDriverType.E2B: {
const apiKey = twentyConfigService.get('E2B_API_KEY');
if (!apiKey) {
throw new Error(
'E2B_API_KEY is required when CODE_INTERPRETER_TYPE is E2B',
);
}
return {
type: CodeInterpreterDriverType.E2B,
options: {
apiKey,
timeoutMs,
},
};
}
default:
throw new Error(
`Invalid code interpreter driver type (${driverType}), check your .env file`,
);
}
};
@@ -0,0 +1,3 @@
export const CODE_INTERPRETER_DRIVER = Symbol('CODE_INTERPRETER_DRIVER');
export const DEFAULT_CODE_INTERPRETER_TIMEOUT_MS = 300_000;
@@ -0,0 +1,30 @@
import { type FactoryProvider, type ModuleMetadata } from '@nestjs/common';
import { type E2BDriverOptions } from './drivers/e2b.driver';
import { type LocalDriverOptions } from './drivers/local.driver';
export enum CodeInterpreterDriverType {
LOCAL = 'LOCAL',
E2B = 'E2B',
}
export type LocalDriverFactoryOptions = {
type: CodeInterpreterDriverType.LOCAL;
options: LocalDriverOptions;
};
export type E2BDriverFactoryOptions = {
type: CodeInterpreterDriverType.E2B;
options: E2BDriverOptions;
};
export type CodeInterpreterModuleOptions =
| LocalDriverFactoryOptions
| E2BDriverFactoryOptions;
export type CodeInterpreterModuleAsyncOptions = {
useFactory: (
...args: unknown[]
) => CodeInterpreterModuleOptions | Promise<CodeInterpreterModuleOptions>;
} & Pick<ModuleMetadata, 'imports'> &
Pick<FactoryProvider, 'inject'>;
@@ -0,0 +1,37 @@
import { type DynamicModule, Global } from '@nestjs/common';
import { CODE_INTERPRETER_DRIVER } from './code-interpreter.constants';
import {
CodeInterpreterDriverType,
type CodeInterpreterModuleAsyncOptions,
} from './code-interpreter.interface';
import { CodeInterpreterService } from './code-interpreter.service';
import { E2BDriver } from './drivers/e2b.driver';
import { LocalDriver } from './drivers/local.driver';
@Global()
export class CodeInterpreterModule {
static forRootAsync(
options: CodeInterpreterModuleAsyncOptions,
): DynamicModule {
const provider = {
provide: CODE_INTERPRETER_DRIVER,
useFactory: async (...args: unknown[]) => {
const config = await options.useFactory(...args);
return config.type === CodeInterpreterDriverType.LOCAL
? new LocalDriver(config.options)
: new E2BDriver(config.options);
},
inject: options.inject ?? [],
};
return {
module: CodeInterpreterModule,
imports: options.imports ?? [],
providers: [CodeInterpreterService, provider],
exports: [CodeInterpreterService],
};
}
}
@@ -0,0 +1,27 @@
import { Inject, Injectable } from '@nestjs/common';
import { CODE_INTERPRETER_DRIVER } from './code-interpreter.constants';
import {
type CodeExecutionResult,
type CodeInterpreterDriver,
type ExecutionContext,
type InputFile,
type StreamCallbacks,
} from './drivers/interfaces/code-interpreter-driver.interface';
@Injectable()
export class CodeInterpreterService implements CodeInterpreterDriver {
constructor(
@Inject(CODE_INTERPRETER_DRIVER) private driver: CodeInterpreterDriver,
) {}
execute(
code: string,
files?: InputFile[],
context?: ExecutionContext,
callbacks?: StreamCallbacks,
): Promise<CodeExecutionResult> {
return this.driver.execute(code, files, context, callbacks);
}
}
@@ -0,0 +1,147 @@
import { promises as fs } from 'fs';
import { join } from 'path';
import { Sandbox } from '@e2b/code-interpreter';
import { DEFAULT_CODE_INTERPRETER_TIMEOUT_MS } from 'src/engine/core-modules/code-interpreter/code-interpreter.constants';
import { getMimeType } from 'src/engine/core-modules/code-interpreter/utils/get-mime-type.util';
import {
type CodeExecutionResult,
type CodeInterpreterDriver,
type ExecutionContext,
type InputFile,
type OutputFile,
type StreamCallbacks,
} from './interfaces/code-interpreter-driver.interface';
export type E2BDriverOptions = {
apiKey: string;
timeoutMs?: number;
};
const SANDBOX_SCRIPTS_PATH = join(__dirname, '..', 'sandbox-scripts');
async function uploadDirectoryToSandbox(
sbx: Sandbox,
localPath: string,
remotePath: string,
) {
const entries = await fs.readdir(localPath, { withFileTypes: true });
for (const entry of entries) {
const localEntryPath = join(localPath, entry.name);
const remoteEntryPath = `${remotePath}/${entry.name}`;
if (entry.isDirectory()) {
await uploadDirectoryToSandbox(sbx, localEntryPath, remoteEntryPath);
} else {
const content = await fs.readFile(localEntryPath);
const arrayBuffer = new Uint8Array(content).buffer;
await sbx.files.write(remoteEntryPath, arrayBuffer);
}
}
}
export class E2BDriver implements CodeInterpreterDriver {
constructor(private options: E2BDriverOptions) {}
async execute(
code: string,
files?: InputFile[],
context?: ExecutionContext,
callbacks?: StreamCallbacks,
): Promise<CodeExecutionResult> {
const sbx = await Sandbox.create({
apiKey: this.options.apiKey,
timeoutMs: this.options.timeoutMs ?? DEFAULT_CODE_INTERPRETER_TIMEOUT_MS,
});
try {
// Upload pre-installed scripts to sandbox
try {
await uploadDirectoryToSandbox(
sbx,
SANDBOX_SCRIPTS_PATH,
'/home/user/scripts',
);
} catch {
// Scripts directory might not exist
}
for (const file of files ?? []) {
const arrayBuffer = new Uint8Array(file.content).buffer;
await sbx.files.write(`/home/user/${file.filename}`, arrayBuffer);
}
const envSetup = context?.env
? `import os\n${Object.entries(context.env)
.map(([key, value]) => {
const escapedValue = value
.replace(/\\/g, '\\\\')
.replace(/'/g, "\\'")
.replace(/\n/g, '\\n')
.replace(/\r/g, '\\r');
return `os.environ['${key}'] = '${escapedValue}'`;
})
.join('\n')}\n\n`
: '';
const outputFiles: OutputFile[] = [];
let chartCounter = 0;
const execution = await sbx.runCode(envSetup + code, {
onStdout: (data) => callbacks?.onStdout?.(data.line),
onStderr: (data) => callbacks?.onStderr?.(data.line),
onResult: (result) => {
if (result.png) {
const outputFile: OutputFile = {
filename: `chart-${chartCounter++}.png`,
content: Buffer.from(result.png, 'base64'),
mimeType: 'image/png',
};
outputFiles.push(outputFile);
callbacks?.onResult?.(outputFile);
}
},
});
try {
const outputDir = await sbx.files.list('/home/user/output');
for (const file of outputDir) {
if (file.type === 'file') {
const content = await sbx.files.read(
`/home/user/output/${file.name}`,
);
const outputFile: OutputFile = {
filename: file.name,
content: Buffer.from(content),
mimeType: getMimeType(file.name),
};
outputFiles.push(outputFile);
callbacks?.onResult?.(outputFile);
}
}
} catch {
// Output directory doesn't exist - that's fine
}
return {
stdout: execution.logs.stdout.join('\n'),
stderr: execution.logs.stderr.join('\n'),
exitCode: execution.error ? 1 : 0,
files: outputFiles,
error: execution.error?.value,
};
} finally {
await sbx.kill();
}
}
}
@@ -0,0 +1,38 @@
export type InputFile = {
filename: string;
content: Buffer;
mimeType: string;
};
export type OutputFile = {
filename: string;
content: Buffer;
mimeType: string;
};
export type CodeExecutionResult = {
stdout: string;
stderr: string;
exitCode: number;
files: OutputFile[];
error?: string;
};
export type ExecutionContext = {
env?: Record<string, string>;
};
export type StreamCallbacks = {
onStdout?: (line: string) => void;
onStderr?: (line: string) => void;
onResult?: (result: OutputFile) => void;
};
export interface CodeInterpreterDriver {
execute(
code: string,
files?: InputFile[],
context?: ExecutionContext,
callbacks?: StreamCallbacks,
): Promise<CodeExecutionResult>;
}
@@ -0,0 +1,212 @@
import { spawn } from 'child_process';
import { promises as fs } from 'fs';
import { tmpdir } from 'os';
import { basename, join } from 'path';
import { DEFAULT_CODE_INTERPRETER_TIMEOUT_MS } from 'src/engine/core-modules/code-interpreter/code-interpreter.constants';
import { getMimeType } from 'src/engine/core-modules/code-interpreter/utils/get-mime-type.util';
import {
type CodeExecutionResult,
type CodeInterpreterDriver,
type ExecutionContext,
type InputFile,
type OutputFile,
type StreamCallbacks,
} from './interfaces/code-interpreter-driver.interface';
export type LocalDriverOptions = {
timeoutMs?: number;
};
const SANDBOX_SCRIPTS_PATH = join(__dirname, '..', 'sandbox-scripts');
async function copyDirectoryRecursive(src: string, dest: string) {
await fs.mkdir(dest, { recursive: true });
const entries = await fs.readdir(src, { withFileTypes: true });
for (const entry of entries) {
const srcPath = join(src, entry.name);
const destPath = join(dest, entry.name);
if (entry.isDirectory()) {
await copyDirectoryRecursive(srcPath, destPath);
} else {
await fs.copyFile(srcPath, destPath);
}
}
}
// WARNING: This driver is UNSAFE and should only be used for development.
// It executes arbitrary Python code on the server without any sandboxing.
export class LocalDriver implements CodeInterpreterDriver {
constructor(private options: LocalDriverOptions = {}) {}
async execute(
code: string,
files?: InputFile[],
context?: ExecutionContext,
callbacks?: StreamCallbacks,
): Promise<CodeExecutionResult> {
const workDir = await fs.mkdtemp(join(tmpdir(), 'code-interpreter-'));
const outputDir = join(workDir, 'output');
const scriptsDir = join(workDir, 'scripts');
await fs.mkdir(outputDir);
// Copy pre-installed scripts to sandbox
try {
await copyDirectoryRecursive(SANDBOX_SCRIPTS_PATH, scriptsDir);
} catch {
// Scripts directory might not exist in dev environment
}
try {
for (const file of files ?? []) {
const safeFilename = basename(file.filename);
await fs.writeFile(join(workDir, safeFilename), file.content);
}
// Rewrite E2B-style paths to local paths for compatibility
const rewrittenCode = code
.replace(/\/home\/user\/scripts\//g, `${scriptsDir}/`)
.replace(/\/home\/user\/scripts/g, scriptsDir)
.replace(/\/home\/user\/output\//g, `${outputDir}/`)
.replace(/\/home\/user\/output/g, outputDir)
.replace(/\/home\/user\//g, `${workDir}/`)
.replace(/\/home\/user/g, workDir);
const scriptPath = join(workDir, 'script.py');
await fs.writeFile(scriptPath, rewrittenCode);
const timeoutMs =
this.options.timeoutMs ?? DEFAULT_CODE_INTERPRETER_TIMEOUT_MS;
const { stdout, stderr, exitCode, error } = await this.runPythonScript(
scriptPath,
workDir,
outputDir,
context?.env,
timeoutMs,
callbacks,
);
const outputFiles: OutputFile[] = [];
try {
const outputEntries = await fs.readdir(outputDir, {
withFileTypes: true,
});
for (const entry of outputEntries) {
if (entry.isFile()) {
const content = await fs.readFile(join(outputDir, entry.name));
const outputFile: OutputFile = {
filename: entry.name,
content,
mimeType: getMimeType(entry.name),
};
outputFiles.push(outputFile);
callbacks?.onResult?.(outputFile);
}
}
} catch {
// Output directory might be empty or not exist
}
return {
stdout,
stderr,
exitCode,
files: outputFiles,
error,
};
} finally {
await fs.rm(workDir, { recursive: true, force: true });
}
}
private runPythonScript(
scriptPath: string,
workDir: string,
outputDir: string,
env?: Record<string, string>,
timeoutMs?: number,
callbacks?: StreamCallbacks,
): Promise<{
stdout: string;
stderr: string;
exitCode: number;
error?: string;
}> {
return new Promise((resolve) => {
const child = spawn('python3', [scriptPath], {
cwd: workDir,
env: {
...process.env,
OUTPUT_DIR: outputDir,
...env,
},
});
let stdout = '';
let stderr = '';
let killed = false;
const timeout = setTimeout(() => {
killed = true;
child.kill('SIGKILL');
}, timeoutMs ?? DEFAULT_CODE_INTERPRETER_TIMEOUT_MS);
child.stdout.on('data', (data: Buffer) => {
const text = data.toString();
stdout += text;
const lines = text.split('\n');
for (const line of lines) {
if (line) {
callbacks?.onStdout?.(line);
}
}
});
child.stderr.on('data', (data: Buffer) => {
const text = data.toString();
stderr += text;
const lines = text.split('\n');
for (const line of lines) {
if (line) {
callbacks?.onStderr?.(line);
}
}
});
child.on('close', (code) => {
clearTimeout(timeout);
resolve({
stdout,
stderr,
exitCode: code ?? 0,
error: killed ? 'Process timed out' : undefined,
});
});
child.on('error', (err) => {
clearTimeout(timeout);
resolve({
stdout,
stderr,
exitCode: 1,
error: err.message,
});
});
});
}
}
@@ -0,0 +1,159 @@
#!/usr/bin/env python3
"""
Tool to pack a directory into a .docx, .pptx, or .xlsx file with XML formatting undone.
Example usage:
python pack.py <input_directory> <office_file> [--force]
"""
import argparse
import shutil
import subprocess
import sys
import tempfile
import defusedxml.minidom
import zipfile
from pathlib import Path
def main():
parser = argparse.ArgumentParser(description="Pack a directory into an Office file")
parser.add_argument("input_directory", help="Unpacked Office document directory")
parser.add_argument("output_file", help="Output Office file (.docx/.pptx/.xlsx)")
parser.add_argument("--force", action="store_true", help="Skip validation")
args = parser.parse_args()
try:
success = pack_document(
args.input_directory, args.output_file, validate=not args.force
)
# Show warning if validation was skipped
if args.force:
print("Warning: Skipped validation, file may be corrupt", file=sys.stderr)
# Exit with error if validation failed
elif not success:
print("Contents would produce a corrupt file.", file=sys.stderr)
print("Please validate XML before repacking.", file=sys.stderr)
print("Use --force to skip validation and pack anyway.", file=sys.stderr)
sys.exit(1)
except ValueError as e:
sys.exit(f"Error: {e}")
def pack_document(input_dir, output_file, validate=False):
"""Pack a directory into an Office file (.docx/.pptx/.xlsx).
Args:
input_dir: Path to unpacked Office document directory
output_file: Path to output Office file
validate: If True, validates with soffice (default: False)
Returns:
bool: True if successful, False if validation failed
"""
input_dir = Path(input_dir)
output_file = Path(output_file)
if not input_dir.is_dir():
raise ValueError(f"{input_dir} is not a directory")
if output_file.suffix.lower() not in {".docx", ".pptx", ".xlsx"}:
raise ValueError(f"{output_file} must be a .docx, .pptx, or .xlsx file")
# Work in temporary directory to avoid modifying original
with tempfile.TemporaryDirectory() as temp_dir:
temp_content_dir = Path(temp_dir) / "content"
shutil.copytree(input_dir, temp_content_dir)
# Process XML files to remove pretty-printing whitespace
for pattern in ["*.xml", "*.rels"]:
for xml_file in temp_content_dir.rglob(pattern):
condense_xml(xml_file)
# Create final Office file as zip archive
output_file.parent.mkdir(parents=True, exist_ok=True)
with zipfile.ZipFile(output_file, "w", zipfile.ZIP_DEFLATED) as zf:
for f in temp_content_dir.rglob("*"):
if f.is_file():
zf.write(f, f.relative_to(temp_content_dir))
# Validate if requested
if validate:
if not validate_document(output_file):
output_file.unlink() # Delete the corrupt file
return False
return True
def validate_document(doc_path):
"""Validate document by converting to HTML with soffice."""
# Determine the correct filter based on file extension
match doc_path.suffix.lower():
case ".docx":
filter_name = "html:HTML"
case ".pptx":
filter_name = "html:impress_html_Export"
case ".xlsx":
filter_name = "html:HTML (StarCalc)"
with tempfile.TemporaryDirectory() as temp_dir:
try:
result = subprocess.run(
[
"soffice",
"--headless",
"--convert-to",
filter_name,
"--outdir",
temp_dir,
str(doc_path),
],
capture_output=True,
timeout=10,
text=True,
)
if not (Path(temp_dir) / f"{doc_path.stem}.html").exists():
error_msg = result.stderr.strip() or "Document validation failed"
print(f"Validation error: {error_msg}", file=sys.stderr)
return False
return True
except FileNotFoundError:
print("Warning: soffice not found. Skipping validation.", file=sys.stderr)
return True
except subprocess.TimeoutExpired:
print("Validation error: Timeout during conversion", file=sys.stderr)
return False
except Exception as e:
print(f"Validation error: {e}", file=sys.stderr)
return False
def condense_xml(xml_file):
"""Strip unnecessary whitespace and remove comments."""
with open(xml_file, "r", encoding="utf-8") as f:
dom = defusedxml.minidom.parse(f)
# Process each element to remove whitespace and comments
for element in dom.getElementsByTagName("*"):
# Skip w:t elements and their processing
if element.tagName.endswith(":t"):
continue
# Remove whitespace-only text nodes and comment nodes
for child in list(element.childNodes):
if (
child.nodeType == child.TEXT_NODE
and child.nodeValue
and child.nodeValue.strip() == ""
) or child.nodeType == child.COMMENT_NODE:
element.removeChild(child)
# Write back the condensed XML
with open(xml_file, "wb") as f:
f.write(dom.toxml(encoding="UTF-8"))
if __name__ == "__main__":
main()
@@ -0,0 +1,29 @@
#!/usr/bin/env python3
"""Unpack and format XML contents of Office files (.docx, .pptx, .xlsx)"""
import random
import sys
import defusedxml.minidom
import zipfile
from pathlib import Path
# Get command line arguments
assert len(sys.argv) == 3, "Usage: python unpack.py <office_file> <output_dir>"
input_file, output_dir = sys.argv[1], sys.argv[2]
# Extract and format
output_path = Path(output_dir)
output_path.mkdir(parents=True, exist_ok=True)
zipfile.ZipFile(input_file).extractall(output_path)
# Pretty print all XML files
xml_files = list(output_path.rglob("*.xml")) + list(output_path.rglob("*.rels"))
for xml_file in xml_files:
content = xml_file.read_text(encoding="utf-8")
dom = defusedxml.minidom.parseString(content)
xml_file.write_bytes(dom.toprettyxml(indent=" ", encoding="ascii"))
# For .docx files, suggest an RSID for tracked changes
if input_file.endswith(".docx"):
suggested_rsid = "".join(random.choices("0123456789ABCDEF", k=8))
print(f"Suggested RSID for edit session: {suggested_rsid}")
@@ -0,0 +1,69 @@
#!/usr/bin/env python3
"""
Command line tool to validate Office document XML files against XSD schemas and tracked changes.
Usage:
python validate.py <dir> --original <original_file>
"""
import argparse
import sys
from pathlib import Path
from validation import DOCXSchemaValidator, PPTXSchemaValidator, RedliningValidator
def main():
parser = argparse.ArgumentParser(description="Validate Office document XML files")
parser.add_argument(
"unpacked_dir",
help="Path to unpacked Office document directory",
)
parser.add_argument(
"--original",
required=True,
help="Path to original file (.docx/.pptx/.xlsx)",
)
parser.add_argument(
"-v",
"--verbose",
action="store_true",
help="Enable verbose output",
)
args = parser.parse_args()
# Validate paths
unpacked_dir = Path(args.unpacked_dir)
original_file = Path(args.original)
file_extension = original_file.suffix.lower()
assert unpacked_dir.is_dir(), f"Error: {unpacked_dir} is not a directory"
assert original_file.is_file(), f"Error: {original_file} is not a file"
assert file_extension in [".docx", ".pptx", ".xlsx"], (
f"Error: {original_file} must be a .docx, .pptx, or .xlsx file"
)
# Run validations
match file_extension:
case ".docx":
validators = [DOCXSchemaValidator, RedliningValidator]
case ".pptx":
validators = [PPTXSchemaValidator]
case _:
print(f"Error: Validation not supported for file type {file_extension}")
sys.exit(1)
# Run validators
success = True
for V in validators:
validator = V(unpacked_dir, original_file, verbose=args.verbose)
if not validator.validate():
success = False
if success:
print("All validations PASSED!")
sys.exit(0 if success else 1)
if __name__ == "__main__":
main()
@@ -0,0 +1,15 @@
"""
Validation modules for Word document processing.
"""
from .base import BaseSchemaValidator
from .docx import DOCXSchemaValidator
from .pptx import PPTXSchemaValidator
from .redlining import RedliningValidator
__all__ = [
"BaseSchemaValidator",
"DOCXSchemaValidator",
"PPTXSchemaValidator",
"RedliningValidator",
]
@@ -0,0 +1,951 @@
"""
Base validator with common validation logic for document files.
"""
import re
from pathlib import Path
import lxml.etree
class BaseSchemaValidator:
"""Base validator with common validation logic for document files."""
# Elements whose 'id' attributes must be unique within their file
# Format: element_name -> (attribute_name, scope)
# scope can be 'file' (unique within file) or 'global' (unique across all files)
UNIQUE_ID_REQUIREMENTS = {
# Word elements
"comment": ("id", "file"), # Comment IDs in comments.xml
"commentrangestart": ("id", "file"), # Must match comment IDs
"commentrangeend": ("id", "file"), # Must match comment IDs
"bookmarkstart": ("id", "file"), # Bookmark start IDs
"bookmarkend": ("id", "file"), # Bookmark end IDs
# Note: ins and del (track changes) can share IDs when part of same revision
# PowerPoint elements
"sldid": ("id", "file"), # Slide IDs in presentation.xml
"sldmasterid": ("id", "global"), # Slide master IDs must be globally unique
"sldlayoutid": ("id", "global"), # Slide layout IDs must be globally unique
"cm": ("authorid", "file"), # Comment author IDs
# Excel elements
"sheet": ("sheetid", "file"), # Sheet IDs in workbook.xml
"definedname": ("id", "file"), # Named range IDs
# Drawing/Shape elements (all formats)
"cxnsp": ("id", "file"), # Connection shape IDs
"sp": ("id", "file"), # Shape IDs
"pic": ("id", "file"), # Picture IDs
"grpsp": ("id", "file"), # Group shape IDs
}
# Mapping of element names to expected relationship types
# Subclasses should override this with format-specific mappings
ELEMENT_RELATIONSHIP_TYPES = {}
# Unified schema mappings for all Office document types
SCHEMA_MAPPINGS = {
# Document type specific schemas
"word": "ISO-IEC29500-4_2016/wml.xsd", # Word documents
"ppt": "ISO-IEC29500-4_2016/pml.xsd", # PowerPoint presentations
"xl": "ISO-IEC29500-4_2016/sml.xsd", # Excel spreadsheets
# Common file types
"[Content_Types].xml": "ecma/fouth-edition/opc-contentTypes.xsd",
"app.xml": "ISO-IEC29500-4_2016/shared-documentPropertiesExtended.xsd",
"core.xml": "ecma/fouth-edition/opc-coreProperties.xsd",
"custom.xml": "ISO-IEC29500-4_2016/shared-documentPropertiesCustom.xsd",
".rels": "ecma/fouth-edition/opc-relationships.xsd",
# Word-specific files
"people.xml": "microsoft/wml-2012.xsd",
"commentsIds.xml": "microsoft/wml-cid-2016.xsd",
"commentsExtensible.xml": "microsoft/wml-cex-2018.xsd",
"commentsExtended.xml": "microsoft/wml-2012.xsd",
# Chart files (common across document types)
"chart": "ISO-IEC29500-4_2016/dml-chart.xsd",
# Theme files (common across document types)
"theme": "ISO-IEC29500-4_2016/dml-main.xsd",
# Drawing and media files
"drawing": "ISO-IEC29500-4_2016/dml-main.xsd",
}
# Unified namespace constants
MC_NAMESPACE = "http://schemas.openxmlformats.org/markup-compatibility/2006"
XML_NAMESPACE = "http://www.w3.org/XML/1998/namespace"
# Common OOXML namespaces used across validators
PACKAGE_RELATIONSHIPS_NAMESPACE = (
"http://schemas.openxmlformats.org/package/2006/relationships"
)
OFFICE_RELATIONSHIPS_NAMESPACE = (
"http://schemas.openxmlformats.org/officeDocument/2006/relationships"
)
CONTENT_TYPES_NAMESPACE = (
"http://schemas.openxmlformats.org/package/2006/content-types"
)
# Folders where we should clean ignorable namespaces
MAIN_CONTENT_FOLDERS = {"word", "ppt", "xl"}
# All allowed OOXML namespaces (superset of all document types)
OOXML_NAMESPACES = {
"http://schemas.openxmlformats.org/officeDocument/2006/math",
"http://schemas.openxmlformats.org/officeDocument/2006/relationships",
"http://schemas.openxmlformats.org/schemaLibrary/2006/main",
"http://schemas.openxmlformats.org/drawingml/2006/main",
"http://schemas.openxmlformats.org/drawingml/2006/chart",
"http://schemas.openxmlformats.org/drawingml/2006/chartDrawing",
"http://schemas.openxmlformats.org/drawingml/2006/diagram",
"http://schemas.openxmlformats.org/drawingml/2006/picture",
"http://schemas.openxmlformats.org/drawingml/2006/spreadsheetDrawing",
"http://schemas.openxmlformats.org/drawingml/2006/wordprocessingDrawing",
"http://schemas.openxmlformats.org/wordprocessingml/2006/main",
"http://schemas.openxmlformats.org/presentationml/2006/main",
"http://schemas.openxmlformats.org/spreadsheetml/2006/main",
"http://schemas.openxmlformats.org/officeDocument/2006/sharedTypes",
"http://www.w3.org/XML/1998/namespace",
}
def __init__(self, unpacked_dir, original_file, verbose=False):
self.unpacked_dir = Path(unpacked_dir).resolve()
self.original_file = Path(original_file)
self.verbose = verbose
# Set schemas directory
self.schemas_dir = Path(__file__).parent.parent.parent / "schemas"
# Get all XML and .rels files
patterns = ["*.xml", "*.rels"]
self.xml_files = [
f for pattern in patterns for f in self.unpacked_dir.rglob(pattern)
]
if not self.xml_files:
print(f"Warning: No XML files found in {self.unpacked_dir}")
def validate(self):
"""Run all validation checks and return True if all pass."""
raise NotImplementedError("Subclasses must implement the validate method")
def validate_xml(self):
"""Validate that all XML files are well-formed."""
errors = []
for xml_file in self.xml_files:
try:
# Try to parse the XML file
lxml.etree.parse(str(xml_file))
except lxml.etree.XMLSyntaxError as e:
errors.append(
f" {xml_file.relative_to(self.unpacked_dir)}: "
f"Line {e.lineno}: {e.msg}"
)
except Exception as e:
errors.append(
f" {xml_file.relative_to(self.unpacked_dir)}: "
f"Unexpected error: {str(e)}"
)
if errors:
print(f"FAILED - Found {len(errors)} XML violations:")
for error in errors:
print(error)
return False
else:
if self.verbose:
print("PASSED - All XML files are well-formed")
return True
def validate_namespaces(self):
"""Validate that namespace prefixes in Ignorable attributes are declared."""
errors = []
for xml_file in self.xml_files:
try:
root = lxml.etree.parse(str(xml_file)).getroot()
declared = set(root.nsmap.keys()) - {None} # Exclude default namespace
for attr_val in [
v for k, v in root.attrib.items() if k.endswith("Ignorable")
]:
undeclared = set(attr_val.split()) - declared
errors.extend(
f" {xml_file.relative_to(self.unpacked_dir)}: "
f"Namespace '{ns}' in Ignorable but not declared"
for ns in undeclared
)
except lxml.etree.XMLSyntaxError:
continue
if errors:
print(f"FAILED - {len(errors)} namespace issues:")
for error in errors:
print(error)
return False
if self.verbose:
print("PASSED - All namespace prefixes properly declared")
return True
def validate_unique_ids(self):
"""Validate that specific IDs are unique according to OOXML requirements."""
errors = []
global_ids = {} # Track globally unique IDs across all files
for xml_file in self.xml_files:
try:
root = lxml.etree.parse(str(xml_file)).getroot()
file_ids = {} # Track IDs that must be unique within this file
# Remove all mc:AlternateContent elements from the tree
mc_elements = root.xpath(
".//mc:AlternateContent", namespaces={"mc": self.MC_NAMESPACE}
)
for elem in mc_elements:
elem.getparent().remove(elem)
# Now check IDs in the cleaned tree
for elem in root.iter():
# Get the element name without namespace
tag = (
elem.tag.split("}")[-1].lower()
if "}" in elem.tag
else elem.tag.lower()
)
# Check if this element type has ID uniqueness requirements
if tag in self.UNIQUE_ID_REQUIREMENTS:
attr_name, scope = self.UNIQUE_ID_REQUIREMENTS[tag]
# Look for the specified attribute
id_value = None
for attr, value in elem.attrib.items():
attr_local = (
attr.split("}")[-1].lower()
if "}" in attr
else attr.lower()
)
if attr_local == attr_name:
id_value = value
break
if id_value is not None:
if scope == "global":
# Check global uniqueness
if id_value in global_ids:
prev_file, prev_line, prev_tag = global_ids[
id_value
]
errors.append(
f" {xml_file.relative_to(self.unpacked_dir)}: "
f"Line {elem.sourceline}: Global ID '{id_value}' in <{tag}> "
f"already used in {prev_file} at line {prev_line} in <{prev_tag}>"
)
else:
global_ids[id_value] = (
xml_file.relative_to(self.unpacked_dir),
elem.sourceline,
tag,
)
elif scope == "file":
# Check file-level uniqueness
key = (tag, attr_name)
if key not in file_ids:
file_ids[key] = {}
if id_value in file_ids[key]:
prev_line = file_ids[key][id_value]
errors.append(
f" {xml_file.relative_to(self.unpacked_dir)}: "
f"Line {elem.sourceline}: Duplicate {attr_name}='{id_value}' in <{tag}> "
f"(first occurrence at line {prev_line})"
)
else:
file_ids[key][id_value] = elem.sourceline
except (lxml.etree.XMLSyntaxError, Exception) as e:
errors.append(
f" {xml_file.relative_to(self.unpacked_dir)}: Error: {e}"
)
if errors:
print(f"FAILED - Found {len(errors)} ID uniqueness violations:")
for error in errors:
print(error)
return False
else:
if self.verbose:
print("PASSED - All required IDs are unique")
return True
def validate_file_references(self):
"""
Validate that all .rels files properly reference files and that all files are referenced.
"""
errors = []
# Find all .rels files
rels_files = list(self.unpacked_dir.rglob("*.rels"))
if not rels_files:
if self.verbose:
print("PASSED - No .rels files found")
return True
# Get all files in the unpacked directory (excluding reference files)
all_files = []
for file_path in self.unpacked_dir.rglob("*"):
if (
file_path.is_file()
and file_path.name != "[Content_Types].xml"
and not file_path.name.endswith(".rels")
): # This file is not referenced by .rels
all_files.append(file_path.resolve())
# Track all files that are referenced by any .rels file
all_referenced_files = set()
if self.verbose:
print(
f"Found {len(rels_files)} .rels files and {len(all_files)} target files"
)
# Check each .rels file
for rels_file in rels_files:
try:
# Parse relationships file
rels_root = lxml.etree.parse(str(rels_file)).getroot()
# Get the directory where this .rels file is located
rels_dir = rels_file.parent
# Find all relationships and their targets
referenced_files = set()
broken_refs = []
for rel in rels_root.findall(
".//ns:Relationship",
namespaces={"ns": self.PACKAGE_RELATIONSHIPS_NAMESPACE},
):
target = rel.get("Target")
if target and not target.startswith(
("http", "mailto:")
): # Skip external URLs
# Resolve the target path relative to the .rels file location
if rels_file.name == ".rels":
# Root .rels file - targets are relative to unpacked_dir
target_path = self.unpacked_dir / target
else:
# Other .rels files - targets are relative to their parent's parent
# e.g., word/_rels/document.xml.rels -> targets relative to word/
base_dir = rels_dir.parent
target_path = base_dir / target
# Normalize the path and check if it exists
try:
target_path = target_path.resolve()
if target_path.exists() and target_path.is_file():
referenced_files.add(target_path)
all_referenced_files.add(target_path)
else:
broken_refs.append((target, rel.sourceline))
except (OSError, ValueError):
broken_refs.append((target, rel.sourceline))
# Report broken references
if broken_refs:
rel_path = rels_file.relative_to(self.unpacked_dir)
for broken_ref, line_num in broken_refs:
errors.append(
f" {rel_path}: Line {line_num}: Broken reference to {broken_ref}"
)
except Exception as e:
rel_path = rels_file.relative_to(self.unpacked_dir)
errors.append(f" Error parsing {rel_path}: {e}")
# Check for unreferenced files (files that exist but are not referenced anywhere)
unreferenced_files = set(all_files) - all_referenced_files
if unreferenced_files:
for unref_file in sorted(unreferenced_files):
unref_rel_path = unref_file.relative_to(self.unpacked_dir)
errors.append(f" Unreferenced file: {unref_rel_path}")
if errors:
print(f"FAILED - Found {len(errors)} relationship validation errors:")
for error in errors:
print(error)
print(
"CRITICAL: These errors will cause the document to appear corrupt. "
+ "Broken references MUST be fixed, "
+ "and unreferenced files MUST be referenced or removed."
)
return False
else:
if self.verbose:
print(
"PASSED - All references are valid and all files are properly referenced"
)
return True
def validate_all_relationship_ids(self):
"""
Validate that all r:id attributes in XML files reference existing IDs
in their corresponding .rels files, and optionally validate relationship types.
"""
import lxml.etree
errors = []
# Process each XML file that might contain r:id references
for xml_file in self.xml_files:
# Skip .rels files themselves
if xml_file.suffix == ".rels":
continue
# Determine the corresponding .rels file
# For dir/file.xml, it's dir/_rels/file.xml.rels
rels_dir = xml_file.parent / "_rels"
rels_file = rels_dir / f"{xml_file.name}.rels"
# Skip if there's no corresponding .rels file (that's okay)
if not rels_file.exists():
continue
try:
# Parse the .rels file to get valid relationship IDs and their types
rels_root = lxml.etree.parse(str(rels_file)).getroot()
rid_to_type = {}
for rel in rels_root.findall(
f".//{{{self.PACKAGE_RELATIONSHIPS_NAMESPACE}}}Relationship"
):
rid = rel.get("Id")
rel_type = rel.get("Type", "")
if rid:
# Check for duplicate rIds
if rid in rid_to_type:
rels_rel_path = rels_file.relative_to(self.unpacked_dir)
errors.append(
f" {rels_rel_path}: Line {rel.sourceline}: "
f"Duplicate relationship ID '{rid}' (IDs must be unique)"
)
# Extract just the type name from the full URL
type_name = (
rel_type.split("/")[-1] if "/" in rel_type else rel_type
)
rid_to_type[rid] = type_name
# Parse the XML file to find all r:id references
xml_root = lxml.etree.parse(str(xml_file)).getroot()
# Find all elements with r:id attributes
for elem in xml_root.iter():
# Check for r:id attribute (relationship ID)
rid_attr = elem.get(f"{{{self.OFFICE_RELATIONSHIPS_NAMESPACE}}}id")
if rid_attr:
xml_rel_path = xml_file.relative_to(self.unpacked_dir)
elem_name = (
elem.tag.split("}")[-1] if "}" in elem.tag else elem.tag
)
# Check if the ID exists
if rid_attr not in rid_to_type:
errors.append(
f" {xml_rel_path}: Line {elem.sourceline}: "
f"<{elem_name}> references non-existent relationship '{rid_attr}' "
f"(valid IDs: {', '.join(sorted(rid_to_type.keys())[:5])}{'...' if len(rid_to_type) > 5 else ''})"
)
# Check if we have type expectations for this element
elif self.ELEMENT_RELATIONSHIP_TYPES:
expected_type = self._get_expected_relationship_type(
elem_name
)
if expected_type:
actual_type = rid_to_type[rid_attr]
# Check if the actual type matches or contains the expected type
if expected_type not in actual_type.lower():
errors.append(
f" {xml_rel_path}: Line {elem.sourceline}: "
f"<{elem_name}> references '{rid_attr}' which points to '{actual_type}' "
f"but should point to a '{expected_type}' relationship"
)
except Exception as e:
xml_rel_path = xml_file.relative_to(self.unpacked_dir)
errors.append(f" Error processing {xml_rel_path}: {e}")
if errors:
print(f"FAILED - Found {len(errors)} relationship ID reference errors:")
for error in errors:
print(error)
print("\nThese ID mismatches will cause the document to appear corrupt!")
return False
else:
if self.verbose:
print("PASSED - All relationship ID references are valid")
return True
def _get_expected_relationship_type(self, element_name):
"""
Get the expected relationship type for an element.
First checks the explicit mapping, then tries pattern detection.
"""
# Normalize element name to lowercase
elem_lower = element_name.lower()
# Check explicit mapping first
if elem_lower in self.ELEMENT_RELATIONSHIP_TYPES:
return self.ELEMENT_RELATIONSHIP_TYPES[elem_lower]
# Try pattern detection for common patterns
# Pattern 1: Elements ending in "Id" often expect a relationship of the prefix type
if elem_lower.endswith("id") and len(elem_lower) > 2:
# e.g., "sldId" -> "sld", "sldMasterId" -> "sldMaster"
prefix = elem_lower[:-2] # Remove "id"
# Check if this might be a compound like "sldMasterId"
if prefix.endswith("master"):
return prefix.lower()
elif prefix.endswith("layout"):
return prefix.lower()
else:
# Simple case like "sldId" -> "slide"
# Common transformations
if prefix == "sld":
return "slide"
return prefix.lower()
# Pattern 2: Elements ending in "Reference" expect a relationship of the prefix type
if elem_lower.endswith("reference") and len(elem_lower) > 9:
prefix = elem_lower[:-9] # Remove "reference"
return prefix.lower()
return None
def validate_content_types(self):
"""Validate that all content files are properly declared in [Content_Types].xml."""
errors = []
# Find [Content_Types].xml file
content_types_file = self.unpacked_dir / "[Content_Types].xml"
if not content_types_file.exists():
print("FAILED - [Content_Types].xml file not found")
return False
try:
# Parse and get all declared parts and extensions
root = lxml.etree.parse(str(content_types_file)).getroot()
declared_parts = set()
declared_extensions = set()
# Get Override declarations (specific files)
for override in root.findall(
f".//{{{self.CONTENT_TYPES_NAMESPACE}}}Override"
):
part_name = override.get("PartName")
if part_name is not None:
declared_parts.add(part_name.lstrip("/"))
# Get Default declarations (by extension)
for default in root.findall(
f".//{{{self.CONTENT_TYPES_NAMESPACE}}}Default"
):
extension = default.get("Extension")
if extension is not None:
declared_extensions.add(extension.lower())
# Root elements that require content type declaration
declarable_roots = {
"sld",
"sldLayout",
"sldMaster",
"presentation", # PowerPoint
"document", # Word
"workbook",
"worksheet", # Excel
"theme", # Common
}
# Common media file extensions that should be declared
media_extensions = {
"png": "image/png",
"jpg": "image/jpeg",
"jpeg": "image/jpeg",
"gif": "image/gif",
"bmp": "image/bmp",
"tiff": "image/tiff",
"wmf": "image/x-wmf",
"emf": "image/x-emf",
}
# Get all files in the unpacked directory
all_files = list(self.unpacked_dir.rglob("*"))
all_files = [f for f in all_files if f.is_file()]
# Check all XML files for Override declarations
for xml_file in self.xml_files:
path_str = str(xml_file.relative_to(self.unpacked_dir)).replace(
"\\", "/"
)
# Skip non-content files
if any(
skip in path_str
for skip in [".rels", "[Content_Types]", "docProps/", "_rels/"]
):
continue
try:
root_tag = lxml.etree.parse(str(xml_file)).getroot().tag
root_name = root_tag.split("}")[-1] if "}" in root_tag else root_tag
if root_name in declarable_roots and path_str not in declared_parts:
errors.append(
f" {path_str}: File with <{root_name}> root not declared in [Content_Types].xml"
)
except Exception:
continue # Skip unparseable files
# Check all non-XML files for Default extension declarations
for file_path in all_files:
# Skip XML files and metadata files (already checked above)
if file_path.suffix.lower() in {".xml", ".rels"}:
continue
if file_path.name == "[Content_Types].xml":
continue
if "_rels" in file_path.parts or "docProps" in file_path.parts:
continue
extension = file_path.suffix.lstrip(".").lower()
if extension and extension not in declared_extensions:
# Check if it's a known media extension that should be declared
if extension in media_extensions:
relative_path = file_path.relative_to(self.unpacked_dir)
errors.append(
f' {relative_path}: File with extension \'{extension}\' not declared in [Content_Types].xml - should add: <Default Extension="{extension}" ContentType="{media_extensions[extension]}"/>'
)
except Exception as e:
errors.append(f" Error parsing [Content_Types].xml: {e}")
if errors:
print(f"FAILED - Found {len(errors)} content type declaration errors:")
for error in errors:
print(error)
return False
else:
if self.verbose:
print(
"PASSED - All content files are properly declared in [Content_Types].xml"
)
return True
def validate_file_against_xsd(self, xml_file, verbose=False):
"""Validate a single XML file against XSD schema, comparing with original.
Args:
xml_file: Path to XML file to validate
verbose: Enable verbose output
Returns:
tuple: (is_valid, new_errors_set) where is_valid is True/False/None (skipped)
"""
# Resolve both paths to handle symlinks
xml_file = Path(xml_file).resolve()
unpacked_dir = self.unpacked_dir.resolve()
# Validate current file
is_valid, current_errors = self._validate_single_file_xsd(
xml_file, unpacked_dir
)
if is_valid is None:
return None, set() # Skipped
elif is_valid:
return True, set() # Valid, no errors
# Get errors from original file for this specific file
original_errors = self._get_original_file_errors(xml_file)
# Compare with original (both are guaranteed to be sets here)
assert current_errors is not None
new_errors = current_errors - original_errors
if new_errors:
if verbose:
relative_path = xml_file.relative_to(unpacked_dir)
print(f"FAILED - {relative_path}: {len(new_errors)} new error(s)")
for error in list(new_errors)[:3]:
truncated = error[:250] + "..." if len(error) > 250 else error
print(f" - {truncated}")
return False, new_errors
else:
# All errors existed in original
if verbose:
print(
f"PASSED - No new errors (original had {len(current_errors)} errors)"
)
return True, set()
def validate_against_xsd(self):
"""Validate XML files against XSD schemas, showing only new errors compared to original."""
new_errors = []
original_error_count = 0
valid_count = 0
skipped_count = 0
for xml_file in self.xml_files:
relative_path = str(xml_file.relative_to(self.unpacked_dir))
is_valid, new_file_errors = self.validate_file_against_xsd(
xml_file, verbose=False
)
if is_valid is None:
skipped_count += 1
continue
elif is_valid and not new_file_errors:
valid_count += 1
continue
elif is_valid:
# Had errors but all existed in original
original_error_count += 1
valid_count += 1
continue
# Has new errors
new_errors.append(f" {relative_path}: {len(new_file_errors)} new error(s)")
for error in list(new_file_errors)[:3]: # Show first 3 errors
new_errors.append(
f" - {error[:250]}..." if len(error) > 250 else f" - {error}"
)
# Print summary
if self.verbose:
print(f"Validated {len(self.xml_files)} files:")
print(f" - Valid: {valid_count}")
print(f" - Skipped (no schema): {skipped_count}")
if original_error_count:
print(f" - With original errors (ignored): {original_error_count}")
print(
f" - With NEW errors: {len(new_errors) > 0 and len([e for e in new_errors if not e.startswith(' ')]) or 0}"
)
if new_errors:
print("\nFAILED - Found NEW validation errors:")
for error in new_errors:
print(error)
return False
else:
if self.verbose:
print("\nPASSED - No new XSD validation errors introduced")
return True
def _get_schema_path(self, xml_file):
"""Determine the appropriate schema path for an XML file."""
# Check exact filename match
if xml_file.name in self.SCHEMA_MAPPINGS:
return self.schemas_dir / self.SCHEMA_MAPPINGS[xml_file.name]
# Check .rels files
if xml_file.suffix == ".rels":
return self.schemas_dir / self.SCHEMA_MAPPINGS[".rels"]
# Check chart files
if "charts/" in str(xml_file) and xml_file.name.startswith("chart"):
return self.schemas_dir / self.SCHEMA_MAPPINGS["chart"]
# Check theme files
if "theme/" in str(xml_file) and xml_file.name.startswith("theme"):
return self.schemas_dir / self.SCHEMA_MAPPINGS["theme"]
# Check if file is in a main content folder and use appropriate schema
if xml_file.parent.name in self.MAIN_CONTENT_FOLDERS:
return self.schemas_dir / self.SCHEMA_MAPPINGS[xml_file.parent.name]
return None
def _clean_ignorable_namespaces(self, xml_doc):
"""Remove attributes and elements not in allowed namespaces."""
# Create a clean copy
xml_string = lxml.etree.tostring(xml_doc, encoding="unicode")
xml_copy = lxml.etree.fromstring(xml_string)
# Remove attributes not in allowed namespaces
for elem in xml_copy.iter():
attrs_to_remove = []
for attr in elem.attrib:
# Check if attribute is from a namespace other than allowed ones
if "{" in attr:
ns = attr.split("}")[0][1:]
if ns not in self.OOXML_NAMESPACES:
attrs_to_remove.append(attr)
# Remove collected attributes
for attr in attrs_to_remove:
del elem.attrib[attr]
# Remove elements not in allowed namespaces
self._remove_ignorable_elements(xml_copy)
return lxml.etree.ElementTree(xml_copy)
def _remove_ignorable_elements(self, root):
"""Recursively remove all elements not in allowed namespaces."""
elements_to_remove = []
# Find elements to remove
for elem in list(root):
# Skip non-element nodes (comments, processing instructions, etc.)
if not hasattr(elem, "tag") or callable(elem.tag):
continue
tag_str = str(elem.tag)
if tag_str.startswith("{"):
ns = tag_str.split("}")[0][1:]
if ns not in self.OOXML_NAMESPACES:
elements_to_remove.append(elem)
continue
# Recursively clean child elements
self._remove_ignorable_elements(elem)
# Remove collected elements
for elem in elements_to_remove:
root.remove(elem)
def _preprocess_for_mc_ignorable(self, xml_doc):
"""Preprocess XML to handle mc:Ignorable attribute properly."""
# Remove mc:Ignorable attributes before validation
root = xml_doc.getroot()
# Remove mc:Ignorable attribute from root
if f"{{{self.MC_NAMESPACE}}}Ignorable" in root.attrib:
del root.attrib[f"{{{self.MC_NAMESPACE}}}Ignorable"]
return xml_doc
def _validate_single_file_xsd(self, xml_file, base_path):
"""Validate a single XML file against XSD schema. Returns (is_valid, errors_set)."""
schema_path = self._get_schema_path(xml_file)
if not schema_path:
return None, None # Skip file
try:
# Load schema
with open(schema_path, "rb") as xsd_file:
parser = lxml.etree.XMLParser()
xsd_doc = lxml.etree.parse(
xsd_file, parser=parser, base_url=str(schema_path)
)
schema = lxml.etree.XMLSchema(xsd_doc)
# Load and preprocess XML
with open(xml_file, "r") as f:
xml_doc = lxml.etree.parse(f)
xml_doc, _ = self._remove_template_tags_from_text_nodes(xml_doc)
xml_doc = self._preprocess_for_mc_ignorable(xml_doc)
# Clean ignorable namespaces if needed
relative_path = xml_file.relative_to(base_path)
if (
relative_path.parts
and relative_path.parts[0] in self.MAIN_CONTENT_FOLDERS
):
xml_doc = self._clean_ignorable_namespaces(xml_doc)
# Validate
if schema.validate(xml_doc):
return True, set()
else:
errors = set()
for error in schema.error_log:
# Store normalized error message (without line numbers for comparison)
errors.add(error.message)
return False, errors
except Exception as e:
return False, {str(e)}
def _get_original_file_errors(self, xml_file):
"""Get XSD validation errors from a single file in the original document.
Args:
xml_file: Path to the XML file in unpacked_dir to check
Returns:
set: Set of error messages from the original file
"""
import tempfile
import zipfile
# Resolve both paths to handle symlinks (e.g., /var vs /private/var on macOS)
xml_file = Path(xml_file).resolve()
unpacked_dir = self.unpacked_dir.resolve()
relative_path = xml_file.relative_to(unpacked_dir)
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
# Extract original file
with zipfile.ZipFile(self.original_file, "r") as zip_ref:
zip_ref.extractall(temp_path)
# Find corresponding file in original
original_xml_file = temp_path / relative_path
if not original_xml_file.exists():
# File didn't exist in original, so no original errors
return set()
# Validate the specific file in original
is_valid, errors = self._validate_single_file_xsd(
original_xml_file, temp_path
)
return errors if errors else set()
def _remove_template_tags_from_text_nodes(self, xml_doc):
"""Remove template tags from XML text nodes and collect warnings.
Template tags follow the pattern {{ ... }} and are used as placeholders
for content replacement. They should be removed from text content before
XSD validation while preserving XML structure.
Returns:
tuple: (cleaned_xml_doc, warnings_list)
"""
warnings = []
template_pattern = re.compile(r"\{\{[^}]*\}\}")
# Create a copy of the document to avoid modifying the original
xml_string = lxml.etree.tostring(xml_doc, encoding="unicode")
xml_copy = lxml.etree.fromstring(xml_string)
def process_text_content(text, content_type):
if not text:
return text
matches = list(template_pattern.finditer(text))
if matches:
for match in matches:
warnings.append(
f"Found template tag in {content_type}: {match.group()}"
)
return template_pattern.sub("", text)
return text
# Process all text nodes in the document
for elem in xml_copy.iter():
# Skip processing if this is a w:t element
if not hasattr(elem, "tag") or callable(elem.tag):
continue
tag_str = str(elem.tag)
if tag_str.endswith("}t") or tag_str == "t":
continue
elem.text = process_text_content(elem.text, "text content")
elem.tail = process_text_content(elem.tail, "tail content")
return lxml.etree.ElementTree(xml_copy), warnings
if __name__ == "__main__":
raise RuntimeError("This module should not be run directly.")
@@ -0,0 +1,274 @@
"""
Validator for Word document XML files against XSD schemas.
"""
import re
import tempfile
import zipfile
import lxml.etree
from .base import BaseSchemaValidator
class DOCXSchemaValidator(BaseSchemaValidator):
"""Validator for Word document XML files against XSD schemas."""
# Word-specific namespace
WORD_2006_NAMESPACE = "http://schemas.openxmlformats.org/wordprocessingml/2006/main"
# Word-specific element to relationship type mappings
# Start with empty mapping - add specific cases as we discover them
ELEMENT_RELATIONSHIP_TYPES = {}
def validate(self):
"""Run all validation checks and return True if all pass."""
# Test 0: XML well-formedness
if not self.validate_xml():
return False
# Test 1: Namespace declarations
all_valid = True
if not self.validate_namespaces():
all_valid = False
# Test 2: Unique IDs
if not self.validate_unique_ids():
all_valid = False
# Test 3: Relationship and file reference validation
if not self.validate_file_references():
all_valid = False
# Test 4: Content type declarations
if not self.validate_content_types():
all_valid = False
# Test 5: XSD schema validation
if not self.validate_against_xsd():
all_valid = False
# Test 6: Whitespace preservation
if not self.validate_whitespace_preservation():
all_valid = False
# Test 7: Deletion validation
if not self.validate_deletions():
all_valid = False
# Test 8: Insertion validation
if not self.validate_insertions():
all_valid = False
# Test 9: Relationship ID reference validation
if not self.validate_all_relationship_ids():
all_valid = False
# Count and compare paragraphs
self.compare_paragraph_counts()
return all_valid
def validate_whitespace_preservation(self):
"""
Validate that w:t elements with whitespace have xml:space='preserve'.
"""
errors = []
for xml_file in self.xml_files:
# Only check document.xml files
if xml_file.name != "document.xml":
continue
try:
root = lxml.etree.parse(str(xml_file)).getroot()
# Find all w:t elements
for elem in root.iter(f"{{{self.WORD_2006_NAMESPACE}}}t"):
if elem.text:
text = elem.text
# Check if text starts or ends with whitespace
if re.match(r"^\s.*", text) or re.match(r".*\s$", text):
# Check if xml:space="preserve" attribute exists
xml_space_attr = f"{{{self.XML_NAMESPACE}}}space"
if (
xml_space_attr not in elem.attrib
or elem.attrib[xml_space_attr] != "preserve"
):
# Show a preview of the text
text_preview = (
repr(text)[:50] + "..."
if len(repr(text)) > 50
else repr(text)
)
errors.append(
f" {xml_file.relative_to(self.unpacked_dir)}: "
f"Line {elem.sourceline}: w:t element with whitespace missing xml:space='preserve': {text_preview}"
)
except (lxml.etree.XMLSyntaxError, Exception) as e:
errors.append(
f" {xml_file.relative_to(self.unpacked_dir)}: Error: {e}"
)
if errors:
print(f"FAILED - Found {len(errors)} whitespace preservation violations:")
for error in errors:
print(error)
return False
else:
if self.verbose:
print("PASSED - All whitespace is properly preserved")
return True
def validate_deletions(self):
"""
Validate that w:t elements are not within w:del elements.
For some reason, XSD validation does not catch this, so we do it manually.
"""
errors = []
for xml_file in self.xml_files:
# Only check document.xml files
if xml_file.name != "document.xml":
continue
try:
root = lxml.etree.parse(str(xml_file)).getroot()
# Find all w:t elements that are descendants of w:del elements
namespaces = {"w": self.WORD_2006_NAMESPACE}
xpath_expression = ".//w:del//w:t"
problematic_t_elements = root.xpath(
xpath_expression, namespaces=namespaces
)
for t_elem in problematic_t_elements:
if t_elem.text:
# Show a preview of the text
text_preview = (
repr(t_elem.text)[:50] + "..."
if len(repr(t_elem.text)) > 50
else repr(t_elem.text)
)
errors.append(
f" {xml_file.relative_to(self.unpacked_dir)}: "
f"Line {t_elem.sourceline}: <w:t> found within <w:del>: {text_preview}"
)
except (lxml.etree.XMLSyntaxError, Exception) as e:
errors.append(
f" {xml_file.relative_to(self.unpacked_dir)}: Error: {e}"
)
if errors:
print(f"FAILED - Found {len(errors)} deletion validation violations:")
for error in errors:
print(error)
return False
else:
if self.verbose:
print("PASSED - No w:t elements found within w:del elements")
return True
def count_paragraphs_in_unpacked(self):
"""Count the number of paragraphs in the unpacked document."""
count = 0
for xml_file in self.xml_files:
# Only check document.xml files
if xml_file.name != "document.xml":
continue
try:
root = lxml.etree.parse(str(xml_file)).getroot()
# Count all w:p elements
paragraphs = root.findall(f".//{{{self.WORD_2006_NAMESPACE}}}p")
count = len(paragraphs)
except Exception as e:
print(f"Error counting paragraphs in unpacked document: {e}")
return count
def count_paragraphs_in_original(self):
"""Count the number of paragraphs in the original docx file."""
count = 0
try:
# Create temporary directory to unpack original
with tempfile.TemporaryDirectory() as temp_dir:
# Unpack original docx
with zipfile.ZipFile(self.original_file, "r") as zip_ref:
zip_ref.extractall(temp_dir)
# Parse document.xml
doc_xml_path = temp_dir + "/word/document.xml"
root = lxml.etree.parse(doc_xml_path).getroot()
# Count all w:p elements
paragraphs = root.findall(f".//{{{self.WORD_2006_NAMESPACE}}}p")
count = len(paragraphs)
except Exception as e:
print(f"Error counting paragraphs in original document: {e}")
return count
def validate_insertions(self):
"""
Validate that w:delText elements are not within w:ins elements.
w:delText is only allowed in w:ins if nested within a w:del.
"""
errors = []
for xml_file in self.xml_files:
if xml_file.name != "document.xml":
continue
try:
root = lxml.etree.parse(str(xml_file)).getroot()
namespaces = {"w": self.WORD_2006_NAMESPACE}
# Find w:delText in w:ins that are NOT within w:del
invalid_elements = root.xpath(
".//w:ins//w:delText[not(ancestor::w:del)]",
namespaces=namespaces
)
for elem in invalid_elements:
text_preview = (
repr(elem.text or "")[:50] + "..."
if len(repr(elem.text or "")) > 50
else repr(elem.text or "")
)
errors.append(
f" {xml_file.relative_to(self.unpacked_dir)}: "
f"Line {elem.sourceline}: <w:delText> within <w:ins>: {text_preview}"
)
except (lxml.etree.XMLSyntaxError, Exception) as e:
errors.append(
f" {xml_file.relative_to(self.unpacked_dir)}: Error: {e}"
)
if errors:
print(f"FAILED - Found {len(errors)} insertion validation violations:")
for error in errors:
print(error)
return False
else:
if self.verbose:
print("PASSED - No w:delText elements within w:ins elements")
return True
def compare_paragraph_counts(self):
"""Compare paragraph counts between original and new document."""
original_count = self.count_paragraphs_in_original()
new_count = self.count_paragraphs_in_unpacked()
diff = new_count - original_count
diff_str = f"+{diff}" if diff > 0 else str(diff)
print(f"\nParagraphs: {original_count}{new_count} ({diff_str})")
if __name__ == "__main__":
raise RuntimeError("This module should not be run directly.")
@@ -0,0 +1,315 @@
"""
Validator for PowerPoint presentation XML files against XSD schemas.
"""
import re
from .base import BaseSchemaValidator
class PPTXSchemaValidator(BaseSchemaValidator):
"""Validator for PowerPoint presentation XML files against XSD schemas."""
# PowerPoint presentation namespace
PRESENTATIONML_NAMESPACE = (
"http://schemas.openxmlformats.org/presentationml/2006/main"
)
# PowerPoint-specific element to relationship type mappings
ELEMENT_RELATIONSHIP_TYPES = {
"sldid": "slide",
"sldmasterid": "slidemaster",
"notesmasterid": "notesmaster",
"sldlayoutid": "slidelayout",
"themeid": "theme",
"tablestyleid": "tablestyles",
}
def validate(self):
"""Run all validation checks and return True if all pass."""
# Test 0: XML well-formedness
if not self.validate_xml():
return False
# Test 1: Namespace declarations
all_valid = True
if not self.validate_namespaces():
all_valid = False
# Test 2: Unique IDs
if not self.validate_unique_ids():
all_valid = False
# Test 3: UUID ID validation
if not self.validate_uuid_ids():
all_valid = False
# Test 4: Relationship and file reference validation
if not self.validate_file_references():
all_valid = False
# Test 5: Slide layout ID validation
if not self.validate_slide_layout_ids():
all_valid = False
# Test 6: Content type declarations
if not self.validate_content_types():
all_valid = False
# Test 7: XSD schema validation
if not self.validate_against_xsd():
all_valid = False
# Test 8: Notes slide reference validation
if not self.validate_notes_slide_references():
all_valid = False
# Test 9: Relationship ID reference validation
if not self.validate_all_relationship_ids():
all_valid = False
# Test 10: Duplicate slide layout references validation
if not self.validate_no_duplicate_slide_layouts():
all_valid = False
return all_valid
def validate_uuid_ids(self):
"""Validate that ID attributes that look like UUIDs contain only hex values."""
import lxml.etree
errors = []
# UUID pattern: 8-4-4-4-12 hex digits with optional braces/hyphens
uuid_pattern = re.compile(
r"^[\{\(]?[0-9A-Fa-f]{8}-?[0-9A-Fa-f]{4}-?[0-9A-Fa-f]{4}-?[0-9A-Fa-f]{4}-?[0-9A-Fa-f]{12}[\}\)]?$"
)
for xml_file in self.xml_files:
try:
root = lxml.etree.parse(str(xml_file)).getroot()
# Check all elements for ID attributes
for elem in root.iter():
for attr, value in elem.attrib.items():
# Check if this is an ID attribute
attr_name = attr.split("}")[-1].lower()
if attr_name == "id" or attr_name.endswith("id"):
# Check if value looks like a UUID (has the right length and pattern structure)
if self._looks_like_uuid(value):
# Validate that it contains only hex characters in the right positions
if not uuid_pattern.match(value):
errors.append(
f" {xml_file.relative_to(self.unpacked_dir)}: "
f"Line {elem.sourceline}: ID '{value}' appears to be a UUID but contains invalid hex characters"
)
except (lxml.etree.XMLSyntaxError, Exception) as e:
errors.append(
f" {xml_file.relative_to(self.unpacked_dir)}: Error: {e}"
)
if errors:
print(f"FAILED - Found {len(errors)} UUID ID validation errors:")
for error in errors:
print(error)
return False
else:
if self.verbose:
print("PASSED - All UUID-like IDs contain valid hex values")
return True
def _looks_like_uuid(self, value):
"""Check if a value has the general structure of a UUID."""
# Remove common UUID delimiters
clean_value = value.strip("{}()").replace("-", "")
# Check if it's 32 hex-like characters (could include invalid hex chars)
return len(clean_value) == 32 and all(c.isalnum() for c in clean_value)
def validate_slide_layout_ids(self):
"""Validate that sldLayoutId elements in slide masters reference valid slide layouts."""
import lxml.etree
errors = []
# Find all slide master files
slide_masters = list(self.unpacked_dir.glob("ppt/slideMasters/*.xml"))
if not slide_masters:
if self.verbose:
print("PASSED - No slide masters found")
return True
for slide_master in slide_masters:
try:
# Parse the slide master file
root = lxml.etree.parse(str(slide_master)).getroot()
# Find the corresponding _rels file for this slide master
rels_file = slide_master.parent / "_rels" / f"{slide_master.name}.rels"
if not rels_file.exists():
errors.append(
f" {slide_master.relative_to(self.unpacked_dir)}: "
f"Missing relationships file: {rels_file.relative_to(self.unpacked_dir)}"
)
continue
# Parse the relationships file
rels_root = lxml.etree.parse(str(rels_file)).getroot()
# Build a set of valid relationship IDs that point to slide layouts
valid_layout_rids = set()
for rel in rels_root.findall(
f".//{{{self.PACKAGE_RELATIONSHIPS_NAMESPACE}}}Relationship"
):
rel_type = rel.get("Type", "")
if "slideLayout" in rel_type:
valid_layout_rids.add(rel.get("Id"))
# Find all sldLayoutId elements in the slide master
for sld_layout_id in root.findall(
f".//{{{self.PRESENTATIONML_NAMESPACE}}}sldLayoutId"
):
r_id = sld_layout_id.get(
f"{{{self.OFFICE_RELATIONSHIPS_NAMESPACE}}}id"
)
layout_id = sld_layout_id.get("id")
if r_id and r_id not in valid_layout_rids:
errors.append(
f" {slide_master.relative_to(self.unpacked_dir)}: "
f"Line {sld_layout_id.sourceline}: sldLayoutId with id='{layout_id}' "
f"references r:id='{r_id}' which is not found in slide layout relationships"
)
except (lxml.etree.XMLSyntaxError, Exception) as e:
errors.append(
f" {slide_master.relative_to(self.unpacked_dir)}: Error: {e}"
)
if errors:
print(f"FAILED - Found {len(errors)} slide layout ID validation errors:")
for error in errors:
print(error)
print(
"Remove invalid references or add missing slide layouts to the relationships file."
)
return False
else:
if self.verbose:
print("PASSED - All slide layout IDs reference valid slide layouts")
return True
def validate_no_duplicate_slide_layouts(self):
"""Validate that each slide has exactly one slideLayout reference."""
import lxml.etree
errors = []
slide_rels_files = list(self.unpacked_dir.glob("ppt/slides/_rels/*.xml.rels"))
for rels_file in slide_rels_files:
try:
root = lxml.etree.parse(str(rels_file)).getroot()
# Find all slideLayout relationships
layout_rels = [
rel
for rel in root.findall(
f".//{{{self.PACKAGE_RELATIONSHIPS_NAMESPACE}}}Relationship"
)
if "slideLayout" in rel.get("Type", "")
]
if len(layout_rels) > 1:
errors.append(
f" {rels_file.relative_to(self.unpacked_dir)}: has {len(layout_rels)} slideLayout references"
)
except Exception as e:
errors.append(
f" {rels_file.relative_to(self.unpacked_dir)}: Error: {e}"
)
if errors:
print("FAILED - Found slides with duplicate slideLayout references:")
for error in errors:
print(error)
return False
else:
if self.verbose:
print("PASSED - All slides have exactly one slideLayout reference")
return True
def validate_notes_slide_references(self):
"""Validate that each notesSlide file is referenced by only one slide."""
import lxml.etree
errors = []
notes_slide_references = {} # Track which slides reference each notesSlide
# Find all slide relationship files
slide_rels_files = list(self.unpacked_dir.glob("ppt/slides/_rels/*.xml.rels"))
if not slide_rels_files:
if self.verbose:
print("PASSED - No slide relationship files found")
return True
for rels_file in slide_rels_files:
try:
# Parse the relationships file
root = lxml.etree.parse(str(rels_file)).getroot()
# Find all notesSlide relationships
for rel in root.findall(
f".//{{{self.PACKAGE_RELATIONSHIPS_NAMESPACE}}}Relationship"
):
rel_type = rel.get("Type", "")
if "notesSlide" in rel_type:
target = rel.get("Target", "")
if target:
# Normalize the target path to handle relative paths
normalized_target = target.replace("../", "")
# Track which slide references this notesSlide
slide_name = rels_file.stem.replace(
".xml", ""
) # e.g., "slide1"
if normalized_target not in notes_slide_references:
notes_slide_references[normalized_target] = []
notes_slide_references[normalized_target].append(
(slide_name, rels_file)
)
except (lxml.etree.XMLSyntaxError, Exception) as e:
errors.append(
f" {rels_file.relative_to(self.unpacked_dir)}: Error: {e}"
)
# Check for duplicate references
for target, references in notes_slide_references.items():
if len(references) > 1:
slide_names = [ref[0] for ref in references]
errors.append(
f" Notes slide '{target}' is referenced by multiple slides: {', '.join(slide_names)}"
)
for slide_name, rels_file in references:
errors.append(f" - {rels_file.relative_to(self.unpacked_dir)}")
if errors:
print(
f"FAILED - Found {len([e for e in errors if not e.startswith(' ')])} notes slide reference validation errors:"
)
for error in errors:
print(error)
print("Each slide may optionally have its own slide file.")
return False
else:
if self.verbose:
print("PASSED - All notes slide references are unique")
return True
if __name__ == "__main__":
raise RuntimeError("This module should not be run directly.")
@@ -0,0 +1,279 @@
"""
Validator for tracked changes in Word documents.
"""
import subprocess
import tempfile
import zipfile
from pathlib import Path
class RedliningValidator:
"""Validator for tracked changes in Word documents."""
def __init__(self, unpacked_dir, original_docx, verbose=False):
self.unpacked_dir = Path(unpacked_dir)
self.original_docx = Path(original_docx)
self.verbose = verbose
self.namespaces = {
"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main"
}
def validate(self):
"""Main validation method that returns True if valid, False otherwise."""
# Verify unpacked directory exists and has correct structure
modified_file = self.unpacked_dir / "word" / "document.xml"
if not modified_file.exists():
print(f"FAILED - Modified document.xml not found at {modified_file}")
return False
# First, check if there are any tracked changes by Claude to validate
try:
import xml.etree.ElementTree as ET
tree = ET.parse(modified_file)
root = tree.getroot()
# Check for w:del or w:ins tags authored by Claude
del_elements = root.findall(".//w:del", self.namespaces)
ins_elements = root.findall(".//w:ins", self.namespaces)
# Filter to only include changes by Claude
claude_del_elements = [
elem
for elem in del_elements
if elem.get(f"{{{self.namespaces['w']}}}author") == "Claude"
]
claude_ins_elements = [
elem
for elem in ins_elements
if elem.get(f"{{{self.namespaces['w']}}}author") == "Claude"
]
# Redlining validation is only needed if tracked changes by Claude have been used.
if not claude_del_elements and not claude_ins_elements:
if self.verbose:
print("PASSED - No tracked changes by Claude found.")
return True
except Exception:
# If we can't parse the XML, continue with full validation
pass
# Create temporary directory for unpacking original docx
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
# Unpack original docx
try:
with zipfile.ZipFile(self.original_docx, "r") as zip_ref:
zip_ref.extractall(temp_path)
except Exception as e:
print(f"FAILED - Error unpacking original docx: {e}")
return False
original_file = temp_path / "word" / "document.xml"
if not original_file.exists():
print(
f"FAILED - Original document.xml not found in {self.original_docx}"
)
return False
# Parse both XML files using xml.etree.ElementTree for redlining validation
try:
import xml.etree.ElementTree as ET
modified_tree = ET.parse(modified_file)
modified_root = modified_tree.getroot()
original_tree = ET.parse(original_file)
original_root = original_tree.getroot()
except ET.ParseError as e:
print(f"FAILED - Error parsing XML files: {e}")
return False
# Remove Claude's tracked changes from both documents
self._remove_claude_tracked_changes(original_root)
self._remove_claude_tracked_changes(modified_root)
# Extract and compare text content
modified_text = self._extract_text_content(modified_root)
original_text = self._extract_text_content(original_root)
if modified_text != original_text:
# Show detailed character-level differences for each paragraph
error_message = self._generate_detailed_diff(
original_text, modified_text
)
print(error_message)
return False
if self.verbose:
print("PASSED - All changes by Claude are properly tracked")
return True
def _generate_detailed_diff(self, original_text, modified_text):
"""Generate detailed word-level differences using git word diff."""
error_parts = [
"FAILED - Document text doesn't match after removing Claude's tracked changes",
"",
"Likely causes:",
" 1. Modified text inside another author's <w:ins> or <w:del> tags",
" 2. Made edits without proper tracked changes",
" 3. Didn't nest <w:del> inside <w:ins> when deleting another's insertion",
"",
"For pre-redlined documents, use correct patterns:",
" - To reject another's INSERTION: Nest <w:del> inside their <w:ins>",
" - To restore another's DELETION: Add new <w:ins> AFTER their <w:del>",
"",
]
# Show git word diff
git_diff = self._get_git_word_diff(original_text, modified_text)
if git_diff:
error_parts.extend(["Differences:", "============", git_diff])
else:
error_parts.append("Unable to generate word diff (git not available)")
return "\n".join(error_parts)
def _get_git_word_diff(self, original_text, modified_text):
"""Generate word diff using git with character-level precision."""
try:
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
# Create two files
original_file = temp_path / "original.txt"
modified_file = temp_path / "modified.txt"
original_file.write_text(original_text, encoding="utf-8")
modified_file.write_text(modified_text, encoding="utf-8")
# Try character-level diff first for precise differences
result = subprocess.run(
[
"git",
"diff",
"--word-diff=plain",
"--word-diff-regex=.", # Character-by-character diff
"-U0", # Zero lines of context - show only changed lines
"--no-index",
str(original_file),
str(modified_file),
],
capture_output=True,
text=True,
)
if result.stdout.strip():
# Clean up the output - remove git diff header lines
lines = result.stdout.split("\n")
# Skip the header lines (diff --git, index, +++, ---, @@)
content_lines = []
in_content = False
for line in lines:
if line.startswith("@@"):
in_content = True
continue
if in_content and line.strip():
content_lines.append(line)
if content_lines:
return "\n".join(content_lines)
# Fallback to word-level diff if character-level is too verbose
result = subprocess.run(
[
"git",
"diff",
"--word-diff=plain",
"-U0", # Zero lines of context
"--no-index",
str(original_file),
str(modified_file),
],
capture_output=True,
text=True,
)
if result.stdout.strip():
lines = result.stdout.split("\n")
content_lines = []
in_content = False
for line in lines:
if line.startswith("@@"):
in_content = True
continue
if in_content and line.strip():
content_lines.append(line)
return "\n".join(content_lines)
except (subprocess.CalledProcessError, FileNotFoundError, Exception):
# Git not available or other error, return None to use fallback
pass
return None
def _remove_claude_tracked_changes(self, root):
"""Remove tracked changes authored by Claude from the XML root."""
ins_tag = f"{{{self.namespaces['w']}}}ins"
del_tag = f"{{{self.namespaces['w']}}}del"
author_attr = f"{{{self.namespaces['w']}}}author"
# Remove w:ins elements
for parent in root.iter():
to_remove = []
for child in parent:
if child.tag == ins_tag and child.get(author_attr) == "Claude":
to_remove.append(child)
for elem in to_remove:
parent.remove(elem)
# Unwrap content in w:del elements where author is "Claude"
deltext_tag = f"{{{self.namespaces['w']}}}delText"
t_tag = f"{{{self.namespaces['w']}}}t"
for parent in root.iter():
to_process = []
for child in parent:
if child.tag == del_tag and child.get(author_attr) == "Claude":
to_process.append((child, list(parent).index(child)))
# Process in reverse order to maintain indices
for del_elem, del_index in reversed(to_process):
# Convert w:delText to w:t before moving
for elem in del_elem.iter():
if elem.tag == deltext_tag:
elem.tag = t_tag
# Move all children of w:del to its parent before removing w:del
for child in reversed(list(del_elem)):
parent.insert(del_index, child)
parent.remove(del_elem)
def _extract_text_content(self, root):
"""Extract text content from Word XML, preserving paragraph structure.
Empty paragraphs are skipped to avoid false positives when tracked
insertions add only structural elements without text content.
"""
p_tag = f"{{{self.namespaces['w']}}}p"
t_tag = f"{{{self.namespaces['w']}}}t"
paragraphs = []
for p_elem in root.findall(f".//{p_tag}"):
# Get all text elements within this paragraph
text_parts = []
for t_elem in p_elem.findall(f".//{t_tag}"):
if t_elem.text:
text_parts.append(t_elem.text)
paragraph_text = "".join(text_parts)
# Skip empty paragraphs - they don't affect content validation
if paragraph_text:
paragraphs.append(paragraph_text)
return "\n".join(paragraphs)
if __name__ == "__main__":
raise RuntimeError("This module should not be run directly.")
@@ -0,0 +1,70 @@
from dataclasses import dataclass
import json
import sys
# Script to check that the `fields.json` file that Claude creates when analyzing PDFs
# does not have overlapping bounding boxes. See forms.md.
@dataclass
class RectAndField:
rect: list[float]
rect_type: str
field: dict
# Returns a list of messages that are printed to stdout for Claude to read.
def get_bounding_box_messages(fields_json_stream) -> list[str]:
messages = []
fields = json.load(fields_json_stream)
messages.append(f"Read {len(fields['form_fields'])} fields")
def rects_intersect(r1, r2):
disjoint_horizontal = r1[0] >= r2[2] or r1[2] <= r2[0]
disjoint_vertical = r1[1] >= r2[3] or r1[3] <= r2[1]
return not (disjoint_horizontal or disjoint_vertical)
rects_and_fields = []
for f in fields["form_fields"]:
rects_and_fields.append(RectAndField(f["label_bounding_box"], "label", f))
rects_and_fields.append(RectAndField(f["entry_bounding_box"], "entry", f))
has_error = False
for i, ri in enumerate(rects_and_fields):
# This is O(N^2); we can optimize if it becomes a problem.
for j in range(i + 1, len(rects_and_fields)):
rj = rects_and_fields[j]
if ri.field["page_number"] == rj.field["page_number"] and rects_intersect(ri.rect, rj.rect):
has_error = True
if ri.field is rj.field:
messages.append(f"FAILURE: intersection between label and entry bounding boxes for `{ri.field['description']}` ({ri.rect}, {rj.rect})")
else:
messages.append(f"FAILURE: intersection between {ri.rect_type} bounding box for `{ri.field['description']}` ({ri.rect}) and {rj.rect_type} bounding box for `{rj.field['description']}` ({rj.rect})")
if len(messages) >= 20:
messages.append("Aborting further checks; fix bounding boxes and try again")
return messages
if ri.rect_type == "entry":
if "entry_text" in ri.field:
font_size = ri.field["entry_text"].get("font_size", 14)
entry_height = ri.rect[3] - ri.rect[1]
if entry_height < font_size:
has_error = True
messages.append(f"FAILURE: entry bounding box height ({entry_height}) for `{ri.field['description']}` is too short for the text content (font size: {font_size}). Increase the box height or decrease the font size.")
if len(messages) >= 20:
messages.append("Aborting further checks; fix bounding boxes and try again")
return messages
if not has_error:
messages.append("SUCCESS: All bounding boxes are valid")
return messages
if __name__ == "__main__":
if len(sys.argv) != 2:
print("Usage: check_bounding_boxes.py [fields.json]")
sys.exit(1)
# Input file should be in the `fields.json` format described in forms.md.
with open(sys.argv[1]) as f:
messages = get_bounding_box_messages(f)
for msg in messages:
print(msg)
@@ -0,0 +1,226 @@
import unittest
import json
import io
from check_bounding_boxes import get_bounding_box_messages
# Currently this is not run automatically in CI; it's just for documentation and manual checking.
class TestGetBoundingBoxMessages(unittest.TestCase):
def create_json_stream(self, data):
"""Helper to create a JSON stream from data"""
return io.StringIO(json.dumps(data))
def test_no_intersections(self):
"""Test case with no bounding box intersections"""
data = {
"form_fields": [
{
"description": "Name",
"page_number": 1,
"label_bounding_box": [10, 10, 50, 30],
"entry_bounding_box": [60, 10, 150, 30]
},
{
"description": "Email",
"page_number": 1,
"label_bounding_box": [10, 40, 50, 60],
"entry_bounding_box": [60, 40, 150, 60]
}
]
}
stream = self.create_json_stream(data)
messages = get_bounding_box_messages(stream)
self.assertTrue(any("SUCCESS" in msg for msg in messages))
self.assertFalse(any("FAILURE" in msg for msg in messages))
def test_label_entry_intersection_same_field(self):
"""Test intersection between label and entry of the same field"""
data = {
"form_fields": [
{
"description": "Name",
"page_number": 1,
"label_bounding_box": [10, 10, 60, 30],
"entry_bounding_box": [50, 10, 150, 30] # Overlaps with label
}
]
}
stream = self.create_json_stream(data)
messages = get_bounding_box_messages(stream)
self.assertTrue(any("FAILURE" in msg and "intersection" in msg for msg in messages))
self.assertFalse(any("SUCCESS" in msg for msg in messages))
def test_intersection_between_different_fields(self):
"""Test intersection between bounding boxes of different fields"""
data = {
"form_fields": [
{
"description": "Name",
"page_number": 1,
"label_bounding_box": [10, 10, 50, 30],
"entry_bounding_box": [60, 10, 150, 30]
},
{
"description": "Email",
"page_number": 1,
"label_bounding_box": [40, 20, 80, 40], # Overlaps with Name's boxes
"entry_bounding_box": [160, 10, 250, 30]
}
]
}
stream = self.create_json_stream(data)
messages = get_bounding_box_messages(stream)
self.assertTrue(any("FAILURE" in msg and "intersection" in msg for msg in messages))
self.assertFalse(any("SUCCESS" in msg for msg in messages))
def test_different_pages_no_intersection(self):
"""Test that boxes on different pages don't count as intersecting"""
data = {
"form_fields": [
{
"description": "Name",
"page_number": 1,
"label_bounding_box": [10, 10, 50, 30],
"entry_bounding_box": [60, 10, 150, 30]
},
{
"description": "Email",
"page_number": 2,
"label_bounding_box": [10, 10, 50, 30], # Same coordinates but different page
"entry_bounding_box": [60, 10, 150, 30]
}
]
}
stream = self.create_json_stream(data)
messages = get_bounding_box_messages(stream)
self.assertTrue(any("SUCCESS" in msg for msg in messages))
self.assertFalse(any("FAILURE" in msg for msg in messages))
def test_entry_height_too_small(self):
"""Test that entry box height is checked against font size"""
data = {
"form_fields": [
{
"description": "Name",
"page_number": 1,
"label_bounding_box": [10, 10, 50, 30],
"entry_bounding_box": [60, 10, 150, 20], # Height is 10
"entry_text": {
"font_size": 14 # Font size larger than height
}
}
]
}
stream = self.create_json_stream(data)
messages = get_bounding_box_messages(stream)
self.assertTrue(any("FAILURE" in msg and "height" in msg for msg in messages))
self.assertFalse(any("SUCCESS" in msg for msg in messages))
def test_entry_height_adequate(self):
"""Test that adequate entry box height passes"""
data = {
"form_fields": [
{
"description": "Name",
"page_number": 1,
"label_bounding_box": [10, 10, 50, 30],
"entry_bounding_box": [60, 10, 150, 30], # Height is 20
"entry_text": {
"font_size": 14 # Font size smaller than height
}
}
]
}
stream = self.create_json_stream(data)
messages = get_bounding_box_messages(stream)
self.assertTrue(any("SUCCESS" in msg for msg in messages))
self.assertFalse(any("FAILURE" in msg for msg in messages))
def test_default_font_size(self):
"""Test that default font size is used when not specified"""
data = {
"form_fields": [
{
"description": "Name",
"page_number": 1,
"label_bounding_box": [10, 10, 50, 30],
"entry_bounding_box": [60, 10, 150, 20], # Height is 10
"entry_text": {} # No font_size specified, should use default 14
}
]
}
stream = self.create_json_stream(data)
messages = get_bounding_box_messages(stream)
self.assertTrue(any("FAILURE" in msg and "height" in msg for msg in messages))
self.assertFalse(any("SUCCESS" in msg for msg in messages))
def test_no_entry_text(self):
"""Test that missing entry_text doesn't cause height check"""
data = {
"form_fields": [
{
"description": "Name",
"page_number": 1,
"label_bounding_box": [10, 10, 50, 30],
"entry_bounding_box": [60, 10, 150, 20] # Small height but no entry_text
}
]
}
stream = self.create_json_stream(data)
messages = get_bounding_box_messages(stream)
self.assertTrue(any("SUCCESS" in msg for msg in messages))
self.assertFalse(any("FAILURE" in msg for msg in messages))
def test_multiple_errors_limit(self):
"""Test that error messages are limited to prevent excessive output"""
fields = []
# Create many overlapping fields
for i in range(25):
fields.append({
"description": f"Field{i}",
"page_number": 1,
"label_bounding_box": [10, 10, 50, 30], # All overlap
"entry_bounding_box": [20, 15, 60, 35] # All overlap
})
data = {"form_fields": fields}
stream = self.create_json_stream(data)
messages = get_bounding_box_messages(stream)
# Should abort after ~20 messages
self.assertTrue(any("Aborting" in msg for msg in messages))
# Should have some FAILURE messages but not hundreds
failure_count = sum(1 for msg in messages if "FAILURE" in msg)
self.assertGreater(failure_count, 0)
self.assertLess(len(messages), 30) # Should be limited
def test_edge_touching_boxes(self):
"""Test that boxes touching at edges don't count as intersecting"""
data = {
"form_fields": [
{
"description": "Name",
"page_number": 1,
"label_bounding_box": [10, 10, 50, 30],
"entry_bounding_box": [50, 10, 150, 30] # Touches at x=50
}
]
}
stream = self.create_json_stream(data)
messages = get_bounding_box_messages(stream)
self.assertTrue(any("SUCCESS" in msg for msg in messages))
self.assertFalse(any("FAILURE" in msg for msg in messages))
if __name__ == '__main__':
unittest.main()
@@ -0,0 +1,12 @@
import sys
from pypdf import PdfReader
# Script for Claude to run to determine whether a PDF has fillable form fields. See forms.md.
reader = PdfReader(sys.argv[1])
if (reader.get_fields()):
print("This PDF has fillable form fields")
else:
print("This PDF does not have fillable form fields; you will need to visually determine where to enter data")
@@ -0,0 +1,35 @@
import os
import sys
from pdf2image import convert_from_path
# Converts each page of a PDF to a PNG image.
def convert(pdf_path, output_dir, max_dim=1000):
images = convert_from_path(pdf_path, dpi=200)
for i, image in enumerate(images):
# Scale image if needed to keep width/height under `max_dim`
width, height = image.size
if width > max_dim or height > max_dim:
scale_factor = min(max_dim / width, max_dim / height)
new_width = int(width * scale_factor)
new_height = int(height * scale_factor)
image = image.resize((new_width, new_height))
image_path = os.path.join(output_dir, f"page_{i+1}.png")
image.save(image_path)
print(f"Saved page {i+1} as {image_path} (size: {image.size})")
print(f"Converted {len(images)} pages to PNG images")
if __name__ == "__main__":
if len(sys.argv) != 3:
print("Usage: convert_pdf_to_images.py [input pdf] [output directory]")
sys.exit(1)
pdf_path = sys.argv[1]
output_directory = sys.argv[2]
convert(pdf_path, output_directory)
@@ -0,0 +1,41 @@
import json
import sys
from PIL import Image, ImageDraw
# Creates "validation" images with rectangles for the bounding box information that
# Claude creates when determining where to add text annotations in PDFs. See forms.md.
def create_validation_image(page_number, fields_json_path, input_path, output_path):
# Input file should be in the `fields.json` format described in forms.md.
with open(fields_json_path, 'r') as f:
data = json.load(f)
img = Image.open(input_path)
draw = ImageDraw.Draw(img)
num_boxes = 0
for field in data["form_fields"]:
if field["page_number"] == page_number:
entry_box = field['entry_bounding_box']
label_box = field['label_bounding_box']
# Draw red rectangle over entry bounding box and blue rectangle over the label.
draw.rectangle(entry_box, outline='red', width=2)
draw.rectangle(label_box, outline='blue', width=2)
num_boxes += 2
img.save(output_path)
print(f"Created validation image at {output_path} with {num_boxes} bounding boxes")
if __name__ == "__main__":
if len(sys.argv) != 5:
print("Usage: create_validation_image.py [page number] [fields.json file] [input image path] [output image path]")
sys.exit(1)
page_number = int(sys.argv[1])
fields_json_path = sys.argv[2]
input_image_path = sys.argv[3]
output_image_path = sys.argv[4]
create_validation_image(page_number, fields_json_path, input_image_path, output_image_path)
@@ -0,0 +1,152 @@
import json
import sys
from pypdf import PdfReader
# Extracts data for the fillable form fields in a PDF and outputs JSON that
# Claude uses to fill the fields. See forms.md.
# This matches the format used by PdfReader `get_fields` and `update_page_form_field_values` methods.
def get_full_annotation_field_id(annotation):
components = []
while annotation:
field_name = annotation.get('/T')
if field_name:
components.append(field_name)
annotation = annotation.get('/Parent')
return ".".join(reversed(components)) if components else None
def make_field_dict(field, field_id):
field_dict = {"field_id": field_id}
ft = field.get('/FT')
if ft == "/Tx":
field_dict["type"] = "text"
elif ft == "/Btn":
field_dict["type"] = "checkbox" # radio groups handled separately
states = field.get("/_States_", [])
if len(states) == 2:
# "/Off" seems to always be the unchecked value, as suggested by
# https://opensource.adobe.com/dc-acrobat-sdk-docs/standards/pdfstandards/pdf/PDF32000_2008.pdf#page=448
# It can be either first or second in the "/_States_" list.
if "/Off" in states:
field_dict["checked_value"] = states[0] if states[0] != "/Off" else states[1]
field_dict["unchecked_value"] = "/Off"
else:
print(f"Unexpected state values for checkbox `${field_id}`. Its checked and unchecked values may not be correct; if you're trying to check it, visually verify the results.")
field_dict["checked_value"] = states[0]
field_dict["unchecked_value"] = states[1]
elif ft == "/Ch":
field_dict["type"] = "choice"
states = field.get("/_States_", [])
field_dict["choice_options"] = [{
"value": state[0],
"text": state[1],
} for state in states]
else:
field_dict["type"] = f"unknown ({ft})"
return field_dict
# Returns a list of fillable PDF fields:
# [
# {
# "field_id": "name",
# "page": 1,
# "type": ("text", "checkbox", "radio_group", or "choice")
# // Per-type additional fields described in forms.md
# },
# ]
def get_field_info(reader: PdfReader):
fields = reader.get_fields()
field_info_by_id = {}
possible_radio_names = set()
for field_id, field in fields.items():
# Skip if this is a container field with children, except that it might be
# a parent group for radio button options.
if field.get("/Kids"):
if field.get("/FT") == "/Btn":
possible_radio_names.add(field_id)
continue
field_info_by_id[field_id] = make_field_dict(field, field_id)
# Bounding rects are stored in annotations in page objects.
# Radio button options have a separate annotation for each choice;
# all choices have the same field name.
# See https://westhealth.github.io/exploring-fillable-forms-with-pdfrw.html
radio_fields_by_id = {}
for page_index, page in enumerate(reader.pages):
annotations = page.get('/Annots', [])
for ann in annotations:
field_id = get_full_annotation_field_id(ann)
if field_id in field_info_by_id:
field_info_by_id[field_id]["page"] = page_index + 1
field_info_by_id[field_id]["rect"] = ann.get('/Rect')
elif field_id in possible_radio_names:
try:
# ann['/AP']['/N'] should have two items. One of them is '/Off',
# the other is the active value.
on_values = [v for v in ann["/AP"]["/N"] if v != "/Off"]
except KeyError:
continue
if len(on_values) == 1:
rect = ann.get("/Rect")
if field_id not in radio_fields_by_id:
radio_fields_by_id[field_id] = {
"field_id": field_id,
"type": "radio_group",
"page": page_index + 1,
"radio_options": [],
}
# Note: at least on macOS 15.7, Preview.app doesn't show selected
# radio buttons correctly. (It does if you remove the leading slash
# from the value, but that causes them not to appear correctly in
# Chrome/Firefox/Acrobat/etc).
radio_fields_by_id[field_id]["radio_options"].append({
"value": on_values[0],
"rect": rect,
})
# Some PDFs have form field definitions without corresponding annotations,
# so we can't tell where they are. Ignore these fields for now.
fields_with_location = []
for field_info in field_info_by_id.values():
if "page" in field_info:
fields_with_location.append(field_info)
else:
print(f"Unable to determine location for field id: {field_info.get('field_id')}, ignoring")
# Sort by page number, then Y position (flipped in PDF coordinate system), then X.
def sort_key(f):
if "radio_options" in f:
rect = f["radio_options"][0]["rect"] or [0, 0, 0, 0]
else:
rect = f.get("rect") or [0, 0, 0, 0]
adjusted_position = [-rect[1], rect[0]]
return [f.get("page"), adjusted_position]
sorted_fields = fields_with_location + list(radio_fields_by_id.values())
sorted_fields.sort(key=sort_key)
return sorted_fields
def write_field_info(pdf_path: str, json_output_path: str):
reader = PdfReader(pdf_path)
field_info = get_field_info(reader)
with open(json_output_path, "w") as f:
json.dump(field_info, f, indent=2)
print(f"Wrote {len(field_info)} fields to {json_output_path}")
if __name__ == "__main__":
if len(sys.argv) != 3:
print("Usage: extract_form_field_info.py [input pdf] [output json]")
sys.exit(1)
write_field_info(sys.argv[1], sys.argv[2])
@@ -0,0 +1,114 @@
import json
import sys
from pypdf import PdfReader, PdfWriter
from extract_form_field_info import get_field_info
# Fills fillable form fields in a PDF. See forms.md.
def fill_pdf_fields(input_pdf_path: str, fields_json_path: str, output_pdf_path: str):
with open(fields_json_path) as f:
fields = json.load(f)
# Group by page number.
fields_by_page = {}
for field in fields:
if "value" in field:
field_id = field["field_id"]
page = field["page"]
if page not in fields_by_page:
fields_by_page[page] = {}
fields_by_page[page][field_id] = field["value"]
reader = PdfReader(input_pdf_path)
has_error = False
field_info = get_field_info(reader)
fields_by_ids = {f["field_id"]: f for f in field_info}
for field in fields:
existing_field = fields_by_ids.get(field["field_id"])
if not existing_field:
has_error = True
print(f"ERROR: `{field['field_id']}` is not a valid field ID")
elif field["page"] != existing_field["page"]:
has_error = True
print(f"ERROR: Incorrect page number for `{field['field_id']}` (got {field['page']}, expected {existing_field['page']})")
else:
if "value" in field:
err = validation_error_for_field_value(existing_field, field["value"])
if err:
print(err)
has_error = True
if has_error:
sys.exit(1)
writer = PdfWriter(clone_from=reader)
for page, field_values in fields_by_page.items():
writer.update_page_form_field_values(writer.pages[page - 1], field_values, auto_regenerate=False)
# This seems to be necessary for many PDF viewers to format the form values correctly.
# It may cause the viewer to show a "save changes" dialog even if the user doesn't make any changes.
writer.set_need_appearances_writer(True)
with open(output_pdf_path, "wb") as f:
writer.write(f)
def validation_error_for_field_value(field_info, field_value):
field_type = field_info["type"]
field_id = field_info["field_id"]
if field_type == "checkbox":
checked_val = field_info["checked_value"]
unchecked_val = field_info["unchecked_value"]
if field_value != checked_val and field_value != unchecked_val:
return f'ERROR: Invalid value "{field_value}" for checkbox field "{field_id}". The checked value is "{checked_val}" and the unchecked value is "{unchecked_val}"'
elif field_type == "radio_group":
option_values = [opt["value"] for opt in field_info["radio_options"]]
if field_value not in option_values:
return f'ERROR: Invalid value "{field_value}" for radio group field "{field_id}". Valid values are: {option_values}'
elif field_type == "choice":
choice_values = [opt["value"] for opt in field_info["choice_options"]]
if field_value not in choice_values:
return f'ERROR: Invalid value "{field_value}" for choice field "{field_id}". Valid values are: {choice_values}'
return None
# pypdf (at least version 5.7.0) has a bug when setting the value for a selection list field.
# In _writer.py around line 966:
#
# if field.get(FA.FT, "/Tx") == "/Ch" and field_flags & FA.FfBits.Combo == 0:
# txt = "\n".join(annotation.get_inherited(FA.Opt, []))
#
# The problem is that for selection lists, `get_inherited` returns a list of two-element lists like
# [["value1", "Text 1"], ["value2", "Text 2"], ...]
# This causes `join` to throw a TypeError because it expects an iterable of strings.
# The horrible workaround is to patch `get_inherited` to return a list of the value strings.
# We call the original method and adjust the return value only if the argument to `get_inherited`
# is `FA.Opt` and if the return value is a list of two-element lists.
def monkeypatch_pydpf_method():
from pypdf.generic import DictionaryObject
from pypdf.constants import FieldDictionaryAttributes
original_get_inherited = DictionaryObject.get_inherited
def patched_get_inherited(self, key: str, default = None):
result = original_get_inherited(self, key, default)
if key == FieldDictionaryAttributes.Opt:
if isinstance(result, list) and all(isinstance(v, list) and len(v) == 2 for v in result):
result = [r[0] for r in result]
return result
DictionaryObject.get_inherited = patched_get_inherited
if __name__ == "__main__":
if len(sys.argv) != 4:
print("Usage: fill_fillable_fields.py [input pdf] [field_values.json] [output pdf]")
sys.exit(1)
monkeypatch_pydpf_method()
input_pdf = sys.argv[1]
fields_json = sys.argv[2]
output_pdf = sys.argv[3]
fill_pdf_fields(input_pdf, fields_json, output_pdf)
@@ -0,0 +1,108 @@
import json
import sys
from pypdf import PdfReader, PdfWriter
from pypdf.annotations import FreeText
# Fills a PDF by adding text annotations defined in `fields.json`. See forms.md.
def transform_coordinates(bbox, image_width, image_height, pdf_width, pdf_height):
"""Transform bounding box from image coordinates to PDF coordinates"""
# Image coordinates: origin at top-left, y increases downward
# PDF coordinates: origin at bottom-left, y increases upward
x_scale = pdf_width / image_width
y_scale = pdf_height / image_height
left = bbox[0] * x_scale
right = bbox[2] * x_scale
# Flip Y coordinates for PDF
top = pdf_height - (bbox[1] * y_scale)
bottom = pdf_height - (bbox[3] * y_scale)
return left, bottom, right, top
def fill_pdf_form(input_pdf_path, fields_json_path, output_pdf_path):
"""Fill the PDF form with data from fields.json"""
# `fields.json` format described in forms.md.
with open(fields_json_path, "r") as f:
fields_data = json.load(f)
# Open the PDF
reader = PdfReader(input_pdf_path)
writer = PdfWriter()
# Copy all pages to writer
writer.append(reader)
# Get PDF dimensions for each page
pdf_dimensions = {}
for i, page in enumerate(reader.pages):
mediabox = page.mediabox
pdf_dimensions[i + 1] = [mediabox.width, mediabox.height]
# Process each form field
annotations = []
for field in fields_data["form_fields"]:
page_num = field["page_number"]
# Get page dimensions and transform coordinates.
page_info = next(p for p in fields_data["pages"] if p["page_number"] == page_num)
image_width = page_info["image_width"]
image_height = page_info["image_height"]
pdf_width, pdf_height = pdf_dimensions[page_num]
transformed_entry_box = transform_coordinates(
field["entry_bounding_box"],
image_width, image_height,
pdf_width, pdf_height
)
# Skip empty fields
if "entry_text" not in field or "text" not in field["entry_text"]:
continue
entry_text = field["entry_text"]
text = entry_text["text"]
if not text:
continue
font_name = entry_text.get("font", "Arial")
font_size = str(entry_text.get("font_size", 14)) + "pt"
font_color = entry_text.get("font_color", "000000")
# Font size/color seems to not work reliably across viewers:
# https://github.com/py-pdf/pypdf/issues/2084
annotation = FreeText(
text=text,
rect=transformed_entry_box,
font=font_name,
font_size=font_size,
font_color=font_color,
border_color=None,
background_color=None,
)
annotations.append(annotation)
# page_number is 0-based for pypdf
writer.add_annotation(page_number=page_num - 1, annotation=annotation)
# Save the filled PDF
with open(output_pdf_path, "wb") as output:
writer.write(output)
print(f"Successfully filled PDF form and saved to {output_pdf_path}")
print(f"Added {len(annotations)} text annotations")
if __name__ == "__main__":
if len(sys.argv) != 4:
print("Usage: fill_pdf_form_with_annotations.py [input pdf] [fields.json] [output pdf]")
sys.exit(1)
input_pdf = sys.argv[1]
fields_json = sys.argv[2]
output_pdf = sys.argv[3]
fill_pdf_form(input_pdf, fields_json, output_pdf)
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,231 @@
#!/usr/bin/env python3
"""
Rearrange PowerPoint slides based on a sequence of indices.
Usage:
python rearrange.py template.pptx output.pptx 0,34,34,50,52
This will create output.pptx using slides from template.pptx in the specified order.
Slides can be repeated (e.g., 34 appears twice).
"""
import argparse
import shutil
import sys
from copy import deepcopy
from pathlib import Path
import six
from pptx import Presentation
def main():
parser = argparse.ArgumentParser(
description="Rearrange PowerPoint slides based on a sequence of indices.",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
python rearrange.py template.pptx output.pptx 0,34,34,50,52
Creates output.pptx using slides 0, 34 (twice), 50, and 52 from template.pptx
python rearrange.py template.pptx output.pptx 5,3,1,2,4
Creates output.pptx with slides reordered as specified
Note: Slide indices are 0-based (first slide is 0, second is 1, etc.)
""",
)
parser.add_argument("template", help="Path to template PPTX file")
parser.add_argument("output", help="Path for output PPTX file")
parser.add_argument(
"sequence", help="Comma-separated sequence of slide indices (0-based)"
)
args = parser.parse_args()
# Parse the slide sequence
try:
slide_sequence = [int(x.strip()) for x in args.sequence.split(",")]
except ValueError:
print(
"Error: Invalid sequence format. Use comma-separated integers (e.g., 0,34,34,50,52)"
)
sys.exit(1)
# Check template exists
template_path = Path(args.template)
if not template_path.exists():
print(f"Error: Template file not found: {args.template}")
sys.exit(1)
# Create output directory if needed
output_path = Path(args.output)
output_path.parent.mkdir(parents=True, exist_ok=True)
try:
rearrange_presentation(template_path, output_path, slide_sequence)
except ValueError as e:
print(f"Error: {e}")
sys.exit(1)
except Exception as e:
print(f"Error processing presentation: {e}")
sys.exit(1)
def duplicate_slide(pres, index):
"""Duplicate a slide in the presentation."""
source = pres.slides[index]
# Use source's layout to preserve formatting
new_slide = pres.slides.add_slide(source.slide_layout)
# Collect all image and media relationships from the source slide
image_rels = {}
for rel_id, rel in six.iteritems(source.part.rels):
if "image" in rel.reltype or "media" in rel.reltype:
image_rels[rel_id] = rel
# CRITICAL: Clear placeholder shapes to avoid duplicates
for shape in new_slide.shapes:
sp = shape.element
sp.getparent().remove(sp)
# Copy all shapes from source
for shape in source.shapes:
el = shape.element
new_el = deepcopy(el)
new_slide.shapes._spTree.insert_element_before(new_el, "p:extLst")
# Handle picture shapes - need to update the blip reference
# Look for all blip elements (they can be in pic or other contexts)
# Using the element's own xpath method without namespaces argument
blips = new_el.xpath(".//a:blip[@r:embed]")
for blip in blips:
old_rId = blip.get(
"{http://schemas.openxmlformats.org/officeDocument/2006/relationships}embed"
)
if old_rId in image_rels:
# Create a new relationship in the destination slide for this image
old_rel = image_rels[old_rId]
# get_or_add returns the rId directly, or adds and returns new rId
new_rId = new_slide.part.rels.get_or_add(
old_rel.reltype, old_rel._target
)
# Update the blip's embed reference to use the new relationship ID
blip.set(
"{http://schemas.openxmlformats.org/officeDocument/2006/relationships}embed",
new_rId,
)
# Copy any additional image/media relationships that might be referenced elsewhere
for rel_id, rel in image_rels.items():
try:
new_slide.part.rels.get_or_add(rel.reltype, rel._target)
except Exception:
pass # Relationship might already exist
return new_slide
def delete_slide(pres, index):
"""Delete a slide from the presentation."""
rId = pres.slides._sldIdLst[index].rId
pres.part.drop_rel(rId)
del pres.slides._sldIdLst[index]
def reorder_slides(pres, slide_index, target_index):
"""Move a slide from one position to another."""
slides = pres.slides._sldIdLst
# Remove slide element from current position
slide_element = slides[slide_index]
slides.remove(slide_element)
# Insert at target position
slides.insert(target_index, slide_element)
def rearrange_presentation(template_path, output_path, slide_sequence):
"""
Create a new presentation with slides from template in specified order.
Args:
template_path: Path to template PPTX file
output_path: Path for output PPTX file
slide_sequence: List of slide indices (0-based) to include
"""
# Copy template to preserve dimensions and theme
if template_path != output_path:
shutil.copy2(template_path, output_path)
prs = Presentation(output_path)
else:
prs = Presentation(template_path)
total_slides = len(prs.slides)
# Validate indices
for idx in slide_sequence:
if idx < 0 or idx >= total_slides:
raise ValueError(f"Slide index {idx} out of range (0-{total_slides - 1})")
# Track original slides and their duplicates
slide_map = [] # List of actual slide indices for final presentation
duplicated = {} # Track duplicates: original_idx -> [duplicate_indices]
# Step 1: DUPLICATE repeated slides
print(f"Processing {len(slide_sequence)} slides from template...")
for i, template_idx in enumerate(slide_sequence):
if template_idx in duplicated and duplicated[template_idx]:
# Already duplicated this slide, use the duplicate
slide_map.append(duplicated[template_idx].pop(0))
print(f" [{i}] Using duplicate of slide {template_idx}")
elif slide_sequence.count(template_idx) > 1 and template_idx not in duplicated:
# First occurrence of a repeated slide - create duplicates
slide_map.append(template_idx)
duplicates = []
count = slide_sequence.count(template_idx) - 1
print(
f" [{i}] Using original slide {template_idx}, creating {count} duplicate(s)"
)
for _ in range(count):
duplicate_slide(prs, template_idx)
duplicates.append(len(prs.slides) - 1)
duplicated[template_idx] = duplicates
else:
# Unique slide or first occurrence already handled, use original
slide_map.append(template_idx)
print(f" [{i}] Using original slide {template_idx}")
# Step 2: DELETE unwanted slides (work backwards)
slides_to_keep = set(slide_map)
print(f"\nDeleting {len(prs.slides) - len(slides_to_keep)} unused slides...")
for i in range(len(prs.slides) - 1, -1, -1):
if i not in slides_to_keep:
delete_slide(prs, i)
# Update slide_map indices after deletion
slide_map = [idx - 1 if idx > i else idx for idx in slide_map]
# Step 3: REORDER to final sequence
print(f"Reordering {len(slide_map)} slides to final sequence...")
for target_pos in range(len(slide_map)):
# Find which slide should be at target_pos
current_pos = slide_map[target_pos]
if current_pos != target_pos:
reorder_slides(prs, current_pos, target_pos)
# Update slide_map: the move shifts other slides
for i in range(len(slide_map)):
if slide_map[i] > current_pos and slide_map[i] <= target_pos:
slide_map[i] -= 1
elif slide_map[i] < current_pos and slide_map[i] >= target_pos:
slide_map[i] += 1
slide_map[target_pos] = target_pos
# Save the presentation
prs.save(output_path)
print(f"\nSaved rearranged presentation to: {output_path}")
print(f"Final presentation has {len(prs.slides)} slides")
if __name__ == "__main__":
main()
@@ -0,0 +1,385 @@
#!/usr/bin/env python3
"""Apply text replacements to PowerPoint presentation.
Usage:
python replace.py <input.pptx> <replacements.json> <output.pptx>
The replacements JSON should have the structure output by inventory.py.
ALL text shapes identified by inventory.py will have their text cleared
unless "paragraphs" is specified in the replacements for that shape.
"""
import json
import sys
from pathlib import Path
from typing import Any, Dict, List
from inventory import InventoryData, extract_text_inventory
from pptx import Presentation
from pptx.dml.color import RGBColor
from pptx.enum.dml import MSO_THEME_COLOR
from pptx.enum.text import PP_ALIGN
from pptx.oxml.xmlchemy import OxmlElement
from pptx.util import Pt
def clear_paragraph_bullets(paragraph):
"""Clear bullet formatting from a paragraph."""
pPr = paragraph._element.get_or_add_pPr()
# Remove existing bullet elements
for child in list(pPr):
if (
child.tag.endswith("buChar")
or child.tag.endswith("buNone")
or child.tag.endswith("buAutoNum")
or child.tag.endswith("buFont")
):
pPr.remove(child)
return pPr
def apply_paragraph_properties(paragraph, para_data: Dict[str, Any]):
"""Apply formatting properties to a paragraph."""
# Get the text but don't set it on paragraph directly yet
text = para_data.get("text", "")
# Get or create paragraph properties
pPr = clear_paragraph_bullets(paragraph)
# Handle bullet formatting
if para_data.get("bullet", False):
level = para_data.get("level", 0)
paragraph.level = level
# Calculate font-proportional indentation
font_size = para_data.get("font_size", 18.0)
level_indent_emu = int((font_size * (1.6 + level * 1.6)) * 12700)
hanging_indent_emu = int(-font_size * 0.8 * 12700)
# Set indentation
pPr.attrib["marL"] = str(level_indent_emu)
pPr.attrib["indent"] = str(hanging_indent_emu)
# Add bullet character
buChar = OxmlElement("a:buChar")
buChar.set("char", "")
pPr.append(buChar)
# Default to left alignment for bullets if not specified
if "alignment" not in para_data:
paragraph.alignment = PP_ALIGN.LEFT
else:
# Remove indentation for non-bullet text
pPr.attrib["marL"] = "0"
pPr.attrib["indent"] = "0"
# Add buNone element
buNone = OxmlElement("a:buNone")
pPr.insert(0, buNone)
# Apply alignment
if "alignment" in para_data:
alignment_map = {
"LEFT": PP_ALIGN.LEFT,
"CENTER": PP_ALIGN.CENTER,
"RIGHT": PP_ALIGN.RIGHT,
"JUSTIFY": PP_ALIGN.JUSTIFY,
}
if para_data["alignment"] in alignment_map:
paragraph.alignment = alignment_map[para_data["alignment"]]
# Apply spacing
if "space_before" in para_data:
paragraph.space_before = Pt(para_data["space_before"])
if "space_after" in para_data:
paragraph.space_after = Pt(para_data["space_after"])
if "line_spacing" in para_data:
paragraph.line_spacing = Pt(para_data["line_spacing"])
# Apply run-level formatting
if not paragraph.runs:
run = paragraph.add_run()
run.text = text
else:
run = paragraph.runs[0]
run.text = text
# Apply font properties
apply_font_properties(run, para_data)
def apply_font_properties(run, para_data: Dict[str, Any]):
"""Apply font properties to a text run."""
if "bold" in para_data:
run.font.bold = para_data["bold"]
if "italic" in para_data:
run.font.italic = para_data["italic"]
if "underline" in para_data:
run.font.underline = para_data["underline"]
if "font_size" in para_data:
run.font.size = Pt(para_data["font_size"])
if "font_name" in para_data:
run.font.name = para_data["font_name"]
# Apply color - prefer RGB, fall back to theme_color
if "color" in para_data:
color_hex = para_data["color"].lstrip("#")
if len(color_hex) == 6:
r = int(color_hex[0:2], 16)
g = int(color_hex[2:4], 16)
b = int(color_hex[4:6], 16)
run.font.color.rgb = RGBColor(r, g, b)
elif "theme_color" in para_data:
# Get theme color by name (e.g., "DARK_1", "ACCENT_1")
theme_name = para_data["theme_color"]
try:
run.font.color.theme_color = getattr(MSO_THEME_COLOR, theme_name)
except AttributeError:
print(f" WARNING: Unknown theme color name '{theme_name}'")
def detect_frame_overflow(inventory: InventoryData) -> Dict[str, Dict[str, float]]:
"""Detect text overflow in shapes (text exceeding shape bounds).
Returns dict of slide_key -> shape_key -> overflow_inches.
Only includes shapes that have text overflow.
"""
overflow_map = {}
for slide_key, shapes_dict in inventory.items():
for shape_key, shape_data in shapes_dict.items():
# Check for frame overflow (text exceeding shape bounds)
if shape_data.frame_overflow_bottom is not None:
if slide_key not in overflow_map:
overflow_map[slide_key] = {}
overflow_map[slide_key][shape_key] = shape_data.frame_overflow_bottom
return overflow_map
def validate_replacements(inventory: InventoryData, replacements: Dict) -> List[str]:
"""Validate that all shapes in replacements exist in inventory.
Returns list of error messages.
"""
errors = []
for slide_key, shapes_data in replacements.items():
if not slide_key.startswith("slide-"):
continue
# Check if slide exists
if slide_key not in inventory:
errors.append(f"Slide '{slide_key}' not found in inventory")
continue
# Check each shape
for shape_key in shapes_data.keys():
if shape_key not in inventory[slide_key]:
# Find shapes without replacements defined and show their content
unused_with_content = []
for k in inventory[slide_key].keys():
if k not in shapes_data:
shape_data = inventory[slide_key][k]
# Get text from paragraphs as preview
paragraphs = shape_data.paragraphs
if paragraphs and paragraphs[0].text:
first_text = paragraphs[0].text[:50]
if len(paragraphs[0].text) > 50:
first_text += "..."
unused_with_content.append(f"{k} ('{first_text}')")
else:
unused_with_content.append(k)
errors.append(
f"Shape '{shape_key}' not found on '{slide_key}'. "
f"Shapes without replacements: {', '.join(sorted(unused_with_content)) if unused_with_content else 'none'}"
)
return errors
def check_duplicate_keys(pairs):
"""Check for duplicate keys when loading JSON."""
result = {}
for key, value in pairs:
if key in result:
raise ValueError(f"Duplicate key found in JSON: '{key}'")
result[key] = value
return result
def apply_replacements(pptx_file: str, json_file: str, output_file: str):
"""Apply text replacements from JSON to PowerPoint presentation."""
# Load presentation
prs = Presentation(pptx_file)
# Get inventory of all text shapes (returns ShapeData objects)
# Pass prs to use same Presentation instance
inventory = extract_text_inventory(Path(pptx_file), prs)
# Detect text overflow in original presentation
original_overflow = detect_frame_overflow(inventory)
# Load replacement data with duplicate key detection
with open(json_file, "r") as f:
replacements = json.load(f, object_pairs_hook=check_duplicate_keys)
# Validate replacements
errors = validate_replacements(inventory, replacements)
if errors:
print("ERROR: Invalid shapes in replacement JSON:")
for error in errors:
print(f" - {error}")
print("\nPlease check the inventory and update your replacement JSON.")
print(
"You can regenerate the inventory with: python inventory.py <input.pptx> <output.json>"
)
raise ValueError(f"Found {len(errors)} validation error(s)")
# Track statistics
shapes_processed = 0
shapes_cleared = 0
shapes_replaced = 0
# Process each slide from inventory
for slide_key, shapes_dict in inventory.items():
if not slide_key.startswith("slide-"):
continue
slide_index = int(slide_key.split("-")[1])
if slide_index >= len(prs.slides):
print(f"Warning: Slide {slide_index} not found")
continue
# Process each shape from inventory
for shape_key, shape_data in shapes_dict.items():
shapes_processed += 1
# Get the shape directly from ShapeData
shape = shape_data.shape
if not shape:
print(f"Warning: {shape_key} has no shape reference")
continue
# ShapeData already validates text_frame in __init__
text_frame = shape.text_frame # type: ignore
text_frame.clear() # type: ignore
shapes_cleared += 1
# Check for replacement paragraphs
replacement_shape_data = replacements.get(slide_key, {}).get(shape_key, {})
if "paragraphs" not in replacement_shape_data:
continue
shapes_replaced += 1
# Add replacement paragraphs
for i, para_data in enumerate(replacement_shape_data["paragraphs"]):
if i == 0:
p = text_frame.paragraphs[0] # type: ignore
else:
p = text_frame.add_paragraph() # type: ignore
apply_paragraph_properties(p, para_data)
# Check for issues after replacements
# Save to a temporary file and reload to avoid modifying the presentation during inventory
# (extract_text_inventory accesses font.color which adds empty <a:solidFill/> elements)
import tempfile
with tempfile.NamedTemporaryFile(suffix=".pptx", delete=False) as tmp:
tmp_path = Path(tmp.name)
prs.save(str(tmp_path))
try:
updated_inventory = extract_text_inventory(tmp_path)
updated_overflow = detect_frame_overflow(updated_inventory)
finally:
tmp_path.unlink() # Clean up temp file
# Check if any text overflow got worse
overflow_errors = []
for slide_key, shape_overflows in updated_overflow.items():
for shape_key, new_overflow in shape_overflows.items():
# Get original overflow (0 if there was no overflow before)
original = original_overflow.get(slide_key, {}).get(shape_key, 0.0)
# Error if overflow increased
if new_overflow > original + 0.01: # Small tolerance for rounding
increase = new_overflow - original
overflow_errors.append(
f'{slide_key}/{shape_key}: overflow worsened by {increase:.2f}" '
f'(was {original:.2f}", now {new_overflow:.2f}")'
)
# Collect warnings from updated shapes
warnings = []
for slide_key, shapes_dict in updated_inventory.items():
for shape_key, shape_data in shapes_dict.items():
if shape_data.warnings:
for warning in shape_data.warnings:
warnings.append(f"{slide_key}/{shape_key}: {warning}")
# Fail if there are any issues
if overflow_errors or warnings:
print("\nERROR: Issues detected in replacement output:")
if overflow_errors:
print("\nText overflow worsened:")
for error in overflow_errors:
print(f" - {error}")
if warnings:
print("\nFormatting warnings:")
for warning in warnings:
print(f" - {warning}")
print("\nPlease fix these issues before saving.")
raise ValueError(
f"Found {len(overflow_errors)} overflow error(s) and {len(warnings)} warning(s)"
)
# Save the presentation
prs.save(output_file)
# Report results
print(f"Saved updated presentation to: {output_file}")
print(f"Processed {len(prs.slides)} slides")
print(f" - Shapes processed: {shapes_processed}")
print(f" - Shapes cleared: {shapes_cleared}")
print(f" - Shapes replaced: {shapes_replaced}")
def main():
"""Main entry point for command-line usage."""
if len(sys.argv) != 4:
print(__doc__)
sys.exit(1)
input_pptx = Path(sys.argv[1])
replacements_json = Path(sys.argv[2])
output_pptx = Path(sys.argv[3])
if not input_pptx.exists():
print(f"Error: Input file '{input_pptx}' not found")
sys.exit(1)
if not replacements_json.exists():
print(f"Error: Replacements JSON file '{replacements_json}' not found")
sys.exit(1)
try:
apply_replacements(str(input_pptx), str(replacements_json), str(output_pptx))
except Exception as e:
print(f"Error applying replacements: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
main()
@@ -0,0 +1,450 @@
#!/usr/bin/env python3
"""
Create thumbnail grids from PowerPoint presentation slides.
Creates a grid layout of slide thumbnails with configurable columns (max 6).
Each grid contains up to cols×(cols+1) images. For presentations with more
slides, multiple numbered grid files are created automatically.
The program outputs the names of all files created.
Output:
- Single grid: {prefix}.jpg (if slides fit in one grid)
- Multiple grids: {prefix}-1.jpg, {prefix}-2.jpg, etc.
Grid limits by column count:
- 3 cols: max 12 slides per grid (3×4)
- 4 cols: max 20 slides per grid (4×5)
- 5 cols: max 30 slides per grid (5×6) [default]
- 6 cols: max 42 slides per grid (6×7)
Usage:
python thumbnail.py input.pptx [output_prefix] [--cols N] [--outline-placeholders]
Examples:
python thumbnail.py presentation.pptx
# Creates: thumbnails.jpg (using default prefix)
# Outputs:
# Created 1 grid(s):
# - thumbnails.jpg
python thumbnail.py large-deck.pptx grid --cols 4
# Creates: grid-1.jpg, grid-2.jpg, grid-3.jpg
# Outputs:
# Created 3 grid(s):
# - grid-1.jpg
# - grid-2.jpg
# - grid-3.jpg
python thumbnail.py template.pptx analysis --outline-placeholders
# Creates thumbnail grids with red outlines around text placeholders
"""
import argparse
import subprocess
import sys
import tempfile
from pathlib import Path
from inventory import extract_text_inventory
from PIL import Image, ImageDraw, ImageFont
from pptx import Presentation
# Constants
THUMBNAIL_WIDTH = 300 # Fixed thumbnail width in pixels
CONVERSION_DPI = 100 # DPI for PDF to image conversion
MAX_COLS = 6 # Maximum number of columns
DEFAULT_COLS = 5 # Default number of columns
JPEG_QUALITY = 95 # JPEG compression quality
# Grid layout constants
GRID_PADDING = 20 # Padding between thumbnails
BORDER_WIDTH = 2 # Border width around thumbnails
FONT_SIZE_RATIO = 0.12 # Font size as fraction of thumbnail width
LABEL_PADDING_RATIO = 0.4 # Label padding as fraction of font size
def main():
parser = argparse.ArgumentParser(
description="Create thumbnail grids from PowerPoint slides."
)
parser.add_argument("input", help="Input PowerPoint file (.pptx)")
parser.add_argument(
"output_prefix",
nargs="?",
default="thumbnails",
help="Output prefix for image files (default: thumbnails, will create prefix.jpg or prefix-N.jpg)",
)
parser.add_argument(
"--cols",
type=int,
default=DEFAULT_COLS,
help=f"Number of columns (default: {DEFAULT_COLS}, max: {MAX_COLS})",
)
parser.add_argument(
"--outline-placeholders",
action="store_true",
help="Outline text placeholders with a colored border",
)
args = parser.parse_args()
# Validate columns
cols = min(args.cols, MAX_COLS)
if args.cols > MAX_COLS:
print(f"Warning: Columns limited to {MAX_COLS} (requested {args.cols})")
# Validate input
input_path = Path(args.input)
if not input_path.exists() or input_path.suffix.lower() != ".pptx":
print(f"Error: Invalid PowerPoint file: {args.input}")
sys.exit(1)
# Construct output path (always JPG)
output_path = Path(f"{args.output_prefix}.jpg")
print(f"Processing: {args.input}")
try:
with tempfile.TemporaryDirectory() as temp_dir:
# Get placeholder regions if outlining is enabled
placeholder_regions = None
slide_dimensions = None
if args.outline_placeholders:
print("Extracting placeholder regions...")
placeholder_regions, slide_dimensions = get_placeholder_regions(
input_path
)
if placeholder_regions:
print(f"Found placeholders on {len(placeholder_regions)} slides")
# Convert slides to images
slide_images = convert_to_images(input_path, Path(temp_dir), CONVERSION_DPI)
if not slide_images:
print("Error: No slides found")
sys.exit(1)
print(f"Found {len(slide_images)} slides")
# Create grids (max cols×(cols+1) images per grid)
grid_files = create_grids(
slide_images,
cols,
THUMBNAIL_WIDTH,
output_path,
placeholder_regions,
slide_dimensions,
)
# Print saved files
print(f"Created {len(grid_files)} grid(s):")
for grid_file in grid_files:
print(f" - {grid_file}")
except Exception as e:
print(f"Error: {e}")
sys.exit(1)
def create_hidden_slide_placeholder(size):
"""Create placeholder image for hidden slides."""
img = Image.new("RGB", size, color="#F0F0F0")
draw = ImageDraw.Draw(img)
line_width = max(5, min(size) // 100)
draw.line([(0, 0), size], fill="#CCCCCC", width=line_width)
draw.line([(size[0], 0), (0, size[1])], fill="#CCCCCC", width=line_width)
return img
def get_placeholder_regions(pptx_path):
"""Extract ALL text regions from the presentation.
Returns a tuple of (placeholder_regions, slide_dimensions).
text_regions is a dict mapping slide indices to lists of text regions.
Each region is a dict with 'left', 'top', 'width', 'height' in inches.
slide_dimensions is a tuple of (width_inches, height_inches).
"""
prs = Presentation(str(pptx_path))
inventory = extract_text_inventory(pptx_path, prs)
placeholder_regions = {}
# Get actual slide dimensions in inches (EMU to inches conversion)
slide_width_inches = (prs.slide_width or 9144000) / 914400.0
slide_height_inches = (prs.slide_height or 5143500) / 914400.0
for slide_key, shapes in inventory.items():
# Extract slide index from "slide-N" format
slide_idx = int(slide_key.split("-")[1])
regions = []
for shape_key, shape_data in shapes.items():
# The inventory only contains shapes with text, so all shapes should be highlighted
regions.append(
{
"left": shape_data.left,
"top": shape_data.top,
"width": shape_data.width,
"height": shape_data.height,
}
)
if regions:
placeholder_regions[slide_idx] = regions
return placeholder_regions, (slide_width_inches, slide_height_inches)
def convert_to_images(pptx_path, temp_dir, dpi):
"""Convert PowerPoint to images via PDF, handling hidden slides."""
# Detect hidden slides
print("Analyzing presentation...")
prs = Presentation(str(pptx_path))
total_slides = len(prs.slides)
# Find hidden slides (1-based indexing for display)
hidden_slides = {
idx + 1
for idx, slide in enumerate(prs.slides)
if slide.element.get("show") == "0"
}
print(f"Total slides: {total_slides}")
if hidden_slides:
print(f"Hidden slides: {sorted(hidden_slides)}")
pdf_path = temp_dir / f"{pptx_path.stem}.pdf"
# Convert to PDF
print("Converting to PDF...")
result = subprocess.run(
[
"soffice",
"--headless",
"--convert-to",
"pdf",
"--outdir",
str(temp_dir),
str(pptx_path),
],
capture_output=True,
text=True,
)
if result.returncode != 0 or not pdf_path.exists():
raise RuntimeError("PDF conversion failed")
# Convert PDF to images
print(f"Converting to images at {dpi} DPI...")
result = subprocess.run(
["pdftoppm", "-jpeg", "-r", str(dpi), str(pdf_path), str(temp_dir / "slide")],
capture_output=True,
text=True,
)
if result.returncode != 0:
raise RuntimeError("Image conversion failed")
visible_images = sorted(temp_dir.glob("slide-*.jpg"))
# Create full list with placeholders for hidden slides
all_images = []
visible_idx = 0
# Get placeholder dimensions from first visible slide
if visible_images:
with Image.open(visible_images[0]) as img:
placeholder_size = img.size
else:
placeholder_size = (1920, 1080)
for slide_num in range(1, total_slides + 1):
if slide_num in hidden_slides:
# Create placeholder image for hidden slide
placeholder_path = temp_dir / f"hidden-{slide_num:03d}.jpg"
placeholder_img = create_hidden_slide_placeholder(placeholder_size)
placeholder_img.save(placeholder_path, "JPEG")
all_images.append(placeholder_path)
else:
# Use the actual visible slide image
if visible_idx < len(visible_images):
all_images.append(visible_images[visible_idx])
visible_idx += 1
return all_images
def create_grids(
image_paths,
cols,
width,
output_path,
placeholder_regions=None,
slide_dimensions=None,
):
"""Create multiple thumbnail grids from slide images, max cols×(cols+1) images per grid."""
# Maximum images per grid is cols × (cols + 1) for better proportions
max_images_per_grid = cols * (cols + 1)
grid_files = []
print(
f"Creating grids with {cols} columns (max {max_images_per_grid} images per grid)"
)
# Split images into chunks
for chunk_idx, start_idx in enumerate(
range(0, len(image_paths), max_images_per_grid)
):
end_idx = min(start_idx + max_images_per_grid, len(image_paths))
chunk_images = image_paths[start_idx:end_idx]
# Create grid for this chunk
grid = create_grid(
chunk_images, cols, width, start_idx, placeholder_regions, slide_dimensions
)
# Generate output filename
if len(image_paths) <= max_images_per_grid:
# Single grid - use base filename without suffix
grid_filename = output_path
else:
# Multiple grids - insert index before extension with dash
stem = output_path.stem
suffix = output_path.suffix
grid_filename = output_path.parent / f"{stem}-{chunk_idx + 1}{suffix}"
# Save grid
grid_filename.parent.mkdir(parents=True, exist_ok=True)
grid.save(str(grid_filename), quality=JPEG_QUALITY)
grid_files.append(str(grid_filename))
return grid_files
def create_grid(
image_paths,
cols,
width,
start_slide_num=0,
placeholder_regions=None,
slide_dimensions=None,
):
"""Create thumbnail grid from slide images with optional placeholder outlining."""
font_size = int(width * FONT_SIZE_RATIO)
label_padding = int(font_size * LABEL_PADDING_RATIO)
# Get dimensions
with Image.open(image_paths[0]) as img:
aspect = img.height / img.width
height = int(width * aspect)
# Calculate grid size
rows = (len(image_paths) + cols - 1) // cols
grid_w = cols * width + (cols + 1) * GRID_PADDING
grid_h = rows * (height + font_size + label_padding * 2) + (rows + 1) * GRID_PADDING
# Create grid
grid = Image.new("RGB", (grid_w, grid_h), "white")
draw = ImageDraw.Draw(grid)
# Load font with size based on thumbnail width
try:
# Use Pillow's default font with size
font = ImageFont.load_default(size=font_size)
except Exception:
# Fall back to basic default font if size parameter not supported
font = ImageFont.load_default()
# Place thumbnails
for i, img_path in enumerate(image_paths):
row, col = i // cols, i % cols
x = col * width + (col + 1) * GRID_PADDING
y_base = (
row * (height + font_size + label_padding * 2) + (row + 1) * GRID_PADDING
)
# Add label with actual slide number
label = f"{start_slide_num + i}"
bbox = draw.textbbox((0, 0), label, font=font)
text_w = bbox[2] - bbox[0]
draw.text(
(x + (width - text_w) // 2, y_base + label_padding),
label,
fill="black",
font=font,
)
# Add thumbnail below label with proportional spacing
y_thumbnail = y_base + label_padding + font_size + label_padding
with Image.open(img_path) as img:
# Get original dimensions before thumbnail
orig_w, orig_h = img.size
# Apply placeholder outlines if enabled
if placeholder_regions and (start_slide_num + i) in placeholder_regions:
# Convert to RGBA for transparency support
if img.mode != "RGBA":
img = img.convert("RGBA")
# Get the regions for this slide
regions = placeholder_regions[start_slide_num + i]
# Calculate scale factors using actual slide dimensions
if slide_dimensions:
slide_width_inches, slide_height_inches = slide_dimensions
else:
# Fallback: estimate from image size at CONVERSION_DPI
slide_width_inches = orig_w / CONVERSION_DPI
slide_height_inches = orig_h / CONVERSION_DPI
x_scale = orig_w / slide_width_inches
y_scale = orig_h / slide_height_inches
# Create a highlight overlay
overlay = Image.new("RGBA", img.size, (255, 255, 255, 0))
overlay_draw = ImageDraw.Draw(overlay)
# Highlight each placeholder region
for region in regions:
# Convert from inches to pixels in the original image
px_left = int(region["left"] * x_scale)
px_top = int(region["top"] * y_scale)
px_width = int(region["width"] * x_scale)
px_height = int(region["height"] * y_scale)
# Draw highlight outline with red color and thick stroke
# Using a bright red outline instead of fill
stroke_width = max(
5, min(orig_w, orig_h) // 150
) # Thicker proportional stroke width
overlay_draw.rectangle(
[(px_left, px_top), (px_left + px_width, px_top + px_height)],
outline=(255, 0, 0, 255), # Bright red, fully opaque
width=stroke_width,
)
# Composite the overlay onto the image using alpha blending
img = Image.alpha_composite(img, overlay)
# Convert back to RGB for JPEG saving
img = img.convert("RGB")
img.thumbnail((width, height), Image.Resampling.LANCZOS)
w, h = img.size
tx = x + (width - w) // 2
ty = y_thumbnail + (height - h) // 2
grid.paste(img, (tx, ty))
# Add border
if BORDER_WIDTH > 0:
draw.rectangle(
[
(tx - BORDER_WIDTH, ty - BORDER_WIDTH),
(tx + w + BORDER_WIDTH - 1, ty + h + BORDER_WIDTH - 1),
],
outline="gray",
width=BORDER_WIDTH,
)
return grid
if __name__ == "__main__":
main()
@@ -0,0 +1,178 @@
#!/usr/bin/env python3
"""
Excel Formula Recalculation Script
Recalculates all formulas in an Excel file using LibreOffice
"""
import json
import sys
import subprocess
import os
import platform
from pathlib import Path
from openpyxl import load_workbook
def setup_libreoffice_macro():
"""Setup LibreOffice macro for recalculation if not already configured"""
if platform.system() == 'Darwin':
macro_dir = os.path.expanduser('~/Library/Application Support/LibreOffice/4/user/basic/Standard')
else:
macro_dir = os.path.expanduser('~/.config/libreoffice/4/user/basic/Standard')
macro_file = os.path.join(macro_dir, 'Module1.xba')
if os.path.exists(macro_file):
with open(macro_file, 'r') as f:
if 'RecalculateAndSave' in f.read():
return True
if not os.path.exists(macro_dir):
subprocess.run(['soffice', '--headless', '--terminate_after_init'],
capture_output=True, timeout=10)
os.makedirs(macro_dir, exist_ok=True)
macro_content = '''<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE script:module PUBLIC "-//OpenOffice.org//DTD OfficeDocument 1.0//EN" "module.dtd">
<script:module xmlns:script="http://openoffice.org/2000/script" script:name="Module1" script:language="StarBasic">
Sub RecalculateAndSave()
ThisComponent.calculateAll()
ThisComponent.store()
ThisComponent.close(True)
End Sub
</script:module>'''
try:
with open(macro_file, 'w') as f:
f.write(macro_content)
return True
except Exception:
return False
def recalc(filename, timeout=30):
"""
Recalculate formulas in Excel file and report any errors
Args:
filename: Path to Excel file
timeout: Maximum time to wait for recalculation (seconds)
Returns:
dict with error locations and counts
"""
if not Path(filename).exists():
return {'error': f'File {filename} does not exist'}
abs_path = str(Path(filename).absolute())
if not setup_libreoffice_macro():
return {'error': 'Failed to setup LibreOffice macro'}
cmd = [
'soffice', '--headless', '--norestore',
'vnd.sun.star.script:Standard.Module1.RecalculateAndSave?language=Basic&location=application',
abs_path
]
# Handle timeout command differences between Linux and macOS
if platform.system() != 'Windows':
timeout_cmd = 'timeout' if platform.system() == 'Linux' else None
if platform.system() == 'Darwin':
# Check if gtimeout is available on macOS
try:
subprocess.run(['gtimeout', '--version'], capture_output=True, timeout=1, check=False)
timeout_cmd = 'gtimeout'
except (FileNotFoundError, subprocess.TimeoutExpired):
pass
if timeout_cmd:
cmd = [timeout_cmd, str(timeout)] + cmd
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0 and result.returncode != 124: # 124 is timeout exit code
error_msg = result.stderr or 'Unknown error during recalculation'
if 'Module1' in error_msg or 'RecalculateAndSave' not in error_msg:
return {'error': 'LibreOffice macro not configured properly'}
else:
return {'error': error_msg}
# Check for Excel errors in the recalculated file - scan ALL cells
try:
wb = load_workbook(filename, data_only=True)
excel_errors = ['#VALUE!', '#DIV/0!', '#REF!', '#NAME?', '#NULL!', '#NUM!', '#N/A']
error_details = {err: [] for err in excel_errors}
total_errors = 0
for sheet_name in wb.sheetnames:
ws = wb[sheet_name]
# Check ALL rows and columns - no limits
for row in ws.iter_rows():
for cell in row:
if cell.value is not None and isinstance(cell.value, str):
for err in excel_errors:
if err in cell.value:
location = f"{sheet_name}!{cell.coordinate}"
error_details[err].append(location)
total_errors += 1
break
wb.close()
# Build result summary
result = {
'status': 'success' if total_errors == 0 else 'errors_found',
'total_errors': total_errors,
'error_summary': {}
}
# Add non-empty error categories
for err_type, locations in error_details.items():
if locations:
result['error_summary'][err_type] = {
'count': len(locations),
'locations': locations[:20] # Show up to 20 locations
}
# Add formula count for context - also check ALL cells
wb_formulas = load_workbook(filename, data_only=False)
formula_count = 0
for sheet_name in wb_formulas.sheetnames:
ws = wb_formulas[sheet_name]
for row in ws.iter_rows():
for cell in row:
if cell.value and isinstance(cell.value, str) and cell.value.startswith('='):
formula_count += 1
wb_formulas.close()
result['total_formulas'] = formula_count
return result
except Exception as e:
return {'error': str(e)}
def main():
if len(sys.argv) < 2:
print("Usage: python recalc.py <excel_file> [timeout_seconds]")
print("\nRecalculates all formulas in an Excel file using LibreOffice")
print("\nReturns JSON with error details:")
print(" - status: 'success' or 'errors_found'")
print(" - total_errors: Total number of Excel errors found")
print(" - total_formulas: Number of formulas in the file")
print(" - error_summary: Breakdown by error type with locations")
print(" - #VALUE!, #DIV/0!, #REF!, #NAME?, #NULL!, #NUM!, #N/A")
sys.exit(1)
filename = sys.argv[1]
timeout = int(sys.argv[2]) if len(sys.argv) > 2 else 30
result = recalc(filename, timeout)
print(json.dumps(result, indent=2))
if __name__ == '__main__':
main()
@@ -0,0 +1,17 @@
const MIME_TYPES: Record<string, string> = {
png: 'image/png',
jpg: 'image/jpeg',
jpeg: 'image/jpeg',
csv: 'text/csv',
xlsx: 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
pptx: 'application/vnd.openxmlformats-officedocument.presentationml.presentation',
pdf: 'application/pdf',
json: 'application/json',
txt: 'text/plain',
};
export const getMimeType = (filename: string): string => {
const ext = filename.split('.').pop()?.toLowerCase();
return MIME_TYPES[ext ?? ''] ?? 'application/octet-stream';
};