118 lines
4.6 KiB
Bash
Executable File
118 lines
4.6 KiB
Bash
Executable File
set -euo pipefail
|
||
|
||
svc_dir="services/chat/src/app/schemas"
|
||
test -d "$svc_dir" || { echo "Not found: $svc_dir"; exit 1; }
|
||
|
||
python3 - <<'PY'
|
||
import re
|
||
from pathlib import Path
|
||
|
||
root = Path("services/chat/src/app/schemas")
|
||
|
||
def ensure_future_and_uuid(text: str) -> str:
|
||
lines = text.splitlines(keepends=True)
|
||
changed = False
|
||
|
||
# Удалим дубли future-импорта
|
||
fut_pat = re.compile(r'^\s*from\s+__future__\s+import\s+annotations\s*(#.*)?$\n?', re.M)
|
||
if fut_pat.search(text):
|
||
lines = [ln for ln in lines if not fut_pat.match(ln)]
|
||
changed = True
|
||
|
||
# Вставим future после докстринга (если он есть)
|
||
i = 0
|
||
while i < len(lines) and (lines[i].strip()=="" or lines[i].lstrip().startswith("#!") or re.match(r'^\s*#.*coding[:=]', lines[i])):
|
||
i += 1
|
||
insert_at = i
|
||
if i < len(lines) and re.match(r'^\s*[rubfRUBF]*[\'"]{3}', lines[i]): # докстринг
|
||
q = '"""' if '"""' in lines[i] else "'''"
|
||
j = i
|
||
while j < len(lines):
|
||
if q in lines[j] and j != i:
|
||
j += 1
|
||
break
|
||
j += 1
|
||
insert_at = min(j, len(lines))
|
||
|
||
if not any("from __future__ import annotations" in ln for ln in lines):
|
||
lines.insert(insert_at, "from __future__ import annotations\n")
|
||
changed = True
|
||
|
||
txt = "".join(lines)
|
||
if "from uuid import UUID" not in txt:
|
||
# вставим сразу после future
|
||
# найдём позицию future
|
||
pos = next((k for k,ln in enumerate(lines) if "from __future__ import annotations" in ln), 0)
|
||
lines.insert(pos+1, "from uuid import UUID\n")
|
||
changed = True
|
||
|
||
return "".join(lines), changed
|
||
|
||
def set_config(text: str) -> str:
|
||
# Pydantic v2: добавим ConfigDict и model_config, если их нет
|
||
if "ConfigDict" not in text:
|
||
text = re.sub(r'from pydantic import ([^\n]+)',
|
||
lambda m: m.group(0).replace(m.group(1), (m.group(1) + ", ConfigDict").replace(", ConfigDict,"," , ConfigDict,")),
|
||
text, count=1) if "from pydantic import" in text else ("from pydantic import BaseModel, ConfigDict\n" + text)
|
||
|
||
# В каждую модель, наследующую BaseModel, которая заканчивается на Read/Out, добавим from_attributes
|
||
def inject_cfg(block: str) -> str:
|
||
# если уже есть model_config/orm_mode — не трогаем
|
||
if "model_config" in block or "class Config" in block:
|
||
return block
|
||
# аккуратно добавим model_config сверху класса
|
||
header, body = block.split(":", 1)
|
||
return header + ":\n model_config = ConfigDict(from_attributes=True)\n" + body
|
||
|
||
# Разобьём по классам
|
||
parts = re.split(r'(\nclass\s+[A-Za-z0-9_]+\s*\([^\)]*BaseModel[^\)]*\)\s*:)', text)
|
||
if len(parts) > 1:
|
||
out = [parts[0]]
|
||
for i in range(1, len(parts), 2):
|
||
head = parts[i]
|
||
body = parts[i+1]
|
||
m = re.search(r'class\s+([A-Za-z0-9_]+)\s*\(', head)
|
||
name = m.group(1) if m else ""
|
||
if name.endswith(("Read","Out")):
|
||
out.append(inject_cfg(head + body))
|
||
else:
|
||
out.append(head + body)
|
||
text = "".join(out)
|
||
return text
|
||
|
||
def fix_uuid_fields(text: str) -> str:
|
||
# Преобразуем только в классах Read/Out
|
||
def repl_in_class(cls_text: str) -> str:
|
||
# заменяем аннотации полей
|
||
repl = {
|
||
r'(^\s*)(id)\s*:\s*str(\b)': r'\1\2: UUID\3',
|
||
r'(^\s*)(pair_id)\s*:\s*str(\b)': r'\1\2: UUID\3',
|
||
r'(^\s*)(room_id)\s*:\s*str(\b)': r'\1\2: UUID\3',
|
||
r'(^\s*)(sender_id)\s*:\s*str(\b)': r'\1\2: UUID\3',
|
||
r'(^\s*)(user_id)\s*:\s*str(\b)': r'\1\2: UUID\3',
|
||
}
|
||
for pat, sub in repl.items():
|
||
cls_text = re.sub(pat, sub, cls_text, flags=re.M)
|
||
return cls_text
|
||
|
||
patt = re.compile(r'(\nclass\s+[A-Za-z0-9_]+(Read|Out)\s*\([^\)]*BaseModel[^\)]*\)\s*:\n(?:\s+.*\n)+)', re.M)
|
||
return patt.sub(lambda m: repl_in_class(m.group(1)), text)
|
||
|
||
changed_any = False
|
||
for p in root.glob("**/*.py"):
|
||
txt = p.read_text()
|
||
new, c1 = ensure_future_and_uuid(txt)
|
||
new = set_config(new)
|
||
new2 = fix_uuid_fields(new)
|
||
if new2 != txt:
|
||
p.write_text(new2)
|
||
print("fixed:", p)
|
||
changed_any = True
|
||
|
||
print("DONE" if changed_any else "NO-CHANGES")
|
||
PY
|
||
|
||
echo "[docker] rebuild & restart chat…"
|
||
docker compose build --no-cache chat >/dev/null
|
||
docker compose restart chat
|