Skip to main content

ALDashboard.docx_wrangling

copy​

docx​

io​

sys​

defaultdict​

ThreadPoolExecutor​

urlsplit​

tiktoken​

json​

OxmlElement​

qn​

re​

chat_completion​

Any​

Dict​

List​

Tuple​

Optional​

Union​

Sequence​

get_docx_prompt_profile​

analyze_docx_template_markup​

get_jinja_template_validation​

__all__​

DEFAULT_DOCX_PROMPT_PROFILE​

DEFAULT_TEMPLATE_HIGHLIGHT_FILL​

NESTED_IF_TEMPLATE_HIGHLIGHT_FILLS​

_JINJA_TAG_PATTERN​

_DOCXTPL_SPECIAL_PREFIX_PATTERN​

_IF_OPEN_PATTERN​

_IF_BRANCH_PATTERN​

_IF_CLOSE_PATTERN​

_XML_INVALID_CONTROL_CHARS​

_sanitize_xml_text​

def _sanitize_xml_text(text: str) -> str

Remove control characters that are invalid in XML 1.0 to prevent lxml ValueError.

_get_docx_label_role_description​

def _get_docx_label_role_description(
        *,
        prompt_profile: str = DEFAULT_DOCX_PROMPT_PROFILE,
        custom_prompt: Optional[str] = None,
        prompt_library_path: Optional[str] = None) -> str

Resolve the DOCX labeler role description for the active prompt profile.

Arguments​

  • prompt_profile - Prompt profile name to resolve.
  • custom_prompt - Optional caller-supplied prompt override.
  • prompt_library_path - Optional prompt library override path.

Returns​

  • str - The role description sent to the labeling model.

_get_docx_label_rules_addendum​

def _get_docx_label_rules_addendum(
        *,
        prompt_profile: str = DEFAULT_DOCX_PROMPT_PROFILE,
        prompt_library_path: Optional[str] = None,
        primary_person_variable: Optional[str] = None) -> str

Resolve extra prompt rules for the active DOCX prompt profile.

Arguments​

  • prompt_profile - Prompt profile name to resolve.
  • prompt_library_path - Optional prompt library override path.

Returns​

  • str - Additional labeling rules appended to the base prompt.

_get_docx_label_temperature​

def _get_docx_label_temperature(
        *,
        prompt_profile: str = DEFAULT_DOCX_PROMPT_PROFILE,
        prompt_library_path: Optional[str] = None) -> float

Resolve the model temperature for the active DOCX prompt profile.

Arguments​

  • prompt_profile - Prompt profile name to resolve.
  • prompt_library_path - Optional prompt library override path.

Returns​

  • float - Temperature value for model calls.

_coerce_modified_run_item​

def _coerce_modified_run_item(
        item: Any) -> Optional[Tuple[int, int, str, int]]

Normalize one model result into (paragraph, run, text, paragraph_delta).

_normalize_modified_runs​

def _normalize_modified_runs(
        modified_runs: Sequence[Any]) -> List[Tuple[int, int, str, int]]

Normalize a heterogeneous suggestion list into docx update tuples.

Arguments​

  • modified_runs - Raw model or API suggestion items.

Returns​

List[Tuple[int, int, str, int]]: Valid normalized run modifications.

_normalize_paragraph_insert_tag​

def _normalize_paragraph_insert_tag(text: str, new_paragraph: int) -> str

Convert paragraph-level control tags into Docassemble {%p ... %} tags.

_normalize_openai_base_url​

def _normalize_openai_base_url(
        openai_base_url: Optional[str]) -> Optional[str]

Ensure Azure resource URLs become SDK-compatible /openai/v1/ bases.

_extract_model_results​

def _extract_model_results(response: Any) -> List[Any]

Extract a best-effort list of run updates from varied model JSON shapes.

_append_text_content​

def _append_text_content(run_element: Any, text: str) -> None

Append text to a w:r element, preserving tabs/newlines in WordprocessingML.

_copy_run_properties​

def _copy_run_properties(source_run: Any) -> Optional[Any]

Clone run properties while clearing existing highlight/shading markup.

_build_run_element​

