feat: implement specialty content module with expanded Page entity fields, database migrations, and data import utilities
This commit is contained in:
@@ -0,0 +1,153 @@
|
||||
import oracledb
|
||||
from bs4 import BeautifulSoup
|
||||
import re
|
||||
|
||||
def transform_html(title, raw_html):
|
||||
soup = BeautifulSoup(raw_html, "html.parser")
|
||||
|
||||
# Extract all images
|
||||
images = []
|
||||
for img in soup.find_all('img'):
|
||||
src = img.get('src')
|
||||
if src:
|
||||
images.append(src)
|
||||
img.decompose() # Remove from flow
|
||||
|
||||
hero_image = images[0] if len(images) > 0 else "https://sisvietnam.vn/wp-content/uploads/2021/07/DSC07517-1024x576.jpg"
|
||||
other_images = images[1:]
|
||||
|
||||
# Extract Contact Info
|
||||
text = soup.get_text()
|
||||
email_match = re.search(r"(?i)Email\s*:\s*([\w\.-]+@[\w\.-]+)", text)
|
||||
phone_match = re.search(r"(?i)(?:SĐT|Điện thoại|ĐT)\s*:\s*([\d\.\s-]+)", text)
|
||||
address_match = re.search(r"(?i)Địa chỉ\s*:\s*([^\n]+)", text)
|
||||
|
||||
contact_email = email_match.group(1).strip() if email_match else None
|
||||
contact_phone = phone_match.group(1).strip() if phone_match else None
|
||||
contact_address = address_match.group(1).strip() if address_match else None
|
||||
|
||||
# Parse sections based on wp:list or headings
|
||||
sections_data = []
|
||||
|
||||
top_uls = soup.find_all('ul', recursive=False)
|
||||
if not top_uls and soup.find('div', class_='wp-content-container'):
|
||||
top_uls = soup.find('div', class_='wp-content-container').find_all('ul', recursive=False)
|
||||
|
||||
for ul in top_uls:
|
||||
heading = "Nội dung"
|
||||
strong_tag = ul.find('strong')
|
||||
if strong_tag:
|
||||
heading = strong_tag.text.strip()
|
||||
strong_tag.decompose()
|
||||
|
||||
content_html = ""
|
||||
for li in ul.find_all('li', recursive=False):
|
||||
inner_uls = li.find_all('ul')
|
||||
for inner in inner_uls:
|
||||
content_html += str(inner)
|
||||
|
||||
if not content_html:
|
||||
content_html = str(ul)
|
||||
|
||||
sections_data.append({
|
||||
'heading': heading,
|
||||
'content': content_html
|
||||
})
|
||||
|
||||
if not sections_data:
|
||||
sections_data.append({
|
||||
'heading': 'Thông tin chi tiết',
|
||||
'content': str(soup)
|
||||
})
|
||||
|
||||
out = ""
|
||||
bg_colors = ["bg-white", "bg-gray-50", "bg-white", "bg-gray-100"]
|
||||
for i, sec in enumerate(sections_data):
|
||||
bg = bg_colors[i % len(bg_colors)]
|
||||
|
||||
img_html = ""
|
||||
if i < len(other_images):
|
||||
img_src = other_images[i]
|
||||
img_html = f"""
|
||||
<div class="w-full">
|
||||
<div class="rounded-2xl overflow-hidden shadow-lg aspect-[4/3] relative">
|
||||
<img src="{img_src}" class="absolute inset-0 w-full h-full object-cover">
|
||||
</div>
|
||||
</div>
|
||||
"""
|
||||
|
||||
out += f"""
|
||||
<!-- SECTION {i+1} -->
|
||||
<section class="xl:py-12 md:py-8 py-4 {bg}">
|
||||
<div class="container mx-auto px-4">
|
||||
<div class="flex flex-col gap-6 lg:items-center">
|
||||
<div class="w-full">
|
||||
<h2 class="display-6 text-primary-600 mb-4 md:mb-6 xl:mb-8">{sec['heading']}</h2>
|
||||
<div class="prose max-w-none text-gray-700 body-2" style="list-style-type: disc;">
|
||||
{sec['content']}
|
||||
</div>
|
||||
</div>
|
||||
{img_html}
|
||||
</div>
|
||||
</div>
|
||||
</section>
|
||||
"""
|
||||
return out, hero_image, contact_email, contact_address, contact_phone
|
||||
|
||||
def main():
|
||||
dsn = "localhost:1521/sisvietnam"
|
||||
conn = oracledb.connect(user="sisvietnam", password="sisvietnam", dsn=dsn)
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute("SELECT slug, title, content FROM sis_page WHERE page_type = 'CUSTOM'")
|
||||
rows = cursor.fetchall()
|
||||
|
||||
count = 0
|
||||
warnings = []
|
||||
|
||||
for row in rows:
|
||||
slug = row[0]
|
||||
title = row[1]
|
||||
raw_content = row[2].read() if hasattr(row[2], "read") else row[2]
|
||||
|
||||
import json
|
||||
if not raw_content:
|
||||
print(f"Skipping {slug} as content is empty/None.")
|
||||
continue
|
||||
|
||||
if raw_content.strip().startswith("{"):
|
||||
try:
|
||||
data = json.loads(raw_content)
|
||||
html_parts = []
|
||||
for b in data.get("blocks", []):
|
||||
if b.get("type") == "raw" or b.get("type") == "snippet":
|
||||
html_parts.append(b.get("data", {}).get("html", b.get("data", {}).get("htmlContent", "")))
|
||||
raw_content = "".join(html_parts)
|
||||
except Exception as e:
|
||||
print(f"Failed to parse JSON for {slug}: {e}")
|
||||
|
||||
transformed_content, hero_image, email, address, phone = transform_html(title, raw_content)
|
||||
|
||||
if "Thông tin chi tiết" in transformed_content and "wp-content-container" not in transformed_content:
|
||||
warnings.append(slug)
|
||||
|
||||
# Update DB
|
||||
update_sql = """
|
||||
UPDATE sis_page
|
||||
SET content = :1, hero_image = :2, contact_email = :3, contact_address = :4, contact_phone = :5
|
||||
WHERE slug = :6
|
||||
"""
|
||||
cursor.execute(update_sql, [transformed_content, hero_image, email, address, phone, slug])
|
||||
count += 1
|
||||
print(f"Updated {slug}")
|
||||
|
||||
conn.commit()
|
||||
print(f"Successfully transformed {count} pages.")
|
||||
|
||||
if warnings:
|
||||
print("\nWARNING: The following pages lacked standard WP lists (<ul>) and were migrated as a single flat section:")
|
||||
for w in warnings:
|
||||
print(f" - {w}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user