|
7 | 7 |
|
8 | 8 | from . import const, grammar |
9 | 9 | from .exceptions import GedcomSerializeError |
| 10 | +from .types import GedcomStructure |
10 | 11 |
|
11 | 12 | if TYPE_CHECKING: |
12 | 13 | from collections.abc import Iterable |
13 | 14 | from typing import BinaryIO |
14 | 15 |
|
15 | | - from .types import GedcomStructure |
16 | | - |
17 | 16 | _EOL = re.compile(r"\r\n|\r|\n") |
18 | 17 | _TAG = re.compile(grammar.tag) |
19 | 18 | _XREF = re.compile(grammar.xref) |
20 | 19 | _POINTER = re.compile(grammar.pointer) |
21 | 20 | _BANNED = re.compile(grammar.banned) |
22 | 21 | _TAGDEF = re.compile(grammar.tagdef) |
| 22 | +_EXTTAG = re.compile(grammar.exttag) |
23 | 23 |
|
24 | 24 | _BOM = "\ufeff" |
25 | 25 |
|
@@ -52,6 +52,97 @@ def _schema(records: Iterable[GedcomStructure]) -> dict[str, str]: |
52 | 52 | return uris |
53 | 53 |
|
54 | 54 |
|
| 55 | +def _extension_tag(uri: str, taken: set[str]) -> str: |
| 56 | + """Invent an extension tag for a URI, avoiding every tag already spoken for. |
| 57 | +
|
| 58 | + The last path segment or fragment of the URI usually reads as a name, so it |
| 59 | + is the basis for the tag; a URI that yields nothing usable falls back to a |
| 60 | + plain counter. A tag already in use is stepped past rather than reused, since |
| 61 | + a tag standing for two things resolves to neither on the way back in. |
| 62 | + """ |
| 63 | + segment = re.split(r"[/#]", uri)[-1] |
| 64 | + base = "_" + re.sub(r"[^A-Z0-9_]", "", segment.upper()) |
| 65 | + if _EXTTAG.fullmatch(base) is None: |
| 66 | + base = "_EXT" |
| 67 | + candidate, suffix = base, 1 |
| 68 | + while candidate in taken: |
| 69 | + suffix += 1 |
| 70 | + candidate = f"{base}{suffix}" |
| 71 | + return candidate |
| 72 | + |
| 73 | + |
| 74 | +def generate_schema(records: Iterable[GedcomStructure]) -> None: |
| 75 | + """Add to HEAD the schema declarations dumps needs to write extension tags. |
| 76 | +
|
| 77 | + Takes a whole dataset, HEAD included, and modifies its structures in place:: |
| 78 | +
|
| 79 | + records = gedcom7.loads(text) |
| 80 | + gedcom7.generate_schema(records) # records[0] gains HEAD.SCHMA.TAG |
| 81 | + gedcom7.dumps(records) |
| 82 | +
|
| 83 | + A tag held as a URI cannot be written until the header abbreviates it. Only |
| 84 | + HEAD changes; the URIs stay on their own structures for :func:`dumps` to |
| 85 | + substitute as it writes. Declarations already there are kept, so a second |
| 86 | + call does nothing. |
| 87 | +
|
| 88 | + Raises :class:`~gedcom7.exceptions.GedcomSerializeError` if there is no HEAD, |
| 89 | + or if a tag is neither writable nor a URI that can be abbreviated. |
| 90 | + """ |
| 91 | + # Read once: the records are walked several times below, and a caller passing |
| 92 | + # a one-shot iterator would otherwise have them silently disappear part way. |
| 93 | + records = list(records) |
| 94 | + head = next((record for record in records if record.tag == const.HEAD), None) |
| 95 | + if head is None: |
| 96 | + raise GedcomSerializeError( |
| 97 | + "a schema is declared in the HEAD pseudo-structure, and these " |
| 98 | + "records do not have one" |
| 99 | + ) |
| 100 | + |
| 101 | + declared = _schema(records) |
| 102 | + # Every literal tag in the document is spoken for as well: were a generated |
| 103 | + # tag to collide with one, that structure would resolve to the extension's |
| 104 | + # URI when the stream was read back. |
| 105 | + taken = set(declared.values()) |
| 106 | + undeclared: dict[str, None] = {} |
| 107 | + |
| 108 | + def visit(structure: GedcomStructure) -> None: |
| 109 | + if _TAG.fullmatch(structure.tag) is None: |
| 110 | + if structure.tag not in declared: |
| 111 | + undeclared.setdefault(structure.tag, None) |
| 112 | + else: |
| 113 | + taken.add(structure.tag) |
| 114 | + for child in structure.children: |
| 115 | + visit(child) |
| 116 | + |
| 117 | + for record in records: |
| 118 | + visit(record) |
| 119 | + |
| 120 | + if not undeclared: |
| 121 | + return |
| 122 | + |
| 123 | + schema = next((c for c in head.children if c.tag == const.SCHMA), None) |
| 124 | + if schema is None: |
| 125 | + schema = GedcomStructure(tag=const.SCHMA) |
| 126 | + schema.parent = head |
| 127 | + gedc = next( |
| 128 | + (i for i, c in enumerate(head.children) if c.tag == const.GEDC), None |
| 129 | + ) |
| 130 | + head.children.insert(0 if gedc is None else gedc + 1, schema) |
| 131 | + |
| 132 | + # One tag per URI and one URI per tag: dumps picks arbitrarily between two |
| 133 | + # tags for a URI, and the parser refuses to resolve a tag that maps to two. |
| 134 | + for uri in undeclared: |
| 135 | + tag = _extension_tag(uri, taken) |
| 136 | + taken.add(tag) |
| 137 | + definition = f"{tag} {uri}" |
| 138 | + if _TAGDEF.fullmatch(definition) is None: |
| 139 | + raise GedcomSerializeError( |
| 140 | + f"{uri!r} cannot be declared in a schema: it is neither a tag " |
| 141 | + "this serializer can write nor a URI it can abbreviate" |
| 142 | + ) |
| 143 | + schema.append_child(GedcomStructure(tag=const.TAG, text=definition)) |
| 144 | + |
| 145 | + |
55 | 146 | def _lines( |
56 | 147 | structure: GedcomStructure, level: int, uris: dict[str, str] |
57 | 148 | ) -> Iterable[str]: |
|
0 commit comments