Files
new.rybstock.ru/tools/generate_old_site_import.py
T
2026-06-05 15:52:20 +03:00

557 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from __future__ import annotations
import json
import re
import shutil
from decimal import Decimal, InvalidOperation
from pathlib import Path
from urllib.parse import urlparse
ROOT = Path(__file__).resolve().parents[1]
OLD_ROOT = ROOT / "old_site"
DUMP = OLD_ROOT / "1780592391_db_u3496304_default_04.06.2026.sql"
OUT_SQL = ROOT / "database" / "import_old_site_generated.sql"
OUT_SUMMARY = ROOT / "storage" / "import_old_site_summary.json"
UPLOADS = ROOT / "public" / "assets" / "uploads" / "old"
def find_statement_end(text: str, start: int) -> int:
in_quote = False
escaped = False
for index in range(start, len(text)):
char = text[index]
if in_quote:
if escaped:
escaped = False
elif char == "\\":
escaped = True
elif char == "'":
in_quote = False
else:
if char == "'":
in_quote = True
elif char == ";":
return index
raise RuntimeError("SQL statement end was not found")
def split_tuples(chunk: str) -> list[str]:
rows: list[str] = []
start: int | None = None
depth = 0
in_quote = False
escaped = False
for index, char in enumerate(chunk):
if in_quote:
if escaped:
escaped = False
elif char == "\\":
escaped = True
elif char == "'":
in_quote = False
continue
if char == "'":
in_quote = True
elif char == "(":
if depth == 0:
start = index
depth += 1
elif char == ")":
depth -= 1
if depth == 0 and start is not None:
rows.append(chunk[start + 1:index])
start = None
return rows
def parse_tuple(row: str) -> list[str | None]:
values: list[str | None] = []
current: list[str] = []
in_quote = False
escaped = False
for char in row:
if in_quote:
if escaped:
current.append({"n": "\n", "r": "\r", "t": "\t", "0": "\0"}.get(char, char))
escaped = False
elif char == "\\":
escaped = True
elif char == "'":
in_quote = False
else:
current.append(char)
continue
if char == "'":
in_quote = True
elif char == ",":
values.append(normalize_value("".join(current)))
current = []
else:
current.append(char)
values.append(normalize_value("".join(current)))
return values
def normalize_value(value: str) -> str | None:
value = value.strip()
if value.upper() == "NULL":
return None
return value
def read_table(text: str, table: str) -> list[list[str | None]]:
rows: list[list[str | None]] = []
pattern = re.compile(rf"INSERT INTO `{re.escape(table)}` VALUES ")
for match in pattern.finditer(text):
start = match.end()
end = find_statement_end(text, start)
rows.extend(parse_tuple(row) for row in split_tuples(text[start:end]))
return rows
def read_sitemap_paths() -> set[str]:
paths: set[str] = set()
for sitemap in OLD_ROOT.glob("sitemap*.xml"):
if not sitemap.is_file():
continue
content = sitemap.read_text(encoding="utf-8", errors="replace")
for match in re.finditer(r"<loc>(.*?)</loc>", content):
parsed = urlparse(match.group(1).strip())
path = parsed.path.rstrip("/") or "/"
paths.add(path)
return paths
def sql_string(value: object) -> str:
if value is None:
return "NULL"
text = str(value)
return "'" + text.replace("\\", "\\\\").replace("'", "''").replace("\0", "") + "'"
def decimal_or_none(value: object) -> Decimal | None:
if value is None:
return None
try:
number = Decimal(str(value).strip())
except (InvalidOperation, ValueError):
return None
return number
def decimal_sql(value: object, default: str = "0.00") -> str:
number = decimal_or_none(value)
if number is None:
return default
return format(number.quantize(Decimal("0.01")), "f")
def int_value(value: object, default: int = 0) -> int:
try:
return int(Decimal(str(value)))
except (InvalidOperation, ValueError, TypeError):
return default
def slugify_fallback(text: str, suffix: str) -> str:
raw = text.lower()
translit = {
"а": "a", "б": "b", "в": "v", "г": "g", "д": "d", "е": "e", "ё": "e",
"ж": "zh", "з": "z", "и": "i", "й": "y", "к": "k", "л": "l", "м": "m",
"н": "n", "о": "o", "п": "p", "р": "r", "с": "s", "т": "t", "у": "u",
"ф": "f", "х": "h", "ц": "c", "ч": "ch", "ш": "sh", "щ": "sch",
"ъ": "", "ы": "y", "ь": "", "э": "e", "ю": "yu", "я": "ya",
}
raw = "".join(translit.get(char, char) for char in raw)
raw = re.sub(r"[^a-z0-9]+", "-", raw).strip("-")
return raw or f"item-{suffix}"
def unique_slug(slug: str, used: set[str], fallback_text: str, suffix: str) -> str:
base = (slug or "").strip() or slugify_fallback(fallback_text, suffix)
candidate = base
counter = 2
while candidate in used:
candidate = f"{base}-{counter}"
counter += 1
used.add(candidate)
return candidate
def public_image_path(kind: str, image_name: str | None) -> str | None:
if not image_name:
return None
return f"/assets/uploads/old/{kind}/{image_name}"
def rewrite_content_images(html: str | None) -> str | None:
if not html:
return html
def replace_path(match: re.Match[str]) -> str:
attribute = match.group(1)
quote = match.group(2)
image_path = match.group(3).lstrip("/")
return f'{attribute}={quote}/assets/uploads/old/content-images/{image_path}{quote}'
return re.sub(
r'(src|href)=(["\'])(?:https?://(?:www\.)?loveriba\.ru/)?/?images/([^"\']+)\2',
replace_path,
html,
flags=re.IGNORECASE,
)
def copy_images(source_dir: Path, destination_dir: Path) -> int:
destination_dir.mkdir(parents=True, exist_ok=True)
copied = 0
if not source_dir.exists():
return copied
for source in source_dir.iterdir():
if not source.is_file():
continue
destination = destination_dir / source.name
if not destination.exists() or destination.stat().st_size != source.stat().st_size:
shutil.copy2(source, destination)
copied += 1
return copied
def copy_content_images(source_dir: Path, destination_dir: Path) -> int:
destination_dir.mkdir(parents=True, exist_ok=True)
copied = 0
allowed_extensions = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".svg"}
if not source_dir.exists():
return copied
for source in source_dir.rglob("*"):
if not source.is_file() or source.suffix.lower() not in allowed_extensions:
continue
relative = source.relative_to(source_dir)
destination = destination_dir / relative
destination.parent.mkdir(parents=True, exist_ok=True)
if not destination.exists() or destination.stat().st_size != source.stat().st_size:
shutil.copy2(source, destination)
copied += 1
return copied
def package_unit_and_quantity(label: str, fallback_quantity: Decimal | None) -> tuple[str, Decimal]:
normalized = label.lower().replace(",", ".").replace(" ", "")
match = re.search(r"(\d+(?:\.\d+)?)(кг|kg|г|гр|g|л|l|мл|ml|шт)", normalized)
unit = "kg"
quantity = fallback_quantity if fallback_quantity and fallback_quantity > 0 else Decimal("1")
if match:
quantity = Decimal(match.group(1))
raw_unit = match.group(2)
if raw_unit in {"г", "гр", "g"}:
unit = "g"
elif raw_unit in {"л", "l"}:
unit = "l"
elif raw_unit in {"мл", "ml"}:
unit = "ml"
elif raw_unit == "шт":
unit = "pcs"
else:
unit = "kg"
return unit, quantity
def main() -> None:
text = DUMP.read_text(encoding="utf-8", errors="replace")
old_categories = read_table(text, "bo9rg_jshopping_categories")
old_products = read_table(text, "bo9rg_jshopping_products")
old_attrs = read_table(text, "bo9rg_jshopping_attr")
old_attr_values = read_table(text, "bo9rg_jshopping_attr_values")
old_product_attrs = read_table(text, "bo9rg_jshopping_products_attr")
old_product_images = read_table(text, "bo9rg_jshopping_products_images")
old_product_categories = read_table(text, "bo9rg_jshopping_products_to_categories")
sitemap_paths = read_sitemap_paths()
copied_products = copy_images(
OLD_ROOT / "components" / "com_jshopping" / "files" / "img_products",
UPLOADS / "products",
)
copied_categories = copy_images(
OLD_ROOT / "components" / "com_jshopping" / "files" / "img_categories",
UPLOADS / "categories",
)
copied_content_images = copy_content_images(
OLD_ROOT / "images",
UPLOADS / "content-images",
)
attr_value_by_id = {int_value(row[0]): row[5] or "" for row in old_attr_values}
attr_group_by_id = {int_value(row[0]): row[9] or f"Атрибут {row[0]}" for row in old_attrs}
product_categories: dict[int, list[tuple[int, int]]] = {}
for row in old_product_categories:
product_id = int_value(row[0])
category_id = int_value(row[1])
sort_order = int_value(row[2], 100)
product_categories.setdefault(product_id, []).append((category_id, sort_order))
product_images: dict[int, list[dict[str, object]]] = {}
for row in old_product_images:
product_id = int_value(row[1])
image_name = row[2] or ""
if image_name == "":
continue
product_images.setdefault(product_id, []).append({
"path": public_image_path("products", image_name),
"alt": row[3] or row[4] or None,
"sort_order": int_value(row[5], 100),
})
product_attrs_by_product: dict[int, list[list[str | None]]] = {}
for row in old_product_attrs:
product_attrs_by_product.setdefault(int_value(row[1]), []).append(row)
sql: list[str] = [
"SET NAMES utf8mb4;",
"SET FOREIGN_KEY_CHECKS = 0;",
"CREATE TABLE IF NOT EXISTS product_category_links (",
" id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,",
" product_id BIGINT UNSIGNED NOT NULL,",
" category_id BIGINT UNSIGNED NOT NULL,",
" sort_order INT NOT NULL DEFAULT 100,",
" created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,",
" PRIMARY KEY (id),",
" UNIQUE KEY uq_product_category_link (product_id, category_id),",
" KEY idx_product_category_links_category (category_id, sort_order)",
") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;",
"TRUNCATE TABLE redirects;",
"TRUNCATE TABLE variant_attribute_values;",
"TRUNCATE TABLE product_attribute_groups;",
"TRUNCATE TABLE attribute_values;",
"TRUNCATE TABLE attribute_groups;",
"TRUNCATE TABLE product_images;",
"TRUNCATE TABLE product_category_links;",
"TRUNCATE TABLE product_variants;",
"TRUNCATE TABLE popular_products;",
"TRUNCATE TABLE promo_block_items;",
"TRUNCATE TABLE products;",
"TRUNCATE TABLE categories;",
]
used_category_slugs: set[str] = set()
category_slug_by_old_id: dict[int, str] = {}
category_old_id_by_slug: dict[str, int] = {}
for row in old_categories:
category_id = int_value(row[0])
parent_id = int_value(row[2])
name = row[19] or f"Категория {category_id}"
slug = unique_slug(row[20] or "", used_category_slugs, name, str(category_id))
category_slug_by_old_id[category_id] = slug
category_old_id_by_slug[slug] = category_id
image_path = public_image_path("categories", row[1] or "")
category_description = rewrite_content_images(row[22] or row[15] or None)
values = [
category_id,
"NULL" if parent_id == 0 else parent_id,
sql_string(name),
sql_string(slug),
sql_string(f"/katalog/category/view/{category_id}"),
sql_string(category_description),
sql_string(name),
sql_string(row[23] or None),
sql_string(row[24] or None),
sql_string(row[25] or None),
sql_string(image_path),
int_value(row[5], 100),
int_value(row[3], 1),
]
sql.append(
"INSERT INTO categories (id, parent_id, name, slug, legacy_path, description, h1, seo_title, seo_description, seo_keywords, image_path, sort_order, is_active) VALUES ("
+ ",".join(map(str, values))
+ ");"
)
sql.append(
"INSERT IGNORE INTO redirects (old_path, new_path, http_code, is_active) VALUES "
f"({sql_string('/katalog/category/view/' + str(category_id))},{sql_string('/catalog/' + slug)},301,1);"
)
sql.append(
"INSERT IGNORE INTO redirects (old_path, new_path, http_code, is_active) VALUES "
f"({sql_string('/katalog')},{sql_string('/catalog')},301,1);"
)
sitemap_category_redirects = 0
for path in sorted(sitemap_paths):
match = re.fullmatch(r"/katalog/([^/]+)", path)
if not match:
continue
slug = match.group(1)
if slug not in category_old_id_by_slug:
continue
sql.append(
"INSERT IGNORE INTO redirects (old_path, new_path, http_code, is_active) VALUES "
f"({sql_string(path)},{sql_string('/catalog/' + slug)},301,1);"
)
sitemap_category_redirects += 1
used_product_slugs: set[str] = set()
product_slug_by_old_id: dict[int, str] = {}
imported_products = 0
for row in old_products:
product_id = int_value(row[0])
name = row[41] or f"Товар {product_id}"
slug = unique_slug(row[42] or "", used_product_slugs, name, str(product_id))
product_slug_by_old_id[product_id] = slug
category_id = int_value(row[33]) or (product_categories.get(product_id, [(0, 100)])[0][0])
if category_id not in category_slug_by_old_id:
category_id = 0
main_image = public_image_path("products", row[20] or "")
short_description = rewrite_content_images(row[43] or None)
description = rewrite_content_images(row[44] or None)
base_price = decimal_or_none(row[16]) or Decimal("0")
min_price = decimal_or_none(row[17]) or Decimal("0")
package_quantity = decimal_or_none(row[28]) or decimal_or_none(row[19]) or Decimal("1")
package_price = min_price if min_price > 0 else (base_price * package_quantity if package_quantity > 0 else base_price)
base_unit = "kg"
display_mode = "auto" if base_price > 0 else "package_only"
values = [
product_id,
"NULL" if category_id == 0 else category_id,
sql_string(name),
sql_string(slug),
sql_string(f"/katalog/product/view/{category_id}/{product_id}"),
sql_string(row[2] if row[2] not in {"", "0"} else None),
sql_string(short_description),
sql_string(description),
sql_string(None),
sql_string(name),
sql_string(row[45] or None),
sql_string(row[46] or None),
sql_string(row[47] or None),
sql_string(main_image),
sql_string(base_unit),
sql_string(display_mode),
1 if base_price > 0 else 0,
int_value(row[9], 0),
1,
int_value(row[27], 0),
int_value(row[26], 100),
sql_string(row[7] if row[7] != "0000-00-00 00:00:00" else None),
sql_string(row[8] if row[8] != "0000-00-00 00:00:00" else None),
]
sql.append(
"INSERT INTO products (id, category_id, name, slug, legacy_path, sku, short_description, description, product_info, h1, seo_title, seo_description, seo_keywords, main_image_path, base_unit, package_display_mode, is_weight_product, is_published, is_available, sales_count, sort_order, published_at, updated_at) VALUES ("
+ ",".join(map(str, values))
+ ");"
)
for old_category_id, sort_order in product_categories.get(product_id, []):
if old_category_id in category_slug_by_old_id:
sql.append(
"INSERT IGNORE INTO product_category_links (product_id, category_id, sort_order) VALUES "
f"({product_id},{old_category_id},{sort_order});"
)
sql.append(
"INSERT IGNORE INTO redirects (old_path, new_path, http_code, is_active) VALUES "
f"({sql_string('/katalog/product/view/' + str(old_category_id) + '/' + str(product_id))},{sql_string('/product/' + slug)},301,1);"
)
for image in product_images.get(product_id, []):
sql.append(
"INSERT INTO product_images (product_id, path, alt, sort_order) VALUES "
f"({product_id},{sql_string(image['path'])},{sql_string(image['alt'])},{int(image['sort_order'])});"
)
variants = product_attrs_by_product.get(product_id, [])
if not variants and package_price > 0:
fallback_variant_id = 100000000 + product_id
price_per_unit = base_price if base_price > 0 else None
sql.append(
"INSERT INTO product_variants (id, product_id, name, unit, package_quantity, step_quantity, price_per_unit, package_price, old_package_price, is_default, is_published, is_available, sort_order) VALUES "
f"({fallback_variant_id},{product_id},{sql_string('Фасовка')},{sql_string(base_unit)},{decimal_sql(package_quantity)},{decimal_sql(package_quantity)},{decimal_sql(price_per_unit, 'NULL')},{decimal_sql(package_price)},{decimal_sql(row[14], 'NULL')},1,1,1,100);"
)
for index, variant in enumerate(variants):
variant_id = int_value(variant[0])
package_label = attr_value_by_id.get(int_value(variant[11]), "Фасовка")
unit, quantity = package_unit_and_quantity(package_label, decimal_or_none(variant[9]))
variant_price = decimal_or_none(variant[3]) or Decimal("0")
price_per_unit = variant_price / quantity if unit == "kg" and quantity > 0 and variant_price > 0 else None
sql.append(
"INSERT INTO product_variants (id, product_id, name, unit, package_quantity, step_quantity, price_per_unit, package_price, old_package_price, is_default, is_published, is_available, sort_order) VALUES "
f"({variant_id},{product_id},{sql_string(package_label or 'Фасовка')},{sql_string(unit)},{decimal_sql(quantity)},{decimal_sql(quantity)},{decimal_sql(price_per_unit, 'NULL')},{decimal_sql(variant_price)},{decimal_sql(variant[4], 'NULL')},{1 if index == 0 else 0},1,1,{100 + index});"
)
for attr_offset, group_id in [(11, 1), (12, 3), (13, 4), (14, 5)]:
value_id = int_value(variant[attr_offset])
if value_id <= 0:
continue
sql.append(
"INSERT IGNORE INTO variant_attribute_values (variant_id, group_id, value_id) VALUES "
f"({variant_id},{group_id},{value_id});"
)
imported_products += 1
for group_id, group_name in attr_group_by_id.items():
sql.append(
"INSERT INTO attribute_groups (id, name, slug, display_type, is_filterable, sort_order) VALUES "
f"({group_id},{sql_string(group_name)},{sql_string(slugify_fallback(group_name, str(group_id)))},'select',1,{group_id * 10});"
)
used_attribute_slugs_by_group: dict[int, set[str]] = {}
for row in old_attr_values:
value_id = int_value(row[0])
group_id = int_value(row[1])
value = row[5] or f"???????? {value_id}"
used_attribute_slugs = used_attribute_slugs_by_group.setdefault(group_id, set())
value_slug = unique_slug("", used_attribute_slugs, value, str(value_id))
sql.append(
"INSERT INTO attribute_values (id, group_id, value, slug, sort_order) VALUES "
f"({value_id},{group_id},{sql_string(value)},{sql_string(value_slug)},{int_value(row[2], 100)});"
)
sql.extend([
"SET FOREIGN_KEY_CHECKS = 1;",
])
OUT_SQL.write_text("\n".join(sql) + "\n", encoding="utf-8")
OUT_SUMMARY.parent.mkdir(parents=True, exist_ok=True)
OUT_SUMMARY.write_text(json.dumps({
"categories": len(old_categories),
"products": imported_products,
"product_variants": len(old_product_attrs),
"product_images_rows": len(old_product_images),
"product_category_links": len(old_product_categories),
"sitemap_category_redirects": sitemap_category_redirects,
"copied_product_images": copied_products,
"copied_category_images": copied_categories,
"copied_content_images": copied_content_images,
"sql_file": str(OUT_SQL.relative_to(ROOT)),
}, ensure_ascii=False, indent=2), encoding="utf-8")
print(OUT_SUMMARY.read_text(encoding="utf-8"))
if __name__ == "__main__":
main()