summaryrefslogtreecommitdiff
path: root/server/szurubooru/func/util.py
blob: dc5ced0981395d2e4c77bc1fb2ccb3bfd4e6c667 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
import hashlib
import os
import re
import tempfile
from contextlib import contextmanager
from datetime import datetime, timedelta
from typing import Any, Dict, Generator, List, Optional, Tuple, TypeVar, Union

from szurubooru import errors

T = TypeVar("T")


def snake_case_to_lower_camel_case(text: str) -> str:
    components = text.split("_")
    return components[0].lower() + "".join(
        word[0].upper() + word[1:].lower() for word in components[1:]
    )


def snake_case_to_upper_train_case(text: str) -> str:
    return "-".join(
        word[0].upper() + word[1:].lower() for word in text.split("_")
    )


def snake_case_to_lower_camel_case_keys(
    source: Dict[str, Any]
) -> Dict[str, Any]:
    target = {}
    for key, value in source.items():
        target[snake_case_to_lower_camel_case(key)] = value
    return target


@contextmanager
def create_temp_file(**kwargs: Any) -> Generator:
    (descriptor, path) = tempfile.mkstemp(**kwargs)
    os.close(descriptor)
    try:
        with open(path, "r+b") as handle:
            yield handle
    finally:
        os.remove(path)


@contextmanager
def create_temp_file_path(**kwargs: Any) -> Generator:
    (descriptor, path) = tempfile.mkstemp(**kwargs)
    os.close(descriptor)
    try:
        yield path
    finally:
        os.remove(path)


def unalias_dict(source: List[Tuple[List[str], T]]) -> Dict[str, T]:
    output_dict = {}  # type: Dict[str, T]
    for aliases, value in source:
        for alias in aliases:
            output_dict[alias] = value
    return output_dict


def flatten_list(source: List[List[T]]) -> List[T]:
    return [item for sublist in source for item in sublist]


def get_md5(source: Union[str, bytes]) -> str:
    if not isinstance(source, bytes):
        source = source.encode("utf-8")
    md5 = hashlib.md5()
    md5.update(source)
    return md5.hexdigest()


def get_sha1(source: Union[str, bytes]) -> str:
    if not isinstance(source, bytes):
        source = source.encode("utf-8")
    sha1 = hashlib.sha1()
    sha1.update(source)
    return sha1.hexdigest()


def flip(source: Dict[Any, Any]) -> Dict[Any, Any]:
    return {v: k for k, v in source.items()}


def is_valid_email(email: Optional[str]) -> bool:
    """Return whether given email address is valid or empty."""
    return not email or re.match(r"^[^@]*@[^@]*\.[^@]*$", email) is not None


class dotdict(dict):
    """dot.notation access to dictionary attributes."""

    def __getattr__(self, attr: str) -> Any:
        return self.get(attr)

    __setattr__ = dict.__setitem__
    __delattr__ = dict.__delitem__


def parse_time_range(value: str) -> Tuple[datetime, datetime]:
    """Return tuple containing min/max time for given text representation."""
    one_day = timedelta(days=1)
    one_second = timedelta(seconds=1)
    almost_one_day = one_day - one_second

    value = value.lower()
    if not value:
        raise errors.ValidationError("Empty date format.")

    if value == "today":
        now = datetime.utcnow()
        return (
            datetime(now.year, now.month, now.day, 0, 0, 0),
            datetime(now.year, now.month, now.day, 0, 0, 0) + almost_one_day,
        )

    if value == "yesterday":
        now = datetime.utcnow()
        return (
            datetime(now.year, now.month, now.day, 0, 0, 0) - one_day,
            datetime(now.year, now.month, now.day, 0, 0, 0) - one_second,
        )

    match = re.match(r"^(\d{4})$", value)
    if match:
        year = int(match.group(1))
        return (datetime(year, 1, 1), datetime(year + 1, 1, 1) - one_second)

    match = re.match(r"^(\d{4})-(\d{1,2})$", value)
    if match:
        year = int(match.group(1))
        month = int(match.group(2))
        return (
            datetime(year, month, 1),
            datetime(year, month + 1, 1) - one_second,
        )

    match = re.match(r"^(\d{4})-(\d{1,2})-(\d{1,2})$", value)
    if match:
        year = int(match.group(1))
        month = int(match.group(2))
        day = int(match.group(3))
        return (
            datetime(year, month, day),
            datetime(year, month, day + 1) - one_second,
        )

    raise errors.ValidationError("Invalid date format: %r." % value)


def icase_unique(source: List[str]) -> List[str]:
    target = []  # type: List[str]
    target_low = []  # type: List[str]
    for source_item in source:
        if source_item.lower() not in target_low:
            target.append(source_item)
            target_low.append(source_item.lower())
    return target


def value_exceeds_column_size(value: Optional[str], column: Any) -> bool:
    if not value:
        return False
    max_length = column.property.columns[0].type.length
    if max_length is None:
        return False
    return len(value) > max_length


def get_column_size(column: Any) -> Optional[int]:
    if not column:
        return None
    return column.property.columns[0].type.length


def chunks(source_list: List[Any], part_size: int) -> Generator:
    for i in range(0, len(source_list), part_size):
        yield source_list[i : i + part_size]