def _build_run_element(source_run: Any,
                       text: str,
                       *,
                       shading_fill: Optional[str] = None) -> Any

Create a w:r element mirroring a source run's formatting.

_if_highlight_fill_for_depth​

def _if_highlight_fill_for_depth(depth: int) -> str

Return a low-saturation teal-family fill for the nesting depth.

_template_tag_highlight_fill​

def _template_tag_highlight_fill(tag_text: str, if_stack: List[str]) -> str

Choose the highlight color for one Jinja tag and update nesting state.

_split_tag_body_for_highlight​

def _split_tag_body_for_highlight(
        tag_text: str, if_stack: List[str]) -> List[Tuple[str, Optional[str]]]

Split one tag into delimiter, body, and delimiter segments.

_split_template_run_segments​

def _split_template_run_segments(
        text: str, if_stack: List[str]) -> List[Tuple[str, Optional[str]]]

Split one run into literal text and separately highlightable Jinja segments.

_collapse_adjacent_segments​

def _collapse_adjacent_segments(
    segments: Sequence[Tuple[str, Optional[str]]]
) -> List[Tuple[str, Optional[str]]]

Merge neighboring segments that share the same highlight fill.

_is_template_delimiter_segment​

def _is_template_delimiter_segment(segment_text: str) -> bool

Return True for the bare opening or closing delimiter pieces.

_replace_run_with_segments​

def _replace_run_with_segments(
        run: Any, segments: Sequence[Tuple[str, Optional[str]]]) -> None

Replace one run with multiple runs that preserve formatting and split tags.

apply_jinja2_highlights​

def apply_jinja2_highlights(
        document: Union[docx.document.Document,
                        str]) -> docx.document.Document

Split Jinja tags into separate runs and shade the tag body text.

The inner Jinja code is placed in its own run between the opening and closing delimiters so python-docx-template can still consume the rendered text safely. Nested if and %p if control blocks receive depth-based colors.

_SAFE_RUN_CHILD_TAGS​

_run_element_is_safe_to_defragment​

def _run_element_is_safe_to_defragment(run_element: Any) -> bool

Only merge plain-text <w:r> elements that contain no complex XML children.

_run_is_safe_to_defragment​

def _run_is_safe_to_defragment(run: Any) -> bool

Only merge plain-text runs that contain no fields, drawings, or other complex XML.

_collect_paragraphs_from_table​

def _collect_paragraphs_from_table(table: Any, collected: List[Any],
                                   seen_elements: set) -> None

_collect_paragraphs_from_container​

def _collect_paragraphs_from_container(container: Any, collected: List[Any],
                                       seen_elements: set) -> None

Collect unique paragraphs from a container and any nested tables.

Arguments​

  • container - Document, cell, header, or footer-like object to inspect.
  • collected - Output list receiving paragraph objects.
  • seen_elements - Set of paragraph element IDs already collected.

_collect_target_paragraphs​

def _collect_target_paragraphs(document: Any) -> List[Any]

Collect paragraphs from body, tables, headers, and footers.

defragment_docx_runs​

def defragment_docx_runs(
    document: Union[docx.document.Document, str],
    paragraph_numbers: Optional[Sequence[int]] = None
) -> Tuple[docx.document.Document, dict]

Merge text-only runs within target paragraphs.

Only consecutive <w:r> siblings with no intervening non-run elements (e.g. hyperlinks) are merged together. This prevents hyperlinks and other inline elements from being displaced from their original position.

Arguments​

  • document - A loaded python-docx document or a path to a DOCX file.
  • paragraph_numbers - Optional paragraph indexes to limit defragmentation.

Returns​

Tuple[docx.document.Document, dict]: The updated document and summary stats.

_clone_document​

def _clone_document(
        document: docx.document.Document) -> docx.document.Document

Create an in-memory clone of a python-docx document.

Arguments​

  • document - Source document to copy.

Returns​

  • docx.document.Document - A detached copy of the input document.

_contains_template_markup​

def _contains_template_markup(text: str) -> bool

Check whether text contains Jinja-style template delimiters.

Arguments​

  • text - Text to inspect.

Returns​

  • bool - True when template delimiters are present.

