1#!/usr/bin/env python3
2"""
3
4"""
5
6# Imports:
7from __future__ import annotations
8
9# ##-- stdlib imports
10import atexit# for @atexit.register
11import collections
12import contextlib
13import datetime
14import enum
15import faulthandler
16import functools as ftz
17import hashlib
18import itertools as itz
19import logging as logmod
20import pathlib as pl
21import re
22import time
23from copy import deepcopy
24from uuid import UUID, uuid1
25from weakref import ref
26
27# ##-- end stdlib imports
28
29# ##-- 3rd party imports
30import bibtexparser
31import bibtexparser.model as model
32from bibtexparser import exceptions as bexp
33from bibtexparser import middlewares as ms
34from bibtexparser.model import MiddlewareErrorBlock
35
36# ##-- end 3rd party imports
37
38# ##-- 1st party imports
39from bibble import _interface as API
40import bibble.model as bmodel
41
42# ##-- end 1st party imports
43
44# ##-- types
45# isort: off
46# General
47import abc
48import collections.abc
49import typing
50import types
51from typing import cast, assert_type, assert_never
52from typing import Generic, NewType, Never
53from typing import no_type_check, final, override, overload
54# Protocols and Interfaces:
55from typing import Protocol, runtime_checkable
56# isort: on
57# ##-- end types
58
59# ##-- type checking
60# isort: off
61if typing.TYPE_CHECKING:
62 from typing import Final, ClassVar, Any, Self
63 from typing import Literal, LiteralString
64 from typing import TypeGuard
65 from collections.abc import Iterable, Iterator, Callable, Generator
66 from collections.abc import Sequence, Mapping, MutableMapping, Hashable
67
68 from jgdv import Maybe, Either, Result, Rx
69 type Middleware = API.Middleware_p | API.BidirectionalMiddleware_p
70
71## isort: on
72# ##-- end type checking
73
74
75##-- logging
76logging = logmod.getLogger(__name__)
77##-- end logging
78
79# Vars:
80EMPTY : Final[Rx] = re.compile(r"^$")
81OR_REG : Final[str] = r"|"
82ANY_REG : Final[Rx] = re.compile(r".")
83# Body:
84
[docs]
85class MiddlewareValidator_m:
86 """ For ensuring the middlewares of a reader/writer are appropriate,
87 by excluding certain middlewares.
88 """
89 _middlewares : list[Middleware]
90
[docs]
91 def exclude_middlewares(self, proto:type):
92 if not issubclass(proto, Protocol):
93 raise TypeError("Tried to validate middlewares with a non-protocol", proto)
94 failures = []
95 for middle in self._middlewares:
96 # note: no 'not'.
97 if isinstance(middle, proto):
98 failures.append(middle)
99 elif not isinstance(middle, API.Middleware_p|API.BidirectionalMiddleware_p):
100 failures.append(middle)
101 else:
102 if bool(failures):
103 raise TypeError("Bad middlewares", failures)
104
[docs]
105class ErrorRaiser_m:
106 """ Mixin for easily combining middleware errors into a block"""
107
[docs]
108 def make_error_block(self, entry:API.Entry, err:Exception) -> MiddlewareErrorBlock:
109 return bmodel.FailedBlock(block=entry, error=err, source=self)
110
[docs]
111class FieldMatcher_m:
112 """ Mixin to process fields if their key matchs a regex
113 defaults are in the attrs _field_blacklist and _field_whitelist, _entry_whitelist
114 Call set_field_matchers to extend.
115 Call match_on_fields to start.
116 Call maybe_skip_entry to compare the lowercase entry type to a whitelist
117 Implement field_handler to use.
118
119 match_on_fields calls entry.set_field on the field_handlers result
120 """
121
[docs]
122 def set_field_matchers(self, *, white:list[str], black:list[str]) -> Self:
123 """ sets the blacklist and whitelist regex's
124 returns self to help in building parse stacks
125 """
126 match white:
127 case []:
128 self._field_white_re = ANY_REG
129 case [*xs]:
130 self._field_white_re = re.compile(OR_REG.join(xs))
131
132 match black:
133 case []:
134 self._field_black_re = EMPTY
135 case [*xs]:
136 self._field_black_re = re.compile(OR_REG.join(xs))
137
138 return self
139
[docs]
140 def match_on_fields(self, entry: API.Entry, library: API.Library) -> Result[API.Entry, Exception]:
141 errors : list[str] = []
142 whitelist, blacklist = self._field_white_re, self._field_black_re
143 for field in entry.fields:
144 match field:
145 case model.Field(key=str() as key) if whitelist.match(key) and not blacklist.match(key):
146 res = self.field_h(field, entry)
147 case _:
148 continue
149
150 match res:
151 case [*xs]:
152 for x in xs:
153 entry.set_field(x)
154 case Exception() as err:
155 errors += err.args
156 case x:
157 raise TypeError(type(x), self)
158 else:
159 if bool(errors):
160 return ValueError(*errors)
161
162 return entry
163
[docs]
164 def field_h(self, field:API.Field, entry:API.Entry) -> Result[list[API.Field], Exception]:
165 raise NotImplementedError("Implement the field handler")
166
[docs]
167class EntrySkipper_m:
168 """
169 Be able to skip entries by their type
170 """
171
[docs]
172 def set_entry_skiplists(self, *, white:Maybe[list[str]]=None, black:Maybe[list[str]]=None) -> None:
173 self._entry_whitelist = [x.lower() for x in white or []]
174 self._entry_blacklist = [x.lower() for x in black or []]
175
[docs]
176 def should_skip_entry(self, entry:API.Entry, library:API.Library) -> bool:
177 low_type = entry.entry_type.lower()
178 match self._entry_blacklist:
179 case list() as xs if low_type in xs:
180 return True
181 case _:
182 pass
183
184 match self._entry_whitelist:
185 case []:
186 return False
187 case list() as xs if low_type in xs:
188 return False
189 case _:
190 return True