111 lines
3.4 KiB
Python
111 lines
3.4 KiB
Python
"""
|
||
Скрипт для массового импорта организаций из .docx файла через API.
|
||
|
||
Использование:
|
||
python import_orgs.py МБОУ.docx
|
||
|
||
Требования:
|
||
pip install python-docx httpx
|
||
|
||
Настройки ниже (или через переменные окружения):
|
||
API_URL — адрес API (по умолчанию http://localhost:8000)
|
||
ADMIN_KEY — значение переменной ADMIN_KEY из .env
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
import httpx
|
||
from docx import Document
|
||
|
||
API_URL = os.getenv("API_URL", "http://localhost:8000")
|
||
ADMIN_KEY = os.getenv("ADMIN_KEY", "")
|
||
|
||
|
||
def extract_names_from_docx(path: str) -> list[str]:
|
||
"""Извлекает непустые строки из параграфов и ячеек таблиц docx."""
|
||
doc = Document(path)
|
||
names = []
|
||
|
||
# Параграфы
|
||
for para in doc.paragraphs:
|
||
text = para.text.strip()
|
||
if text:
|
||
names.append(text)
|
||
|
||
# Таблицы (если список оформлен таблицей)
|
||
for table in doc.tables:
|
||
for row in table.rows:
|
||
for cell in row.cells:
|
||
text = cell.text.strip()
|
||
if text and text not in names:
|
||
names.append(text)
|
||
|
||
return names
|
||
|
||
|
||
def import_organizations(names: list[str]) -> None:
|
||
if not ADMIN_KEY:
|
||
print("ОШИБКА: задайте ADMIN_KEY=... перед запуском")
|
||
print(" Пример: ADMIN_KEY=mykey python import_orgs.py МБОУ.docx")
|
||
sys.exit(1)
|
||
|
||
headers = {"X-Admin-Key": ADMIN_KEY, "Content-Type": "application/json"}
|
||
added = 0
|
||
skipped = 0
|
||
errors = 0
|
||
|
||
with httpx.Client(base_url=API_URL, headers=headers, timeout=30) as client:
|
||
for name in names:
|
||
try:
|
||
resp = client.post("/organizations/", json={"name_organization": name})
|
||
if resp.status_code == 201:
|
||
added += 1
|
||
print(f" [+] {name}")
|
||
elif resp.status_code == 400:
|
||
skipped += 1
|
||
print(f" [=] УЖЕ ЕСТЬ: {name}")
|
||
else:
|
||
errors += 1
|
||
print(f" [!] ОШИБКА {resp.status_code}: {name} — {resp.text}")
|
||
except Exception as e:
|
||
errors += 1
|
||
print(f" [!] ОШИБКА соединения: {name} — {e}")
|
||
|
||
print(f"\nГотово: добавлено {added}, пропущено {skipped}, ошибок {errors}")
|
||
|
||
|
||
def main():
|
||
if len(sys.argv) < 2:
|
||
print("Использование: python import_orgs.py <файл.docx>")
|
||
sys.exit(1)
|
||
|
||
docx_path = sys.argv[1]
|
||
if not os.path.exists(docx_path):
|
||
print(f"Файл не найден: {docx_path}")
|
||
sys.exit(1)
|
||
|
||
print(f"Читаю: {docx_path}")
|
||
names = extract_names_from_docx(docx_path)
|
||
print(f"Найдено строк: {len(names)}")
|
||
print()
|
||
|
||
# Показываем первые 10 для проверки
|
||
print("Первые 10 записей:")
|
||
for n in names[:10]:
|
||
print(f" • {n}")
|
||
if len(names) > 10:
|
||
print(f" ... и ещё {len(names) - 10}")
|
||
|
||
print()
|
||
confirm = input("Начать импорт? (y/n): ").strip().lower()
|
||
if confirm != "y":
|
||
print("Отменено.")
|
||
sys.exit(0)
|
||
|
||
print()
|
||
import_organizations(names)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|