_has_balanced_template_delimiters​

def _has_balanced_template_delimiters(text: str) -> bool

Check whether Jinja delimiter pairs are balanced in a text snippet.

Arguments​

  • text - Text to inspect.

Returns​

  • bool - True when opening and closing delimiters are balanced.

_normalize_docassemble_template_source​

def _normalize_docassemble_template_source(text: str) -> str

Convert docassemble paragraph tags into plain Jinja tags for AST parsing.

_classify_if_control_tag​

def _classify_if_control_tag(raw_tag: str) -> Optional[Dict[str, Any]]

_append_validation_flag​

def _append_validation_flag(flags_by_index: Dict[int, List[Dict[str, str]]],
                            index: int, code: str, message: str) -> None

_validate_if_control_tag_parity​

def _validate_if_control_tag_parity(
    suggestions: Sequence[Tuple[int, int, str, int]]
) -> Dict[int, List[Dict[str, str]]]

_should_validate_snippet_as_template​

def _should_validate_snippet_as_template(text: str) -> bool

apply_docx_label_renames​

def apply_docx_label_renames(document: docx.document.Document,
                             renames: Sequence[Dict[str, Any]]) -> int

Apply label find/replace operations across all paragraphs in the DOCX.

validate_docx_template_syntax​

def validate_docx_template_syntax(
        document: Union[docx.document.Document, str],
        *,
        suggestions: Sequence[Any] = (),
        renames: Sequence[Dict[str, Any]] = (),
        defragment_runs: bool = False) -> Dict[str, Any]

Validate the Jinja syntax of a DOCX after simulated edits are applied.

_has_placeholder_markers​

def _has_placeholder_markers(text: str) -> bool

Detect obvious placeholder markers such as long underscores or tab runs.

Arguments​

  • text - Text to inspect.

Returns​

  • bool - True when placeholder markers are present.

_has_adjacent_word_fragments​

def _has_adjacent_word_fragments(text: str) -> bool

Detect template markup that is glued directly to surrounding letters.

Arguments​

  • text - Text to inspect.

Returns​

  • bool - True when markup appears attached to word fragments.

_looks_inline_placeholder​

def _looks_inline_placeholder(source_run_text: str,
                              source_paragraph_text: str) -> bool

Decide whether the source text looks like an inline placeholder target.

Arguments​

  • source_run_text - Original text from the targeted run.
  • source_paragraph_text - Original paragraph text containing the run.

Returns​

  • bool - True when the text should likely be replaced in place.

_filter_noop_suggestions​

def _filter_noop_suggestions(
    document: docx.document.Document, suggestions: Sequence[Tuple[int, int,
                                                                  str, int]]
) -> List[Tuple[int, int, str, int]]

Drop suggestions that leave the targeted run or paragraph unchanged.

_run_has_fragmented_word_boundary​

def _run_has_fragmented_word_boundary(paragraph: Any, run_number: int) -> bool

Check whether a run is embedded inside a split alphabetical word.

Arguments​

  • paragraph - Paragraph containing the run.
  • run_number - Index of the run to inspect.

Returns​

  • bool - True when neighboring runs split a word around the target run.

validate_docx_label_suggestions​

def validate_docx_label_suggestions(
        document: Union[docx.document.Document,
                        str], suggestions: Sequence[Any]) -> Dict[str, Any]

Run deterministic checks over model suggestions and the simulated output.

Arguments​

  • document - A loaded python-docx document or a path to a DOCX file.
  • suggestions - Raw suggestion items returned by labeling helpers or models.

Returns​

Dict[str, Any]: Per-suggestion validation results and aggregate counts.

review_flagged_docx_label_suggestions​

def review_flagged_docx_label_suggestions(
        document: Union[docx.document.Document, str],
        suggestions: Sequence[Any],
        deterministic_validation: Dict[str, Any],
        *,
        openai_client: Optional[Any] = None,
        openai_api: Optional[str] = None,
        openai_base_url: Optional[str] = None,
        model: str = "gpt-5-mini",
        max_output_tokens: Optional[int] = None) -> Dict[str, Any]

Ask an LLM to review only deterministic-validator flagged suggestions.

