import sys, os, json, re from datetime import datetime from typing import Any, Dict, List import requests import openpyxl from PyQt6 import QtWidgets from PyQt6.QtWidgets import ( QApplication, QMainWindow, QFileDialog, QVBoxLayout, QHBoxLayout, QWidget, QPushButton, QLabel, QComboBox, QTextEdit, QTableWidget, QTableWidgetItem, QHeaderView ) # === ПУТИ === BASE_DIR = os.path.dirname(os.path.abspath(__file__)) MAPPINGS_DIR = os.path.join(BASE_DIR, "mappings") HISTORY_FILE = os.path.join(BASE_DIR, "history.json") def ts() -> str: return datetime.now().strftime("%H:%M:%S") def split_list_cell(val: str) -> List[str]: if val is None: return [] s = str(val).strip() if not s: return [] parts = re.split(r"[\n,;|]+", s) return [p.strip() for p in parts if p.strip()] def cast_value(raw: Any, t: str): if raw is None: return None s = str(raw).strip() if t == "int": try: return int(float(s)) except: return None if t == "float": try: return float(str(s).replace(",", ".")) except: return None if t == "bool": s_low = s.lower() return s_low in ("1", "true", "yes", "y", "да", "истина") if t == "list" or t == "list[string]": return split_list_cell(s) # string return s def resolve_endpoint(selected_base_url: str, mapping_endpoint: str) -> str: """ Правила: - Если endpoint начинается с '/', приклеиваем к выбранному base_url. - Если endpoint начинается с 'http', заменяем схему+хост на выбранный base_url (оставляем path). - Иначе считаем это относительным путём от base_url. """ base = selected_base_url.rstrip("/") ep = mapping_endpoint.strip() if ep.startswith("/"): return f"{base}{ep}" if ep.startswith("http://") or ep.startswith("https://"): # вытащим только путь try: from urllib.parse import urlparse parsed = urlparse(ep) path = parsed.path if parsed.query: path += f"?{parsed.query}" return f"{base}{path}" except: return ep # fallback return f"{base}/{ep.lstrip('/')}" class ImportApp(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle("🧩 Import Client v0.3") self.resize(1200, 800) self.api_history = self.load_history() self.current_mapping: Dict[str, Any] = {} self.loaded_data: List[List[Any]] = [] # с заголовком self.headers: List[str] = [] # заголовки из Excel self.map_pairs: List[Dict[str, str]] = [] # [{"json_field": "...", "excel_col": "..."}] # === UI === main_widget = QWidget() main_layout = QVBoxLayout(main_widget) self.setCentralWidget(main_widget) # Верхняя панель: API URL top_layout = QHBoxLayout() self.api_label = QLabel("API URL:") self.api_box = QComboBox() self.api_box.setEditable(True) self.api_box.addItems(self.api_history or ["http://localhost:3005"]) self.btn_save_url = QPushButton("💾 Сохранить") self.btn_save_url.clicked.connect(self.save_current_url) top_layout.addWidget(self.api_label) top_layout.addWidget(self.api_box) top_layout.addWidget(self.btn_save_url) main_layout.addLayout(top_layout) # Тип импорта import_layout = QHBoxLayout() self.import_label = QLabel("Тип импорта:") self.import_type = QComboBox() self.import_type.addItems(["brands", "categories", "products"]) self.import_type.currentIndexChanged.connect(self.load_mapping) import_layout.addWidget(self.import_label) import_layout.addWidget(self.import_type) # === parserName (если предусмотрен шаблоном) === self.parser_label = QLabel("Parser name:") self.parser_box = QComboBox() self.parser_box.setEditable(True) # можно вписать вручную self.parser_label.hide() self.parser_box.hide() parser_layout = QHBoxLayout() parser_layout.addWidget(self.parser_label) parser_layout.addWidget(self.parser_box) main_layout.addLayout(parser_layout) # Кнопки Excel self.btn_load_excel = QPushButton("📂 Загрузить Excel-файл") self.btn_load_excel.clicked.connect(self.load_excel) import_layout.addWidget(self.btn_load_excel) main_layout.addLayout(import_layout) # Таблица предпросмотра Excel self.table = QTableWidget() main_layout.addWidget(self.table) # Таблица сопоставлений main_layout.addWidget(QLabel("Сопоставление полей: JSON ↔ Колонка Excel")) self.map_table = QTableWidget() self.map_table.setColumnCount(3) self.map_table.setHorizontalHeaderLabels(["Поле JSON", "Тип", "Колонка Excel"]) self.map_table.horizontalHeader().setSectionResizeMode(0, QHeaderView.ResizeMode.Stretch) self.map_table.horizontalHeader().setSectionResizeMode(1, QHeaderView.ResizeMode.ResizeToContents) self.map_table.horizontalHeader().setSectionResizeMode(2, QHeaderView.ResizeMode.Stretch) main_layout.addWidget(self.map_table) # Кнопки действий action_layout = QHBoxLayout() self.btn_autofill = QPushButton("✨ Автосопоставление") self.btn_autofill.clicked.connect(self.autofill_mapping) self.btn_generate = QPushButton("🧱 Сформировать JSON") self.btn_generate.clicked.connect(self.generate_json) self.btn_send = QPushButton("📤 Отправить в API") self.btn_send.clicked.connect(self.send_to_api) action_layout.addWidget(self.btn_autofill) action_layout.addWidget(self.btn_generate) action_layout.addWidget(self.btn_send) main_layout.addLayout(action_layout) # Предпросмотр JSON main_layout.addWidget(QLabel("Предпросмотр JSON:")) self.json_preview = QTextEdit() self.json_preview.setReadOnly(True) main_layout.addWidget(self.json_preview) # Лог main_layout.addWidget(QLabel("Лог:")) self.log_box = QTextEdit() self.log_box.setReadOnly(True) main_layout.addWidget(self.log_box) # Загрузим первый шаблон self.load_mapping() # ---------- History ---------- def load_history(self): if os.path.exists(HISTORY_FILE): try: with open(HISTORY_FILE, "r", encoding="utf-8") as f: return json.load(f) except: return [] return [] def save_current_url(self): url = self.api_box.currentText().strip() if not url: return if url not in self.api_history: self.api_history.insert(0, url) with open(HISTORY_FILE, "w", encoding="utf-8") as f: json.dump(self.api_history[:10], f, ensure_ascii=False, indent=2) self.log(f"✅ Сохранен URL: {url}") # ---------- Mapping ---------- def load_mapping(self): import_type = self.import_type.currentText() path = os.path.join(MAPPINGS_DIR, f"{import_type}.json") if not os.path.exists(path): self.current_mapping = {} self.log(f"⚠️ Не найден шаблон: {path}") return try: with open(path, "r", encoding="utf-8") as f: self.current_mapping = json.load(f) self.log(f"📘 Загружен шаблон: {import_type}") # Если шаблон содержит parserName с choices — показать выбор fields = self.current_mapping.get("fields", {}) parser_meta = fields.get("parserName", {}) if isinstance(parser_meta, dict) and "choices" in parser_meta: self.parser_label.show() self.parser_box.show() self.parser_box.clear() for opt in parser_meta["choices"]: self.parser_box.addItem(opt) # выбираем первый как дефолт self.parser_box.setCurrentIndex(0) else: self.parser_label.hide() self.parser_box.hide() self.populate_map_table() except Exception as e: self.current_mapping = {} self.log(f"❌ Ошибка чтения шаблона: {e}") def populate_map_table(self): self.map_table.setRowCount(0) self.map_pairs.clear() fields = (self.current_mapping.get("fields") or {}) # Список кандидатов полей JSON для сопоставления # Пропускаем поля с "fixed" (их не мапим) — они выставляются автоматически candidates = [] for k, meta in fields.items(): if isinstance(meta, dict) and "fixed" in meta: continue candidates.append((k, meta)) self.map_table.setRowCount(len(candidates)) for i, (json_field, meta) in enumerate(candidates): # колонка 0: имя поля json self.map_table.setItem(i, 0, QTableWidgetItem(json_field)) # колонка 1: тип t = meta["type"] if isinstance(meta, dict) and "type" in meta else "string" self.map_table.setItem(i, 1, QTableWidgetItem(t)) # колонка 2: выпадающий список — заголовки Excel combo = QComboBox() combo.addItem("") # пусто = не сопоставлено for h in self.headers: combo.addItem(h) self.map_table.setCellWidget(i, 2, combo) def autofill_mapping(self): """Автосопоставление только при полном совпадении названия поля JSON и колонки Excel.""" rows = self.map_table.rowCount() matched = 0 for i in range(rows): json_field = self.map_table.item(i, 0).text().strip() combo = self.map_table.cellWidget(i, 2) if not combo: continue # точное совпадение (с учётом регистра) if json_field in self.headers: idx = combo.findText(json_field) if idx >= 0: combo.setCurrentIndex(idx) matched += 1 # допускаем вариант без учёта регистра (если нужно) else: for h in self.headers: if h.lower() == json_field.lower(): idx = combo.findText(h) if idx >= 0: combo.setCurrentIndex(idx) matched += 1 break self.log(f"✨ Автосопоставление завершено: найдено точных совпадений {matched}") # ---------- Excel ---------- def load_excel(self): path, _ = QFileDialog.getOpenFileName(self, "Выбери Excel-файл", "", "Excel Files (*.xlsx)") if not path: return try: wb = openpyxl.load_workbook(path, data_only=True) sheet = wb.active # === читаем ВСЕ строки для обработки === all_data = [list(r) for r in sheet.iter_rows(values_only=True)] if not all_data: self.log("⚠️ Файл пуст.") return self.headers = [str(h) if h else "" for h in all_data[0]] self.loaded_data = all_data # ВСЁ содержимое для дальнейшей работы # === только первые 50 строк показываем в таблице === preview_data = all_data[:51] self.table.clear() self.table.setRowCount(0) self.table.setColumnCount(len(self.headers)) self.table.setHorizontalHeaderLabels(self.headers) for r_idx, row in enumerate(preview_data[1:], start=0): self.table.insertRow(r_idx) for c_idx, value in enumerate(row): self.table.setItem(r_idx, c_idx, QTableWidgetItem("" if value is None else str(value))) self.log(f"📄 Загружен файл: {os.path.basename(path)} ({sheet.max_row} строк)") self.populate_map_table() except Exception as e: self.log(f"❌ Ошибка при чтении файла: {e}") def display_sheet(self, sheet): self.table.clear() self.table.setRowCount(0) self.table.setColumnCount(0) data = [] for row in sheet.iter_rows(values_only=True): data.append(list(row)) if len(data) > 50: # покажем первые 50 строк break if not data: return self.headers = [str(h) if h else "" for h in data[0]] self.loaded_data = data self.table.setColumnCount(len(self.headers)) self.table.setHorizontalHeaderLabels(self.headers) for r_idx, row in enumerate(data[1:], start=0): self.table.insertRow(r_idx) for c_idx, value in enumerate(row): self.table.setItem(r_idx, c_idx, QTableWidgetItem("" if value is None else str(value))) # ---------- Генерация JSON ---------- def collect_mapping(self) -> Dict[str, str]: """Собираем словарь {json_field -> excel_header} для выбранных пользователем маппингов.""" mapping = {} rows = self.map_table.rowCount() for i in range(rows): json_field = self.map_table.item(i, 0).text() combo = self.map_table.cellWidget(i, 2) header = combo.currentText().strip() if combo else "" if header: mapping[json_field] = header return mapping def generate_json(self): if not self.loaded_data: self.log("⚠️ Сначала загрузите Excel-файл.") return if not self.current_mapping: self.log("⚠️ Не загружен шаблон JSON.") return import_type = self.import_type.currentText() fields_meta = self.current_mapping.get("fields") or {} mapping = self.collect_mapping() rows = self.loaded_data[1:] # без заголовка result: Any # ===== BRANDS ===== if import_type == "brands": # ожидается список строк в ключе "brands" json_key = "brands" selected_col = mapping.get(json_key) values = [] if selected_col: col_idx = self.headers.index(selected_col) seen = set() for r in rows: if col_idx < len(r) and r[col_idx]: val = str(r[col_idx]).strip() if val and val not in seen: seen.add(val) values.append(val) result = {json_key: values} # ===== CATEGORIES ===== elif import_type == "categories": # ожидаем объекты {"originPath": "...", "path": "..."} origin_col = mapping.get("originPath", "") path_col = mapping.get("path", "") result = [] seen_origins = set() idx_origin = self.headers.index(origin_col) if origin_col in self.headers else -1 idx_path = self.headers.index(path_col) if path_col in self.headers else -1 for r in rows: origin_val = ( str(r[idx_origin]).strip() if idx_origin >= 0 and idx_origin < len(r) and r[idx_origin] else "" ) path_val = ( str(r[idx_path]).strip() if idx_path >= 0 and idx_path < len(r) and r[idx_path] else "" ) # пропускаем полностью пустые строки if not origin_val and not path_val: continue # если originPath уже был, пропускаем (оставляем только первое вхождение) if origin_val in seen_origins: continue seen_origins.add(origin_val) # нормализуем разделители (чисто косметически) path_val = " > ".join([p.strip() for p in path_val.split(">")]) result.append({"originPath": origin_val, "path": path_val}) self.log(f"🧩 Найдено уникальных originPath: {len(result)}") # ===== PRODUCTS ===== else: # структура: # { # "parserName": fixed или из поля, # "items": [ { category:{name}, brand:{name}, variant:{...} } ] # } parser_meta = fields_meta.get("parserName", {}) parser_name = "unknown" # если есть choices — берём выбранное из UI if isinstance(parser_meta, dict) and "choices" in parser_meta: parser_name = self.parser_box.currentText().strip() or parser_meta["choices"][0] # иначе если fixed elif isinstance(parser_meta, dict) and "fixed" in parser_meta: parser_name = parser_meta["fixed"] # fallback else: parser_name = "xlsx-parser" result = {"parserName": parser_name, "items": []} # подготовим индекс колонок col_index = {h: i for i, h in enumerate(self.headers)} for r in rows: # helpers для получения значения по json-полю def get_cell(json_field: str): header = mapping.get(json_field) if not header: return None idx = col_index.get(header, -1) if idx < 0 or idx >= len(r): return None return r[idx] # категори/бренд cat_name = get_cell("category.name") brand_name = get_cell("brand.name") # variant поля с типами v = {} for jf, meta in fields_meta.items(): if jf in ("parserName", "category.name", "brand.name"): continue if isinstance(meta, dict) and "fixed" in meta: v_key = jf.split("variant.", 1)[-1] if jf.startswith("variant.") else jf v[v_key] = meta["fixed"] continue header = mapping.get(jf) if not header: # оставим default если указан default = meta.get("default") if isinstance(meta, dict) else None if default is not None and jf.startswith("variant."): v_key = jf.split("variant.", 1)[-1] v[v_key] = default continue raw = get_cell(jf) t = meta.get("type", "string") if isinstance(meta, dict) else "string" # особый случай: images -> список ссылок if jf.endswith(".images"): v["images"] = cast_value(raw, "list") else: val = cast_value(raw, t) v_key = jf.split("variant.", 1)[-1] if jf.startswith("variant.") else jf v[v_key] = val item = { "category": {"name": cast_value(cat_name, "string") or ""}, "brand": {"name": cast_value(brand_name, "string") or ""}, "variant": { "status_id": v.get("status_id") if v.get("status_id") is not None else 1, "color": v.get("color") if v.get("color") is not None else "", "sku": v.get("sku") if v.get("sku") is not None else "", "size": v.get("size") if v.get("size") is not None else "", "cost": v.get("cost") if v.get("cost") is not None else 0, "originalUrl": v.get("originalUrl") if v.get("originalUrl") is not None else "", "originalName": v.get("originalName") if v.get("originalName") is not None else "", # КЛЮЧЕВОЕ: пустые строки вместо null "originalDescription": v.get("originalDescription") or "", "originalComposition": v.get("originalComposition") or "", "images": v.get("images") if v.get("images") is not None else [], "inStock": v.get("inStock") if v.get("inStock") is not None else True, "weight": v.get("weight") if v.get("weight") is not None else 0 } } # фильтр пустых строк: нужен хотя бы sku или originalUrl или название if any([item["variant"]["sku"], item["variant"]["originalUrl"], item["variant"]["originalName"]]): result["items"].append(item) # Вывод self.json_preview.setText(json.dumps(result, ensure_ascii=False, indent=2)) self.log(f"🧱 Сформирован JSON для: {self.import_type.currentText()}") # ---------- Отправка ---------- def send_to_api(self): if not self.current_mapping: self.log("⚠️ Не загружен шаблон JSON.") return text = self.json_preview.toPlainText().strip() if not text: self.log("⚠️ Сначала сформируйте JSON.") return try: payload = json.loads(text) except Exception as e: self.log(f"❌ Некорректный JSON: {e}") return base_url = self.api_box.currentText().strip() or "http://localhost:3005" endpoint = self.current_mapping.get("endpoint", "/") url = resolve_endpoint(base_url, endpoint) try: self.log(f"➡️ POST {url}") r = requests.post(url, json=payload, headers={"Content-Type": "application/json"}, timeout=60) # plain text ответ content_type = r.headers.get("Content-Type", "") body = r.text if "text" in content_type or "html" in content_type or content_type == "" else r.content[:1000] self.log(f"⬅️ {r.status_code} {r.reason}\n{body}") except Exception as e: self.log(f"❌ Ошибка запроса: {e}") # ---------- Лог ---------- def log(self, text): self.log_box.append(f"{ts()} - {text}") if __name__ == "__main__": app = QApplication(sys.argv) window = ImportApp() window.show() sys.exit(app.exec())