"""مهاجرت‌های سبک دیتابیس — DDL و تبدیل داده‌های قدیمی؛ همه idempotent. هم main.py (startup) و هم seed.py این‌ها را صدا می‌زنند تا هر مسیر اجرایی دیتابیس موجود را بدون از دست رفتن داده ارتقا دهد. """ from sqlalchemy import text from .database import SessionLocal, engine # جدول‌های مجموعه‌ای که ستون position دارند POSITION_TABLES = [ "services", "projects", "articles", "testimonials", "faqs", "team", "timeline", "stats", "why_us", "investment_models", "process_steps", "offices", "gallery_images", "nav_items", ] # ستون‌های انگلیسی — (جدول، ستون، نوع). ستون‌های JSON بدون default اند تا # سطرهای موجود NULL بگیرند (یعنی «ترجمه ثبت نشده») و فرانت به فارسی برگردد. EN_COLUMNS: list[tuple[str, str, str]] = [ ("brand", "name_en", "VARCHAR(128)"), ("brand", "address_en", "TEXT"), ("services", "title_en", "VARCHAR(256)"), ("services", "desc_en", "TEXT"), ("projects", "title_en", "VARCHAR(256)"), ("projects", "location_en", "VARCHAR(128)"), ("projects", "year_en", "VARCHAR(64)"), ("projects", "area_en", "VARCHAR(64)"), ("projects", "floors_en", "VARCHAR(64)"), ("projects", "units_en", "VARCHAR(64)"), ("projects", "client_en", "VARCHAR(128)"), ("projects", "summary_en", "TEXT"), ("projects", "body_en", "JSON"), ("articles", "title_en", "VARCHAR(256)"), ("articles", "date_en", "VARCHAR(64)"), ("articles", "category_en", "VARCHAR(64)"), ("articles", "read_en", "VARCHAR(32)"), ("articles", "excerpt_en", "TEXT"), ("articles", "body_en", "JSON"), ("testimonials", "name_en", "VARCHAR(128)"), ("testimonials", "role_en", "VARCHAR(128)"), ("testimonials", "quote_en", "TEXT"), ("faqs", "question_en", "TEXT"), ("faqs", "answer_en", "TEXT"), ("team", "name_en", "VARCHAR(128)"), ("team", "role_en", "VARCHAR(128)"), ("team", "bio_en", "TEXT"), ("timeline", "year_en", "VARCHAR(32)"), ("timeline", "title_en", "VARCHAR(256)"), ("timeline", "desc_en", "TEXT"), ("stats", "suffix_en", "VARCHAR(16)"), ("stats", "label_en", "VARCHAR(128)"), ("why_us", "title_en", "VARCHAR(256)"), ("why_us", "desc_en", "TEXT"), ("before_after", "title_en", "VARCHAR(256)"), ("before_after", "lead_en", "TEXT"), ("before_after", "before_label_en", "VARCHAR(128)"), ("before_after", "after_label_en", "VARCHAR(128)"), ("nav_items", "label_en", "VARCHAR(64)"), ("hero", "kicker_en", "TEXT"), ("hero", "title_en", "TEXT"), ("hero", "title_highlight_en", "VARCHAR(128)"), ("hero", "subtitle_en", "TEXT"), ("hero", "badge_value_en", "VARCHAR(32)"), ("hero", "badge_label_en", "TEXT"), ("hero", "cta_primary_label_en", "VARCHAR(128)"), ("hero", "cta_secondary_label_en", "VARCHAR(128)"), ("hero", "scroll_label_en", "VARCHAR(64)"), ("story", "title_en", "VARCHAR(256)"), ("story", "paragraphs_en", "JSON"), ("story", "quote_en", "TEXT"), ("story", "quote_author_en", "TEXT"), ("story", "stat1_value_en", "VARCHAR(32)"), ("story", "stat1_label_en", "VARCHAR(128)"), ("story", "stat2_value_en", "VARCHAR(32)"), ("story", "stat2_label_en", "VARCHAR(128)"), ("story", "link_label_en", "VARCHAR(128)"), ("story", "bento_en", "JSON"), ("section_headings", "label_en", "VARCHAR(64)"), ("section_headings", "title_en", "VARCHAR(256)"), ("section_headings", "lead_en", "TEXT"), ("section_headings", "cta_label_en", "VARCHAR(128)"), ("page_heroes", "title_en", "VARCHAR(256)"), ("page_heroes", "lead_en", "TEXT"), ("investment_models", "title_en", "VARCHAR(256)"), ("investment_models", "desc_en", "TEXT"), ("investment_models", "points_en", "JSON"), ("process_steps", "step_en", "VARCHAR(16)"), ("process_steps", "title_en", "VARCHAR(256)"), ("process_steps", "desc_en", "TEXT"), ("offices", "city_en", "VARCHAR(64)"), ("offices", "address_en", "TEXT"), ("offices", "hours_en", "VARCHAR(128)"), ("gallery_images", "alt_en", "TEXT"), ("status_labels", "label_en", "VARCHAR(64)"), ("contact_section", "label_en", "VARCHAR(64)"), ("contact_section", "title_en", "VARCHAR(256)"), ("contact_section", "lead_en", "TEXT"), ("contact_section", "address_label_en", "VARCHAR(64)"), ("contact_section", "phone_label_en", "VARCHAR(64)"), ("contact_section", "email_label_en", "VARCHAR(64)"), ("contact_section", "name_label_en", "VARCHAR(64)"), ("contact_section", "name_placeholder_en", "VARCHAR(128)"), ("contact_section", "phone_field_label_en", "VARCHAR(64)"), ("contact_section", "phone_placeholder_en", "VARCHAR(64)"), ("contact_section", "note_label_en", "VARCHAR(64)"), ("contact_section", "note_optional_label_en", "VARCHAR(64)"), ("contact_section", "note_placeholder_en", "VARCHAR(256)"), ("contact_section", "submit_label_en", "VARCHAR(128)"), ("contact_section", "sending_label_en", "VARCHAR(64)"), ("contact_section", "success_title_en", "VARCHAR(256)"), ("contact_section", "success_text_en", "TEXT"), ("contact_section", "success_button_label_en", "VARCHAR(128)"), ("newsletter", "title_en", "VARCHAR(256)"), ("newsletter", "lead_en", "TEXT"), ("newsletter", "placeholder_en", "VARCHAR(128)"), ("newsletter", "button_label_en", "VARCHAR(64)"), ("newsletter", "sending_label_en", "VARCHAR(64)"), ("newsletter", "success_text_en", "VARCHAR(256)"), ("footer", "about_text_en", "TEXT"), ("footer", "pages_heading_en", "VARCHAR(64)"), ("footer", "services_heading_en", "VARCHAR(64)"), ("footer", "contact_heading_en", "VARCHAR(64)"), ("footer", "copyright_en", "VARCHAR(256)"), ("footer", "tagline_en", "VARCHAR(256)"), ("footer", "socials_en", "JSON"), ("seo_meta", "title_en", "VARCHAR(256)"), ("seo_meta", "description_en", "TEXT"), ] def run_ddl_migrations() -> None: """ستون‌های جدید — برای PostgreSQL با ADD COLUMN IF NOT EXISTS.""" if engine.dialect.name != "postgresql": return with engine.begin() as conn: for table in POSITION_TABLES: conn.execute( text(f"ALTER TABLE {table} ADD COLUMN IF NOT EXISTS position INTEGER NOT NULL DEFAULT 0") ) # مقالات موجود منتشر شده‌اند؛ فقط مقاله‌های تازه می‌توانند پیش‌نویس باشند conn.execute( text("ALTER TABLE articles ADD COLUMN IF NOT EXISTS status VARCHAR(16) NOT NULL DEFAULT 'published'") ) conn.execute( text("ALTER TABLE articles ADD COLUMN IF NOT EXISTS seo_keyphrase VARCHAR(128) NOT NULL DEFAULT ''") ) # ستون‌های انگلیسی — رشته‌ها با پیش‌فرض خالی، JSON بدون پیش‌فرض (NULL = بدون ترجمه) for table, column, col_type in EN_COLUMNS: if col_type == "JSON": conn.execute(text(f"ALTER TABLE {table} ADD COLUMN IF NOT EXISTS {column} JSON")) else: conn.execute( text(f"ALTER TABLE {table} ADD COLUMN IF NOT EXISTS {column} {col_type} NOT NULL DEFAULT ''") ) def paragraphs_to_doc(paragraphs: list) -> dict: """تبدیل لیست پاراگراف‌های متنی (فرمت قدیمی بدنه مقاله) به سند TipTap.""" content = [] for para in paragraphs: if isinstance(para, str) and para.strip(): content.append({ "type": "paragraph", "content": [{"type": "text", "text": para}], }) if not content: content = [{"type": "paragraph"}] return {"type": "doc", "content": content} def migrate_article_bodies() -> None: """تبدیل بدنه مقالات از لیست پاراگراف به سند JSON — idempotent.""" from . import models # lazy — جلوگیری از import حلقه‌ای db = SessionLocal() try: changed = 0 for article in db.query(models.Article).all(): if isinstance(article.body, list): article.body = paragraphs_to_doc(article.body) changed += 1 db.commit() if changed: print(f"مقالات تبدیل‌شده به فرمت سند: {changed}") finally: db.close() def _is_empty(value: object) -> bool: """مقدار خالی انگلیسی — یعنی هنوز ترجمه‌ای ثبت نشده و قابل پر کردن است.""" if value is None or value == "" or value == [] or value == {}: return True if isinstance(value, dict) and value.get("content") == [{"type": "paragraph"}]: return True return False def _backfill_row(row, en_fields: dict) -> int: """پر کردن فیلدهای خالی *_en یک سطر از دیکشنری ترجمه — تعداد فیلدهای پرشده.""" filled = 0 for key, value in en_fields.items(): if not key.endswith("_en"): continue if not _is_empty(getattr(row, key, None)): continue setattr(row, key, value) filled += 1 return filled def backfill_en_content() -> None: """پر کردن ترجمه‌های انگلیسی سطرهای موجود از seed_data_en — idempotent. تطبیق سطرها: جدول‌های تک‌سطری مستقیم؛ جدول‌های دارای کلید طبیعی (slug/key/ page_key) با همان کلید؛ بقیه فقط وقتی سطرها با seed دست‌نخورده‌اند (تعداد و idها مطابق) و به‌ترتیب. هیچ مقدار انگلیسیِ ثبت‌شده بازنویسی نمی‌شود. """ from . import models, seed_data, seed_data_en # lazy — جلوگیری از import حلقه‌ای # (مدل، فهرست فارسی، فهرست انگلیسی، کلید تطبیق یا None برای ترتیبی) PLAN: list[tuple[object, list[dict], list[dict], str | None]] = [ (models.Brand, [seed_data.BRAND], [seed_data_en.BRAND], None), (models.Hero, [seed_data.HERO], [seed_data_en.HERO], None), (models.StorySection, [seed_data.STORY], [seed_data_en.STORY], None), (models.SectionHeadingContent, seed_data.SECTION_HEADINGS, seed_data_en.SECTION_HEADINGS, "key"), (models.PageHeroContent, seed_data.PAGE_HEROES, seed_data_en.PAGE_HEROES, "key"), (models.NavItem, seed_data.NAV_ITEMS, seed_data_en.NAV_ITEMS, None), (models.InvestmentModel, seed_data.INVESTMENT_MODELS, seed_data_en.INVESTMENT_MODELS, None), (models.ProcessStep, seed_data.PROCESS_STEPS, seed_data_en.PROCESS_STEPS, None), (models.Office, seed_data.OFFICES, seed_data_en.OFFICES, None), (models.GalleryImage, seed_data.GALLERY_IMAGES, seed_data_en.GALLERY_IMAGES, None), (models.StatusLabel, seed_data.STATUS_LABELS, seed_data_en.STATUS_LABELS, "key"), (models.ContactSectionContent, [seed_data.CONTACT_SECTION], [seed_data_en.CONTACT_SECTION], None), (models.NewsletterSectionContent, [seed_data.NEWSLETTER], [seed_data_en.NEWSLETTER], None), (models.FooterContent, [seed_data.FOOTER], [seed_data_en.FOOTER], None), (models.Service, seed_data.SERVICES, seed_data_en.SERVICES, None), (models.Stat, seed_data.STATS, seed_data_en.STATS, None), (models.Project, seed_data.PROJECTS, seed_data_en.PROJECTS, "slug"), (models.Article, seed_data.ARTICLES, seed_data_en.ARTICLES, "slug"), (models.Testimonial, seed_data.TESTIMONIALS, seed_data_en.TESTIMONIALS, None), (models.Faq, seed_data.FAQS, seed_data_en.FAQS, None), (models.TeamMember, seed_data.TEAM, seed_data_en.TEAM, None), (models.TimelineItem, seed_data.TIMELINE, seed_data_en.TIMELINE, None), (models.WhyUsItem, seed_data.WHY, seed_data_en.WHY, None), (models.BeforeAfter, [seed_data.BEFORE_AFTER], [seed_data_en.BEFORE_AFTER], None), (models.SeoMeta, seed_data.SEO_META, seed_data_en.SEO_META, "page_key"), ] db = SessionLocal() try: total_rows = 0 total_fields = 0 skipped: list[str] = [] for model, fa_rows, en_rows, match_key in PLAN: rows = db.query(model).all() if not rows or not en_rows: continue if match_key is not None: en_by_key = {r[match_key]: r for r in en_rows if match_key in r} matched = [(row, en_by_key.get(getattr(row, match_key, None))) for row in rows] pairs = [(row, en) for row, en in matched if en is not None] elif len(rows) == 1 and len(en_rows) == 1: pairs = [(rows[0], en_rows[0])] else: # تطبیق ترتیبی فقط برای جدول‌های دست‌نخورده — idها از ۱ پیوسته ids = [row.id for row in rows] if ids != list(range(1, len(rows) + 1)) or len(rows) != len(en_rows): skipped.append(getattr(model, "__tablename__", str(model))) continue pairs = list(zip(rows, en_rows)) for row, en_row in pairs: # بدنه مقاله انگلیسی هم مثل فارسی از پاراگراف به سند تبدیل می‌شود en_fields = dict(en_row) if model is models.Article and isinstance(en_fields.get("body_en"), list): en_fields["body_en"] = paragraphs_to_doc(en_fields["body_en"]) filled = _backfill_row(row, en_fields) total_fields += filled if filled: total_rows += 1 db.commit() if total_fields: print(f"ترجمه انگلیسی پر شد: {total_fields} فیلد در {total_rows} سطر.") if skipped: print(f"جدول‌های تغییرکرده (بدون backfill خودکار — از پنل مدیریت ترجمه کنید): {', '.join(skipped)}") finally: db.close() def run_all() -> None: run_ddl_migrations() migrate_article_bodies() backfill_en_content()