zenfyr.dev
social media crossposting tool. 3rd time's the charm
mastodon
misskey
crossposting
bluesky
1import logging
2import os
3import sys
4from collections.abc import Callable
5from typing import Any, cast
6
7import env
8
9
10shutdown_hook: list[Callable[[], None]] = []
11
12logging.basicConfig(stream=sys.stderr, level=logging.DEBUG if env.DEV else logging.INFO)
13logging.getLogger("httpx").setLevel(logging.WARNING)
14logging.getLogger("httpcore").setLevel(logging.WARNING)
15LOGGER = logging.getLogger("XPost")
16
17
18class Result[V, E]:
19 _value: V
20 _error: E | None
21
22 def __init__(self, value: V, err: E) -> None:
23 self._value = value
24 self._error = err
25
26 def error(self) -> E:
27 if self._error is None:
28 raise ValueError("self._error not set!")
29 return self._error
30
31 def value(self) -> V:
32 return self._value
33
34 def is_ok(self) -> bool:
35 return self._error is None
36
37 @classmethod
38 def err(cls, err: E) -> "Result[V, E]":
39 return cast("Result[V, E]", Result(None, err))
40
41 @classmethod
42 def ok(cls, val: V) -> "Result[V, E]":
43 return cast("Result[V, E]", Result(val, None))
44
45
46def normalize_service_url(url: str) -> str:
47 if not url.startswith("https://") and not url.startswith("http://"):
48 raise ValueError(f"Invalid service url {url}! Must start with http/https!")
49
50 return url[:-1] if url.endswith("/") else url
51
52
53def _read_env(data: Any) -> None:
54 match data:
55 case dict():
56 read_env(data)
57 case list():
58 for v in data:
59 _read_env(v)
60 case _:
61 pass
62
63
64def read_env(data: dict[str, Any]) -> None:
65 keys = list(data.keys())
66 for key in keys:
67 val = data[key]
68 match val:
69 case str():
70 if val.startswith("env:"):
71 envval = os.environ.get(val[4:])
72 if envval is None:
73 del data[key]
74 else:
75 data[key] = envval
76 case _:
77 _read_env(val)