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
|
"""スタジオ空き枠一覧のローカルサーバー。
python3 server.py # http://localhost:8000
python3 server.py 8080 # ポート指定
HOST=0.0.0.0 python3 server.py # コンテナ内など外から受ける場合
リンキィディンクの取得には RINKY_ID / RINKY_PASSWORD が要る（.env に書けば読み込む）。
"""
import json
import os
import sys
from concurrent.futures import ThreadPoolExecutor
from datetime import date, timedelta
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, urlparse
import providers
ROOT = Path(__file__).resolve().parent
STATIC = ROOT / "static"
MAX_DAYS_AHEAD = 92 # リンキィディンクは 3 ヶ月先まで予約可
def load_env():
env = ROOT / ".env"
if not env.exists():
return
for line in env.read_text(encoding="utf-8").splitlines():
if "=" in line and not line.lstrip().startswith("#"):
key, value = line.split("=", 1)
os.environ.setdefault(key.strip(), value.strip())
def load_config():
return json.loads((ROOT / "studios.json").read_text(encoding="utf-8"))
def studio_info(studio_id, studio):
return {
"id": studio_id,
"name": studio["name"],
"area": studio["area"],
"url": studio["url"],
"phone": studio["phone"],
"booking": providers.booking_url(studio),
"online": bool(studio.get("provider")),
}
def slots_for(set_id, day):
config = load_config()
target = next((s for s in config["sets"] if s["id"] == set_id), None)
if not target:
return None
def one(studio_id):
studio = config["studios"][studio_id]
info = studio_info(studio_id, studio)
try:
rooms, fetched_at = providers.fetch(studio_id, studio, day)
return {**info, "rooms": rooms, "fetched_at": fetched_at, "error": None}
except Exception as e: # 1軒の取得失敗で一覧全体を落とさない
return {**info, "rooms": None, "fetched_at": None, "error": str(e)}
with ThreadPoolExecutor(max_workers=4) as pool:
studios = list(pool.map(one, target["studios"]))
return {"set": target["id"], "date": day.isoformat(), "studios": studios}
class Handler(SimpleHTTPRequestHandler):
def __init__(self, *args, **kwargs):
super().__init__(*args, directory=str(STATIC), **kwargs)
def do_GET(self):
url = urlparse(self.path)
qs = parse_qs(url.query)
if url.path == "/healthz": # 死活監視用。Caddy の Basic 認証の対象外
return self.send_json({"ok": True})
if url.path == "/api/sets":
config = load_config()
return self.send_json([{"id": s["id"], "name": s["name"]} for s in config["sets"]])
if url.path != "/api/slots":
return super().do_GET()
try:
day = date.fromisoformat(qs.get("date", [date.today().isoformat()])[0])
except ValueError:
return self.send_json({"error": "date は YYYY-MM-DD で指定してください"}, 400)
if not date.today() - timedelta(days=1) <= day <= date.today() + timedelta(days=MAX_DAYS_AHEAD):
return self.send_json({"error": f"date は今日から {MAX_DAYS_AHEAD} 日先までです"}, 400)
body = slots_for(qs.get("set", [""])[0], day)
if body is None:
return self.send_json({"error": "set が見つかりません"}, 404)
self.send_json(body)
def send_json(self, body, status=200):
data = json.dumps(body, ensure_ascii=False).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
if __name__ == "__main__":
load_env()
port = int(sys.argv[1]) if len(sys.argv) > 1 else 8000
host = os.environ.get("HOST", "127.0.0.1")
print(f"http://{host}:{port}", flush=True)
ThreadingHTTPServer((host, port), Handler).serve_forever()
|
