This commit is contained in:
@@ -1,481 +0,0 @@
|
||||
"""
|
||||
Recursive location index system for dynamic content positioning.
|
||||
|
||||
This module provides a flexible, hierarchical position tracking system that can
|
||||
reference any type of content (words, images, table cells, list items, etc.)
|
||||
in a nested document structure.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
from dataclasses import dataclass, field
|
||||
from typing import List, Dict, Any, Optional, Union, Tuple
|
||||
from enum import Enum
|
||||
import json
|
||||
import pickle
|
||||
import shelve
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class ContentType(Enum):
|
||||
"""Types of content that can be referenced in the position index"""
|
||||
DOCUMENT = "document"
|
||||
CHAPTER = "chapter"
|
||||
BLOCK = "block"
|
||||
PARAGRAPH = "paragraph"
|
||||
HEADING = "heading"
|
||||
TABLE = "table"
|
||||
TABLE_ROW = "table_row"
|
||||
TABLE_CELL = "table_cell"
|
||||
LIST = "list"
|
||||
LIST_ITEM = "list_item"
|
||||
WORD = "word"
|
||||
IMAGE = "image"
|
||||
LINK = "link"
|
||||
BUTTON = "button"
|
||||
FORM_FIELD = "form_field"
|
||||
LINE = "line" # Rendered line of text
|
||||
PAGE = "page" # Rendered page
|
||||
|
||||
|
||||
@dataclass
|
||||
class LocationNode:
|
||||
"""
|
||||
A single node in the recursive location index.
|
||||
Each node represents a position within a specific content type.
|
||||
"""
|
||||
content_type: ContentType
|
||||
index: int = 0 # Position within this content type
|
||||
offset: int = 0 # Offset within the indexed item (e.g., character offset in word)
|
||||
metadata: Dict[str, Any] = field(default_factory=dict) # Additional context
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
"""Serialize node to dictionary"""
|
||||
return {
|
||||
'content_type': self.content_type.value,
|
||||
'index': self.index,
|
||||
'offset': self.offset,
|
||||
'metadata': self.metadata
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict[str, Any]) -> 'LocationNode':
|
||||
"""Deserialize node from dictionary"""
|
||||
return cls(
|
||||
content_type=ContentType(data['content_type']),
|
||||
index=data['index'],
|
||||
offset=data['offset'],
|
||||
metadata=data.get('metadata', {})
|
||||
)
|
||||
|
||||
def __str__(self) -> str:
|
||||
"""Human-readable representation"""
|
||||
if self.offset > 0:
|
||||
return f"{self.content_type.value}[{self.index}]+{self.offset}"
|
||||
return f"{self.content_type.value}[{self.index}]"
|
||||
|
||||
|
||||
@dataclass
|
||||
class RecursivePosition:
|
||||
"""
|
||||
Hierarchical position that can reference any nested content structure.
|
||||
|
||||
The path represents a traversal from document root to the specific location:
|
||||
- Document -> Chapter[2] -> Block[5] -> Paragraph -> Word[12] -> Character[3]
|
||||
- Document -> Chapter[1] -> Block[3] -> Table -> Row[2] -> Cell[1] -> Word[0]
|
||||
- Document -> Chapter[0] -> Block[1] -> Image
|
||||
"""
|
||||
path: List[LocationNode] = field(default_factory=list)
|
||||
rendering_metadata: Dict[str, Any] = field(default_factory=dict) # Font scale, page size, etc.
|
||||
|
||||
def __post_init__(self):
|
||||
"""Ensure we always have at least a document root"""
|
||||
if not self.path:
|
||||
self.path = [LocationNode(ContentType.DOCUMENT)]
|
||||
|
||||
def copy(self) -> 'RecursivePosition':
|
||||
"""Create a deep copy of this position"""
|
||||
return RecursivePosition(
|
||||
path=[LocationNode(node.content_type, node.index, node.offset, node.metadata.copy())
|
||||
for node in self.path],
|
||||
rendering_metadata=self.rendering_metadata.copy()
|
||||
)
|
||||
|
||||
def get_node(self, content_type: ContentType) -> Optional[LocationNode]:
|
||||
"""Get the first node of a specific content type in the path"""
|
||||
for node in self.path:
|
||||
if node.content_type == content_type:
|
||||
return node
|
||||
return None
|
||||
|
||||
def get_nodes(self, content_type: ContentType) -> List[LocationNode]:
|
||||
"""Get all nodes of a specific content type in the path"""
|
||||
return [node for node in self.path if node.content_type == content_type]
|
||||
|
||||
def add_node(self, node: LocationNode) -> 'RecursivePosition':
|
||||
"""Add a node to the path (returns self for chaining)"""
|
||||
self.path.append(node)
|
||||
return self
|
||||
|
||||
def pop_node(self) -> Optional[LocationNode]:
|
||||
"""Remove and return the last node in the path"""
|
||||
if len(self.path) > 1: # Keep at least document root
|
||||
return self.path.pop()
|
||||
return None
|
||||
|
||||
def get_depth(self) -> int:
|
||||
"""Get the depth of the position (number of nodes)"""
|
||||
return len(self.path)
|
||||
|
||||
def get_leaf_node(self) -> LocationNode:
|
||||
"""Get the deepest (most specific) node in the path"""
|
||||
return self.path[-1] if self.path else LocationNode(ContentType.DOCUMENT)
|
||||
|
||||
def truncate_to_type(self, content_type: ContentType) -> 'RecursivePosition':
|
||||
"""Truncate path to end at the first occurrence of the given content type"""
|
||||
for i, node in enumerate(self.path):
|
||||
if node.content_type == content_type:
|
||||
self.path = self.path[:i+1]
|
||||
break
|
||||
return self
|
||||
|
||||
def is_ancestor_of(self, other: 'RecursivePosition') -> bool:
|
||||
"""Check if this position is an ancestor of another position"""
|
||||
if len(self.path) >= len(other.path):
|
||||
return False
|
||||
|
||||
for i, node in enumerate(self.path):
|
||||
if i >= len(other.path):
|
||||
return False
|
||||
other_node = other.path[i]
|
||||
if (node.content_type != other_node.content_type or
|
||||
node.index != other_node.index):
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def is_descendant_of(self, other: 'RecursivePosition') -> bool:
|
||||
"""Check if this position is a descendant of another position"""
|
||||
return other.is_ancestor_of(self)
|
||||
|
||||
def get_common_ancestor(self, other: 'RecursivePosition') -> 'RecursivePosition':
|
||||
"""Find the deepest common ancestor with another position"""
|
||||
common_path = []
|
||||
min_length = min(len(self.path), len(other.path))
|
||||
|
||||
for i in range(min_length):
|
||||
if (self.path[i].content_type == other.path[i].content_type and
|
||||
self.path[i].index == other.path[i].index):
|
||||
common_path.append(self.path[i])
|
||||
else:
|
||||
break
|
||||
|
||||
return RecursivePosition(path=common_path)
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
"""Serialize position to dictionary for JSON storage"""
|
||||
return {
|
||||
'path': [node.to_dict() for node in self.path],
|
||||
'rendering_metadata': self.rendering_metadata
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict[str, Any]) -> 'RecursivePosition':
|
||||
"""Deserialize position from dictionary"""
|
||||
return cls(
|
||||
path=[LocationNode.from_dict(node_data) for node_data in data['path']],
|
||||
rendering_metadata=data.get('rendering_metadata', {})
|
||||
)
|
||||
|
||||
def to_json(self) -> str:
|
||||
"""Serialize to JSON string"""
|
||||
return json.dumps(self.to_dict(), indent=2)
|
||||
|
||||
@classmethod
|
||||
def from_json(cls, json_str: str) -> 'RecursivePosition':
|
||||
"""Deserialize from JSON string"""
|
||||
return cls.from_dict(json.loads(json_str))
|
||||
|
||||
def __str__(self) -> str:
|
||||
"""Human-readable path representation"""
|
||||
return " -> ".join(str(node) for node in self.path)
|
||||
|
||||
def __eq__(self, other) -> bool:
|
||||
"""Check equality with another position"""
|
||||
if not isinstance(other, RecursivePosition):
|
||||
return False
|
||||
return (self.path == other.path and
|
||||
self.rendering_metadata == other.rendering_metadata)
|
||||
|
||||
def __hash__(self) -> int:
|
||||
"""Make position hashable for use as dict key"""
|
||||
path_tuple = tuple((node.content_type, node.index, node.offset) for node in self.path)
|
||||
return hash(path_tuple)
|
||||
|
||||
|
||||
class PositionBuilder:
|
||||
"""
|
||||
Builder class for constructing RecursivePosition objects fluently.
|
||||
|
||||
Example usage:
|
||||
position = (PositionBuilder()
|
||||
.chapter(2)
|
||||
.block(5)
|
||||
.paragraph()
|
||||
.word(12, offset=3)
|
||||
.build())
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self._position = RecursivePosition()
|
||||
|
||||
def document(self, index: int = 0, **metadata) -> 'PositionBuilder':
|
||||
"""Add document node"""
|
||||
self._position.add_node(LocationNode(ContentType.DOCUMENT, index, metadata=metadata))
|
||||
return self
|
||||
|
||||
def chapter(self, index: int, **metadata) -> 'PositionBuilder':
|
||||
"""Add chapter node"""
|
||||
self._position.add_node(LocationNode(ContentType.CHAPTER, index, metadata=metadata))
|
||||
return self
|
||||
|
||||
def block(self, index: int, **metadata) -> 'PositionBuilder':
|
||||
"""Add block node"""
|
||||
self._position.add_node(LocationNode(ContentType.BLOCK, index, metadata=metadata))
|
||||
return self
|
||||
|
||||
def paragraph(self, index: int = 0, **metadata) -> 'PositionBuilder':
|
||||
"""Add paragraph node"""
|
||||
self._position.add_node(LocationNode(ContentType.PARAGRAPH, index, metadata=metadata))
|
||||
return self
|
||||
|
||||
def heading(self, index: int = 0, **metadata) -> 'PositionBuilder':
|
||||
"""Add heading node"""
|
||||
self._position.add_node(LocationNode(ContentType.HEADING, index, metadata=metadata))
|
||||
return self
|
||||
|
||||
def table(self, index: int = 0, **metadata) -> 'PositionBuilder':
|
||||
"""Add table node"""
|
||||
self._position.add_node(LocationNode(ContentType.TABLE, index, metadata=metadata))
|
||||
return self
|
||||
|
||||
def table_row(self, index: int, **metadata) -> 'PositionBuilder':
|
||||
"""Add table row node"""
|
||||
self._position.add_node(LocationNode(ContentType.TABLE_ROW, index, metadata=metadata))
|
||||
return self
|
||||
|
||||
def table_cell(self, index: int, **metadata) -> 'PositionBuilder':
|
||||
"""Add table cell node"""
|
||||
self._position.add_node(LocationNode(ContentType.TABLE_CELL, index, metadata=metadata))
|
||||
return self
|
||||
|
||||
def list(self, index: int = 0, **metadata) -> 'PositionBuilder':
|
||||
"""Add list node"""
|
||||
self._position.add_node(LocationNode(ContentType.LIST, index, metadata=metadata))
|
||||
return self
|
||||
|
||||
def list_item(self, index: int, **metadata) -> 'PositionBuilder':
|
||||
"""Add list item node"""
|
||||
self._position.add_node(LocationNode(ContentType.LIST_ITEM, index, metadata=metadata))
|
||||
return self
|
||||
|
||||
def word(self, index: int, offset: int = 0, **metadata) -> 'PositionBuilder':
|
||||
"""Add word node"""
|
||||
self._position.add_node(LocationNode(ContentType.WORD, index, offset, metadata=metadata))
|
||||
return self
|
||||
|
||||
def image(self, index: int = 0, **metadata) -> 'PositionBuilder':
|
||||
"""Add image node"""
|
||||
self._position.add_node(LocationNode(ContentType.IMAGE, index, metadata=metadata))
|
||||
return self
|
||||
|
||||
def link(self, index: int, **metadata) -> 'PositionBuilder':
|
||||
"""Add link node"""
|
||||
self._position.add_node(LocationNode(ContentType.LINK, index, metadata=metadata))
|
||||
return self
|
||||
|
||||
def button(self, index: int, **metadata) -> 'PositionBuilder':
|
||||
"""Add button node"""
|
||||
self._position.add_node(LocationNode(ContentType.BUTTON, index, metadata=metadata))
|
||||
return self
|
||||
|
||||
def form_field(self, index: int, **metadata) -> 'PositionBuilder':
|
||||
"""Add form field node"""
|
||||
self._position.add_node(LocationNode(ContentType.FORM_FIELD, index, metadata=metadata))
|
||||
return self
|
||||
|
||||
def line(self, index: int, **metadata) -> 'PositionBuilder':
|
||||
"""Add rendered line node"""
|
||||
self._position.add_node(LocationNode(ContentType.LINE, index, metadata=metadata))
|
||||
return self
|
||||
|
||||
def page(self, index: int, **metadata) -> 'PositionBuilder':
|
||||
"""Add page node"""
|
||||
self._position.add_node(LocationNode(ContentType.PAGE, index, metadata=metadata))
|
||||
return self
|
||||
|
||||
def with_rendering_metadata(self, **metadata) -> 'PositionBuilder':
|
||||
"""Add rendering metadata (font scale, page size, etc.)"""
|
||||
self._position.rendering_metadata.update(metadata)
|
||||
return self
|
||||
|
||||
def build(self) -> RecursivePosition:
|
||||
"""Build and return the final position"""
|
||||
return self._position
|
||||
|
||||
|
||||
class PositionStorage:
|
||||
"""
|
||||
Storage manager for recursive positions supporting both JSON and shelf formats.
|
||||
"""
|
||||
|
||||
def __init__(self, storage_dir: str = "positions", use_shelf: bool = False):
|
||||
"""
|
||||
Initialize position storage.
|
||||
|
||||
Args:
|
||||
storage_dir: Directory to store position files
|
||||
use_shelf: If True, use Python shelf format; if False, use JSON
|
||||
"""
|
||||
self.storage_dir = Path(storage_dir)
|
||||
self.storage_dir.mkdir(exist_ok=True)
|
||||
self.use_shelf = use_shelf
|
||||
|
||||
def save_position(self, document_id: str, position_name: str, position: RecursivePosition):
|
||||
"""Save a position to storage"""
|
||||
if self.use_shelf:
|
||||
self._save_to_shelf(document_id, position_name, position)
|
||||
else:
|
||||
self._save_to_json(document_id, position_name, position)
|
||||
|
||||
def load_position(self, document_id: str, position_name: str) -> Optional[RecursivePosition]:
|
||||
"""Load a position from storage"""
|
||||
if self.use_shelf:
|
||||
return self._load_from_shelf(document_id, position_name)
|
||||
else:
|
||||
return self._load_from_json(document_id, position_name)
|
||||
|
||||
def list_positions(self, document_id: str) -> List[str]:
|
||||
"""List all saved positions for a document"""
|
||||
if self.use_shelf:
|
||||
return self._list_shelf_positions(document_id)
|
||||
else:
|
||||
return self._list_json_positions(document_id)
|
||||
|
||||
def delete_position(self, document_id: str, position_name: str) -> bool:
|
||||
"""Delete a position from storage"""
|
||||
if self.use_shelf:
|
||||
return self._delete_from_shelf(document_id, position_name)
|
||||
else:
|
||||
return self._delete_from_json(document_id, position_name)
|
||||
|
||||
def _save_to_json(self, document_id: str, position_name: str, position: RecursivePosition):
|
||||
"""Save position as JSON file"""
|
||||
file_path = self.storage_dir / f"{document_id}_{position_name}.json"
|
||||
with open(file_path, 'w') as f:
|
||||
json.dump(position.to_dict(), f, indent=2)
|
||||
|
||||
def _load_from_json(self, document_id: str, position_name: str) -> Optional[RecursivePosition]:
|
||||
"""Load position from JSON file"""
|
||||
file_path = self.storage_dir / f"{document_id}_{position_name}.json"
|
||||
if not file_path.exists():
|
||||
return None
|
||||
|
||||
try:
|
||||
with open(file_path, 'r') as f:
|
||||
data = json.load(f)
|
||||
return RecursivePosition.from_dict(data)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def _list_json_positions(self, document_id: str) -> List[str]:
|
||||
"""List JSON position files for a document"""
|
||||
pattern = f"{document_id}_*.json"
|
||||
files = list(self.storage_dir.glob(pattern))
|
||||
return [f.stem.replace(f"{document_id}_", "") for f in files]
|
||||
|
||||
def _delete_from_json(self, document_id: str, position_name: str) -> bool:
|
||||
"""Delete JSON position file"""
|
||||
file_path = self.storage_dir / f"{document_id}_{position_name}.json"
|
||||
if file_path.exists():
|
||||
file_path.unlink()
|
||||
return True
|
||||
return False
|
||||
|
||||
def _save_to_shelf(self, document_id: str, position_name: str, position: RecursivePosition):
|
||||
"""Save position to shelf database"""
|
||||
shelf_path = str(self.storage_dir / f"{document_id}.shelf")
|
||||
with shelve.open(shelf_path) as shelf:
|
||||
shelf[position_name] = position
|
||||
|
||||
def _load_from_shelf(self, document_id: str, position_name: str) -> Optional[RecursivePosition]:
|
||||
"""Load position from shelf database"""
|
||||
shelf_path = str(self.storage_dir / f"{document_id}.shelf")
|
||||
try:
|
||||
with shelve.open(shelf_path) as shelf:
|
||||
return shelf.get(position_name)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def _list_shelf_positions(self, document_id: str) -> List[str]:
|
||||
"""List positions in shelf database"""
|
||||
shelf_path = str(self.storage_dir / f"{document_id}.shelf")
|
||||
try:
|
||||
with shelve.open(shelf_path) as shelf:
|
||||
return list(shelf.keys())
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
def _delete_from_shelf(self, document_id: str, position_name: str) -> bool:
|
||||
"""Delete position from shelf database"""
|
||||
shelf_path = str(self.storage_dir / f"{document_id}.shelf")
|
||||
try:
|
||||
with shelve.open(shelf_path) as shelf:
|
||||
if position_name in shelf:
|
||||
del shelf[position_name]
|
||||
return True
|
||||
except Exception:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
# Convenience functions for common position patterns
|
||||
def create_word_position(chapter: int, block: int, word: int, char_offset: int = 0) -> RecursivePosition:
|
||||
"""Create a position pointing to a specific word and character"""
|
||||
return (PositionBuilder()
|
||||
.chapter(chapter)
|
||||
.block(block)
|
||||
.paragraph()
|
||||
.word(word, offset=char_offset)
|
||||
.build())
|
||||
|
||||
|
||||
def create_image_position(chapter: int, block: int, image_index: int = 0) -> RecursivePosition:
|
||||
"""Create a position pointing to an image"""
|
||||
return (PositionBuilder()
|
||||
.chapter(chapter)
|
||||
.block(block)
|
||||
.image(image_index)
|
||||
.build())
|
||||
|
||||
|
||||
def create_table_cell_position(chapter: int, block: int, row: int, col: int, word: int = 0) -> RecursivePosition:
|
||||
"""Create a position pointing to content in a table cell"""
|
||||
return (PositionBuilder()
|
||||
.chapter(chapter)
|
||||
.block(block)
|
||||
.table()
|
||||
.table_row(row)
|
||||
.table_cell(col)
|
||||
.word(word)
|
||||
.build())
|
||||
|
||||
|
||||
def create_list_item_position(chapter: int, block: int, item: int, word: int = 0) -> RecursivePosition:
|
||||
"""Create a position pointing to content in a list item"""
|
||||
return (PositionBuilder()
|
||||
.chapter(chapter)
|
||||
.block(block)
|
||||
.list()
|
||||
.list_item(item)
|
||||
.word(word)
|
||||
.build())
|
||||
Reference in New Issue
Block a user