Source code for ts_ids_core.mapping_table.sync

from pathlib import Path
from typing import List, TextIO, Union

from jsonref import replace_refs

from ts_ids_core.mapping_table.convert import (
    json_string_to_dict,
    mapping_table_to_markdown_lines,
)
from ts_ids_core.mapping_table.locate import get_table_indexes
from ts_ids_core.mapping_table.mapping_table import (
    MappingTable,
    sync_mapping_table_with_schema,
)




def create_mapping_table_json(
    schema_file: Union[Path, TextIO, dict] = Path("schema.json"),
    mapping_table_path: Path = Path("mapping_table.json"),
    default_record_visibility: bool = True,
) -> str:
    """Create the mapping table json for the field paths in schema.json
    Steps
        - Read the field paths from the schema
        - Create mapping for the old records in the mapping table: dict[path, element]
        - For each property, generate a template record
        - Transfer data from old record to new record, if the old exists
    """
    if isinstance(schema_file, dict):
        schema = replace_refs(schema_file)
    else:
        if isinstance(schema_file, Path):
            schema_string = schema_file.read_text(encoding="utf-8")
        else:
            schema_string = schema_file.read()
        schema = json_string_to_dict(schema_string)

    if mapping_table_path.is_file():
        original_mapping_table = MappingTable.model_validate_json(
            mapping_table_path.read_text(encoding="utf-8")
        )
    else:
        original_mapping_table = MappingTable()

    return sync_mapping_table_with_schema(
        schema, original_mapping_table, default_record_visibility
    ).model_dump_json(indent=2)





def create_readme(
    readme_path: Path,
    mapping_table_path: Path,
    mapping_table_section_heading: str = "Raw to IDS Mapping",
) -> str:
    """Insert the mapping table into the README
    Steps:
        - Generate a markdown table from the mapping_table.json
        - Find the existing mapping table in the README
        - Remove the existing table
        - Insert the new table
    """
    mapping_table = MappingTable.model_validate_json(
        mapping_table_path.read_text(encoding="utf-8")
    )

    readme_lines = readme_path.read_text(encoding="utf-8").splitlines(keepends=True)

    return "".join(
        replace_mapping_table_in_readme_lines(
            readme_lines, mapping_table, mapping_table_section_heading
        )
    )





def replace_mapping_table_in_readme_lines(
    readme_lines: List[str],
    mapping_table_json: MappingTable,
    mapping_table_section_heading: str = "Raw to IDS Mapping",
) -> List[str]:
    """Replace the mapping table lines in the readme with lines generated from
    a MappingTable instance.
    """

    mapping_table_lines = mapping_table_to_markdown_lines(mapping_table_json)

    header_line_index, body_end_index = get_table_indexes(
        readme_lines, mapping_table_section_heading
    )

    readme_lines_with_new_table = (
        readme_lines[:header_line_index]
        + mapping_table_lines
        + readme_lines[body_end_index:]
    )
    return readme_lines_with_new_table