Arguments​

  • document - A loaded python-docx document or a path to a DOCX file.
  • suggestions - Raw suggestion items returned by a model or API.
  • deterministic_validation - Validator output describing flagged suggestions.
  • openai_client - Optional initialized OpenAI client.
  • openai_api - Optional API key override.
  • openai_base_url - Optional OpenAI-compatible base URL override.
  • model - Model name to use for the review step.
  • max_output_tokens - Optional token limit for the review call.

Returns​

Dict[str, Any]: Review status, any error text, and normalized review items.

_suggestion_confidence_tier​

def _suggestion_confidence_tier(clean_vote_count: int,
                                total_generations: int) -> str

Classify suggestion confidence from clean votes across generations.

Arguments​

  • clean_vote_count - Number of generations that agreed without validation flags.
  • total_generations - Total number of candidate generations considered.

Returns​

  • str - high, medium, or low confidence.

_litigation_template_paragraph_likely_templated​

def _litigation_template_paragraph_likely_templated(text: str) -> bool

Heuristically detect pleading text that likely needs templating.

Arguments​

  • text - Paragraph text to inspect.

Returns​

  • bool - True when the paragraph looks like drafting or placeholder text.

_litigation_alignment_heavy_line​

def _litigation_alignment_heavy_line(text: str) -> bool

Detect caption-style lines where tabs and parentheticals drive layout.

Arguments​

  • text - Paragraph text to inspect.

Returns​

  • bool - True when the line is alignment-heavy.

_candidate_flag_codes​

def _candidate_flag_codes(candidate: Dict[str, Any]) -> set[str]

Extract validation flag codes from an aggregated candidate record.

Arguments​

  • candidate - Candidate suggestion metadata.

Returns​

  • set[str] - Unique flag codes present on the candidate.

_effective_candidate_flags​

def _effective_candidate_flags(
        candidate: Dict[str, Any],
        group: Dict[str, Any],
        *,
        prompt_profile: str = DEFAULT_DOCX_PROMPT_PROFILE
) -> List[Dict[str, Any]]

Adjust candidate validation flags for prompt-profile-specific heuristics.

Arguments​

  • candidate - Candidate suggestion metadata.
  • group - Aggregated source-group metadata.
  • prompt_profile - Prompt profile name driving validation behavior.

Returns​

List[Dict[str, Any]]: The effective validation flags to use for ranking.

_effective_clean_vote_count​

def _effective_clean_vote_count(
        candidate: Dict[str, Any],
        group: Dict[str, Any],
        *,
        prompt_profile: str = DEFAULT_DOCX_PROMPT_PROFILE) -> int

Compute the effective clean-vote count after profile-specific adjustments.

Arguments​

  • candidate - Candidate suggestion metadata.
  • group - Aggregated source-group metadata.
  • prompt_profile - Prompt profile name driving validation behavior.

Returns​

  • int - Effective count of clean votes for candidate ranking.

_candidate_priority_key​

def _candidate_priority_key(
        candidate: Dict[str, Any]) -> Tuple[int, int, int, int]

Build a sort key that favors consensus, cleanliness, and concise labels.

Arguments​

  • candidate - Candidate suggestion metadata.

Returns​

Tuple[int, int, int, int]: Sort key used to rank competing candidates.

review_docx_label_candidate_groups​

def review_docx_label_candidate_groups(
        candidate_groups: Sequence[Dict[str, Any]],
        *,
        openai_client: Optional[Any] = None,
        openai_api: Optional[str] = None,
        openai_base_url: Optional[str] = None,
        model: str = "gpt-5-mini",
        max_output_tokens: Optional[int] = None,
        prompt_profile: str = DEFAULT_DOCX_PROMPT_PROFILE) -> Dict[str, Any]

Ask an LLM judge to choose the best candidate per ambiguous position.

Arguments​

  • candidate_groups - Candidate groups requiring judge review.
  • openai_client - Optional initialized OpenAI client.
  • openai_api - Optional API key override.
  • openai_base_url - Optional OpenAI-compatible base URL override.
  • model - Model name to use for judge review.
  • max_output_tokens - Optional token limit for the judge call.
  • prompt_profile - Prompt profile name used for litigation-specific heuristics.

