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
|
"""Keep app/converter/config.py and server.json in sync with setup choices."""
import json
import re
import urllib.parse
from pathlib import Path
from typing import Optional
from backends import common
from backends.common import CONFIG_PATH, url_with_port
from converter import config
from . import build
from .constants import FALLBACK_PORT
def config_port() -> int:
"""Return the port of AUDIOCPP_API_URL in app/converter/config.py."""
try:
return urllib.parse.urlsplit(config.AUDIOCPP_API_URL).port or FALLBACK_PORT
except ValueError:
return FALLBACK_PORT
def update_config_api_url_port(port: int, config_path: Optional[Path] = None) -> bool:
"""Rewrite the port inside AUDIOCPP_API_URL in app/converter/config.py.
Reads the configured URL from the file (not from the imported module,
which a long hub session can leave behind), swaps its port for PORT,
and writes it back through ``common.update_config_value`` so the
imported module mirrors the change immediately. Returns True when the
file now holds the new URL.
"""
path = Path(config_path) if config_path is not None else CONFIG_PATH
try:
text = path.read_text(encoding="utf-8")
except OSError:
return False
match = re.search(r'(?m)^\s*AUDIOCPP_API_URL\s*=\s*"([^"]*)"', text)
if not match:
return False
return common.update_config_value("AUDIOCPP_API_URL",
url_with_port(match.group(1), port),
config_path=path)
def update_server_config_port(port: int) -> bool:
"""Rewrite the 'port' in the audio.cpp checkout's server.json.
Loads ``<checkout>/server.json``, sets its ``port`` to PORT, and
rewrites it with the same ``json.dump`` formatting the wizard uses.
Returns True when the file now carries PORT (a no-op when it already
does), and False when there is no checkout/server.json or the file
cannot be read or written.
"""
checkout = build.find_local_checkout()
if checkout is None:
return False
server_json = checkout / "server.json"
if not server_json.exists():
return False
try:
data = json.loads(server_json.read_text(encoding="utf-8"))
except (OSError, ValueError):
return False
if not isinstance(data, dict):
return False
if data.get("port") == port:
return True
data["port"] = port
try:
with server_json.open("w", encoding="utf-8") as handle:
json.dump(data, handle, indent=2, ensure_ascii=False)
handle.write("\n")
except OSError:
return False
return True
def update_config_model_ids(model_id: str,
clone_model_id: Optional[str] = None,
config_path: Optional[Path] = None) -> bool:
"""Rewrite AUDIOCPP_MODEL_ID (and AUDIOCPP_CLONE_MODEL_ID when given).
Goes through ``common.update_config_value`` so the imported config
module mirrors the change immediately. Returns True when every named
key now holds its value in the file.
"""
path = Path(config_path) if config_path is not None else CONFIG_PATH
ok = common.update_config_value("AUDIOCPP_MODEL_ID", model_id,
config_path=path)
if clone_model_id is not None:
ok = common.update_config_value("AUDIOCPP_CLONE_MODEL_ID",
clone_model_id,
config_path=path) and ok
return ok
def _apply_port_sync(port: int, accepted: bool) -> None:
"""Write the port into app/converter/config.py, or report when declined."""
if accepted:
if not update_config_api_url_port(port):
print(f"[WARNING] Could not update {CONFIG_PATH}; edit "
"AUDIOCPP_API_URL by hand so audiobook.py uses the "
"new port")
else:
print("[WARNING] Left AUDIOCPP_API_URL unchanged; audiobook.py "
f"will still use port {config_port()}")
def _offer_config_model_id_sync(model_id: str, accepted: Optional[bool]) -> None:
"""Point app/converter/config.py at a single hosted model entry.
The converter requests the model id configured in AUDIOCPP_MODEL_ID,
and single-model servers use the same id for the clone entry, so both
ids are rewritten together. ACCEPTED is True/False (apply/skip the
rewrite) or None when no single-entry sync applies (nothing to do).
"""
if config.AUDIOCPP_MODEL_ID == model_id \
and config.AUDIOCPP_CLONE_MODEL_ID == model_id:
return
if accepted is None:
return
if accepted:
if not update_config_model_ids(model_id, model_id):
print(f"[WARNING] Could not update {CONFIG_PATH}; edit "
"AUDIOCPP_MODEL_ID and AUDIOCPP_CLONE_MODEL_ID by hand so "
"audiobook.py uses this model")
else:
print("[WARNING] Left the model ids unchanged; audiobook.py will "
f"still request model '{config.AUDIOCPP_MODEL_ID}'")
def update_server_backend(backend: str) -> bool:
"""Rewrite the 'backend' in the checkout's server.json, or True when none.
Sets ``backend`` to BACKEND in ``<checkout>/server.json`` (same
``json.dump`` formatting as the wizard). Returns True when the file now
carries BACKEND, when there is no server.json (nothing to sync), or when
it already does; False when the file exists but cannot be read/written.
"""
checkout = build.find_local_checkout()
if checkout is None:
return True
server_json = checkout / "server.json"
if not server_json.exists():
return True
try:
data = json.loads(server_json.read_text(encoding="utf-8"))
except (OSError, ValueError):
return False
if not isinstance(data, dict):
return False
if data.get("backend") == backend:
return True
data["backend"] = backend
try:
with server_json.open("w", encoding="utf-8") as handle:
json.dump(data, handle, indent=2, ensure_ascii=False)
handle.write("\n")
except OSError:
return False
return True
|
