harness_zbt/backend/app/commands/build_analytics.py

from datetime import UTC, date, datetime

from sqlalchemy.dialects.mysql import insert

from app.core.config import Settings
from app.infrastructure.mysql.analytics_models import (
    AnalyticsSyncOffsetRecord,
    DimDateRecord,
    MetricDefinitionRecord,
)
from app.infrastructure.mysql.sessions import (
    DatabaseName,
    create_mysql_engine,
    create_session_factory,
)


def build_analytics_foundation(settings: Settings | None = None) -> None:
    resolved_settings = settings or Settings()
    engine = create_mysql_engine(resolved_settings, DatabaseName.ANALYTICS)
    factory = create_session_factory(engine)
    today = date.today()
    date_key = int(today.strftime("%Y%m%d"))
    with factory.begin() as session:
        session.execute(
            insert(DimDateRecord)
            .values(
                date_key=date_key,
                full_date=today,
                year=today.year,
                month=today.month,
                day=today.day,
            )
            .on_duplicate_key_update(full_date=today)
        )
        session.execute(
            insert(MetricDefinitionRecord)
            .values(
                metric_code="ORDER_COUNT",
                name="订单量",
                description="按筛选条件统计的订单数量",
                unit="笔",
                expression="COUNT(order_id)",
            )
            .on_duplicate_key_update(name="订单量")
        )
        session.execute(
            insert(AnalyticsSyncOffsetRecord)
            .values(source_name="core_outbox", updated_at=datetime.now(UTC))
            .on_duplicate_key_update(updated_at=datetime.now(UTC))
        )
    engine.dispose()


if __name__ == "__main__":
    build_analytics_foundation()
    print("Analytics 基础维度、指标定义和同步游标已构建。")