Returns​

Dict[str, Any]: Judge review status and normalized review items.

aggregate_docx_label_suggestion_runs​

def aggregate_docx_label_suggestion_runs(
        document: Union[docx.document.Document, str],
        suggestion_runs: Sequence[Dict[str, Any]],
        *,
        judge_model: Optional[str] = None,
        openai_client: Optional[Any] = None,
        openai_api: Optional[str] = None,
        openai_base_url: Optional[str] = None,
        judge_max_output_tokens: Optional[int] = None,
        prompt_profile: str = DEFAULT_DOCX_PROMPT_PROFILE) -> Dict[str, Any]

Combine repeated suggestion runs into one ranked set with alternates.

Arguments​

  • document - A loaded python-docx document or a path to a DOCX file.
  • suggestion_runs - Multiple labeling result sets to combine.
  • judge_model - Optional model override for ambiguous-group judging.
  • openai_client - Optional initialized OpenAI client for judge review.
  • openai_api - Optional API key override.
  • openai_base_url - Optional OpenAI-compatible base URL override.
  • judge_max_output_tokens - Optional token limit for judge calls.
  • prompt_profile - Prompt profile name used for ranking heuristics.

Returns​

Dict[str, Any]: Aggregated suggestions, vote data, and judge metadata.

get_voted_docx_label_suggestions​

def get_voted_docx_label_suggestions(
        docx_path: str,
        *,
        custom_people_names: Optional[List[Tuple[str, str]]] = None,
        primary_person_variable: Optional[str] = None,
        preferred_variable_names: Optional[Sequence[str]] = None,
        openai_client: Optional[Any] = None,
        openai_api: Optional[str] = None,
        openai_base_url: Optional[str] = None,
        model: str = "gpt-5-mini",
        generator_models: Optional[Sequence[str]] = None,
        judge_model: Optional[str] = None,
        prompt_profile: str = DEFAULT_DOCX_PROMPT_PROFILE,
        prompt_library_path: Optional[str] = None,
        optional_context: Optional[str] = None,
        custom_prompt: Optional[str] = None,
        additional_instructions: Optional[str] = None,
        max_output_tokens: Optional[int] = None,
        judge_max_output_tokens: Optional[int] = None,
        defragment_runs: bool = False) -> Dict[str, Any]

Run repeated generations and aggregate them into one ranked suggestion set.

Arguments​

  • docx_path - Path to the DOCX file to label.
  • custom_people_names - Optional list of custom people-variable descriptions.
  • primary_person_variable - Optional override for the main 'users' equivalent variable.
  • preferred_variable_names - Optional preferred variable names to bias prompts.
  • openai_client - Optional initialized OpenAI client.
  • openai_api - Optional API key override.
  • openai_base_url - Optional OpenAI-compatible base URL override.
  • model - Default generator model to use.
  • generator_models - Optional explicit sequence of generator models.
  • judge_model - Optional model override for ambiguous-group judging.
  • prompt_profile - Prompt profile name used for prompt/ranking heuristics.
  • prompt_library_path - Optional prompt library override path.
  • optional_context - Optional extra source context for the prompt.
  • custom_prompt - Optional full prompt override.
  • additional_instructions - Optional prompt suffix for extra guidance.
  • max_output_tokens - Optional token limit for generation calls.
  • judge_max_output_tokens - Optional token limit for judge calls.
  • defragment_runs - Whether to merge safe split runs before labeling.

Returns​

Dict[str, Any]: Aggregated suggestions and generation metadata.

_build_paragraph_with_text​

def _build_paragraph_with_text(source_paragraph: Any, text: str) -> Any

Build a new paragraph element that mirrors a source paragraph's style.

Arguments​

  • source_paragraph - Existing paragraph used as a formatting reference.
  • text - Text content for the new paragraph.

Returns​

  • Any - A w:p XML element ready to insert into the document.

add_paragraph_after​

def add_paragraph_after(paragraph: Any, text: str) -> None

Insert a new paragraph after an existing paragraph.

Arguments​

  • paragraph - Existing paragraph that should receive a successor.
  • text - Text content for the inserted paragraph.

