新增增长全局地图

.gitignore

@@ -23,6 +23,7 @@ tests/*
 tests/api/*
 !tests/api/__init__.py
 !tests/api/test_demand_feedback.py
+!tests/api/test_growth_category_tree.py
 !tests/api/test_video_discovery_records.py
 !tests/supply_agent/
 !tests/supply_infra/

alembic/versions/20260820_16_add_global_v2_snapshot_tables.py

@@ -0,0 +1,117 @@
+"""add global V2 snapshot tables
+
+Revision ID: 20260820_16
+Revises: 20260813_15
+Create Date: 2026-08-20
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "20260820_16"
+down_revision: str | None = "20260813_15"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def _timestamps() -> tuple[sa.Column, sa.Column]:
+    return (
+        sa.Column(
+            "create_time",
+            sa.DateTime(),
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+            nullable=False,
+        ),
+        sa.Column(
+            "update_time",
+            sa.DateTime(),
+            server_default=sa.text("CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"),
+            nullable=False,
+        ),
+    )
+
+
+def upgrade() -> None:
+    op.create_table(
+        "global_category_v2",
+        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
+        sa.Column("stable_id", sa.BigInteger(), nullable=True),
+        sa.Column("name", sa.String(length=512), nullable=True),
+        sa.Column("description", sa.Text(), nullable=True),
+        sa.Column("source_type", sa.String(length=64), nullable=True),
+        sa.Column("level", sa.BigInteger(), nullable=True),
+        sa.Column("parent_stable_id", sa.BigInteger(), nullable=True),
+        *_timestamps(),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("idx_global_category_v2_stable_id", "global_category_v2", ["stable_id"])
+    op.create_index(
+        "idx_global_category_v2_parent_stable_id",
+        "global_category_v2",
+        ["parent_stable_id"],
+    )
+
+    op.create_table(
+        "global_element_v2",
+        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
+        sa.Column("element_id", sa.BigInteger(), nullable=True),
+        sa.Column("name", sa.String(length=512), nullable=True),
+        sa.Column("description", sa.Text(), nullable=True),
+        sa.Column("belong_category_stable_id", sa.BigInteger(), nullable=True),
+        sa.Column("source_type", sa.String(length=64), nullable=True),
+        sa.Column("element_sub_type", sa.String(length=128), nullable=True),
+        *_timestamps(),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        "idx_global_element_v2_element_id",
+        "global_element_v2",
+        ["element_id"],
+    )
+    op.create_index(
+        "idx_global_element_v2_category_stable_id",
+        "global_element_v2",
+        ["belong_category_stable_id"],
+    )
+
+    op.create_table(
+        "global_source_element_data",
+        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
+        sa.Column("global_element_id", sa.BigInteger(), nullable=True),
+        sa.Column("global_category_stable_id", sa.BigInteger(), nullable=True),
+        sa.Column("element_type", sa.String(length=64), nullable=True),
+        sa.Column("source_element_id", sa.BigInteger(), nullable=True),
+        sa.Column("post_id", sa.String(length=128), nullable=True),
+        sa.Column("element_name", sa.String(length=512), nullable=True),
+        sa.Column("contribution", sa.Numeric(38, 18), nullable=True),
+        sa.Column("consumption_intent", sa.Numeric(38, 18), nullable=True),
+        sa.Column("click_intent", sa.Numeric(38, 18), nullable=True),
+        sa.Column("share_intent", sa.Numeric(38, 18), nullable=True),
+        *_timestamps(),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        "idx_global_source_element_global_element",
+        "global_source_element_data",
+        ["global_element_id"],
+    )
+    op.create_index(
+        "idx_global_source_element_category",
+        "global_source_element_data",
+        ["global_category_stable_id"],
+    )
+    op.create_index(
+        "idx_global_source_element_post_source",
+        "global_source_element_data",
+        ["post_id", "source_element_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("global_source_element_data")
+    op.drop_table("global_element_v2")
+    op.drop_table("global_category_v2")

alembic/versions/20260820_17_add_global_v2_unique_keys.py

@@ -0,0 +1,106 @@
+"""add global V2 unique business keys
+
+Revision ID: 20260820_17
+Revises: 20260820_16
+Create Date: 2026-08-20
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "20260820_17"
+down_revision: str | None = "20260820_16"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.alter_column(
+        "global_category_v2",
+        "stable_id",
+        existing_type=sa.BigInteger(),
+        nullable=False,
+    )
+    op.drop_index("idx_global_category_v2_stable_id", table_name="global_category_v2")
+    op.create_unique_constraint(
+        "uk_global_category_v2_stable_id",
+        "global_category_v2",
+        ["stable_id"],
+    )
+
+    op.alter_column(
+        "global_element_v2",
+        "element_id",
+        existing_type=sa.BigInteger(),
+        nullable=False,
+    )
+    op.drop_index("idx_global_element_v2_element_id", table_name="global_element_v2")
+    op.create_unique_constraint(
+        "uk_global_element_v2_element_id",
+        "global_element_v2",
+        ["element_id"],
+    )
+
+    op.alter_column(
+        "global_source_element_data",
+        "source_element_id",
+        existing_type=sa.BigInteger(),
+        nullable=False,
+    )
+    op.create_unique_constraint(
+        "uk_global_source_element_source_element_id",
+        "global_source_element_data",
+        ["source_element_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint(
+        "uk_global_source_element_source_element_id",
+        "global_source_element_data",
+        type_="unique",
+    )
+    op.alter_column(
+        "global_source_element_data",
+        "source_element_id",
+        existing_type=sa.BigInteger(),
+        nullable=True,
+    )
+
+    op.drop_constraint(
+        "uk_global_element_v2_element_id",
+        "global_element_v2",
+        type_="unique",
+    )
+    op.create_index(
+        "idx_global_element_v2_element_id",
+        "global_element_v2",
+        ["element_id"],
+    )
+    op.alter_column(
+        "global_element_v2",
+        "element_id",
+        existing_type=sa.BigInteger(),
+        nullable=True,
+    )
+
+    op.drop_constraint(
+        "uk_global_category_v2_stable_id",
+        "global_category_v2",
+        type_="unique",
+    )
+    op.create_index(
+        "idx_global_category_v2_stable_id",
+        "global_category_v2",
+        ["stable_id"],
+    )
+    op.alter_column(
+        "global_category_v2",
+        "stable_id",
+        existing_type=sa.BigInteger(),
+        nullable=True,
+    )

alembic/versions/20260820_18_add_channel_content_data.py

@@ -0,0 +1,42 @@
+"""add channel content data
+
+Revision ID: 20260820_18
+Revises: 20260820_17
+Create Date: 2026-08-20
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "20260820_18"
+down_revision: str | None = "20260820_17"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "channel_content_data",
+        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
+        sa.Column("channel_content_id", sa.String(length=128), nullable=False),
+        sa.Column("read_cnt", sa.BigInteger(), nullable=True),
+        sa.Column("like_cnt", sa.BigInteger(), nullable=True),
+        sa.Column("avg_read_cnt_30d", sa.Float(), nullable=True),
+        sa.Column("cal_fans_num", sa.Float(), nullable=True),
+        sa.Column("dt", sa.String(length=8), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "channel_content_id",
+            "dt",
+            name="uk_channel_content_data_content_dt",
+        ),
+    )
+    op.create_index("idx_channel_content_data_dt", "channel_content_data", ["dt"])
+
+
+def downgrade() -> None:
+    op.drop_table("channel_content_data")

alembic/versions/20260820_19_use_double_for_metric_fields.py

@@ -0,0 +1,87 @@
+"""use DOUBLE for global source and channel content metrics
+
+Revision ID: 20260820_19
+Revises: 20260820_18
+Create Date: 2026-08-20
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "20260820_19"
+down_revision: str | None = "20260820_18"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    for column_name in (
+        "contribution",
+        "consumption_intent",
+        "click_intent",
+        "share_intent",
+    ):
+        op.alter_column(
+            "global_source_element_data",
+            column_name,
+            existing_type=sa.Numeric(38, 18),
+            type_=sa.Double(),
+            existing_nullable=True,
+        )
+
+    for column_name, existing_type in (
+        ("read_cnt", sa.BigInteger()),
+        ("like_cnt", sa.BigInteger()),
+        ("avg_read_cnt_30d", sa.Float()),
+        ("cal_fans_num", sa.Float()),
+    ):
+        op.alter_column(
+            "channel_content_data",
+            column_name,
+            existing_type=existing_type,
+            type_=sa.Double(),
+            existing_nullable=True,
+        )
+
+
+def downgrade() -> None:
+    op.alter_column(
+        "channel_content_data",
+        "cal_fans_num",
+        existing_type=sa.Double(),
+        type_=sa.Float(),
+        existing_nullable=True,
+    )
+    op.alter_column(
+        "channel_content_data",
+        "avg_read_cnt_30d",
+        existing_type=sa.Double(),
+        type_=sa.Float(),
+        existing_nullable=True,
+    )
+    for column_name in ("like_cnt", "read_cnt"):
+        op.alter_column(
+            "channel_content_data",
+            column_name,
+            existing_type=sa.Double(),
+            type_=sa.BigInteger(),
+            existing_nullable=True,
+        )
+
+    for column_name in (
+        "share_intent",
+        "click_intent",
+        "consumption_intent",
+        "contribution",
+    ):
+        op.alter_column(
+            "global_source_element_data",
+            column_name,
+            existing_type=sa.Double(),
+            type_=sa.Numeric(38, 18),
+            existing_nullable=True,
+        )

alembic/versions/20260820_20_add_channel_content_rates.py

@@ -0,0 +1,36 @@
+"""add channel content rate fields
+
+Revision ID: 20260820_20
+Revises: 20260820_19
+Create Date: 2026-08-20
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "20260820_20"
+down_revision: str | None = "20260820_19"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    for column_name in ("avg_read_rate", "like_rate", "read_rate"):
+        op.add_column(
+            "channel_content_data",
+            sa.Column(
+                column_name,
+                sa.Double(),
+                server_default=sa.text("0"),
+                nullable=False,
+            ),
+        )
+
+
+def downgrade() -> None:
+    for column_name in ("read_rate", "like_rate", "avg_read_rate"):
+        op.drop_column("channel_content_data", column_name)

alembic/versions/20260820_21_add_global_category_content_weights.py

@@ -0,0 +1,68 @@
+"""add resumable daily global category content weights
+
+Revision ID: 20260820_21
+Revises: 20260820_20
+Create Date: 2026-08-20
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "20260820_21"
+down_revision: str | None = "20260820_20"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "global_category_content_weight_di",
+        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
+        sa.Column("stable_id", sa.BigInteger(), nullable=False),
+        sa.Column("parent_stable_id", sa.BigInteger(), nullable=True),
+        sa.Column("level", sa.BigInteger(), nullable=True),
+        sa.Column("biz_dt", sa.String(length=8), nullable=False),
+        sa.Column("status", sa.String(length=16), server_default="pending", nullable=False),
+        sa.Column("attempt_count", sa.Integer(), server_default="0", nullable=False),
+        sa.Column(
+            "direct_source_element_count", sa.BigInteger(), server_default="0", nullable=False
+        ),
+        sa.Column("source_element_count", sa.BigInteger(), server_default="0", nullable=False),
+        sa.Column("direct_contribution_sum", sa.Double(), server_default="0", nullable=False),
+        sa.Column("contribution_sum", sa.Double(), server_default="0", nullable=False),
+        sa.Column(
+            "direct_avg_read_rate_weighted_sum", sa.Double(), server_default="0", nullable=False
+        ),
+        sa.Column("avg_read_rate_weighted_sum", sa.Double(), server_default="0", nullable=False),
+        sa.Column("avg_read_rate_score", sa.Double(), server_default="0", nullable=False),
+        sa.Column("direct_like_rate_weighted_sum", sa.Double(), server_default="0", nullable=False),
+        sa.Column("like_rate_weighted_sum", sa.Double(), server_default="0", nullable=False),
+        sa.Column("like_rate_score", sa.Double(), server_default="0", nullable=False),
+        sa.Column("direct_read_rate_weighted_sum", sa.Double(), server_default="0", nullable=False),
+        sa.Column("read_rate_weighted_sum", sa.Double(), server_default="0", nullable=False),
+        sa.Column("read_rate_score", sa.Double(), server_default="0", nullable=False),
+        sa.Column("error_message", sa.Text(), nullable=True),
+        sa.Column("computed_at", sa.DateTime(), nullable=True),
+        sa.Column("create_time", sa.DateTime(), server_default=sa.func.now(), nullable=False),
+        sa.Column("update_time", sa.DateTime(), server_default=sa.func.now(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("stable_id", "biz_dt", name="uk_global_category_content_weight"),
+    )
+    op.create_index(
+        "idx_global_category_content_weight_dt_status",
+        "global_category_content_weight_di",
+        ["biz_dt", "status"],
+    )
+    op.create_index(
+        "idx_global_category_content_weight_parent",
+        "global_category_content_weight_di",
+        ["biz_dt", "parent_stable_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("global_category_content_weight_di")

alembic/versions/20260820_22_add_weight_table_column_comments.py

@@ -0,0 +1,181 @@
+"""add Chinese comments to category content weight columns
+
+Revision ID: 20260820_22
+Revises: 20260820_21
+Create Date: 2026-08-20
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "20260820_22"
+down_revision: str | None = "20260820_21"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+TABLE_NAME = "global_category_content_weight_di"
+
+COLUMNS = (
+    ("id", sa.BigInteger(), False, None, "自增主键", True),
+    ("stable_id", sa.BigInteger(), False, None, "全局分类节点稳定ID", False),
+    ("parent_stable_id", sa.BigInteger(), True, None, "父分类节点稳定ID；根节点为空", False),
+    ("level", sa.BigInteger(), True, None, "分类节点层级", False),
+    ("biz_dt", sa.String(length=8), False, None, "业务日期，格式YYYYMMDD", False),
+    (
+        "status",
+        sa.String(length=16),
+        False,
+        sa.text("'pending'"),
+        "计算状态：pending待计算、completed已完成、failed失败",
+        False,
+    ),
+    ("attempt_count", sa.Integer(), False, sa.text("0"), "计算尝试次数", False),
+    (
+        "direct_source_element_count",
+        sa.BigInteger(),
+        False,
+        sa.text("0"),
+        "直接归属当前节点且参与计算的明细数量",
+        False,
+    ),
+    (
+        "source_element_count",
+        sa.BigInteger(),
+        False,
+        sa.text("0"),
+        "当前节点及全部后代节点参与计算的明细数量",
+        False,
+    ),
+    (
+        "direct_contribution_sum",
+        sa.Double(),
+        False,
+        sa.text("0"),
+        "直接归属当前节点明细的contribution之和",
+        False,
+    ),
+    (
+        "contribution_sum",
+        sa.Double(),
+        False,
+        sa.text("0"),
+        "当前节点及全部后代节点的contribution之和（加权分母）",
+        False,
+    ),
+    (
+        "direct_avg_read_rate_weighted_sum",
+        sa.Double(),
+        False,
+        sa.text("0"),
+        "直接归属当前节点的avg_read_rate乘contribution之和",
+        False,
+    ),
+    (
+        "avg_read_rate_weighted_sum",
+        sa.Double(),
+        False,
+        sa.text("0"),
+        "当前节点及全部后代节点的avg_read_rate加权和",
+        False,
+    ),
+    (
+        "avg_read_rate_score",
+        sa.Double(),
+        False,
+        sa.text("0"),
+        "avg_read_rate加权平均值",
+        False,
+    ),
+    (
+        "direct_like_rate_weighted_sum",
+        sa.Double(),
+        False,
+        sa.text("0"),
+        "直接归属当前节点的like_rate乘contribution之和",
+        False,
+    ),
+    (
+        "like_rate_weighted_sum",
+        sa.Double(),
+        False,
+        sa.text("0"),
+        "当前节点及全部后代节点的like_rate加权和",
+        False,
+    ),
+    (
+        "like_rate_score",
+        sa.Double(),
+        False,
+        sa.text("0"),
+        "like_rate加权平均值",
+        False,
+    ),
+    (
+        "direct_read_rate_weighted_sum",
+        sa.Double(),
+        False,
+        sa.text("0"),
+        "直接归属当前节点的read_rate乘contribution之和",
+        False,
+    ),
+    (
+        "read_rate_weighted_sum",
+        sa.Double(),
+        False,
+        sa.text("0"),
+        "当前节点及全部后代节点的read_rate加权和",
+        False,
+    ),
+    (
+        "read_rate_score",
+        sa.Double(),
+        False,
+        sa.text("0"),
+        "read_rate加权平均值",
+        False,
+    ),
+    ("error_message", sa.Text(), True, None, "最近一次计算失败信息", False),
+    ("computed_at", sa.DateTime(), True, None, "节点计算完成时间", False),
+    (
+        "create_time",
+        sa.DateTime(),
+        False,
+        sa.text("CURRENT_TIMESTAMP"),
+        "创建时间",
+        False,
+    ),
+    (
+        "update_time",
+        sa.DateTime(),
+        False,
+        sa.text("CURRENT_TIMESTAMP"),
+        "更新时间",
+        False,
+    ),
+)
+
+
+def _set_comments(*, remove: bool) -> None:
+    for name, column_type, nullable, server_default, comment, autoincrement in COLUMNS:
+        op.alter_column(
+            TABLE_NAME,
+            name,
+            existing_type=column_type,
+            existing_nullable=nullable,
+            existing_server_default=server_default,
+            existing_autoincrement=autoincrement,
+            comment=None if remove else comment,
+            existing_comment=comment if remove else None,
+        )
+
+
+def upgrade() -> None:
+    _set_comments(remove=False)
+
+
+def downgrade() -> None:
+    _set_comments(remove=True)

api/app.py

@@ -25,6 +25,7 @@ from api.services.agent_catalog import (
 )
 from api.services.auth import ensure_bootstrap_admin
 from api.services.category_tree import build_category_tree
+from api.services.growth_category_tree import build_growth_category_tree
 from api.services.demand_belong_category import list_demand_belong_categories
 from api.services.demand_grade import list_demand_grades
 from api.services.demand_grade_videos import list_videos_for_demand_grade
@@ -223,6 +224,18 @@ def category_tree(
     return build_category_tree(biz_dt=biz_dt)
 
 
+@app.get("/api/growth-category-tree")
+def growth_category_tree(
+    biz_dt: str | None = Query(
+        default=None,
+        pattern=r"^\d{8}$",
+        description="业务日 YYYYMMDD；省略则取最新一个全部节点计算完成的日期",
+    ),
+) -> dict:
+    """Return the V2 category tree with three growth heat dimensions."""
+    return build_growth_category_tree(biz_dt=biz_dt)
+
+
 @app.get("/api/demand-belong-category")
 def demand_belong_category() -> dict:
     """Return all active demand_belong_category rows in one response."""

api/auth_middleware.py

@@ -19,6 +19,7 @@ _AUTHENTICATED_USER_PATHS = {
 }
 _NORMAL_USER_PATHS = {
     ("GET", "/api/category-tree"),
+    ("GET", "/api/growth-category-tree"),
     ("GET", "/api/demand-grade"),
     ("GET", "/api/video-discovery/demands"),
     ("GET", "/api/video-discovery/runs"),

api/services/growth_category_tree.py

@@ -0,0 +1,95 @@
+"""Build the growth heat map from global_category_v2 and its daily weights."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy import select
+
+from supply_infra.db.models.global_category_content_weight import (
+    GlobalCategoryContentWeight,
+)
+from supply_infra.db.models.global_v2 import GlobalCategoryV2
+from supply_infra.db.repositories.global_category_content_weight_repo import (
+    GlobalCategoryContentWeightRepository,
+)
+from supply_infra.db.session import get_session
+
+GROWTH_DIM_META: list[dict[str, str]] = [
+    {"key": "read_rate", "label": "阅读率"},
+    {"key": "avg_read_rate", "label": "平均阅读率"},
+    {"key": "like_rate", "label": "点赞率"},
+]
+
+
+def _normalize_parent_id(parent_id: int | None, category_ids: set[int]) -> int | None:
+    if parent_id in (None, 0) or int(parent_id) not in category_ids:
+        return None
+    return int(parent_id)
+
+
+def _build_growth_tree(
+    categories: list[GlobalCategoryV2],
+    weights: list[GlobalCategoryContentWeight],
+) -> list[dict[str, Any]]:
+    category_ids = {int(category.stable_id) for category in categories}
+    children_by_parent: dict[int | None, list[GlobalCategoryV2]] = {}
+    for category in categories:
+        parent_id = _normalize_parent_id(category.parent_stable_id, category_ids)
+        children_by_parent.setdefault(parent_id, []).append(category)
+    for children in children_by_parent.values():
+        children.sort(key=lambda row: (row.level or 0, int(row.stable_id)))
+
+    weight_by_stable_id = {int(row.stable_id): row for row in weights}
+
+    def to_node(category: GlobalCategoryV2, ancestors: frozenset[int]) -> dict[str, Any]:
+        stable_id = int(category.stable_id)
+        weight = weight_by_stable_id.get(stable_id)
+        count = int(weight.source_element_count or 0) if weight else 0
+        next_ancestors = ancestors | {stable_id}
+        children = [
+            to_node(child, next_ancestors)
+            for child in children_by_parent.get(stable_id, [])
+            if int(child.stable_id) not in next_ancestors
+        ]
+        return {
+            "id": stable_id,
+            "name": category.name,
+            "level": category.level,
+            "description": category.description,
+            "weights": {
+                "read_rate": float(weight.read_rate_score) if weight else None,
+                "avg_read_rate": float(weight.avg_read_rate_score) if weight else None,
+                "like_rate": float(weight.like_rate_score) if weight else None,
+            },
+            "counts": {
+                "read_rate": count,
+                "avg_read_rate": count,
+                "like_rate": count,
+            },
+            "children": children,
+        }
+
+    return [to_node(root, frozenset()) for root in children_by_parent.get(None, [])]
+
+
+def build_growth_category_tree(biz_dt: str | None = None) -> dict[str, Any]:
+    """Return the V2 category tree with the requested/latest complete daily scores."""
+    with get_session() as session:
+        categories = list(
+            session.scalars(
+                select(GlobalCategoryV2).order_by(
+                    GlobalCategoryV2.level, GlobalCategoryV2.stable_id
+                )
+            ).all()
+        )
+        weight_repo = GlobalCategoryContentWeightRepository(session)
+        resolved_dt = biz_dt or weight_repo.get_latest_completed_biz_dt()
+        weights = weight_repo.list_completed_models_by_biz_dt(resolved_dt) if resolved_dt else []
+        nodes = _build_growth_tree(categories, weights)
+
+    return {
+        "biz_dt": resolved_dt,
+        "dims": GROWTH_DIM_META,
+        "nodes": nodes,
+    }

api/services/scheduler.py

@@ -11,6 +11,8 @@ from supply_infra.pipeline.run_service import submit_pipeline_run
 from supply_infra.scheduler.constants import (
     AIGC_PUBLISH_JOB_ID,
     AIGC_PUBLISH_JOB_NAME,
+    GLOBAL_V2_SNAPSHOT_JOB_ID,
+    GLOBAL_V2_SNAPSHOT_JOB_NAME,
     SUPPLY_PIPELINE_JOB_ID,
     SUPPLY_PIPELINE_JOB_NAME,
 )
@@ -30,6 +32,14 @@ def list_triggerable_jobs() -> list[dict[str, Any]]:
             "deprecated": False,
             "steps": [step.key for step in PIPELINE_STEPS],
         },
+        {
+            "id": GLOBAL_V2_SNAPSHOT_JOB_ID,
+            "name": GLOBAL_V2_SNAPSHOT_JOB_NAME,
+            "description": "独立读取昨天全量 ODPS 数据，按唯一 key 增量写入四张业务表",
+            "accepts_biz_dt": True,
+            "deprecated": False,
+            "steps": [],
+        },
         {
             "id": AIGC_PUBLISH_JOB_ID,
             "name": AIGC_PUBLISH_JOB_NAME,
@@ -61,6 +71,12 @@ def run_scheduler_job(
             skip_published=True,
             any_biz_dt=biz_dt is None,
         )
+    if job_id == GLOBAL_V2_SNAPSHOT_JOB_ID:
+        from supply_infra.scheduler.jobs.sync_global_v2_snapshot import (
+            sync_global_v2_snapshot,
+        )
+
+        return sync_global_v2_snapshot(partition_date=biz_dt)
     raise KeyError(job_id)
 
 
@@ -89,6 +105,12 @@ def scheduler_status() -> dict[str, Any]:
                 .replace(tzinfo=CHINA_TIMEZONE)
                 .isoformat(),
             },
+            {
+                "id": GLOBAL_V2_SNAPSHOT_JOB_ID,
+                "name": GLOBAL_V2_SNAPSHOT_JOB_NAME,
+                "cron": "0 6 * * *",
+                "timezone": "Asia/Shanghai",
+            },
             {
                 "id": AIGC_PUBLISH_JOB_ID,
                 "name": AIGC_PUBLISH_JOB_NAME,

supply_infra/db/models/__init__.py

@@ -4,6 +4,7 @@ from supply_infra.db.models.agent_document_injection import AgentDocumentInjecti
 from supply_infra.db.models.auth_session import AuthSession
 from supply_infra.db.models.auth_user import AuthUser
 from supply_infra.db.models.category_tree_weight import CategoryTreeWeight
+from supply_infra.db.models.channel_content_data import ChannelContentData
 from supply_infra.db.models.demand_belong_category import DemandBelongCategory
 from supply_infra.db.models.demand_belong_pool_rel import DemandBelongPoolRel
 from supply_infra.db.models.demand_feedback import DemandFeedback
@@ -22,6 +23,14 @@ from supply_infra.db.models.demand_video_expansion import (
 from supply_infra.db.models.generated_demand import GeneratedDemand
 from supply_infra.db.models.global_tree_category import GlobalTreeCategory
 from supply_infra.db.models.global_tree_element import GlobalTreeElement
+from supply_infra.db.models.global_category_content_weight import (
+    GlobalCategoryContentWeight,
+)
+from supply_infra.db.models.global_v2 import (
+    GlobalCategoryV2,
+    GlobalElementV2,
+    GlobalSourceElementData,
+)
 from supply_infra.db.models.multi_demand_pool_di import MultiDemandPoolDi
 from supply_infra.db.models.multi_demand_video_detail import MultiDemandVideoDetail
 from supply_infra.db.models.multi_demand_video_point import MultiDemandVideoPoint
@@ -44,6 +53,7 @@ __all__ = [
     "AuthSession",
     "AuthUser",
     "CategoryTreeWeight",
+    "ChannelContentData",
     "DemandBelongCategory",
     "DemandBelongPoolRel",
     "DemandFeedback",
@@ -58,6 +68,10 @@ __all__ = [
     "GeneratedDemand",
     "GlobalTreeCategory",
     "GlobalTreeElement",
+    "GlobalCategoryContentWeight",
+    "GlobalCategoryV2",
+    "GlobalElementV2",
+    "GlobalSourceElementData",
     "MultiDemandPoolDi",
     "MultiDemandVideoDetail",
     "MultiDemandVideoPoint",

supply_infra/db/models/channel_content_data.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from sqlalchemy import BigInteger, Double, Index, String, UniqueConstraint, text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from supply_infra.db.base import Base
+
+
+class ChannelContentData(Base):
+    """Daily WeChat article read-rate snapshot imported from ODPS."""
+
+    __tablename__ = "channel_content_data"
+    __table_args__ = (
+        UniqueConstraint(
+            "channel_content_id",
+            "dt",
+            name="uk_channel_content_data_content_dt",
+        ),
+        Index("idx_channel_content_data_dt", "dt"),
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    channel_content_id: Mapped[str] = mapped_column(String(128), nullable=False)
+    read_cnt: Mapped[float | None] = mapped_column(Double, nullable=True)
+    like_cnt: Mapped[float | None] = mapped_column(Double, nullable=True)
+    avg_read_cnt_30d: Mapped[float | None] = mapped_column(Double, nullable=True)
+    cal_fans_num: Mapped[float | None] = mapped_column(Double, nullable=True)
+    avg_read_rate: Mapped[float] = mapped_column(
+        Double, nullable=False, server_default=text("0")
+    )
+    like_rate: Mapped[float] = mapped_column(
+        Double, nullable=False, server_default=text("0")
+    )
+    read_rate: Mapped[float] = mapped_column(
+        Double, nullable=False, server_default=text("0")
+    )
+    dt: Mapped[str] = mapped_column(String(8), nullable=False)

supply_infra/db/models/global_category_content_weight.py

@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import (
+    BigInteger,
+    DateTime,
+    Double,
+    Index,
+    Integer,
+    String,
+    Text,
+    UniqueConstraint,
+    func,
+    text,
+)
+from sqlalchemy.orm import Mapped, mapped_column
+
+from supply_infra.db.base import Base
+
+
+class GlobalCategoryContentWeight(Base):
+    """Daily, resumable content-weight rollup for one global category node."""
+
+    __tablename__ = "global_category_content_weight_di"
+    __table_args__ = (
+        UniqueConstraint(
+            "stable_id",
+            "biz_dt",
+            name="uk_global_category_content_weight",
+        ),
+        Index(
+            "idx_global_category_content_weight_dt_status",
+            "biz_dt",
+            "status",
+        ),
+        Index(
+            "idx_global_category_content_weight_parent",
+            "biz_dt",
+            "parent_stable_id",
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(
+        BigInteger, primary_key=True, autoincrement=True, comment="自增主键"
+    )
+    stable_id: Mapped[int] = mapped_column(BigInteger, nullable=False, comment="全局分类节点稳定ID")
+    parent_stable_id: Mapped[int | None] = mapped_column(
+        BigInteger, nullable=True, comment="父分类节点稳定ID；根节点为空"
+    )
+    level: Mapped[int | None] = mapped_column(BigInteger, nullable=True, comment="分类节点层级")
+    biz_dt: Mapped[str] = mapped_column(String(8), nullable=False, comment="业务日期，格式YYYYMMDD")
+    status: Mapped[str] = mapped_column(
+        String(16),
+        nullable=False,
+        server_default=text("'pending'"),
+        comment="计算状态：pending待计算、completed已完成、failed失败",
+    )
+    attempt_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, server_default=text("0"), comment="计算尝试次数"
+    )
+
+    direct_source_element_count: Mapped[int] = mapped_column(
+        BigInteger,
+        nullable=False,
+        server_default=text("0"),
+        comment="直接归属当前节点且参与计算的明细数量",
+    )
+    source_element_count: Mapped[int] = mapped_column(
+        BigInteger,
+        nullable=False,
+        server_default=text("0"),
+        comment="当前节点及全部后代节点参与计算的明细数量",
+    )
+    direct_contribution_sum: Mapped[float] = mapped_column(
+        Double,
+        nullable=False,
+        server_default=text("0"),
+        comment="直接归属当前节点明细的contribution之和",
+    )
+    contribution_sum: Mapped[float] = mapped_column(
+        Double,
+        nullable=False,
+        server_default=text("0"),
+        comment="当前节点及全部后代节点的contribution之和（加权分母）",
+    )
+
+    direct_avg_read_rate_weighted_sum: Mapped[float] = mapped_column(
+        Double,
+        nullable=False,
+        server_default=text("0"),
+        comment="直接归属当前节点的avg_read_rate乘contribution之和",
+    )
+    avg_read_rate_weighted_sum: Mapped[float] = mapped_column(
+        Double,
+        nullable=False,
+        server_default=text("0"),
+        comment="当前节点及全部后代节点的avg_read_rate加权和",
+    )
+    avg_read_rate_score: Mapped[float] = mapped_column(
+        Double,
+        nullable=False,
+        server_default=text("0"),
+        comment="avg_read_rate加权平均值",
+    )
+
+    direct_like_rate_weighted_sum: Mapped[float] = mapped_column(
+        Double,
+        nullable=False,
+        server_default=text("0"),
+        comment="直接归属当前节点的like_rate乘contribution之和",
+    )
+    like_rate_weighted_sum: Mapped[float] = mapped_column(
+        Double,
+        nullable=False,
+        server_default=text("0"),
+        comment="当前节点及全部后代节点的like_rate加权和",
+    )
+    like_rate_score: Mapped[float] = mapped_column(
+        Double,
+        nullable=False,
+        server_default=text("0"),
+        comment="like_rate加权平均值",
+    )
+
+    direct_read_rate_weighted_sum: Mapped[float] = mapped_column(
+        Double,
+        nullable=False,
+        server_default=text("0"),
+        comment="直接归属当前节点的read_rate乘contribution之和",
+    )
+    read_rate_weighted_sum: Mapped[float] = mapped_column(
+        Double,
+        nullable=False,
+        server_default=text("0"),
+        comment="当前节点及全部后代节点的read_rate加权和",
+    )
+    read_rate_score: Mapped[float] = mapped_column(
+        Double,
+        nullable=False,
+        server_default=text("0"),
+        comment="read_rate加权平均值",
+    )
+
+    error_message: Mapped[str | None] = mapped_column(
+        Text, nullable=True, comment="最近一次计算失败信息"
+    )
+    computed_at: Mapped[datetime | None] = mapped_column(
+        DateTime, nullable=True, comment="节点计算完成时间"
+    )
+    create_time: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.now(), comment="创建时间"
+    )
+    update_time: Mapped[datetime] = mapped_column(
+        DateTime,
+        nullable=False,
+        server_default=func.now(),
+        onupdate=func.now(),
+        comment="更新时间",
+    )

supply_infra/db/models/global_v2.py

@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+from datetime import datetime
+from sqlalchemy import (
+    BigInteger,
+    DateTime,
+    Double,
+    Index,
+    String,
+    Text,
+    UniqueConstraint,
+    func,
+)
+from sqlalchemy.orm import Mapped, mapped_column
+
+from supply_infra.db.base import Base
+
+
+class GlobalCategoryV2(Base):
+    """Latest active substantive category snapshot from ODPS global_category."""
+
+    __tablename__ = "global_category_v2"
+    __table_args__ = (
+        UniqueConstraint("stable_id", name="uk_global_category_v2_stable_id"),
+        Index("idx_global_category_v2_parent_stable_id", "parent_stable_id"),
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    stable_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
+    name: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    source_type: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    level: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
+    parent_stable_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
+    create_time: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.now()
+    )
+    update_time: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.now(), onupdate=func.now()
+    )
+
+
+class GlobalElementV2(Base):
+    """Latest active substantive element snapshot from ODPS global_element."""
+
+    __tablename__ = "global_element_v2"
+    __table_args__ = (
+        UniqueConstraint("element_id", name="uk_global_element_v2_element_id"),
+        Index(
+            "idx_global_element_v2_category_stable_id",
+            "belong_category_stable_id",
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    element_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
+    name: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    belong_category_stable_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
+    source_type: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    element_sub_type: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    create_time: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.now()
+    )
+    update_time: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.now(), onupdate=func.now()
+    )
+
+
+class GlobalSourceElementData(Base):
+    """Latest source-element mapping and weight snapshot from ODPS."""
+
+    __tablename__ = "global_source_element_data"
+    __table_args__ = (
+        UniqueConstraint(
+            "source_element_id", name="uk_global_source_element_source_element_id"
+        ),
+        Index("idx_global_source_element_global_element", "global_element_id"),
+        Index("idx_global_source_element_category", "global_category_stable_id"),
+        Index(
+            "idx_global_source_element_post_source",
+            "post_id",
+            "source_element_id",
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    global_element_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
+    global_category_stable_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
+    element_type: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    source_element_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
+    post_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    element_name: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    contribution: Mapped[float | None] = mapped_column(Double, nullable=True)
+    consumption_intent: Mapped[float | None] = mapped_column(Double, nullable=True)
+    click_intent: Mapped[float | None] = mapped_column(Double, nullable=True)
+    share_intent: Mapped[float | None] = mapped_column(Double, nullable=True)
+    create_time: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.now()
+    )
+    update_time: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.now(), onupdate=func.now()
+    )

supply_infra/db/repositories/channel_content_data_repo.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy.dialects.mysql import insert
+from sqlalchemy.orm import Session
+
+from supply_infra.db.models.channel_content_data import ChannelContentData
+
+_BATCH_SIZE = 1000
+
+
+class ChannelContentDataRepository:
+    def __init__(self, session: Session) -> None:
+        self.session = session
+
+    def insert_new(self, rows: list[dict[str, Any]]) -> int:
+        inserted = 0
+        for start in range(0, len(rows), _BATCH_SIZE):
+            batch = rows[start : start + _BATCH_SIZE]
+            result = self.session.execute(
+                insert(ChannelContentData).values(batch).prefix_with("IGNORE")
+            )
+            inserted += int(result.rowcount or 0)
+        return inserted

supply_infra/db/repositories/global_category_content_weight_repo.py

@@ -0,0 +1,175 @@
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy import case, func, select
+from sqlalchemy.dialects.mysql import insert
+from sqlalchemy.orm import Session
+
+from supply_infra.db.models.channel_content_data import ChannelContentData
+from supply_infra.db.models.global_category_content_weight import (
+    GlobalCategoryContentWeight,
+)
+from supply_infra.db.models.global_v2 import GlobalCategoryV2, GlobalSourceElementData
+
+_INSERT_BATCH_SIZE = 500
+
+
+class GlobalCategoryContentWeightRepository:
+    def __init__(self, session: Session) -> None:
+        self.session = session
+
+    def count_by_biz_dt(self, biz_dt: str) -> int:
+        stmt = select(func.count()).where(GlobalCategoryContentWeight.biz_dt == biz_dt)
+        return int(self.session.scalar(stmt) or 0)
+
+    def get_latest_completed_biz_dt(self) -> str | None:
+        """Return the newest date whose node rows are all completed."""
+        completed_count = func.sum(
+            case((GlobalCategoryContentWeight.status == "completed", 1), else_=0)
+        )
+        stmt = (
+            select(GlobalCategoryContentWeight.biz_dt)
+            .group_by(GlobalCategoryContentWeight.biz_dt)
+            .having(completed_count == func.count())
+            .order_by(GlobalCategoryContentWeight.biz_dt.desc())
+            .limit(1)
+        )
+        value = self.session.scalar(stmt)
+        return str(value) if value else None
+
+    def list_completed_models_by_biz_dt(self, biz_dt: str) -> list[GlobalCategoryContentWeight]:
+        return list(
+            self.session.scalars(
+                select(GlobalCategoryContentWeight).where(
+                    GlobalCategoryContentWeight.biz_dt == biz_dt,
+                    GlobalCategoryContentWeight.status == "completed",
+                )
+            ).all()
+        )
+
+    def direct_stats(self, biz_dt: str) -> dict[int, dict[str, float | int]]:
+        contribution = GlobalSourceElementData.contribution
+        stmt = (
+            select(
+                GlobalSourceElementData.global_category_stable_id.label("stable_id"),
+                func.count().label("source_element_count"),
+                func.sum(contribution).label("contribution_sum"),
+                func.sum(ChannelContentData.avg_read_rate * contribution).label(
+                    "avg_read_rate_weighted_sum"
+                ),
+                func.sum(ChannelContentData.like_rate * contribution).label(
+                    "like_rate_weighted_sum"
+                ),
+                func.sum(ChannelContentData.read_rate * contribution).label(
+                    "read_rate_weighted_sum"
+                ),
+            )
+            .join(
+                ChannelContentData,
+                (ChannelContentData.channel_content_id == GlobalSourceElementData.post_id)
+                & (ChannelContentData.dt == biz_dt),
+            )
+            .where(
+                GlobalSourceElementData.contribution.is_not(None),
+                GlobalSourceElementData.global_category_stable_id.is_not(None),
+            )
+            .group_by(GlobalSourceElementData.global_category_stable_id)
+        )
+        return {
+            int(row.stable_id): {
+                "source_element_count": int(row.source_element_count or 0),
+                "contribution_sum": float(row.contribution_sum or 0.0),
+                "avg_read_rate_weighted_sum": float(row.avg_read_rate_weighted_sum or 0.0),
+                "like_rate_weighted_sum": float(row.like_rate_weighted_sum or 0.0),
+                "read_rate_weighted_sum": float(row.read_rate_weighted_sum or 0.0),
+            }
+            for row in self.session.execute(stmt).all()
+        }
+
+    def initialize_day(self, biz_dt: str) -> dict[str, int]:
+        existing = self.count_by_biz_dt(biz_dt)
+        if existing:
+            return {"initialized": 0, "existing": existing}
+
+        categories = self.session.execute(
+            select(
+                GlobalCategoryV2.stable_id,
+                GlobalCategoryV2.parent_stable_id,
+                GlobalCategoryV2.level,
+            )
+        ).all()
+        direct_by_id = self.direct_stats(biz_dt)
+        rows: list[dict[str, Any]] = []
+        for category in categories:
+            stable_id = int(category.stable_id)
+            direct = direct_by_id.get(stable_id, {})
+            rows.append(
+                {
+                    "stable_id": stable_id,
+                    "parent_stable_id": (
+                        int(category.parent_stable_id)
+                        if category.parent_stable_id not in (None, 0)
+                        else None
+                    ),
+                    "level": int(category.level) if category.level is not None else None,
+                    "biz_dt": biz_dt,
+                    "status": "pending",
+                    "direct_source_element_count": int(direct.get("source_element_count", 0)),
+                    "direct_contribution_sum": float(direct.get("contribution_sum", 0.0)),
+                    "direct_avg_read_rate_weighted_sum": float(
+                        direct.get("avg_read_rate_weighted_sum", 0.0)
+                    ),
+                    "direct_like_rate_weighted_sum": float(
+                        direct.get("like_rate_weighted_sum", 0.0)
+                    ),
+                    "direct_read_rate_weighted_sum": float(
+                        direct.get("read_rate_weighted_sum", 0.0)
+                    ),
+                }
+            )
+
+        inserted = 0
+        for start in range(0, len(rows), _INSERT_BATCH_SIZE):
+            batch = rows[start : start + _INSERT_BATCH_SIZE]
+            result = self.session.execute(
+                insert(GlobalCategoryContentWeight).values(batch).prefix_with("IGNORE")
+            )
+            inserted += int(result.rowcount or 0)
+        return {"initialized": inserted, "existing": 0}
+
+    def list_day(self, biz_dt: str) -> list[dict[str, Any]]:
+        rows = self.session.scalars(
+            select(GlobalCategoryContentWeight)
+            .where(GlobalCategoryContentWeight.biz_dt == biz_dt)
+            .order_by(
+                GlobalCategoryContentWeight.level.desc(),
+                GlobalCategoryContentWeight.stable_id,
+            )
+        ).all()
+        fields = (
+            "id",
+            "stable_id",
+            "parent_stable_id",
+            "level",
+            "status",
+            "attempt_count",
+            "direct_source_element_count",
+            "source_element_count",
+            "direct_contribution_sum",
+            "contribution_sum",
+            "direct_avg_read_rate_weighted_sum",
+            "avg_read_rate_weighted_sum",
+            "avg_read_rate_score",
+            "direct_like_rate_weighted_sum",
+            "like_rate_weighted_sum",
+            "like_rate_score",
+            "direct_read_rate_weighted_sum",
+            "read_rate_weighted_sum",
+            "read_rate_score",
+        )
+        return [{field: getattr(row, field) for field in fields} for row in rows]
+
+    def save_completed(self, rows: list[dict[str, Any]]) -> None:
+        if rows:
+            self.session.bulk_update_mappings(GlobalCategoryContentWeight, rows)

supply_infra/db/repositories/global_v2_repo.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy.dialects.mysql import insert
+from sqlalchemy.orm import Session
+
+from supply_infra.db.models.global_v2 import (
+    GlobalCategoryV2,
+    GlobalElementV2,
+    GlobalSourceElementData,
+)
+
+_BATCH_SIZE = 1000
+
+
+class GlobalV2SnapshotRepository:
+    """Insert unseen ODPS rows and ignore keys already present in MySQL."""
+
+    def __init__(self, session: Session) -> None:
+        self.session = session
+
+    def _insert_new(self, model: type, rows: list[dict[str, Any]]) -> int:
+        inserted = 0
+        for start in range(0, len(rows), _BATCH_SIZE):
+            batch = rows[start : start + _BATCH_SIZE]
+            result = self.session.execute(
+                insert(model).values(batch).prefix_with("IGNORE")
+            )
+            inserted += int(result.rowcount or 0)
+        return inserted
+
+    def insert_new(
+        self,
+        *,
+        categories: list[dict[str, Any]],
+        elements: list[dict[str, Any]],
+        source_elements: list[dict[str, Any]],
+    ) -> dict[str, int]:
+        return {
+            "global_category_v2": self._insert_new(GlobalCategoryV2, categories),
+            "global_element_v2": self._insert_new(GlobalElementV2, elements),
+            "global_source_element_data": self._insert_new(
+                GlobalSourceElementData, source_elements
+            ),
+        }

supply_infra/odps/client.py

@@ -256,6 +256,104 @@ class ODPSClient:
         """
         return self.execute_sql(sql)
 
+    def fetch_global_categories_v2(self, bizdate: str) -> list[dict[str, Any]]:
+        """Fetch the active substantive global-category snapshot."""
+        sql = f"""
+        SELECT  stable_id
+                ,name
+                ,description
+                ,source_type
+                ,level
+                ,parent_stable_id
+        FROM    loghubods.global_category
+        WHERE   dt = '{bizdate}'
+        AND     retired_at_execution_id IS NULL
+        AND     source_type = '实质'
+        ORDER BY level
+        """
+        return self.execute_sql(sql)
+
+    def fetch_global_elements_v2(self, bizdate: str) -> list[dict[str, Any]]:
+        """Fetch the active substantive global-element snapshot."""
+        sql = f"""
+        SELECT  id AS element_id
+                ,name
+                ,description
+                ,belong_category_stable_id
+                ,source_type
+                ,element_sub_type
+        FROM    loghubods.global_element
+        WHERE   dt = '{bizdate}'
+        AND     retired_at_execution_id IS NULL
+        AND     source_type = '实质'
+        """
+        return self.execute_sql(sql)
+
+    def fetch_global_source_element_data(
+        self, bizdate: str
+    ) -> list[dict[str, Any]]:
+        """Fetch source-element mappings and their intent weights."""
+        sql = f"""
+        SELECT  t1.global_element_id
+                ,t1.global_category_stable_id
+                ,t1.element_type
+                ,t1.source_element_id
+                ,t1.post_id
+                ,t1.element_name
+                ,t2.contribution
+                ,t2.consumption_intent
+                ,t2.click_intent
+                ,t2.share_intent
+        FROM    loghubods.element_classification_mapping t1
+        INNER JOIN loghubods.post t3
+        ON      t1.post_id = t3.post_id
+        AND     t3.dt = '{bizdate}'
+        AND     t3.platform = 'gongzhonghao'
+        INNER JOIN loghubods.public_post_element_weight t2
+        ON      t1.post_id = t2.post_id
+        AND     t1.source_element_id = t2.element_id
+        AND     t2.dt = '{bizdate}'
+        WHERE   t1.dt = '{bizdate}'
+        AND     t1.element_type = '实质'
+        AND     t1.source_table = 'post_decode_topic_point_element'
+        """
+        return self.execute_sql(sql)
+
+    def fetch_channel_content_data(self, bizdate: str) -> list[dict[str, Any]]:
+        """Fetch one daily WeChat article read-rate partition."""
+        sql = f"""
+        SELECT  channel_content_id
+                ,read_cnt
+                ,like_cnt
+                ,avg_read_cnt_30d
+                ,cal_fans_num
+                ,CASE
+                    WHEN read_cnt IS NULL
+                      OR avg_read_cnt_30d IS NULL
+                      OR avg_read_cnt_30d = 0
+                    THEN 0.0
+                    ELSE CAST(read_cnt AS DOUBLE) / avg_read_cnt_30d
+                 END AS avg_read_rate
+                ,CASE
+                    WHEN like_cnt IS NULL
+                      OR read_cnt IS NULL
+                      OR read_cnt = 0
+                    THEN 0.0
+                    ELSE CAST(like_cnt AS DOUBLE) / CAST(read_cnt AS DOUBLE)
+                 END AS like_rate
+                ,CASE
+                    WHEN read_cnt IS NULL
+                      OR cal_fans_num IS NULL
+                      OR cal_fans_num = 0
+                    THEN 0.0
+                    ELSE CAST(read_cnt AS DOUBLE) / cal_fans_num
+                 END AS read_rate
+                ,dt
+        FROM    loghubods.wechat_article_read_rate_analysis
+        WHERE   dt = '{bizdate}'
+        """
+        return self.execute_sql(sql)
+
 
 @lru_cache
 def get_odps_client() -> ODPSClient:

supply_infra/scheduler/app.py

@@ -12,9 +12,14 @@ from supply_infra.pipeline.run_service import submit_pipeline_run
 from supply_infra.scheduler.constants import (
     AIGC_PUBLISH_JOB_ID,
     AIGC_PUBLISH_JOB_NAME,
+    GLOBAL_V2_SNAPSHOT_CRON_HOUR,
+    GLOBAL_V2_SNAPSHOT_CRON_MINUTE,
+    GLOBAL_V2_SNAPSHOT_JOB_ID,
+    GLOBAL_V2_SNAPSHOT_JOB_NAME,
     SUPPLY_PIPELINE_JOB_ID,
     SUPPLY_PIPELINE_JOB_NAME,
 )
+from supply_infra.pipeline.dates import CHINA_TIMEZONE
 
 if TYPE_CHECKING:
     from apscheduler.schedulers.base import BaseScheduler
@@ -52,8 +57,17 @@ def submit_aigc_publish() -> dict:
     return result
 
 
+def submit_global_v2_snapshot() -> dict:
+    """Independent Cron callback for the previous day's global V2 snapshot."""
+    from supply_infra.scheduler.jobs.sync_global_v2_snapshot import (
+        sync_global_v2_snapshot,
+    )
+
+    return sync_global_v2_snapshot()
+
+
 def create_scheduler() -> BackgroundScheduler:
-    """Create a Scheduler for the daily pipeline and AIGC publish polling."""
+    """Create the independent daily and interval jobs owned by this process."""
     settings = get_infra_settings()
     tz = settings.scheduler_timezone
     scheduler = BackgroundScheduler(timezone=tz)
@@ -85,15 +99,31 @@ def create_scheduler() -> BackgroundScheduler:
         coalesce=True,
         misfire_grace_time=300,
     )
+    scheduler.add_job(
+        submit_global_v2_snapshot,
+        trigger=CronTrigger(
+            hour=GLOBAL_V2_SNAPSHOT_CRON_HOUR,
+            minute=GLOBAL_V2_SNAPSHOT_CRON_MINUTE,
+            timezone=CHINA_TIMEZONE,
+        ),
+        id=GLOBAL_V2_SNAPSHOT_JOB_ID,
+        name=GLOBAL_V2_SNAPSHOT_JOB_NAME,
+        replace_existing=True,
+        max_instances=1,
+        coalesce=True,
+        misfire_grace_time=3600,
+    )
 
     logger.info(
         "Scheduler configured with %d job(s) | timezone=%s | pipeline_cron=%02d:%02d "
-        "| aigc_publish_interval=%dm",
+        "| aigc_publish_interval=%dm | global_v2_cron=%02d:%02d Asia/Shanghai",
         len(scheduler.get_jobs()),
         tz,
         settings.scheduler_cron_hour,
         settings.scheduler_cron_minute,
         settings.scheduler_aigc_publish_interval_minutes,
+        GLOBAL_V2_SNAPSHOT_CRON_HOUR,
+        GLOBAL_V2_SNAPSHOT_CRON_MINUTE,
     )
     return scheduler
 

supply_infra/scheduler/constants.py

@@ -7,6 +7,12 @@ SUPPLY_PIPELINE_JOB_NAME = "供给数据流水线"
 AIGC_PUBLISH_JOB_ID = "publish_videos_from_discovery"
 AIGC_PUBLISH_JOB_NAME = "AIGC候选分发"
 
+# 独立于供给流水线，每天北京时间 06:00 增量写入三张全局 V2 表。
+GLOBAL_V2_SNAPSHOT_JOB_ID = "sync_global_v2_snapshot"
+GLOBAL_V2_SNAPSHOT_JOB_NAME = "全局V2快照同步"
+GLOBAL_V2_SNAPSHOT_CRON_HOUR = 6
+GLOBAL_V2_SNAPSHOT_CRON_MINUTE = 0
+
 # find_agent：当日 S/A/B 需求（有拓展点位），任务队列 + 固定 worker 并发找视频
 PIPELINE_FIND_AGENT_WORKERS = 5
 FIND_AND_EXPAND_GRADES = ("S", "A", "B")

supply_infra/scheduler/jobs/compute_global_category_content_weights.py

@@ -0,0 +1,174 @@
+"""Resumable leaf-to-root daily global-category content weight rollup."""
+
+from __future__ import annotations
+
+import argparse
+import logging
+from collections import defaultdict
+from datetime import timedelta
+from typing import Any
+
+from supply_infra.db.repositories.global_category_content_weight_repo import (
+    GlobalCategoryContentWeightRepository,
+)
+from supply_infra.db.session import get_session
+from supply_infra.pipeline.dates import china_now, validate_biz_dt
+from supply_infra.scheduler.cli_result import run_cli
+
+logger = logging.getLogger(__name__)
+
+_SAVE_BATCH_SIZE = 100
+_WEIGHTED_SUM_FIELDS = (
+    "avg_read_rate_weighted_sum",
+    "like_rate_weighted_sum",
+    "read_rate_weighted_sum",
+)
+
+
+def _resolve_biz_dt(biz_dt: str | None) -> str:
+    if biz_dt:
+        return validate_biz_dt(biz_dt)
+    return (china_now() - timedelta(days=1)).strftime("%Y%m%d")
+
+
+def _postorder(rows_by_id: dict[int, dict[str, Any]]) -> list[int]:
+    children: dict[int, list[int]] = defaultdict(list)
+    for stable_id, row in rows_by_id.items():
+        parent = row.get("parent_stable_id")
+        if parent is not None and int(parent) in rows_by_id:
+            children[int(parent)].append(stable_id)
+    for ids in children.values():
+        ids.sort()
+
+    order: list[int] = []
+    visiting: set[int] = set()
+    visited: set[int] = set()
+
+    def visit(stable_id: int) -> None:
+        if stable_id in visited:
+            return
+        if stable_id in visiting:
+            raise RuntimeError(f"Category cycle detected at stable_id={stable_id}")
+        visiting.add(stable_id)
+        for child_id in children.get(stable_id, []):
+            visit(child_id)
+        visiting.remove(stable_id)
+        visited.add(stable_id)
+        order.append(stable_id)
+
+    for stable_id in sorted(rows_by_id):
+        visit(stable_id)
+    return order
+
+
+def _score(numerator: float, contribution_sum: float) -> float:
+    if contribution_sum == 0:
+        return 0.0
+    return numerator / contribution_sum
+
+
+def compute_global_category_content_weights(
+    biz_dt: str | None = None,
+) -> dict[str, Any]:
+    resolved_dt = _resolve_biz_dt(biz_dt)
+    with get_session() as session:
+        initialized = GlobalCategoryContentWeightRepository(session).initialize_day(resolved_dt)
+
+    with get_session() as session:
+        snapshot = GlobalCategoryContentWeightRepository(session).list_day(resolved_dt)
+    if not snapshot:
+        raise RuntimeError(f"No category nodes initialized for biz_dt={resolved_dt}")
+
+    rows_by_id = {int(row["stable_id"]): row for row in snapshot}
+    children: dict[int, list[int]] = defaultdict(list)
+    for stable_id, row in rows_by_id.items():
+        parent = row.get("parent_stable_id")
+        if parent is not None and int(parent) in rows_by_id:
+            children[int(parent)].append(stable_id)
+
+    order = _postorder(rows_by_id)
+    completed_before = sum(row["status"] == "completed" for row in snapshot)
+    pending_updates: list[dict[str, Any]] = []
+    completed_now = 0
+
+    def flush() -> None:
+        if not pending_updates:
+            return
+        with get_session() as session:
+            GlobalCategoryContentWeightRepository(session).save_completed(pending_updates)
+        pending_updates.clear()
+
+    for stable_id in order:
+        row = rows_by_id[stable_id]
+        if row["status"] == "completed":
+            continue
+
+        child_rows = [rows_by_id[child_id] for child_id in children.get(stable_id, [])]
+        incomplete_children = [
+            int(child["stable_id"]) for child in child_rows if child["status"] != "completed"
+        ]
+        if incomplete_children:
+            raise RuntimeError(
+                f"Node {stable_id} has incomplete children: {incomplete_children[:10]}"
+            )
+
+        source_count = int(row["direct_source_element_count"] or 0) + sum(
+            int(child["source_element_count"] or 0) for child in child_rows
+        )
+        contribution_sum = float(row["direct_contribution_sum"] or 0.0) + sum(
+            float(child["contribution_sum"] or 0.0) for child in child_rows
+        )
+        totals: dict[str, float] = {}
+        for field in _WEIGHTED_SUM_FIELDS:
+            direct_field = f"direct_{field}"
+            totals[field] = float(row[direct_field] or 0.0) + sum(
+                float(child[field] or 0.0) for child in child_rows
+            )
+
+        update = {
+            "id": int(row["id"]),
+            "status": "completed",
+            "attempt_count": int(row["attempt_count"] or 0) + 1,
+            "source_element_count": source_count,
+            "contribution_sum": contribution_sum,
+            **totals,
+            "avg_read_rate_score": _score(totals["avg_read_rate_weighted_sum"], contribution_sum),
+            "like_rate_score": _score(totals["like_rate_weighted_sum"], contribution_sum),
+            "read_rate_score": _score(totals["read_rate_weighted_sum"], contribution_sum),
+            "error_message": None,
+            "computed_at": china_now(),
+        }
+        row.update(update)
+        pending_updates.append(update)
+        completed_now += 1
+        if len(pending_updates) >= _SAVE_BATCH_SIZE:
+            flush()
+            logger.info(
+                "Category content weight progress: biz_dt=%s completed=%d/%d",
+                resolved_dt,
+                completed_before + completed_now,
+                len(snapshot),
+            )
+    flush()
+
+    return {
+        "success": True,
+        "biz_dt": resolved_dt,
+        "nodes": len(snapshot),
+        "initialized": initialized["initialized"],
+        "completed_before": completed_before,
+        "completed_now": completed_now,
+        "completed_total": completed_before + completed_now,
+    }
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description="Compute resumable global category content weights"
+    )
+    parser.add_argument("biz_dt", nargs="?", help="Business date YYYYMMDD")
+    args = parser.parse_args()
+    run_cli(
+        lambda: compute_global_category_content_weights(args.biz_dt),
+        label="compute_global_category_content_weights",
+    )

supply_infra/scheduler/jobs/sync_global_v2_snapshot.py

@@ -0,0 +1,96 @@
+"""Independent daily ODPS -> MySQL refresh for the global V2 snapshot tables."""
+from __future__ import annotations
+
+import argparse
+import logging
+from datetime import timedelta
+from typing import Any
+
+from supply_infra.db.repositories.global_v2_repo import GlobalV2SnapshotRepository
+from supply_infra.db.repositories.channel_content_data_repo import (
+    ChannelContentDataRepository,
+)
+from supply_infra.db.session import get_session
+from supply_infra.odps.client import get_odps_client
+from supply_infra.pipeline.dates import china_now, validate_biz_dt
+from supply_infra.scheduler.cli_result import run_cli
+
+logger = logging.getLogger(__name__)
+
+
+def _resolve_partition_date(partition_date: str | None) -> str:
+    if partition_date:
+        return validate_biz_dt(partition_date)
+    return (china_now() - timedelta(days=1)).strftime("%Y%m%d")
+
+
+def sync_global_v2_snapshot(partition_date: str | None = None) -> dict[str, Any]:
+    """Insert keys not yet present from one frozen full ODPS snapshot."""
+    resolved_date = _resolve_partition_date(partition_date)
+    logger.info("Starting global V2 snapshot sync: partition=%s", resolved_date)
+
+    odps = get_odps_client()
+    categories = odps.fetch_global_categories_v2(resolved_date)
+    elements = odps.fetch_global_elements_v2(resolved_date)
+    source_elements = odps.fetch_global_source_element_data(resolved_date)
+    channel_content = odps.fetch_channel_content_data(resolved_date)
+    fetched = {
+        "global_category_v2": len(categories),
+        "global_element_v2": len(elements),
+        "global_source_element_data": len(source_elements),
+        "channel_content_data": len(channel_content),
+    }
+
+    empty_tables = [
+        name
+        for name in (
+            "global_category_v2",
+            "global_element_v2",
+            "global_source_element_data",
+        )
+        if fetched[name] == 0
+    ]
+    if empty_tables:
+        raise RuntimeError(
+            "ODPS snapshot is incomplete; existing MySQL data was preserved: "
+            + ", ".join(empty_tables)
+        )
+
+    with get_session() as session:
+        inserted = GlobalV2SnapshotRepository(session).insert_new(
+            categories=categories,
+            elements=elements,
+            source_elements=source_elements,
+        )
+        inserted["channel_content_data"] = ChannelContentDataRepository(
+            session
+        ).insert_new(channel_content)
+
+    from supply_infra.scheduler.jobs.compute_global_category_content_weights import (
+        compute_global_category_content_weights,
+    )
+
+    weight_result = compute_global_category_content_weights(resolved_date)
+
+    result = {
+        "success": True,
+        "partition_date": resolved_date,
+        "fetched": fetched,
+        "inserted": inserted,
+        "skipped_existing": {
+            table: fetched[table] - inserted[table] for table in fetched
+        },
+        "category_content_weights": weight_result,
+    }
+    logger.info("Global V2 snapshot sync completed: %s", result)
+    return result
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Sync global V2 ODPS snapshots")
+    parser.add_argument("partition_date", nargs="?", help="ODPS partition YYYYMMDD")
+    args = parser.parse_args()
+    run_cli(
+        lambda: sync_global_v2_snapshot(args.partition_date),
+        label="sync_global_v2_snapshot",
+    )

tests/api/test_growth_category_tree.py

@@ -0,0 +1,134 @@
+from collections.abc import Generator
+from contextlib import contextmanager
+from types import SimpleNamespace
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from api.auth_middleware import normal_user_can_access
+from api.services import growth_category_tree as service
+from api.services.growth_category_tree import _build_growth_tree
+from supply_infra.db.base import Base
+from supply_infra.db.models.global_category_content_weight import (
+    GlobalCategoryContentWeight,
+)
+from supply_infra.db.models.global_v2 import GlobalCategoryV2
+
+
+def test_growth_tree_api_is_available_to_normal_users() -> None:
+    assert normal_user_can_access("GET", "/api/growth-category-tree")
+
+
+def test_build_growth_tree_uses_stable_ids_and_daily_scores() -> None:
+    categories = [
+        SimpleNamespace(
+            stable_id=10,
+            parent_stable_id=None,
+            level=1,
+            name="根分类",
+            description="根描述",
+        ),
+        SimpleNamespace(
+            stable_id=11,
+            parent_stable_id=10,
+            level=2,
+            name="子分类",
+            description=None,
+        ),
+        SimpleNamespace(
+            stable_id=12,
+            parent_stable_id=999,
+            level=2,
+            name="孤立分类",
+            description=None,
+        ),
+    ]
+    weights = [
+        SimpleNamespace(
+            stable_id=10,
+            source_element_count=8,
+            read_rate_score=0.3,
+            avg_read_rate_score=1.2,
+            like_rate_score=0.04,
+        ),
+        SimpleNamespace(
+            stable_id=11,
+            source_element_count=0,
+            read_rate_score=0.0,
+            avg_read_rate_score=0.0,
+            like_rate_score=0.0,
+        ),
+    ]
+
+    nodes = _build_growth_tree(categories, weights)
+
+    assert [node["id"] for node in nodes] == [10, 12]
+    root = nodes[0]
+    assert root["weights"] == {
+        "read_rate": 0.3,
+        "avg_read_rate": 1.2,
+        "like_rate": 0.04,
+    }
+    assert root["counts"] == {
+        "read_rate": 8,
+        "avg_read_rate": 8,
+        "like_rate": 8,
+    }
+    assert root["children"][0]["id"] == 11
+    assert root["children"][0]["counts"]["read_rate"] == 0
+    assert "hung_word_count" not in root
+
+
+def test_build_growth_category_tree_serializes_before_session_closes(monkeypatch) -> None:
+    engine = create_engine("sqlite+pysqlite:///:memory:")
+    Base.metadata.create_all(
+        engine,
+        tables=[
+            GlobalCategoryV2.__table__,
+            GlobalCategoryContentWeight.__table__,
+        ],
+    )
+    factory = sessionmaker(bind=engine, expire_on_commit=True)
+    with factory() as session:
+        session.add(
+            GlobalCategoryV2(
+                id=1,
+                stable_id=10,
+                name="根分类",
+                description=None,
+                source_type="实质",
+                level=1,
+                parent_stable_id=None,
+            )
+        )
+        session.add(
+            GlobalCategoryContentWeight(
+                id=1,
+                stable_id=10,
+                parent_stable_id=None,
+                level=1,
+                biz_dt="20260819",
+                status="completed",
+                source_element_count=5,
+                read_rate_score=0.2,
+                avg_read_rate_score=1.5,
+                like_rate_score=0.03,
+            )
+        )
+        session.commit()
+
+    @contextmanager
+    def get_test_session() -> Generator[Session, None, None]:
+        session = factory()
+        try:
+            yield session
+            session.commit()
+        finally:
+            session.close()
+
+    monkeypatch.setattr(service, "get_session", get_test_session)
+
+    payload = service.build_growth_category_tree()
+
+    assert payload["biz_dt"] == "20260819"
+    assert payload["nodes"][0]["weights"]["avg_read_rate"] == 1.5

tests/supply_infra/scheduler/test_global_category_content_weights.py

@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+from copy import deepcopy
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from supply_infra.scheduler.jobs.compute_global_category_content_weights import (
+    _postorder,
+    compute_global_category_content_weights,
+)
+
+
+def _row(
+    row_id: int,
+    stable_id: int,
+    parent_stable_id: int | None,
+    *,
+    direct_count: int,
+    contribution: float,
+    avg_numerator: float,
+    like_numerator: float,
+    read_numerator: float,
+) -> dict:
+    return {
+        "id": row_id,
+        "stable_id": stable_id,
+        "parent_stable_id": parent_stable_id,
+        "level": row_id,
+        "status": "pending",
+        "attempt_count": 0,
+        "direct_source_element_count": direct_count,
+        "source_element_count": 0,
+        "direct_contribution_sum": contribution,
+        "contribution_sum": 0.0,
+        "direct_avg_read_rate_weighted_sum": avg_numerator,
+        "avg_read_rate_weighted_sum": 0.0,
+        "avg_read_rate_score": 0.0,
+        "direct_like_rate_weighted_sum": like_numerator,
+        "like_rate_weighted_sum": 0.0,
+        "like_rate_score": 0.0,
+        "direct_read_rate_weighted_sum": read_numerator,
+        "read_rate_weighted_sum": 0.0,
+        "read_rate_score": 0.0,
+    }
+
+
+def test_postorder_places_descendants_before_parent() -> None:
+    rows = {
+        1: _row(
+            1,
+            1,
+            None,
+            direct_count=0,
+            contribution=0,
+            avg_numerator=0,
+            like_numerator=0,
+            read_numerator=0,
+        ),
+        2: _row(
+            2,
+            2,
+            1,
+            direct_count=0,
+            contribution=0,
+            avg_numerator=0,
+            like_numerator=0,
+            read_numerator=0,
+        ),
+        3: _row(
+            3,
+            3,
+            2,
+            direct_count=0,
+            contribution=0,
+            avg_numerator=0,
+            like_numerator=0,
+            read_numerator=0,
+        ),
+    }
+    assert _postorder(rows) == [3, 2, 1]
+
+
+@patch(
+    "supply_infra.scheduler.jobs.compute_global_category_content_weights.GlobalCategoryContentWeightRepository"
+)
+@patch("supply_infra.scheduler.jobs.compute_global_category_content_weights.get_session")
+def test_rollup_uses_raw_weighted_sums_from_leaf_to_root(
+    mock_get_session,
+    repo_cls,
+) -> None:
+    mock_get_session.return_value.__enter__.return_value = MagicMock()
+    snapshot = [
+        _row(
+            1,
+            1,
+            None,
+            direct_count=1,
+            contribution=0.8,
+            avg_numerator=0.4,
+            like_numerator=0.016,
+            read_numerator=0.08,
+        ),
+        _row(
+            2,
+            2,
+            1,
+            direct_count=1,
+            contribution=0.2,
+            avg_numerator=0.24,
+            like_numerator=0.008,
+            read_numerator=0.04,
+        ),
+    ]
+    repo = repo_cls.return_value
+    repo.initialize_day.return_value = {"initialized": 2, "existing": 0}
+    repo.list_day.return_value = snapshot
+    saved: list[dict] = []
+    repo.save_completed.side_effect = lambda rows: saved.extend(deepcopy(rows))
+
+    result = compute_global_category_content_weights("20260819")
+
+    saved_by_id = {row["id"]: row for row in saved}
+    root = saved_by_id[1]
+    assert root["source_element_count"] == 2
+    assert root["contribution_sum"] == pytest.approx(1.0)
+    assert root["avg_read_rate_weighted_sum"] == pytest.approx(0.64)
+    assert root["avg_read_rate_score"] == pytest.approx(0.64)
+    assert root["like_rate_score"] == pytest.approx(0.024)
+    assert root["read_rate_score"] == pytest.approx(0.12)
+    assert result["completed_total"] == 2

tests/supply_infra/scheduler/test_global_v2_snapshot.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from supply_infra.scheduler.app import create_scheduler
+from supply_infra.scheduler.constants import GLOBAL_V2_SNAPSHOT_JOB_ID
+from supply_infra.scheduler.jobs.sync_global_v2_snapshot import sync_global_v2_snapshot
+from supply_infra.odps.client import ODPSClient
+
+
+def test_scheduler_registers_global_v2_at_six_china_time() -> None:
+    scheduler = create_scheduler()
+    job = scheduler.get_job(GLOBAL_V2_SNAPSHOT_JOB_ID)
+    assert job is not None
+    assert str(job.trigger) == "cron[hour='6', minute='0']"
+    assert str(job.trigger.timezone) == "Asia/Shanghai"
+
+
+def test_channel_content_query_guards_zero_and_null_denominators() -> None:
+    client = ODPSClient("id", "key", "project", "endpoint")
+    with patch.object(client, "execute_sql", return_value=[]) as execute_sql:
+        client.fetch_channel_content_data("20260819")
+
+    sql = execute_sql.call_args.args[0]
+    assert "avg_read_cnt_30d IS NULL" in sql
+    assert "avg_read_cnt_30d = 0" in sql
+    assert "read_cnt = 0" in sql
+    assert "cal_fans_num = 0" in sql
+    assert "END AS avg_read_rate" in sql
+    assert "END AS like_rate" in sql
+    assert "END AS read_rate" in sql
+
+
+@patch("supply_infra.scheduler.jobs.sync_global_v2_snapshot.get_session")
+@patch("supply_infra.scheduler.jobs.sync_global_v2_snapshot.get_odps_client")
+def test_snapshot_inserts_new_rows_for_all_three_tables(mock_get_odps, mock_get_session) -> None:
+    odps = mock_get_odps.return_value
+    odps.fetch_global_categories_v2.return_value = [{"stable_id": 1}]
+    odps.fetch_global_elements_v2.return_value = [{"element_id": 10, "name": "element"}]
+    odps.fetch_global_source_element_data.return_value = [{"post_id": "post"}]
+    odps.fetch_channel_content_data.return_value = [
+        {"channel_content_id": "content", "dt": "20260819"}
+    ]
+    session = MagicMock()
+    mock_get_session.return_value.__enter__.return_value = session
+
+    with patch(
+        "supply_infra.scheduler.jobs.sync_global_v2_snapshot.GlobalV2SnapshotRepository"
+    ) as repo_cls:
+        repo_cls.return_value.insert_new.return_value = {
+            "global_category_v2": 1,
+            "global_element_v2": 1,
+            "global_source_element_data": 1,
+        }
+        with patch(
+            "supply_infra.scheduler.jobs.sync_global_v2_snapshot.ChannelContentDataRepository"
+        ) as channel_repo_cls:
+            channel_repo_cls.return_value.insert_new.return_value = 1
+            with patch(
+                "supply_infra.scheduler.jobs.compute_global_category_content_weights."
+                "compute_global_category_content_weights",
+                return_value={"success": True, "completed_total": 1},
+            ):
+                result = sync_global_v2_snapshot("20260819")
+
+    assert result["success"] is True
+    odps.fetch_global_categories_v2.assert_called_once_with("20260819")
+    odps.fetch_global_elements_v2.assert_called_once_with("20260819")
+    odps.fetch_global_source_element_data.assert_called_once_with("20260819")
+    odps.fetch_channel_content_data.assert_called_once_with("20260819")
+    assert result["skipped_existing"] == {
+        "global_category_v2": 0,
+        "global_element_v2": 0,
+        "global_source_element_data": 0,
+        "channel_content_data": 0,
+    }
+    repo_cls.return_value.insert_new.assert_called_once()
+
+
+@patch("supply_infra.scheduler.jobs.sync_global_v2_snapshot.get_session")
+@patch("supply_infra.scheduler.jobs.sync_global_v2_snapshot.get_odps_client")
+def test_empty_odps_snapshot_preserves_mysql(mock_get_odps, mock_get_session) -> None:
+    odps = mock_get_odps.return_value
+    odps.fetch_global_categories_v2.return_value = [{"stable_id": 1}]
+    odps.fetch_global_elements_v2.return_value = []
+    odps.fetch_global_source_element_data.return_value = [{"post_id": "post"}]
+    odps.fetch_channel_content_data.return_value = []
+
+    with pytest.raises(RuntimeError, match="global_element_v2"):
+        sync_global_v2_snapshot("20260819")
+
+    mock_get_session.assert_not_called()

web/src/App.vue

@@ -13,6 +13,7 @@ const SIDEBAR_STORAGE_KEY = 'supply-agent-sidebar-collapsed'
 const adminNavItems = [
   { to: '/', label: '供给总览', icon: '◫', admin: true },
   { to: '/demand-map', label: '全局需求地图', icon: '⌁', admin: false },
+  { to: '/growth-heat-map', label: '增长全局热度地图', icon: '♨', admin: false },
   { to: '/video-discovery', label: '需求汇总', icon: '▷', admin: false },
   { to: '/pipeline-runs', label: '定时任务', icon: '◷', admin: true },
   { to: '/llm-billing', label: 'LLM 费用', icon: '＄', admin: true },

web/src/api/growthCategory.ts

@@ -0,0 +1,18 @@
+import type { CategoryTreeResponse } from '../types/category'
+
+export async function fetchGrowthCategoryTree(
+  bizDt?: string | null,
+): Promise<CategoryTreeResponse> {
+  const params = new URLSearchParams()
+  if (bizDt) params.set('biz_dt', bizDt)
+  const query = params.toString()
+  const response = await fetch(
+    `/api/growth-category-tree${query ? `?${query}` : ''}`,
+  )
+  if (!response.ok) {
+    throw new Error(
+      `加载增长全局热度地图失败: ${response.status} ${response.statusText}`,
+    )
+  }
+  return response.json()
+}

web/src/components/IcicleHeatTree.vue

@@ -10,7 +10,11 @@ import {
 import DemandPathPanel from './DemandPathPanel.vue'
 import DemandGradeListPanel from './DemandGradeListPanel.vue'
 import type { DemandGradeListCard } from './DemandGradeListPanel.vue'
-import type { CategoryNode, WeightDimKey } from '../types/category'
+import type {
+  CategoryNode,
+  WeightDimKey,
+  WeightDimMeta,
+} from '../types/category'
 import {
   COLUMN_WIDTH,
   FULL_TREE_TAB,
@@ -61,18 +65,34 @@ function minLeafInLeftTwoColumns(roots: PreparedNode[]): number {
   return Number.isFinite(minLeaf) && minLeaf > 0 ? minLeaf : 1
 }
 
-const props = defineProps<{
-  nodes: CategoryNode[]
-  bizDt?: string | null
-  demandsByCategory?: DemandsByCategory
-}>()
+const props = withDefaults(
+  defineProps<{
+    nodes: CategoryNode[]
+    bizDt?: string | null
+    demandsByCategory?: DemandsByCategory
+    title?: string
+    subtitle?: string
+    heatTabs?: WeightDimMeta[]
+    initialTab?: HeatTabKey
+    heatOnly?: boolean
+  }>(),
+  {
+    bizDt: null,
+    demandsByCategory: () => ({}),
+    title: '平台全局需求地图',
+    subtitle: '完整分类树 · 横向为层级，纵向为分支规模',
+    heatTabs: () => HEAT_DIM_TABS,
+    initialTab: 'total_score',
+    heatOnly: false,
+  },
+)
 
 const canvasRef = ref<HTMLCanvasElement | null>(null)
 const viewportRef = ref<HTMLElement | null>(null)
 const wrapRef = ref<HTMLElement | null>(null)
 
 const prepared = shallowRef(prepareTree([]))
-const activeTab = ref<HeatTabKey>('total_score')
+const activeTab = ref<HeatTabKey>(props.initialTab)
 const startDepth = ref(0)
 const focus = ref<PreparedNode | null>(null)
 const selectedNode = ref<PreparedNode | null>(null)
@@ -357,14 +377,18 @@ const weightScale = computed(() => {
   return collectDimScale(prepared.value.flat, activeDim.value)
 })
 
-const treeTabs = computed(() => [
-  { key: FULL_TREE_TAB as HeatTabKey, label: '全局树' },
-  ...HEAT_DIM_TABS.map((d) => ({ key: d.key as HeatTabKey, label: d.label })),
-])
+const treeTabs = computed(() => {
+  const heatTabs = props.heatTabs.map((dim) => ({
+    key: dim.key as HeatTabKey,
+    label: dim.label,
+  }))
+  if (props.heatOnly) return heatTabs
+  return [{ key: FULL_TREE_TAB as HeatTabKey, label: '全局树' }, ...heatTabs]
+})
 
 const activeDimLabel = computed(() => {
   if (!activeDim.value) return null
-  return HEAT_DIM_TABS.find((d) => d.key === activeDim.value)?.label ?? activeDim.value
+  return props.heatTabs.find((d) => d.key === activeDim.value)?.label ?? activeDim.value
 })
 
 const mapContext = computed(() => {
@@ -493,6 +517,11 @@ function selectNode(
     viewportRef.value.scrollLeft = 0
   }
   // Auto-open path panel when this category has demand words (same as 🔍 nodes).
+  if (props.heatOnly) {
+    closeInspect()
+    scheduleDraw()
+    return
+  }
   const items = props.demandsByCategory?.[node.id] ?? []
   if (items.length) {
     inspectNode.value = node
@@ -923,16 +952,23 @@ onUnmounted(() => {
   <div class="icicle-tree">
     <header class="toolbar">
       <div class="title-row">
-        <h1>平台全局需求地图</h1>
+        <h1>{{ title }}</h1>
         <span v-if="bizDt" class="biz-dt">biz_dt {{ bizDt }}</span>
       </div>
-      <p class="subtitle">完整分类树 · 横向为层级，纵向为分支规模</p>
+      <p class="subtitle">{{ subtitle }}</p>
     </header>
 
     <div
       class="content-grid"
-      :class="{ 'demand-list-resizing': demandListResizing }"
-      :style="{ gridTemplateColumns: `minmax(0, 1fr) ${demandListWidth}px` }"
+      :class="{
+        'demand-list-resizing': demandListResizing,
+        'heat-only': heatOnly,
+      }"
+      :style="
+        heatOnly
+          ? { gridTemplateColumns: 'minmax(0, 1fr)' }
+          : { gridTemplateColumns: `minmax(0, 1fr) ${demandListWidth}px` }
+      "
     >
       <div class="left-panel">
         <section class="map-panel">
@@ -1056,7 +1092,7 @@ onUnmounted(() => {
         </section>
 
         <div
-          v-if="inspectOpen && inspectNode"
+          v-if="!heatOnly && inspectOpen && inspectNode"
           class="path-dock"
           :class="{ resizing: pathDockResizing }"
           :style="{ height: `${pathDockHeight}px` }"
@@ -1099,7 +1135,11 @@ onUnmounted(() => {
         </div>
       </div>
 
-      <div class="demand-list-shell" :class="{ resizing: demandListResizing }">
+      <div
+        v-if="!heatOnly"
+        class="demand-list-shell"
+        :class="{ resizing: demandListResizing }"
+      >
         <div
           class="demand-list-resize"
           role="separator"
@@ -1190,6 +1230,10 @@ onUnmounted(() => {
   user-select: none;
 }
 
+.content-grid.heat-only .left-panel {
+  padding-right: 0;
+}
+
 .left-panel {
   display: flex;
   flex-direction: column;

web/src/router.ts

@@ -2,6 +2,7 @@ import { createRouter, createWebHistory } from 'vue-router'
 import CategoryTreeView from './views/CategoryTreeView.vue'
 import DemandProcessView from './views/DemandProcessView.vue'
 import GlobalDemandMapView from './views/GlobalDemandMapView.vue'
+import GrowthHeatMapView from './views/GrowthHeatMapView.vue'
 import ForbiddenView from './views/ForbiddenView.vue'
 import FindAgentRecordsView from './views/FindAgentRecordsView.vue'
 import FindAgentFrameworkView from './views/FindAgentFrameworkView.vue'
@@ -35,6 +36,12 @@ export const router = createRouter({
       component: GlobalDemandMapView,
       meta: { title: '平台全局需求地图', userAllowed: true },
     },
+    {
+      path: '/growth-heat-map',
+      name: 'growth-heat-map',
+      component: GrowthHeatMapView,
+      meta: { title: '增长全局热度地图', userAllowed: true },
+    },
     {
       path: '/demand-tree',
       name: 'category-tree',

web/src/types/category.ts

@@ -6,6 +6,9 @@ export type WeightDimKey =
   | 'recent_pop'
   | 'real_rov_7d'
   | 'real_vov_7d'
+  | 'read_rate'
+  | 'avg_read_rate'
+  | 'like_rate'
 
 export interface WeightDimMeta {
   key: WeightDimKey

web/src/views/GrowthHeatMapView.vue

@@ -0,0 +1,65 @@
+<script setup lang="ts">
+import { onMounted, ref } from 'vue'
+import IcicleHeatTree from '../components/IcicleHeatTree.vue'
+import { fetchGrowthCategoryTree } from '../api/growthCategory'
+import type { CategoryNode, WeightDimMeta } from '../types/category'
+
+const GROWTH_HEAT_TABS: WeightDimMeta[] = [
+  { key: 'read_rate', label: '阅读率' },
+  { key: 'avg_read_rate', label: '平均阅读率' },
+  { key: 'like_rate', label: '点赞率' },
+]
+
+const nodes = ref<CategoryNode[]>([])
+const bizDt = ref<string | null>(null)
+const loading = ref(true)
+const error = ref<string | null>(null)
+
+onMounted(async () => {
+  try {
+    const tree = await fetchGrowthCategoryTree()
+    nodes.value = tree.nodes ?? []
+    bizDt.value = tree.biz_dt ?? null
+  } catch (cause) {
+    error.value = cause instanceof Error ? cause.message : String(cause)
+  } finally {
+    loading.value = false
+  }
+})
+</script>
+
+<template>
+  <div class="view">
+    <div v-if="loading" class="state">正在加载增长分类树与权重…</div>
+    <div v-else-if="error" class="state error">{{ error }}</div>
+    <IcicleHeatTree
+      v-else
+      :nodes="nodes"
+      :biz-dt="bizDt"
+      title="增长全局热度地图"
+      subtitle="global_category_v2 分类树 · 横向为层级，纵向为分支规模"
+      :heat-tabs="GROWTH_HEAT_TABS"
+      initial-tab="read_rate"
+      heat-only
+    />
+  </div>
+</template>
+
+<style scoped>
+.view {
+  height: max(760px, calc(100vh - 106px));
+  min-height: 760px;
+  padding-bottom: 8px;
+}
+
+.state {
+  padding: 64px 24px;
+  color: #64748b;
+  font-size: 15px;
+  text-align: center;
+}
+
+.state.error {
+  color: #b91c1c;
+}
+</style>