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
|
import json
import os
import threading
import urllib.error
import urllib.request
from pathlib import Path
ALAP = os.environ.get("RAVEOS_JEGYEK_URL", "https://jegyek.rp1.hu/api/v1")
KONFIG = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) / "raveos-center"
KULCSFAJL = KONFIG / "kliens.json"
class ApiHiba(Exception):
def __init__(self, kod, uzenet):
super().__init__(uzenet)
self.kod = kod
def _keres(modszer, ut, adat=None, nyers=None, tipus="application/json", kulcs=None, ido=30):
fejlec = {"User-Agent": "raveos-center"}
torzs = None
if adat is not None:
torzs = json.dumps(adat).encode()
fejlec["Content-Type"] = "application/json"
elif nyers is not None:
torzs = nyers
fejlec["Content-Type"] = tipus
if kulcs:
fejlec["Authorization"] = f"Bearer {kulcs}"
keres = urllib.request.Request(ALAP + ut, data=torzs, headers=fejlec, method=modszer)
try:
with urllib.request.urlopen(keres, timeout=ido) as valasz:
return json.loads(valasz.read() or b"{}")
except urllib.error.HTTPError as e:
try:
uzenet = json.loads(e.read()).get("hiba", str(e))
except ValueError:
uzenet = str(e)
raise ApiHiba(e.code, uzenet) from None
except (urllib.error.URLError, OSError) as e:
raise ApiHiba(0, "nincs_kapcsolat") from e
_kulcs_zar = threading.Lock()
def kulcs():
with _kulcs_zar:
try:
return json.loads(KULCSFAJL.read_text())["kulcs"]
except (OSError, ValueError, KeyError):
pass
uj = _keres("POST", "/kliens")["kulcs"]
KONFIG.mkdir(parents=True, exist_ok=True)
os.chmod(KONFIG, 0o700)
ideiglenes = KULCSFAJL.with_suffix(".uj")
fd = os.open(ideiglenes, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(fd, "w") as f:
json.dump({"kulcs": uj}, f)
os.replace(ideiglenes, KULCSFAJL)
return uj
def _hitelesitett(modszer, ut, **kw):
try:
return _keres(modszer, ut, kulcs=kulcs(), **kw)
except ApiHiba as e:
if e.kod != 401:
raise
with _kulcs_zar:
KULCSFAJL.unlink(missing_ok=True)
return _keres(modszer, ut, kulcs=kulcs(), **kw)
def jegyeim():
return _hitelesitett("GET", "/jegyek")["jegyek"]
def jegy(jegy_id):
return _hitelesitett("GET", f"/jegyek/{jegy_id}")
def uj_jegy(tipus, cim, szoveg, rendszer=None, naplo=None):
adat = {"tipus": tipus, "cim": cim, "szoveg": szoveg}
if rendszer:
adat["rendszer"] = rendszer
if naplo:
adat["naplo"] = naplo
return _hitelesitett("POST", "/jegyek", adat=adat)
def valasz(jegy_id, szoveg):
return _hitelesitett("POST", f"/jegyek/{jegy_id}/uzenetek", adat={"szoveg": szoveg})
def otletek():
return _hitelesitett("GET", "/otletek")["otletek"]
def szavaz(jegy_id, tamogat):
return _hitelesitett("PUT" if tamogat else "DELETE", f"/otletek/{jegy_id}/szavazat")
def ertekeles():
return _hitelesitett("GET", "/ertekeles")
def ertekel(csillag, szoveg):
return _hitelesitett("PUT", "/ertekeles", adat={"csillag": csillag, "szoveg": szoveg})
def allapot():
return _hitelesitett("GET", "/allapot")
def naplo_feltolt(szoveg):
return _hitelesitett(
"POST", "/naplok?forras=center", nyers=szoveg.encode(), tipus="text/plain; charset=utf-8", ido=60
)["azonosito"]
HIBAUZENETEK = {
"nincs_kapcsolat": "Nem sikerült elérni a szervert. Ellenőrizd az internetkapcsolatot.",
"tul_sok_keres": "Túl sok kérés rövid idő alatt. Próbáld újra kicsit később.",
"lezart_jegy": "Ez a jegy le van zárva, már nem lehet rá válaszolni.",
"nincs_jegy": "Ez a jegy már nem létezik.",
"nincs_ilyen_naplo": "A megadott naplóazonosító nem létezik.",
"kitiltva": "Ez a gép ki van tiltva a Visszajelzés használatából.",
}
def emberi(hiba):
if isinstance(hiba, ApiHiba):
return HIBAUZENETEK.get(str(hiba), f"A szerver hibát jelzett ({hiba}).")
return str(hiba)
|