add_paragraph_before​

def add_paragraph_before(paragraph: Any, text: str) -> None

Insert a new paragraph before an existing paragraph.

Arguments​

  • paragraph - Existing paragraph that should receive a predecessor.
  • text - Text content for the inserted paragraph.

get_docx_run_text​

def get_docx_run_text(document: Union[docx.document.Document, str],
                      paragraph_number: int, run_number: int) -> str

Get run text by unified paragraph index across body, tables, headers, and footers.

Arguments​

  • document - A loaded python-docx document or a path to a DOCX file.
  • paragraph_number - Unified paragraph index in the flattened traversal.
  • run_number - Run index within the selected paragraph.

Returns​

  • str - The run text, or an empty string when the coordinates are invalid.

get_docx_run_items​

def get_docx_run_items(document: Union[docx.document.Document, str],
                       defragment_runs: bool = False) -> List[List[Any]]

Return [paragraph_index, run_index, run_text] across all document parts.

Arguments​

  • document - A loaded python-docx document or a path to a DOCX file.
  • defragment_runs - Whether to merge safe split runs before traversal.

Returns​

  • List[List[Any]] - Flattened run coordinates and text for the document.

update_docx​

def update_docx(
        document: Union[docx.document.Document, str],
        modified_runs: List[Tuple[int, int, str, int]],
        defragment_runs: bool = False,
        apply_jinja_highlights: bool = False) -> docx.document.Document

Update the document with modified runs.

Arguments​

  • document - The python-docx document object, or the path to the DOCX file.
  • modified_runs - Tuples of paragraph number, run number, modified text, and paragraph insertion indicator.
  • defragment_runs - Whether to merge safe split runs before applying edits.
  • apply_jinja_highlights - Whether to split and shade all Jinja tags after edits are applied.

Returns​

  • docx.document.Document - The modified document.

get_labeled_docx_runs​

def get_labeled_docx_runs(
        docx_path: str,
        *,
        custom_people_names: Optional[List[Tuple[str, str]]] = None,
        primary_person_variable: Optional[str] = None,
        preferred_variable_names: Optional[Sequence[str]] = None,
        openai_client: Optional[Any] = None,
        openai_api: Optional[str] = None,
        openai_base_url: Optional[str] = None,
        model: str = "gpt-5-mini",
        prompt_profile: str = DEFAULT_DOCX_PROMPT_PROFILE,
        prompt_library_path: Optional[str] = None,
        optional_context: Optional[str] = None,
        custom_prompt: Optional[str] = None,
        additional_instructions: Optional[str] = None,
        max_output_tokens: Optional[int] = None,
        defragment_runs: bool = False) -> List[Tuple[int, int, str, int]]

Scan the DOCX and return a list of modified text with Jinja2 variable names inserted.

Arguments​

  • docx_path - Path to the DOCX file.
  • custom_people_names - Optional list of custom (name, description) pairs.
  • primary_person_variable - Optional override for the main 'users' equivalent variable.
  • preferred_variable_names - Optional preferred variable names to bias prompts.
  • openai_client - Optional preconfigured OpenAI client.
  • openai_api - Optional API key override.
  • openai_base_url - Optional OpenAI-compatible base URL override.
  • model - OpenAI model to use.
  • prompt_profile - Prompt profile name used to select prompt text and rules.
  • prompt_library_path - Optional prompt library override path.
  • optional_context - Optional extra source context included in the prompt.
  • custom_prompt - Optional full prompt override.
  • additional_instructions - Optional extra instructions appended to the prompt.
  • max_output_tokens - Optional token limit passed to the chat completion helper.
  • defragment_runs - Whether to merge safe split runs before labeling.

Returns​

List[Tuple[int, int, str, int]]: Suggested DOCX run replacements.

modify_docx_with_openai_guesses​

def modify_docx_with_openai_guesses(docx_path: str) -> docx.document.Document

Uses OpenAI to guess the variable names for a document and then modifies the document with the guesses.

Arguments​

  • docx_path str - Path to the DOCX file to modify.

Returns​

  • docx.Document - The modified document, ready to be saved to the same or a new path