external_demand/app/hot_content/repository.py

"""热点内容 MySQL 仓储。"""

from __future__ import annotations

import hashlib
import json
from datetime import date, datetime, timedelta
from typing import Any

try:
    import pymysql
    from pymysql.cursors import DictCursor
except ImportError:  # pragma: no cover - runtime dependency check
    pymysql = None
    DictCursor = None

from app.hot_content.exceptions import HotContentFlowError
from app.hot_content.status import ExecutionStatus, PostprocessStatus
from app.hot_content.timezone import SHANGHAI_TZ
from app.hot_content.types import MysqlConfig


def _json_dumps(data: Any) -> str:
    return json.dumps(data, ensure_ascii=False, separators=(",", ":"))


def _nullable_bool(value: Any) -> int | None:
    if value is None:
        return None
    return 1 if value else 0


def _nullable_str(value: Any) -> str | None:
    if value is None:
        return None
    text = str(value).strip()
    return text if text else None


def _nullable_json_dumps(value: Any) -> str | None:
    if value is None:
        return None
    return _json_dumps(value)


def _wxindex_word_record_row(payload: dict[str, Any]) -> tuple[Any, ...]:
    name = str(payload.get("name") or "").strip()
    analyze_ymd = str(payload.get("analyze_ymd") or "").strip()
    if not name or not analyze_ymd:
        raise HotContentFlowError("invalid wxindex word record payload")
    return (
        name,
        payload.get("meta_id"),
        analyze_ymd,
        payload.get("fetch_start_ymd"),
        payload.get("fetch_end_ymd"),
        payload.get("data_start_ymd"),
        payload.get("data_end_ymd"),
        payload.get("data_days"),
        _nullable_bool(payload.get("is_sustained_high")),
        _nullable_bool(payload.get("is_rising")),
        _nullable_bool(payload.get("is_spike")),
        payload.get("retain_reason"),
        _nullable_bool(payload.get("is_internal_demand_matched")),
        _nullable_str(payload.get("matched_demand")),
        payload.get("demand_cache_run_id"),
        _nullable_json_dumps(payload.get("internal_demand_match_json")),
        payload.get("senior_fit_score"),
        _nullable_json_dumps(payload.get("demand_senior_fit_json")),
        _nullable_bool(payload.get("is_final_retained")),
        payload.get("min_score"),
        payload.get("max_score"),
        payload.get("avg_score"),
        _nullable_json_dumps(payload.get("detail_json")),
    )


_WXINDEX_WORD_RECORD_UPSERT_SQL = """
    INSERT INTO hot_content_wxindex_word_records (
        name,
        meta_id,
        analyze_ymd,
        fetch_start_ymd,
        fetch_end_ymd,
        data_start_ymd,
        data_end_ymd,
        data_days,
        is_sustained_high,
        is_rising,
        is_spike,
        retain_reason,
        is_internal_demand_matched,
        matched_demand,
        demand_cache_run_id,
        internal_demand_match_json,
        senior_fit_score,
        demand_senior_fit_json,
        is_final_retained,
        min_score,
        max_score,
        avg_score,
        detail_json
    )
    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
    ON DUPLICATE KEY UPDATE
        meta_id = VALUES(meta_id),
        fetch_start_ymd = VALUES(fetch_start_ymd),
        fetch_end_ymd = VALUES(fetch_end_ymd),
        data_start_ymd = VALUES(data_start_ymd),
        data_end_ymd = VALUES(data_end_ymd),
        data_days = VALUES(data_days),
        is_sustained_high = VALUES(is_sustained_high),
        is_rising = VALUES(is_rising),
        is_spike = VALUES(is_spike),
        retain_reason = VALUES(retain_reason),
        is_internal_demand_matched = VALUES(is_internal_demand_matched),
        matched_demand = VALUES(matched_demand),
        demand_cache_run_id = VALUES(demand_cache_run_id),
        internal_demand_match_json = VALUES(internal_demand_match_json),
        senior_fit_score = VALUES(senior_fit_score),
        demand_senior_fit_json = VALUES(demand_senior_fit_json),
        is_final_retained = VALUES(is_final_retained),
        min_score = VALUES(min_score),
        max_score = VALUES(max_score),
        avg_score = VALUES(avg_score),
        detail_json = VALUES(detail_json)
"""

_WXINDEX_WORD_RECORD_INIT_UPSERT_SQL = """
    INSERT INTO hot_content_wxindex_word_records (
        name,
        meta_id,
        analyze_ymd,
        fetch_start_ymd,
        fetch_end_ymd,
        data_start_ymd,
        data_end_ymd,
        data_days,
        is_sustained_high,
        is_rising,
        is_spike,
        retain_reason,
        is_internal_demand_matched,
        matched_demand,
        demand_cache_run_id,
        internal_demand_match_json,
        senior_fit_score,
        demand_senior_fit_json,
        is_final_retained,
        min_score,
        max_score,
        avg_score,
        detail_json
    )
    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
    ON DUPLICATE KEY UPDATE
        meta_id = VALUES(meta_id),
        fetch_start_ymd = VALUES(fetch_start_ymd),
        fetch_end_ymd = VALUES(fetch_end_ymd),
        demand_cache_run_id = VALUES(demand_cache_run_id)
"""


def _wxindex_word_stats_row(payload: dict[str, Any]) -> tuple[Any, ...]:
    name = str(payload.get("name") or "").strip()
    analyze_ymd = str(payload.get("analyze_ymd") or "").strip()
    if not name or not analyze_ymd:
        raise HotContentFlowError("invalid wxindex word stats payload")
    return (
        name,
        payload.get("meta_id"),
        analyze_ymd,
        payload.get("wxindex_word_record_id"),
        payload.get("retain_reason"),
        payload.get("senior_fit_score"),
        payload.get("data_start_ymd"),
        payload.get("data_end_ymd"),
        payload.get("data_days"),
        payload.get("min_score"),
        payload.get("max_score"),
        payload.get("avg_score"),
        _nullable_json_dumps(payload.get("detail_json")),
    )


_WXINDEX_WORD_STATS_UPSERT_SQL = """
    INSERT INTO hot_content_wxindex_word_stats (
        name,
        meta_id,
        analyze_ymd,
        wxindex_word_record_id,
        retain_reason,
        senior_fit_score,
        data_start_ymd,
        data_end_ymd,
        data_days,
        min_score,
        max_score,
        avg_score,
        detail_json
    )
    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
    ON DUPLICATE KEY UPDATE
        meta_id = VALUES(meta_id),
        wxindex_word_record_id = VALUES(wxindex_word_record_id),
        retain_reason = VALUES(retain_reason),
        senior_fit_score = VALUES(senior_fit_score),
        data_start_ymd = VALUES(data_start_ymd),
        data_end_ymd = VALUES(data_end_ymd),
        data_days = VALUES(data_days),
        min_score = VALUES(min_score),
        max_score = VALUES(max_score),
        avg_score = VALUES(avg_score),
        detail_json = VALUES(detail_json)
"""


def _json_loads(value: Any) -> Any:
    if value is None:
        return None
    if isinstance(value, (dict, list)):
        return value
    if isinstance(value, (bytes, bytearray)):
        value = value.decode("utf-8")
    if isinstance(value, str):
        return json.loads(value)
    return value


def _normalize_demand_names(demand_name_set: list[str]) -> list[str]:
    names: list[str] = []
    seen: set[str] = set()
    for item in demand_name_set:
        name = str(item).strip()
        if not name or name in seen:
            continue
        seen.add(name)
        names.append(name)
    return names


def unique_title_key(source: str, title: str) -> str:
    return hashlib.sha256(f"{source}\n{title}".encode("utf-8")).hexdigest()


class HotContentRepository:
    def __init__(self, config: MysqlConfig):
        if pymysql is None or DictCursor is None:
            raise HotContentFlowError("missing dependency: pip install pymysql")
        self.conn = pymysql.connect(
            host=config.host,
            port=config.port,
            user=config.user,
            password=config.password,
            database=config.database,
            charset=config.charset,
            autocommit=True,
            cursorclass=DictCursor,
        )

    def close(self) -> None:
        self.conn.close()

    def upsert_record(self, *, source: str, title: str, rank: int | None) -> dict[str, Any]:
        key = unique_title_key(source, title)
        sql = """
            INSERT INTO hot_content_records (
                unique_key, source, title, hot_rank, execution_status, created_at, updated_at
            )
            VALUES (%s, %s, %s, %s, %s, NOW(), NOW())
            ON DUPLICATE KEY UPDATE
                id=LAST_INSERT_ID(id),
                hot_rank=VALUES(hot_rank),
                updated_at=NOW()
        """
        with self.conn.cursor() as cursor:
            cursor.execute(
                sql,
                (
                    key,
                    source,
                    title,
                    rank,
                    ExecutionStatus.HOT_SAVED,
                ),
            )
            record_id = int(cursor.lastrowid)
            cursor.execute(
                """
                SELECT
                    id,
                    unique_key,
                    execution_status,
                    article_title,
                    article_body,
                    article_url,
                    decode_request_result IS NOT NULL AS has_decode_request,
                    contribution_points_json IS NOT NULL AS has_contribution_points
                FROM hot_content_records
                WHERE id = %s
                """,
                (record_id,),
            )
            row = cursor.fetchone()
        if not row:
            raise HotContentFlowError(f"missing hot_content_records id={record_id}")
        return {
            "id": int(row["id"]),
            "unique_key": str(row["unique_key"]),
            "execution_status": int(row["execution_status"]),
            "article_title": row.get("article_title"),
            "article_body": row.get("article_body"),
            "article_url": row.get("article_url"),
            "has_decode_request": bool(row.get("has_decode_request")),
            "has_contribution_points": bool(row.get("has_contribution_points")),
        }

    def update_status(
        self,
        *,
        record_id: int,
        status: int,
        error_message: str | None = None,
    ) -> None:
        sql = """
            UPDATE hot_content_records
            SET execution_status=%s, error_reason=%s, updated_at=NOW()
            WHERE id=%s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (status, error_message, record_id))

    def update_article(
        self,
        *,
        record_id: int,
        article_title: str,
        article_body: str,
        url: str,
    ) -> None:
        sql = """
            UPDATE hot_content_records
            SET article_title=%s,
                article_body=%s,
                article_url=%s,
                execution_status=%s,
                error_reason=NULL,
                updated_at=NOW()
            WHERE id=%s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(
                sql,
                (
                    article_title,
                    article_body,
                    url,
                    ExecutionStatus.CONTENT_OK,
                    record_id,
                ),
            )

    def mark_no_valid_content(
        self,
        *,
        record_id: int,
        reason: str,
    ) -> None:
        """搜不到文章或缺标题/正文：仅更新 execution_status + error_reason，不动分类字段。"""
        sql = """
            UPDATE hot_content_records
            SET execution_status=%s,
                error_reason=%s,
                updated_at=NOW()
            WHERE id=%s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(
                sql,
                (
                    ExecutionStatus.NO_VALID_CONTENT,
                    str(reason or "no valid content").strip(),
                    record_id,
                ),
            )

    def update_category_filter_result(
        self,
        *,
        record_id: int,
        passed: bool,
        result_json: dict[str, Any],
    ) -> None:
        self._ensure_category_filter_columns()
        status = (
            ExecutionStatus.CATEGORY_FILTER_PASSED
            if passed
            else ExecutionStatus.CATEGORY_FILTER_REJECTED
        )
        reason = str(result_json.get("reason") or "").strip()
        error_message = None if passed else (reason or "category filter rejected")
        sql = """
            UPDATE hot_content_records
            SET execution_status=%s,
                category_filter_passed=%s,
                category_filter_reason=%s,
                category_filter_json=%s,
                error_reason=%s,
                updated_at=NOW()
            WHERE id=%s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(
                sql,
                (
                    status,
                    1 if passed else 0,
                    reason or None,
                    _json_dumps(result_json),
                    error_message,
                    record_id,
                ),
            )

    def get_record_for_category_filter(self, record_id: int) -> dict[str, Any] | None:
        self._ensure_category_filter_columns()
        sql = """
            SELECT
                id,
                source,
                title,
                article_title,
                article_body,
                article_url,
                execution_status,
                category_filter_passed,
                category_filter_reason,
                category_filter_json
            FROM hot_content_records
            WHERE id = %s
            LIMIT 1
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (record_id,))
            row = cursor.fetchone()
        if not row:
            return None
        category_filter_json = _json_loads(row.get("category_filter_json"))
        passed_raw = row.get("category_filter_passed")
        passed: bool | None
        if passed_raw is None:
            passed = None
        else:
            passed = bool(int(passed_raw))
        return {
            "id": int(row["id"]),
            "source": str(row.get("source") or ""),
            "title": str(row.get("title") or ""),
            "article_title": str(row.get("article_title") or ""),
            "article_body": str(row.get("article_body") or ""),
            "article_url": str(row.get("article_url") or ""),
            "execution_status": int(row.get("execution_status") or 0),
            "category_filter_passed": passed,
            "category_filter_reason": str(row.get("category_filter_reason") or ""),
            "category_filter_json": (
                category_filter_json if isinstance(category_filter_json, dict) else {}
            ),
        }

    def get_category_filter_status(self, record_id: int) -> dict[str, Any] | None:
        record = self.get_record_for_category_filter(record_id)
        if not record:
            return None
        matched_category = None
        category_filter_json = record.get("category_filter_json") or {}
        if isinstance(category_filter_json, dict):
            matched_category = category_filter_json.get("matched_category")
        return {
            "id": record["id"],
            "passed": record["category_filter_passed"],
            "reason": record["category_filter_reason"],
            "matched_category": matched_category,
            "execution_status": record["execution_status"],
        }

    def update_decode_result(
        self,
        *,
        record_id: int,
        status: int,
        request_json: dict[str, Any],
        response_json: dict[str, Any] | None,
        error_message: str | None = None,
    ) -> None:
        decode_request_result = {
            "request": request_json,
            "response": response_json,
        }
        sql = """
            UPDATE hot_content_records
            SET decode_request_result=%s,
                execution_status=%s,
                error_reason=%s,
                updated_at=NOW()
            WHERE id=%s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(
                sql,
                (
                    _json_dumps(decode_request_result),
                    status,
                    error_message,
                    record_id,
                ),
            )

    def list_decode_result_candidates(self, *, limit: int) -> list[dict[str, Any]]:
        sql = """
            SELECT id, unique_key
            FROM hot_content_records
            WHERE execution_status IN (%s, %s, %s)
              AND contribution_points_json IS NULL
            ORDER BY updated_at ASC, id ASC
            LIMIT %s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(
                sql,
                (
                    ExecutionStatus.DECODE_SUBMITTED,
                    ExecutionStatus.DECODE_SUCCESS,
                    ExecutionStatus.DECODE_PENDING,
                    limit,
                ),
            )
            rows = cursor.fetchall()
        return [
            {
                "id": int(row["id"]),
                "unique_key": str(row["unique_key"]),
            }
            for row in rows
        ]

    def save_decode_result_export(
        self,
        *,
        record_id: int,
        decode_result_json: dict[str, Any],
        contribution_points_json: dict[str, Any],
    ) -> None:
        sql = """
            UPDATE hot_content_records
            SET decode_result_json=%s,
                contribution_points_json=%s,
                execution_status=%s,
                error_reason=NULL,
                updated_at=NOW()
            WHERE id=%s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(
                sql,
                (
                    _json_dumps(decode_result_json),
                    _json_dumps(contribution_points_json),
                    ExecutionStatus.CONTRIBUTION_EXTRACTED,
                    record_id,
                ),
            )

    def get_demand_cache_by_hour(self, *, cache_hour: datetime) -> dict[str, Any] | None:
        sql = """
            SELECT
                id,
                cache_hour,
                source_table,
                partition_dt,
                demand_name_set_json,
                item_count,
                updated_at
            FROM demand_pool_hourly_cache
            WHERE cache_hour=%s
            LIMIT 1
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (cache_hour,))
            row = cursor.fetchone()
        if not row:
            return None
        demand_name_set = _json_loads(row.get("demand_name_set_json")) or []
        if not isinstance(demand_name_set, list):
            demand_name_set = []
        return {
            "id": int(row["id"]),
            "cache_hour": row.get("cache_hour"),
            "source_table": str(row["source_table"]),
            "partition_dt": row.get("partition_dt"),
            "demand_name_set": [
                str(name).strip()
                for name in demand_name_set
                if str(name).strip()
            ],
            "item_count": int(row.get("item_count") or 0),
            "updated_at": row.get("updated_at"),
        }

    def save_demand_cache_set(
        self,
        *,
        cache_hour: datetime,
        source_table: str,
        partition_dt: str | None,
        excluded_strategy: str,
        top_n: int,
        demand_name_set: list[str],
    ) -> int:
        sql = """
            INSERT INTO demand_pool_hourly_cache (
                cache_hour,
                source_table,
                partition_dt,
                excluded_strategy,
                top_n,
                demand_name_set_json,
                item_count,
                created_at,
                updated_at
            )
            VALUES (%s, %s, %s, %s, %s, %s, %s, NOW(), NOW())
            ON DUPLICATE KEY UPDATE
                id=LAST_INSERT_ID(id),
                source_table=VALUES(source_table),
                partition_dt=VALUES(partition_dt),
                excluded_strategy=VALUES(excluded_strategy),
                top_n=VALUES(top_n),
                demand_name_set_json=VALUES(demand_name_set_json),
                item_count=VALUES(item_count),
                updated_at=NOW()
        """
        normalized_names = _normalize_demand_names(demand_name_set)
        with self.conn.cursor() as cursor:
            cursor.execute(
                sql,
                (
                    cache_hour,
                    source_table,
                    partition_dt,
                    excluded_strategy,
                    top_n,
                    _json_dumps(normalized_names),
                    len(normalized_names),
                ),
            )
            return int(cursor.lastrowid)

    def list_postprocess_candidates(self, *, limit: int) -> list[dict[str, Any]]:
        sql = """
            SELECT
                id,
                unique_key,
                source,
                title,
                created_at,
                article_title,
                article_body,
                demand_cache_run_id,
                postprocess_status,
                decode_result_json,
                contribution_points_json,
                contribution_demand_match_json,
                wxindex_trend_json,
                demand_event_sense_json,
                demand_senior_fit_json
            FROM hot_content_records
            WHERE contribution_points_json IS NOT NULL
              AND postprocess_status IN (%s, %s, %s, %s)
            ORDER BY updated_at ASC, id ASC
            LIMIT %s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(
                sql,
                (
                    PostprocessStatus.PENDING,
                    PostprocessStatus.DEMAND_MATCHED,
                    PostprocessStatus.WXINDEX_DONE,
                    PostprocessStatus.FAILED,
                    limit,
                ),
            )
            rows = cursor.fetchall()
        return [
            {
                "id": int(row["id"]),
                "unique_key": str(row["unique_key"]),
                "source": str(row.get("source") or ""),
                "title": str(row.get("title") or ""),
                "created_at": row.get("created_at"),
                "article_title": row.get("article_title"),
                "article_body": row.get("article_body"),
                "demand_cache_run_id": row.get("demand_cache_run_id"),
                "postprocess_status": int(row.get("postprocess_status") or 0),
                "decode_result_json": _json_loads(row.get("decode_result_json")),
                "contribution_points_json": _json_loads(row.get("contribution_points_json")),
                "contribution_demand_match_json": _json_loads(
                    row.get("contribution_demand_match_json")
                ),
                "wxindex_trend_json": _json_loads(row.get("wxindex_trend_json")),
                "demand_event_sense_json": _json_loads(
                    row.get("demand_event_sense_json")
                ),
                "demand_senior_fit_json": _json_loads(row.get("demand_senior_fit_json")),
            }
            for row in rows
        ]

    def save_contribution_demand_match(
        self,
        *,
        record_id: int,
        demand_cache_run_id: int,
        match_json: dict[str, Any],
    ) -> None:
        sql = """
            UPDATE hot_content_records
            SET demand_cache_run_id=%s,
                contribution_demand_match_json=%s,
                postprocess_status=%s,
                postprocess_error_reason=NULL,
                updated_at=NOW()
            WHERE id=%s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(
                sql,
                (
                    demand_cache_run_id,
                    _json_dumps(match_json),
                    PostprocessStatus.DEMAND_MATCHED,
                    record_id,
                ),
            )

    def save_wxindex_trend(
        self,
        *,
        record_id: int,
        trend_json: dict[str, Any],
    ) -> None:
        sql = """
            UPDATE hot_content_records
            SET wxindex_trend_json=%s,
                postprocess_status=%s,
                postprocess_error_reason=NULL,
                updated_at=NOW()
            WHERE id=%s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(
                sql,
                (
                    _json_dumps(trend_json),
                    PostprocessStatus.WXINDEX_DONE,
                    record_id,
                ),
            )

    def save_demand_quality(
        self,
        *,
        record_id: int,
        event_sense_json: dict[str, Any],
        senior_fit_json: dict[str, Any],
        update_status: bool = True,
    ) -> None:
        self._ensure_record_quality_columns()
        if update_status:
            sql = """
                UPDATE hot_content_records
                SET demand_event_sense_json=%s,
                    demand_senior_fit_json=%s,
                    postprocess_status=%s,
                    postprocess_error_reason=NULL,
                    updated_at=NOW()
                WHERE id=%s
            """
            params = (
                _json_dumps(event_sense_json),
                _json_dumps(senior_fit_json),
                PostprocessStatus.QUALITY_DONE,
                record_id,
            )
        else:
            sql = """
                UPDATE hot_content_records
                SET demand_event_sense_json=%s,
                    demand_senior_fit_json=%s,
                    updated_at=NOW()
                WHERE id=%s
            """
            params = (
                _json_dumps(event_sense_json),
                _json_dumps(senior_fit_json),
                record_id,
            )
        with self.conn.cursor() as cursor:
            cursor.execute(sql, params)

    def update_postprocess_status(
        self,
        *,
        record_id: int,
        status: int,
        error_message: str | None = None,
    ) -> None:
        sql = """
            UPDATE hot_content_records
            SET postprocess_status=%s,
                postprocess_error_reason=%s,
                updated_at=NOW()
            WHERE id=%s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (status, error_message, record_id))

    def replace_demand_export_rows(
        self,
        *,
        record_id: int,
        source: str,
        hot_title: str,
        article_title: str,
        rows: list[dict[str, Any]],
    ) -> None:
        self._ensure_demand_export_table()
        delete_sql = "DELETE FROM hot_content_demand_exports WHERE record_id=%s"
        insert_sql = """
            INSERT INTO hot_content_demand_exports (
                record_id,
                source,
                hot_title,
                article_title,
                item_type,
                item_text,
                point_category,
                matched_demand,
                contribution_score,
                wxindex_keyword,
                all_hot_keywords,
                wxindex_latest_score,
                wxindex_trend,
                is_as_demand,
                event_sense_score,
                senior_fit_score,
                created_at,
                updated_at
            )
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW())
        """
        with self.conn.cursor() as cursor:
            cursor.execute(delete_sql, (record_id,))
            insert_rows = [
                (
                    record_id,
                    source,
                    hot_title,
                    article_title,
                    str(item.get("item_type") or ""),
                    str(item.get("item_text") or ""),
                    str(item.get("point_category") or ""),
                    str(item.get("matched_demand") or ""),
                    item.get("contribution_score"),
                    str(item.get("wxindex_keyword") or ""),
                    str(item.get("all_hot_keywords") or ""),
                    float(item.get("wxindex_latest_score") or 0),
                    str(item.get("wxindex_trend") or ""),
                    int(item.get("is_as_demand") or 0),
                    item.get("event_sense_score"),
                    item.get("senior_fit_score"),
                )
                for item in rows
                if str(item.get("item_type") or "").strip()
                and str(item.get("item_text") or "").strip()
            ]
            if insert_rows:
                cursor.executemany(insert_sql, insert_rows)

    def list_odps_sync_records(self, *, partition_dt: str | None = None) -> list[dict[str, Any]]:
        """读取指定创建日已完成质量判断的记录，供 ODPS 同步（按 created_at，不按 updated_at）。"""
        self._ensure_record_quality_columns()
        if partition_dt:
            created_day_start = datetime.strptime(partition_dt, "%Y%m%d")
        else:
            created_day_start = datetime.now(SHANGHAI_TZ).replace(
                hour=0,
                minute=0,
                second=0,
                microsecond=0,
                tzinfo=None,
            )
        created_day_end = created_day_start + timedelta(days=1)
        sql = """
            SELECT
                id,
                contribution_points_json,
                contribution_demand_match_json,
                wxindex_trend_json,
                demand_event_sense_json,
                demand_senior_fit_json
            FROM hot_content_records
            WHERE created_at >= %s
              AND created_at < %s
              AND postprocess_status = %s
              AND contribution_demand_match_json IS NOT NULL
              AND TRIM(CAST(contribution_demand_match_json AS CHAR)) <> ''
            ORDER BY id ASC
        """
        with self.conn.cursor() as cursor:
            cursor.execute(
                sql,
                (created_day_start, created_day_end, PostprocessStatus.QUALITY_DONE),
            )
            rows = cursor.fetchall()

        records: list[dict[str, Any]] = []
        for row in rows:
            records.append(
                {
                    "id": int(row["id"]),
                    "contribution_points_json": _json_loads(
                        row.get("contribution_points_json")
                    ),
                    "contribution_demand_match_json": _json_loads(
                        row.get("contribution_demand_match_json")
                    ),
                    "wxindex_trend_json": _json_loads(row.get("wxindex_trend_json")),
                    "demand_event_sense_json": _json_loads(
                        row.get("demand_event_sense_json")
                    ),
                    "demand_senior_fit_json": _json_loads(
                        row.get("demand_senior_fit_json")
                    ),
                }
            )
        return records

    def list_demand_export_groups(self) -> list[dict[str, Any]]:
        """读取主表当天创建的 record 对应导出分组，仅供 ODPS 当天分区同步（不跨天）。"""
        self._ensure_demand_export_table()
        today_start = datetime.now(SHANGHAI_TZ).replace(
            hour=0,
            minute=0,
            second=0,
            microsecond=0,
            tzinfo=None,
        )
        today_end = today_start + timedelta(days=1)
        sql = """
            SELECT
                e.record_id,
                e.item_type,
                e.item_text,
                e.point_category,
                e.matched_demand,
                e.wxindex_latest_score
            FROM hot_content_demand_exports e
            INNER JOIN hot_content_records r ON r.id = e.record_id
            WHERE r.created_at >= %s
              AND r.created_at < %s
            ORDER BY e.record_id ASC, e.id ASC
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (today_start, today_end))
            rows = cursor.fetchall()

        grouped: dict[int, list[dict[str, Any]]] = {}
        for row in rows:
            record_id = int(row["record_id"])
            grouped.setdefault(record_id, []).append(
                {
                    "item_type": str(row.get("item_type") or ""),
                    "item_text": str(row.get("item_text") or ""),
                    "point_category": str(row.get("point_category") or ""),
                    "matched_demand": str(row.get("matched_demand") or ""),
                    "wxindex_latest_score": float(row.get("wxindex_latest_score") or 0),
                }
            )
        return [
            {"record_id": record_id, "export_rows": export_rows}
            for record_id, export_rows in grouped.items()
        ]

    def _ensure_demand_export_table(self) -> None:
        sql = """
            CREATE TABLE IF NOT EXISTS hot_content_demand_exports (
                id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
                record_id BIGINT UNSIGNED NOT NULL,
                source VARCHAR(64) NOT NULL DEFAULT '',
                hot_title VARCHAR(1024) NOT NULL DEFAULT '',
                article_title VARCHAR(1024) NOT NULL DEFAULT '',
                item_type VARCHAR(32) NOT NULL COMMENT '元素/短语',
                item_text VARCHAR(1024) NOT NULL,
                point_category VARCHAR(32) NOT NULL DEFAULT '' COMMENT '点类型：灵感点/目的点/关键点',
                matched_demand VARCHAR(1024) NOT NULL DEFAULT '' COMMENT '匹配到的需求',
                contribution_score DOUBLE NULL COMMENT '贡献分，仅元素有值',
                wxindex_keyword VARCHAR(1024) NOT NULL DEFAULT '' COMMENT '获取微信指数的元素',
                all_hot_keywords VARCHAR(1024) NOT NULL DEFAULT '' COMMENT '全部热点词',
                wxindex_latest_score DOUBLE NOT NULL DEFAULT 0,
                wxindex_trend VARCHAR(32) NOT NULL DEFAULT '' COMMENT '微信指数趋势',
                is_as_demand TINYINT NOT NULL DEFAULT 0 COMMENT '是否作为需求：0否 1是',
                created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
                updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                PRIMARY KEY (id),
                KEY idx_record_id (record_id),
                KEY idx_source_type (source, item_type)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql)
            self._ensure_demand_export_column(
                cursor,
                "matched_demand",
                """
                ALTER TABLE hot_content_demand_exports
                ADD COLUMN matched_demand VARCHAR(1024) NOT NULL DEFAULT ''
                COMMENT '匹配到的需求'
                AFTER item_text
                """,
            )
            self._ensure_demand_export_column(
                cursor,
                "point_category",
                """
                ALTER TABLE hot_content_demand_exports
                ADD COLUMN point_category VARCHAR(32) NOT NULL DEFAULT ''
                COMMENT '点类型：灵感点/目的点/关键点'
                AFTER item_text
                """,
            )
            self._ensure_demand_export_column(
                cursor,
                "contribution_score",
                """
                ALTER TABLE hot_content_demand_exports
                ADD COLUMN contribution_score DOUBLE NULL
                COMMENT '贡献分，仅词有值'
                AFTER matched_demand
                """,
            )
            self._ensure_demand_export_column(
                cursor,
                "wxindex_trend",
                """
                ALTER TABLE hot_content_demand_exports
                ADD COLUMN wxindex_trend VARCHAR(32) NOT NULL DEFAULT ''
                COMMENT '微信指数趋势'
                AFTER wxindex_latest_score
                """,
            )
            self._ensure_demand_export_column(
                cursor,
                "wxindex_keyword",
                """
                ALTER TABLE hot_content_demand_exports
                ADD COLUMN wxindex_keyword VARCHAR(1024) NOT NULL DEFAULT ''
                COMMENT '获取微信指数的词'
                AFTER contribution_score
                """,
            )
            self._ensure_demand_export_column(
                cursor,
                "all_hot_keywords",
                """
                ALTER TABLE hot_content_demand_exports
                ADD COLUMN all_hot_keywords VARCHAR(1024) NOT NULL DEFAULT ''
                COMMENT '全部热点词'
                AFTER wxindex_keyword
                """,
            )
            self._ensure_demand_export_column(
                cursor,
                "is_as_demand",
                """
                ALTER TABLE hot_content_demand_exports
                ADD COLUMN is_as_demand TINYINT NOT NULL DEFAULT 0
                COMMENT '是否作为需求：0否 1是'
                AFTER wxindex_trend
                """,
            )
            self._ensure_demand_export_column(
                cursor,
                "event_sense_score",
                """
                ALTER TABLE hot_content_demand_exports
                ADD COLUMN event_sense_score DOUBLE NULL
                COMMENT '事件性得分 0-10'
                AFTER is_as_demand
                """,
            )
            self._ensure_demand_export_column(
                cursor,
                "senior_fit_score",
                """
                ALTER TABLE hot_content_demand_exports
                ADD COLUMN senior_fit_score DOUBLE NULL
                COMMENT '老年性得分 0-10'
                AFTER event_sense_score
                """,
            )

    def _ensure_record_quality_columns(self) -> None:
        with self.conn.cursor() as cursor:
            for column_name, alter_sql in (
                (
                    "demand_event_sense_json",
                    """
                    ALTER TABLE hot_content_records
                    ADD COLUMN demand_event_sense_json JSON NULL
                    COMMENT '需求事件性 LLM 评分结果'
                    AFTER wxindex_trend_json
                    """,
                ),
                (
                    "demand_senior_fit_json",
                    """
                    ALTER TABLE hot_content_records
                    ADD COLUMN demand_senior_fit_json JSON NULL
                    COMMENT '需求老年性 LLM 评分结果'
                    AFTER demand_event_sense_json
                    """,
                ),
            ):
                cursor.execute(
                    """
                    SELECT COUNT(*) AS cnt
                    FROM information_schema.COLUMNS
                    WHERE TABLE_SCHEMA = DATABASE()
                      AND TABLE_NAME = 'hot_content_records'
                      AND COLUMN_NAME = %s
                    """,
                    (column_name,),
                )
                if int((cursor.fetchone() or {}).get("cnt") or 0) == 0:
                    cursor.execute(alter_sql)

    def _ensure_category_filter_columns(self) -> None:
        with self.conn.cursor() as cursor:
            for column_name, alter_sql in (
                (
                    "category_filter_json",
                    """
                    ALTER TABLE hot_content_records
                    ADD COLUMN category_filter_json JSON NULL
                    COMMENT '老年人兴趣分类筛选 LLM 结果'
                    AFTER article_url
                    """,
                ),
                (
                    "category_filter_passed",
                    """
                    ALTER TABLE hot_content_records
                    ADD COLUMN category_filter_passed TINYINT NULL
                    COMMENT '分类筛选是否通过：1通过 0不通过 NULL未筛选'
                    AFTER category_filter_json
                    """,
                ),
                (
                    "category_filter_reason",
                    """
                    ALTER TABLE hot_content_records
                    ADD COLUMN category_filter_reason TEXT NULL
                    COMMENT '分类筛选原因'
                    AFTER category_filter_passed
                    """,
                ),
            ):
                cursor.execute(
                    """
                    SELECT COUNT(*) AS cnt
                    FROM information_schema.COLUMNS
                    WHERE TABLE_SCHEMA = DATABASE()
                      AND TABLE_NAME = 'hot_content_records'
                      AND COLUMN_NAME = %s
                    """,
                    (column_name,),
                )
                if int((cursor.fetchone() or {}).get("cnt") or 0) == 0:
                    cursor.execute(alter_sql)

    def _ensure_demand_export_column(
        self,
        cursor: Any,
        column_name: str,
        alter_sql: str,
    ) -> None:
        cursor.execute(
            """
            SELECT COUNT(*) AS cnt
            FROM information_schema.COLUMNS
            WHERE TABLE_SCHEMA = DATABASE()
              AND TABLE_NAME = 'hot_content_demand_exports'
              AND COLUMN_NAME = %s
            """,
            (column_name,),
        )
        if int((cursor.fetchone() or {}).get("cnt") or 0) == 0:
            cursor.execute(alter_sql)

    def list_synced_odps_demand_ids(
        self,
        *,
        partition_dt: str,
    ) -> set[str]:
        """返回 hot_content_odps_sync_log 中指定分区日已有的 demand_id（跨流程去重用）。"""
        normalized_partition_dt = str(partition_dt or "").strip()
        if not normalized_partition_dt:
            return set()
        self._ensure_odps_sync_log_table()
        sql = """
            SELECT demand_id
            FROM hot_content_odps_sync_log
            WHERE partition_dt = %s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (normalized_partition_dt,))
            rows = cursor.fetchall()
        return {
            str(row.get("demand_id") or "").strip()
            for row in rows
            if str(row.get("demand_id") or "").strip()
        }

    def count_odps_sync_log_rows(self, *, partition_dt: str) -> int:
        """统计 hot_content_odps_sync_log 指定分区日已写入条数（供日限额计算）。"""
        normalized_partition_dt = str(partition_dt or "").strip()
        if not normalized_partition_dt:
            return 0
        self._ensure_odps_sync_log_table()
        sql = """
            SELECT COUNT(*) AS cnt
            FROM hot_content_odps_sync_log
            WHERE partition_dt = %s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (normalized_partition_dt,))
            row = cursor.fetchone() or {}
        return int(row.get("cnt") or 0)

    def save_odps_sync_logs(self, rows: list[dict[str, Any]]) -> int:
        if not rows:
            return 0
        self._ensure_odps_sync_log_table()
        sql = """
            INSERT INTO hot_content_odps_sync_log (
                partition_dt,
                strategy,
                demand_id,
                demand_name,
                demand_type,
                record_id,
                weight,
                extend
            )
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
            ON DUPLICATE KEY UPDATE
                demand_name = VALUES(demand_name),
                demand_type = VALUES(demand_type),
                record_id = VALUES(record_id),
                weight = VALUES(weight),
                extend = VALUES(extend),
                synced_at = CURRENT_TIMESTAMP
        """
        insert_rows = [
            (
                str(item.get("partition_dt") or ""),
                str(item.get("strategy") or ""),
                str(item.get("demand_id") or ""),
                str(item.get("demand_name") or ""),
                str(item.get("demand_type") or ""),
                int(item.get("record_id") or 0),
                float(item["weight"]) if item.get("weight") is not None else None,
                str(item.get("extend") or "{}"),
            )
            for item in rows
            if str(item.get("demand_id") or "").strip()
        ]
        with self.conn.cursor() as cursor:
            cursor.executemany(sql, insert_rows)
        return len(insert_rows)

    def _ensure_odps_sync_log_weight_column(self, cursor: Any) -> None:
        cursor.execute(
            """
            SELECT COUNT(*) AS cnt
            FROM information_schema.COLUMNS
            WHERE TABLE_SCHEMA = DATABASE()
              AND TABLE_NAME = 'hot_content_odps_sync_log'
              AND COLUMN_NAME = 'weight'
            """,
        )
        if int((cursor.fetchone() or {}).get("cnt") or 0) == 0:
            cursor.execute(
                """
                ALTER TABLE hot_content_odps_sync_log
                    ADD COLUMN weight DOUBLE NULL DEFAULT NULL
                        COMMENT 'ODPS 需求权重（记录 wxindex 最高分 / 1000000）'
                        AFTER record_id
                """
            )

    def _ensure_odps_sync_log_extend_column(self, cursor: Any) -> None:
        cursor.execute(
            """
            SELECT COUNT(*) AS cnt
            FROM information_schema.COLUMNS
            WHERE TABLE_SCHEMA = DATABASE()
              AND TABLE_NAME = 'hot_content_odps_sync_log'
              AND COLUMN_NAME = 'extend'
            """,
        )
        if int((cursor.fetchone() or {}).get("cnt") or 0) == 0:
            cursor.execute(
                """
                ALTER TABLE hot_content_odps_sync_log
                    ADD COLUMN extend TEXT NULL DEFAULT NULL
                        COMMENT 'ODPS extend 扩展字段 JSON'
                        AFTER weight
                """
            )

    def list_wxindex_word_scores(self, name: str) -> list[dict[str, Any]]:
        word = str(name or "").strip()
        if not word:
            return []
        self._ensure_wxindex_words_table()
        sql = """
            SELECT dt, total_score
            FROM hot_content_wxindex_words
            WHERE name = %s
            ORDER BY dt ASC
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (word,))
            rows = cursor.fetchall()
        scores: list[dict[str, Any]] = []
        for row in rows:
            dt = str(row.get("dt") or "").strip()
            if not dt:
                continue
            try:
                total_score = float(row["total_score"])
            except (TypeError, ValueError, KeyError):
                continue
            scores.append({"ymd": dt, "total_score": total_score})
        return scores

    def list_wxindex_word_scores_in_range(
        self,
        name: str,
        *,
        start_ymd: str,
        end_ymd: str,
    ) -> list[dict[str, Any]]:
        word = str(name or "").strip()
        start = str(start_ymd or "").strip()
        end = str(end_ymd or "").strip()
        if not word or not start or not end:
            return []
        self._ensure_wxindex_words_table()
        sql = """
            SELECT dt, total_score
            FROM hot_content_wxindex_words
            WHERE name = %s
              AND dt >= %s
              AND dt <= %s
            ORDER BY dt ASC
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (word, start, end))
            rows = cursor.fetchall()
        scores: list[dict[str, Any]] = []
        for row in rows:
            dt = str(row.get("dt") or "").strip()
            if not dt:
                continue
            try:
                total_score = float(row["total_score"])
            except (TypeError, ValueError, KeyError):
                continue
            scores.append({"ymd": dt, "total_score": total_score})
        return scores

    def list_active_wxindex_word_meta(
        self,
        *,
        today: date | None = None,
    ) -> list[dict[str, Any]]:
        """返回当天仍在抓取窗口内的 meta（today < fetch_end_ymd）。"""
        current = today or datetime.now(SHANGHAI_TZ).date()
        today_ymd = current.strftime("%Y%m%d")
        self._ensure_wxindex_word_meta_table()
        sql = """
            SELECT id, name, event_created_at, fetch_start_ymd, fetch_end_ymd
            FROM hot_content_wxindex_word_meta
            WHERE fetch_end_ymd > %s
            ORDER BY id ASC
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (today_ymd,))
            rows = cursor.fetchall()
        result: list[dict[str, Any]] = []
        for row in rows:
            meta = self._normalize_wxindex_word_meta_row(row)
            if meta is not None:
                result.append(meta)
        return result

    def save_wxindex_word_record(self, payload: dict[str, Any]) -> int:
        row = _wxindex_word_record_row(payload)
        name = str(row[0])
        analyze_ymd = str(row[2])
        self._ensure_wxindex_word_records_table()
        with self.conn.cursor() as cursor:
            cursor.execute(_WXINDEX_WORD_RECORD_UPSERT_SQL, row)
            cursor.execute(
                """
                SELECT id
                FROM hot_content_wxindex_word_records
                WHERE name = %s
                  AND analyze_ymd = %s
                """,
                (name, analyze_ymd),
            )
            record_row = cursor.fetchone() or {}
        return int(record_row.get("id") or 0)

    def init_wxindex_word_records(
        self,
        payloads: list[dict[str, Any]],
    ) -> dict[str, int]:
        """批量 init 追溯记录：一次 executemany，重复跑只更新 meta/抓取窗口。"""
        if not payloads:
            return {}
        rows: list[tuple[Any, ...]] = []
        analyze_ymd = ""
        names: list[str] = []
        for payload in payloads:
            row = _wxindex_word_record_row(payload)
            rows.append(row)
            analyze_ymd = str(row[2])
            names.append(str(row[0]))
        self._ensure_wxindex_word_records_table()
        with self.conn.cursor() as cursor:
            cursor.executemany(_WXINDEX_WORD_RECORD_INIT_UPSERT_SQL, rows)
        return self.map_wxindex_word_record_ids(analyze_ymd=analyze_ymd, names=names)

    def map_wxindex_word_record_ids(
        self,
        *,
        analyze_ymd: str,
        names: list[str],
    ) -> dict[str, int]:
        normalized_analyze_ymd = str(analyze_ymd or "").strip()
        normalized_names = [
            str(name or "").strip() for name in names if str(name or "").strip()
        ]
        if not normalized_analyze_ymd or not normalized_names:
            return {}
        placeholders = ", ".join(["%s"] * len(normalized_names))
        sql = f"""
            SELECT id, name
            FROM hot_content_wxindex_word_records
            WHERE analyze_ymd = %s
              AND name IN ({placeholders})
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, [normalized_analyze_ymd, *normalized_names])
            rows = cursor.fetchall()
        return {
            str(row.get("name") or "").strip(): int(row.get("id") or 0)
            for row in rows
            if str(row.get("name") or "").strip()
        }

    def list_wxindex_word_records_by_analyze_ymd(
        self,
        *,
        analyze_ymd: str,
        names: list[str] | None = None,
    ) -> dict[str, dict[str, Any]]:
        """按 analyze_ymd 批量读取追溯记录，供同日重跑跳过已完成阶段。"""
        normalized_analyze_ymd = str(analyze_ymd or "").strip()
        if not normalized_analyze_ymd:
            return {}
        normalized_names = [
            str(name or "").strip() for name in (names or []) if str(name or "").strip()
        ]
        self._ensure_wxindex_word_records_table()
        params: list[Any] = [normalized_analyze_ymd]
        name_filter = ""
        if normalized_names:
            placeholders = ", ".join(["%s"] * len(normalized_names))
            name_filter = f" AND name IN ({placeholders})"
            params.extend(normalized_names)
        sql = f"""
            SELECT
                id,
                name,
                meta_id,
                analyze_ymd,
                fetch_start_ymd,
                fetch_end_ymd,
                data_start_ymd,
                data_end_ymd,
                data_days,
                is_sustained_high,
                is_rising,
                is_spike,
                retain_reason,
                is_internal_demand_matched,
                matched_demand,
                demand_cache_run_id,
                internal_demand_match_json,
                senior_fit_score,
                demand_senior_fit_json,
                is_final_retained,
                min_score,
                max_score,
                avg_score,
                detail_json
            FROM hot_content_wxindex_word_records
            WHERE analyze_ymd = %s{name_filter}
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, params)
            rows = cursor.fetchall()
        result: dict[str, dict[str, Any]] = {}
        for row in rows or []:
            name = str(row.get("name") or "").strip()
            if not name:
                continue
            record = dict(row)
            record["internal_demand_match_json"] = _json_loads(
                record.get("internal_demand_match_json")
            )
            record["demand_senior_fit_json"] = _json_loads(
                record.get("demand_senior_fit_json")
            )
            record["detail_json"] = _json_loads(record.get("detail_json"))
            result[name] = record
        return result

    def list_wxindex_word_stats_names(
        self,
        *,
        analyze_ymd: str,
        names: list[str] | None = None,
    ) -> set[str]:
        """返回指定 analyze_ymd 下已写入 stats 表的词名集合。"""
        normalized_analyze_ymd = str(analyze_ymd or "").strip()
        if not normalized_analyze_ymd:
            return set()
        normalized_names = [
            str(name or "").strip() for name in (names or []) if str(name or "").strip()
        ]
        self._ensure_wxindex_word_stats_table()
        params: list[Any] = [normalized_analyze_ymd]
        name_filter = ""
        if normalized_names:
            placeholders = ", ".join(["%s"] * len(normalized_names))
            name_filter = f" AND name IN ({placeholders})"
            params.extend(normalized_names)
        sql = f"""
            SELECT name
            FROM hot_content_wxindex_word_stats
            WHERE analyze_ymd = %s{name_filter}
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, params)
            rows = cursor.fetchall()
        return {
            str(row.get("name") or "").strip()
            for row in rows or []
            if str(row.get("name") or "").strip()
        }

    def save_wxindex_word_stats_batch(
        self,
        payloads: list[dict[str, Any]],
    ) -> int:
        """批量写入通过热度+老年性筛选的词统计。"""
        if not payloads:
            return 0
        rows = [_wxindex_word_stats_row(payload) for payload in payloads]
        self._ensure_wxindex_word_stats_table()
        with self.conn.cursor() as cursor:
            cursor.executemany(_WXINDEX_WORD_STATS_UPSERT_SQL, rows)
        return len(rows)

    def _ensure_wxindex_word_stats_table(self) -> None:
        sql = """
            CREATE TABLE IF NOT EXISTS hot_content_wxindex_word_stats (
                id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '主键',
                name VARCHAR(256) NOT NULL COMMENT '词',
                meta_id BIGINT UNSIGNED NULL COMMENT '关联 meta.id',
                analyze_ymd VARCHAR(8) NOT NULL COMMENT '分析日期 yyyymmdd',
                wxindex_word_record_id BIGINT UNSIGNED NULL COMMENT '关联 records.id',
                retain_reason VARCHAR(64) NULL COMMENT '热度保留原因',
                senior_fit_score DOUBLE NULL COMMENT '老年性得分 0-10',
                data_start_ymd VARCHAR(8) NULL COMMENT '分析数据起始日',
                data_end_ymd VARCHAR(8) NULL COMMENT '分析数据结束日',
                data_days INT NULL COMMENT '分析天数',
                min_score DOUBLE NULL COMMENT '区间最低热度',
                max_score DOUBLE NULL COMMENT '区间最高热度',
                avg_score DOUBLE NULL COMMENT '区间平均热度',
                detail_json JSON NULL COMMENT '扩展详情',
                created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
                updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
                    ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
                PRIMARY KEY (id),
                UNIQUE KEY uk_name_analyze_ymd (name, analyze_ymd),
                KEY idx_analyze_ymd (analyze_ymd),
                KEY idx_retain_reason (retain_reason)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql)

    def _ensure_wxindex_word_records_table(self) -> None:
        sql = """
            CREATE TABLE IF NOT EXISTS hot_content_wxindex_word_records (
                id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '主键',
                name VARCHAR(256) NOT NULL COMMENT '词',
                meta_id BIGINT UNSIGNED NULL COMMENT '关联 meta.id',
                analyze_ymd VARCHAR(8) NOT NULL COMMENT '分析日期 yyyymmdd',
                fetch_start_ymd VARCHAR(8) NULL COMMENT 'meta 抓取起始日',
                fetch_end_ymd VARCHAR(8) NULL COMMENT 'meta 抓取结束日',
                data_start_ymd VARCHAR(8) NULL COMMENT '实际分析数据起始日',
                data_end_ymd VARCHAR(8) NULL COMMENT '实际分析数据结束日',
                data_days INT NULL COMMENT '实际分析天数',
                is_sustained_high TINYINT(1) NULL COMMENT '持续热度>1000万',
                is_rising TINYINT(1) NULL COMMENT '热度持续上涨',
                is_spike TINYINT(1) NULL COMMENT '最近3天突然暴涨',
                retain_reason VARCHAR(64) NULL COMMENT '保留原因（按2->3->1优先级）',
                is_internal_demand_matched TINYINT(1) NULL COMMENT '是否匹配票圈内部需求',
                matched_demand VARCHAR(1024) NULL COMMENT '匹配到的内部需求',
                demand_cache_run_id BIGINT UNSIGNED NULL COMMENT '需求池缓存ID',
                internal_demand_match_json JSON NULL COMMENT '内部需求匹配详情',
                senior_fit_score DOUBLE NULL COMMENT '老年性得分 0-10',
                demand_senior_fit_json JSON NULL COMMENT '老年性 LLM 评分结果',
                is_final_retained TINYINT(1) NULL COMMENT '老年性达标且最终保留',
                min_score DOUBLE NULL COMMENT '区间最低热度',
                max_score DOUBLE NULL COMMENT '区间最高热度',
                avg_score DOUBLE NULL COMMENT '区间平均热度',
                detail_json JSON NULL COMMENT '分析详情',
                created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
                updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
                    ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
                PRIMARY KEY (id),
                UNIQUE KEY uk_name_analyze_ymd (name, analyze_ymd),
                KEY idx_analyze_ymd (analyze_ymd),
                KEY idx_patterns (is_sustained_high, is_rising, is_spike),
                KEY idx_retain_reason (retain_reason)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql)
            self._ensure_wxindex_word_records_retain_columns(cursor)
            self._ensure_wxindex_word_records_senior_columns(cursor)
            self._ensure_wxindex_word_records_nullable_columns(cursor)

    def _ensure_wxindex_word_records_nullable_columns(self, cursor: Any) -> None:
        nullable_specs = {
            "fetch_start_ymd": (
                "MODIFY COLUMN fetch_start_ymd VARCHAR(8) NULL "
                "COMMENT 'meta 抓取起始日'"
            ),
            "fetch_end_ymd": (
                "MODIFY COLUMN fetch_end_ymd VARCHAR(8) NULL "
                "COMMENT 'meta 抓取结束日'"
            ),
            "data_start_ymd": (
                "MODIFY COLUMN data_start_ymd VARCHAR(8) NULL "
                "COMMENT '实际分析数据起始日'"
            ),
            "data_end_ymd": (
                "MODIFY COLUMN data_end_ymd VARCHAR(8) NULL "
                "COMMENT '实际分析数据结束日'"
            ),
            "data_days": (
                "MODIFY COLUMN data_days INT NULL COMMENT '实际分析天数'"
            ),
            "is_sustained_high": (
                "MODIFY COLUMN is_sustained_high TINYINT(1) NULL "
                "COMMENT '持续热度>1000万'"
            ),
            "is_rising": (
                "MODIFY COLUMN is_rising TINYINT(1) NULL "
                "COMMENT '热度持续上涨'"
            ),
            "is_spike": (
                "MODIFY COLUMN is_spike TINYINT(1) NULL "
                "COMMENT '最近3天突然暴涨'"
            ),
            "matched_demand": (
                "MODIFY COLUMN matched_demand VARCHAR(1024) NULL "
                "COMMENT '匹配到的内部需求'"
            ),
            "is_final_retained": (
                "MODIFY COLUMN is_final_retained TINYINT(1) NULL "
                "COMMENT '老年性达标且最终保留'"
            ),
        }
        column_names = list(nullable_specs.keys())
        placeholders = ", ".join(["%s"] * len(column_names))
        cursor.execute(
            f"""
            SELECT COLUMN_NAME, IS_NULLABLE
            FROM information_schema.COLUMNS
            WHERE TABLE_SCHEMA = DATABASE()
              AND TABLE_NAME = 'hot_content_wxindex_word_records'
              AND COLUMN_NAME IN ({placeholders})
            """,
            column_names,
        )
        nullable_map = {
            str(row.get("COLUMN_NAME") or ""): str(row.get("IS_NULLABLE") or "").upper()
            for row in (cursor.fetchall() or [])
        }
        alters = [
            sql
            for column_name, sql in nullable_specs.items()
            if nullable_map.get(column_name) == "NO"
        ]
        if alters:
            cursor.execute(
                "ALTER TABLE hot_content_wxindex_word_records " + ", ".join(alters)
            )

    def _ensure_wxindex_word_records_senior_columns(self, cursor: Any) -> None:
        columns = {
            "senior_fit_score": (
                "ALTER TABLE hot_content_wxindex_word_records "
                "ADD COLUMN senior_fit_score DOUBLE NULL "
                "COMMENT '老年性得分 0-10' "
                "AFTER internal_demand_match_json"
            ),
            "demand_senior_fit_json": (
                "ALTER TABLE hot_content_wxindex_word_records "
                "ADD COLUMN demand_senior_fit_json JSON NULL "
                "COMMENT '老年性 LLM 评分结果' "
                "AFTER senior_fit_score"
            ),
            "is_final_retained": (
                "ALTER TABLE hot_content_wxindex_word_records "
                "ADD COLUMN is_final_retained TINYINT(1) NULL "
                "COMMENT '老年性达标且最终保留' "
                "AFTER demand_senior_fit_json"
            ),
        }
        for column_name, alter_sql in columns.items():
            cursor.execute(
                """
                SELECT COUNT(*) AS cnt
                FROM information_schema.COLUMNS
                WHERE TABLE_SCHEMA = DATABASE()
                  AND TABLE_NAME = 'hot_content_wxindex_word_records'
                  AND COLUMN_NAME = %s
                """,
                (column_name,),
            )
            if int((cursor.fetchone() or {}).get("cnt") or 0) == 0:
                cursor.execute(alter_sql)

    def _ensure_wxindex_word_records_retain_columns(self, cursor: Any) -> None:
        columns = {
            "retain_reason": (
                "ALTER TABLE hot_content_wxindex_word_records "
                "ADD COLUMN retain_reason VARCHAR(64) NULL "
                "COMMENT '保留原因（按2->3->1优先级）' "
                "AFTER is_spike"
            ),
            "is_internal_demand_matched": (
                "ALTER TABLE hot_content_wxindex_word_records "
                "ADD COLUMN is_internal_demand_matched TINYINT(1) NULL "
                "COMMENT '是否匹配票圈内部需求' "
                "AFTER retain_reason"
            ),
            "matched_demand": (
                "ALTER TABLE hot_content_wxindex_word_records "
                "ADD COLUMN matched_demand VARCHAR(1024) NULL "
                "COMMENT '匹配到的内部需求' "
                "AFTER is_internal_demand_matched"
            ),
            "demand_cache_run_id": (
                "ALTER TABLE hot_content_wxindex_word_records "
                "ADD COLUMN demand_cache_run_id BIGINT UNSIGNED NULL "
                "COMMENT '需求池缓存ID' "
                "AFTER matched_demand"
            ),
            "internal_demand_match_json": (
                "ALTER TABLE hot_content_wxindex_word_records "
                "ADD COLUMN internal_demand_match_json JSON NULL "
                "COMMENT '内部需求匹配详情' "
                "AFTER demand_cache_run_id"
            ),
        }
        for column_name, alter_sql in columns.items():
            cursor.execute(
                """
                SELECT COUNT(*) AS cnt
                FROM information_schema.COLUMNS
                WHERE TABLE_SCHEMA = DATABASE()
                  AND TABLE_NAME = 'hot_content_wxindex_word_records'
                  AND COLUMN_NAME = %s
                """,
                (column_name,),
            )
            if int((cursor.fetchone() or {}).get("cnt") or 0) == 0:
                cursor.execute(alter_sql)

    def list_stale_wxindex_words(
        self,
        *,
        end_ymd: str,
        update_window_days: int = 7,
        today: date | None = None,
    ) -> list[dict[str, Any]]:
        """返回更新窗口内、仍缺近 7 日区间数据的词。"""
        target_end = str(end_ymd or "").strip()
        if not target_end:
            return []
        current = today or datetime.now(SHANGHAI_TZ).date()
        active_since = current - timedelta(days=max(update_window_days, 0))
        self._ensure_wxindex_word_meta_table()
        self._ensure_wxindex_words_table()
        sql = """
            SELECT
                m.name,
                m.event_created_at,
                m.fetch_start_ymd,
                MIN(w.dt) AS earliest_dt,
                MAX(w.dt) AS latest_dt
            FROM hot_content_wxindex_word_meta m
            INNER JOIN hot_content_wxindex_words w ON w.name = m.name
            WHERE DATE(m.event_created_at) >= %s
            GROUP BY m.name, m.event_created_at, m.fetch_start_ymd
            HAVING MAX(w.dt) < %s OR MIN(w.dt) > m.fetch_start_ymd
            ORDER BY m.name ASC
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (active_since, target_end))
            rows = cursor.fetchall()
        stale_words: list[dict[str, Any]] = []
        for row in rows:
            name = str(row.get("name") or "").strip()
            fetch_start_ymd = str(row.get("fetch_start_ymd") or "").strip()
            earliest_dt = str(row.get("earliest_dt") or "").strip()
            latest_dt = str(row.get("latest_dt") or "").strip()
            event_created_at = row.get("event_created_at")
            if name and fetch_start_ymd and earliest_dt and latest_dt and event_created_at:
                stale_words.append(
                    {
                        "name": name,
                        "event_created_at": event_created_at,
                        "fetch_start_ymd": fetch_start_ymd,
                        "earliest_dt": earliest_dt,
                        "latest_dt": latest_dt,
                    }
                )
        return stale_words

    def list_word_earliest_event_times(
        self,
        *,
        since_dt: datetime,
    ) -> dict[str, datetime]:
        """从 wxindex_trend_json 汇总近期间每个检索词的最早事件时间。"""
        self._ensure_record_quality_columns()
        sql = """
            SELECT
                word_name,
                MIN(event_created_at) AS event_created_at
            FROM (
                SELECT
                    TRIM(searches.keyword) AS word_name,
                    r.created_at AS event_created_at
                FROM hot_content_records r
                JOIN JSON_TABLE(
                    r.wxindex_trend_json,
                    '$.wxindex_searches[*]' COLUMNS (
                        keyword VARCHAR(256) PATH '$.keyword'
                    )
                ) AS searches
                WHERE r.created_at >= %s
                  AND r.wxindex_trend_json IS NOT NULL
                  AND TRIM(searches.keyword) <> ''

                UNION ALL

                SELECT
                    TRIM(JSON_UNQUOTE(JSON_EXTRACT(r.wxindex_trend_json, '$.llm_selected_word'))) AS word_name,
                    r.created_at AS event_created_at
                FROM hot_content_records r
                WHERE r.created_at >= %s
                  AND r.wxindex_trend_json IS NOT NULL
                  AND TRIM(JSON_UNQUOTE(JSON_EXTRACT(r.wxindex_trend_json, '$.llm_selected_word'))) <> ''
            ) AS word_events
            WHERE word_name IS NOT NULL
              AND word_name <> ''
            GROUP BY word_name
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (since_dt, since_dt))
            rows = cursor.fetchall()

        event_map: dict[str, datetime] = {}
        for row in rows:
            name = str(row.get("word_name") or "").strip()
            event_created_at = row.get("event_created_at")
            if name and isinstance(event_created_at, datetime):
                event_map[name] = event_created_at
        return event_map

    def list_wxindex_word_bounds_without_meta(self) -> list[dict[str, Any]]:
        self._ensure_wxindex_word_meta_table()
        self._ensure_wxindex_words_table()
        sql = """
            SELECT
                w.name,
                MIN(w.dt) AS earliest_dt,
                MIN(w.created_at) AS first_created_at
            FROM hot_content_wxindex_words w
            LEFT JOIN hot_content_wxindex_word_meta m ON m.name = w.name
            WHERE m.name IS NULL
            GROUP BY w.name
            ORDER BY w.name ASC
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql)
            rows = cursor.fetchall()
        bounds: list[dict[str, Any]] = []
        for row in rows:
            name = str(row.get("name") or "").strip()
            earliest_dt = str(row.get("earliest_dt") or "").strip()
            first_created_at = row.get("first_created_at")
            if name and earliest_dt:
                bounds.append(
                    {
                        "name": name,
                        "earliest_dt": earliest_dt,
                        "first_created_at": first_created_at,
                    }
                )
        return bounds

    def list_wxindex_word_names_without_meta(self) -> list[str]:
        self._ensure_wxindex_word_meta_table()
        self._ensure_wxindex_words_table()
        sql = """
            SELECT DISTINCT w.name
            FROM hot_content_wxindex_words w
            LEFT JOIN hot_content_wxindex_word_meta m ON m.name = w.name
            WHERE m.name IS NULL
            ORDER BY w.name ASC
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql)
            rows = cursor.fetchall()
        return [
            str(row.get("name") or "").strip()
            for row in rows
            if str(row.get("name") or "").strip()
        ]

    def get_wxindex_word_first_row_created_at(self, name: str) -> datetime | None:
        word = str(name or "").strip()
        if not word:
            return None
        self._ensure_wxindex_words_table()
        sql = """
            SELECT MIN(created_at) AS first_created_at
            FROM hot_content_wxindex_words
            WHERE name = %s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (word,))
            row = cursor.fetchone() or {}
        first_created_at = row.get("first_created_at")
        return first_created_at if isinstance(first_created_at, datetime) else None

    def list_all_wxindex_word_meta(self) -> list[dict[str, Any]]:
        self._ensure_wxindex_word_meta_table()
        sql = """
            SELECT id, name, event_created_at, fetch_start_ymd, fetch_end_ymd
            FROM hot_content_wxindex_word_meta
            ORDER BY id ASC
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql)
            rows = cursor.fetchall()
        result: list[dict[str, Any]] = []
        for row in rows:
            meta = self._normalize_wxindex_word_meta_row(row)
            if meta is not None:
                result.append(meta)
        return result

    def update_wxindex_word_meta_fetch_start(
        self,
        *,
        name: str,
        fetch_start_ymd: str,
    ) -> None:
        word = str(name or "").strip()
        target_start = str(fetch_start_ymd or "").strip()
        if not word or not target_start:
            raise HotContentFlowError("invalid wxindex word meta fetch_start_ymd payload")
        self._ensure_wxindex_word_meta_table()
        sql = """
            UPDATE hot_content_wxindex_word_meta
            SET fetch_start_ymd = %s
            WHERE name = %s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (target_start, word))

    def update_wxindex_word_meta(
        self,
        *,
        name: str,
        event_created_at: datetime,
        fetch_start_ymd: str,
        fetch_end_ymd: str,
    ) -> None:
        word = str(name or "").strip()
        target_start = str(fetch_start_ymd or "").strip()
        target_end = str(fetch_end_ymd or "").strip()
        if not word or not target_start or not target_end:
            raise HotContentFlowError("invalid wxindex word meta payload")
        self._ensure_wxindex_word_meta_table()
        event_at = event_created_at
        if event_at.tzinfo is not None:
            event_at = event_at.astimezone(SHANGHAI_TZ).replace(tzinfo=None)
        sql = """
            UPDATE hot_content_wxindex_word_meta
            SET event_created_at = %s,
                fetch_start_ymd = %s,
                fetch_end_ymd = %s
            WHERE name = %s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (event_at, target_start, target_end, word))

    def get_wxindex_word_meta(self, name: str) -> dict[str, Any] | None:
        word = str(name or "").strip()
        if not word:
            return None
        self._ensure_wxindex_word_meta_table()
        sql = """
            SELECT id, name, event_created_at, fetch_start_ymd, fetch_end_ymd
            FROM hot_content_wxindex_word_meta
            WHERE name = %s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (word,))
            row = cursor.fetchone()
        if not row:
            return None
        return self._normalize_wxindex_word_meta_row(row)

    def ensure_wxindex_word_meta(
        self,
        *,
        name: str,
        event_created_at: datetime,
        fetch_start_ymd: str,
        fetch_end_ymd: str,
    ) -> dict[str, Any]:
        word = str(name or "").strip()
        target_start = str(fetch_start_ymd or "").strip()
        target_end = str(fetch_end_ymd or "").strip()
        if not word or not target_start or not target_end:
            raise HotContentFlowError("invalid wxindex word meta payload")
        self._ensure_wxindex_word_meta_table()
        event_at = event_created_at
        if event_at.tzinfo is not None:
            event_at = event_at.astimezone(SHANGHAI_TZ).replace(tzinfo=None)
        sql = """
            INSERT INTO hot_content_wxindex_word_meta (
                name,
                event_created_at,
                fetch_start_ymd,
                fetch_end_ymd
            )
            VALUES (%s, %s, %s, %s)
            ON DUPLICATE KEY UPDATE
                event_created_at = VALUES(event_created_at),
                fetch_start_ymd = VALUES(fetch_start_ymd),
                fetch_end_ymd = VALUES(fetch_end_ymd)
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (word, event_at, target_start, target_end))
        meta = self.get_wxindex_word_meta(word)
        if meta is None:
            raise HotContentFlowError(f"failed to persist wxindex word meta: {word}")
        return meta

    def delete_wxindex_word_meta_by_names(self, names: list[str]) -> int:
        words = [str(name or "").strip() for name in names]
        words = [name for name in words if name]
        if not words:
            return 0
        self._ensure_wxindex_word_meta_table()
        placeholders = ", ".join(["%s"] * len(words))
        sql = f"DELETE FROM hot_content_wxindex_word_meta WHERE name IN ({placeholders})"
        with self.conn.cursor() as cursor:
            cursor.execute(sql, tuple(words))
            return int(cursor.rowcount or 0)

    def list_low_max_wxindex_words(
        self,
        *,
        min_max_score: float,
    ) -> list[dict[str, Any]]:
        """按 name 聚合，返回最大值低于阈值的词。"""
        self._ensure_wxindex_words_table()
        sql = """
            SELECT
                name,
                MAX(total_score) AS max_score,
                COUNT(*) AS row_count
            FROM hot_content_wxindex_words
            GROUP BY name
            HAVING MAX(total_score) < %s
            ORDER BY name ASC
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (min_max_score,))
            rows = cursor.fetchall()

        low_words: list[dict[str, Any]] = []
        for row in rows:
            name = str(row.get("name") or "").strip()
            if not name:
                continue
            try:
                max_score = float(row["max_score"])
                row_count = int(row["row_count"])
            except (TypeError, ValueError, KeyError):
                continue
            low_words.append(
                {
                    "name": name,
                    "max_score": max_score,
                    "row_count": row_count,
                }
            )
        return low_words

    def count_wxindex_words_outside_event_window(
        self,
        *,
        window_days: int = 7,
    ) -> int:
        self._ensure_wxindex_word_meta_table()
        self._ensure_wxindex_words_table()
        sql = """
            SELECT COUNT(*) AS row_count
            FROM hot_content_wxindex_words w
            INNER JOIN hot_content_wxindex_word_meta m ON m.name = w.name
            WHERE w.dt < m.fetch_start_ymd
               OR w.dt > m.fetch_end_ymd
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql)
            row = cursor.fetchone() or {}
        return int(row.get("row_count") or 0)

    def list_wxindex_words_outside_event_window_samples(
        self,
        *,
        window_days: int = 7,
        limit: int = 20,
    ) -> list[dict[str, Any]]:
        self._ensure_wxindex_word_meta_table()
        self._ensure_wxindex_words_table()
        sql = """
            SELECT
                w.name,
                w.dt,
                m.event_created_at,
                m.fetch_start_ymd AS start_ymd,
                m.fetch_end_ymd AS end_ymd
            FROM hot_content_wxindex_words w
            INNER JOIN hot_content_wxindex_word_meta m ON m.name = w.name
            WHERE w.dt < m.fetch_start_ymd
               OR w.dt > m.fetch_end_ymd
            ORDER BY w.name ASC, w.dt ASC
            LIMIT %s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (limit,))
            rows = cursor.fetchall()
        samples: list[dict[str, Any]] = []
        for row in rows:
            name = str(row.get("name") or "").strip()
            dt = str(row.get("dt") or "").strip()
            if name and dt:
                samples.append(
                    {
                        "name": name,
                        "dt": dt,
                        "event_created_at": row.get("event_created_at"),
                        "start_ymd": str(row.get("start_ymd") or "").strip(),
                        "end_ymd": str(row.get("end_ymd") or "").strip(),
                    }
                )
        return samples

    def delete_wxindex_words_outside_event_window(
        self,
        *,
        window_days: int = 7,
    ) -> int:
        self._ensure_wxindex_word_meta_table()
        self._ensure_wxindex_words_table()
        sql = """
            DELETE w
            FROM hot_content_wxindex_words w
            INNER JOIN hot_content_wxindex_word_meta m ON m.name = w.name
            WHERE w.dt < m.fetch_start_ymd
               OR w.dt > m.fetch_end_ymd
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql)
            return int(cursor.rowcount or 0)

    def delete_wxindex_words_without_meta(self) -> int:
        self._ensure_wxindex_word_meta_table()
        self._ensure_wxindex_words_table()
        sql = """
            DELETE w
            FROM hot_content_wxindex_words w
            LEFT JOIN hot_content_wxindex_word_meta m ON m.name = w.name
            WHERE m.name IS NULL
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql)
            return int(cursor.rowcount or 0)

    def count_wxindex_words_without_meta(self) -> int:
        self._ensure_wxindex_word_meta_table()
        self._ensure_wxindex_words_table()
        sql = """
            SELECT COUNT(*) AS row_count
            FROM hot_content_wxindex_words w
            LEFT JOIN hot_content_wxindex_word_meta m ON m.name = w.name
            WHERE m.name IS NULL
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql)
            row = cursor.fetchone() or {}
        return int(row.get("row_count") or 0)

    def delete_wxindex_words_by_names(self, names: list[str]) -> int:
        cleaned = [str(name or "").strip() for name in names if str(name or "").strip()]
        if not cleaned:
            return 0
        self._ensure_wxindex_words_table()
        placeholders = ", ".join(["%s"] * len(cleaned))
        sql = f"""
            DELETE FROM hot_content_wxindex_words
            WHERE name IN ({placeholders})
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, tuple(cleaned))
            return int(cursor.rowcount or 0)

    def list_wxindex_word_names_with_dt(
        self,
        names: list[str],
        *,
        dt: str,
    ) -> set[str]:
        """返回在 hot_content_wxindex_words 中存在指定日期数据的词名集合。"""
        target_dt = str(dt or "").strip()
        normalized_names = [
            str(name or "").strip() for name in names if str(name or "").strip()
        ]
        if not target_dt or not normalized_names:
            return set()
        self._ensure_wxindex_words_table()
        placeholders = ", ".join(["%s"] * len(normalized_names))
        sql = f"""
            SELECT DISTINCT name
            FROM hot_content_wxindex_words
            WHERE dt = %s
              AND name IN ({placeholders})
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, [target_dt, *normalized_names])
            rows = cursor.fetchall()
        return {
            str(row.get("name") or "").strip()
            for row in rows or []
            if str(row.get("name") or "").strip()
        }

    def has_wxindex_word(self, name: str) -> bool:
        return self.get_wxindex_word_latest_dt(name) is not None

    def get_wxindex_word_latest_dt(self, name: str) -> str | None:
        word = str(name or "").strip()
        if not word:
            return None
        self._ensure_wxindex_words_table()
        sql = """
            SELECT MAX(dt) AS latest_dt
            FROM hot_content_wxindex_words
            WHERE name = %s
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (word,))
            row = cursor.fetchone() or {}
        latest_dt = str(row.get("latest_dt") or "").strip()
        return latest_dt or None

    def save_wxindex_daily_scores(
        self,
        *,
        name: str,
        scores: list[dict[str, Any]],
    ) -> tuple[int, int]:
        """按词+日期写入每日指数，重复行跳过。返回 (inserted, skipped)。"""
        word = str(name or "").strip()
        if not word or not scores:
            return 0, 0
        self._ensure_wxindex_words_table()
        sql = """
            INSERT IGNORE INTO hot_content_wxindex_words (
                name,
                dt,
                total_score
            )
            VALUES (%s, %s, %s)
        """
        rows: list[tuple[str, str, float]] = []
        seen: set[tuple[str, str]] = set()
        for item in scores:
            if not isinstance(item, dict):
                continue
            dt = str(item.get("ymd") or item.get("dt") or "").strip()
            if not dt:
                continue
            try:
                total_score = float(item["total_score"])
            except (TypeError, ValueError, KeyError):
                continue
            key = (word, dt)
            if key in seen:
                continue
            seen.add(key)
            rows.append((word, dt, total_score))
        if not rows:
            return 0, 0

        inserted = 0
        with self.conn.cursor() as cursor:
            for row in rows:
                cursor.execute(sql, row)
                inserted += int(cursor.rowcount or 0)
        skipped = len(rows) - inserted
        return inserted, skipped

    def list_records_with_wxindex_trend_after(
        self,
        *,
        after_created_at: datetime,
    ) -> list[dict[str, Any]]:
        sql = """
            SELECT id, created_at, wxindex_trend_json
            FROM hot_content_records
            WHERE created_at > %s
              AND wxindex_trend_json IS NOT NULL
              AND TRIM(CAST(wxindex_trend_json AS CHAR)) <> ''
            ORDER BY id ASC
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (after_created_at,))
            rows = cursor.fetchall()

        records: list[dict[str, Any]] = []
        for row in rows:
            records.append(
                {
                    "id": int(row["id"]),
                    "created_at": row.get("created_at"),
                    "wxindex_trend_json": _json_loads(row.get("wxindex_trend_json")),
                }
            )
        return records

    def list_records_with_wxindex_trend(
        self,
        *,
        since_dt: datetime,
    ) -> list[dict[str, Any]]:
        sql = """
            SELECT id, created_at, wxindex_trend_json
            FROM hot_content_records
            WHERE created_at >= %s
              AND wxindex_trend_json IS NOT NULL
              AND TRIM(CAST(wxindex_trend_json AS CHAR)) <> ''
            ORDER BY id ASC
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql, (since_dt,))
            rows = cursor.fetchall()

        records: list[dict[str, Any]] = []
        for row in rows:
            records.append(
                {
                    "id": int(row["id"]),
                    "created_at": row.get("created_at"),
                    "wxindex_trend_json": _json_loads(row.get("wxindex_trend_json")),
                }
            )
        return records

    def _ensure_wxindex_word_meta_table(self) -> None:
        sql = """
            CREATE TABLE IF NOT EXISTS hot_content_wxindex_word_meta (
                id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '主键',
                name VARCHAR(256) NOT NULL COMMENT '词',
                event_created_at DATETIME NOT NULL COMMENT '首次关联热点事件创建时间',
                fetch_start_ymd VARCHAR(8) NOT NULL COMMENT '数据窗口左边界：事件创建日往前7天',
                fetch_end_ymd VARCHAR(8) NOT NULL DEFAULT '' COMMENT '数据窗口右边界：事件创建日后7天',
                meta_created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '元数据创建时间',
                PRIMARY KEY (id),
                UNIQUE KEY uk_name (name),
                KEY idx_event_created_at (event_created_at)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql)
            self._ensure_wxindex_word_meta_id_column(cursor)
            self._ensure_wxindex_word_meta_fetch_end_column(cursor)

    def _ensure_wxindex_word_meta_fetch_end_column(self, cursor: Any) -> None:
        cursor.execute(
            """
            SELECT COUNT(*) AS cnt
            FROM information_schema.COLUMNS
            WHERE TABLE_SCHEMA = DATABASE()
              AND TABLE_NAME = 'hot_content_wxindex_word_meta'
              AND COLUMN_NAME = 'fetch_end_ymd'
            """
        )
        if int((cursor.fetchone() or {}).get("cnt") or 0) == 0:
            cursor.execute(
                """
                ALTER TABLE hot_content_wxindex_word_meta
                    ADD COLUMN fetch_end_ymd VARCHAR(8) NOT NULL DEFAULT ''
                        COMMENT '数据窗口右边界：事件创建日后7天'
                        AFTER fetch_start_ymd
                """
            )
        cursor.execute(
            """
            UPDATE hot_content_wxindex_word_meta
            SET fetch_end_ymd = DATE_FORMAT(
                DATE_ADD(DATE(event_created_at), INTERVAL 7 DAY),
                '%Y%m%d'
            )
            WHERE fetch_end_ymd IS NULL
               OR TRIM(fetch_end_ymd) = ''
            """
        )

    def _ensure_wxindex_word_meta_id_column(self, cursor: Any) -> None:
        cursor.execute(
            """
            SELECT COUNT(*) AS cnt
            FROM information_schema.COLUMNS
            WHERE TABLE_SCHEMA = DATABASE()
              AND TABLE_NAME = 'hot_content_wxindex_word_meta'
              AND COLUMN_NAME = 'id'
            """
        )
        if int((cursor.fetchone() or {}).get("cnt") or 0) > 0:
            return
        cursor.execute(
            """
            ALTER TABLE hot_content_wxindex_word_meta
                DROP PRIMARY KEY,
                ADD COLUMN id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY FIRST,
                ADD UNIQUE KEY uk_name (name)
            """
        )

    @staticmethod
    def _normalize_wxindex_word_meta_row(row: dict[str, Any]) -> dict[str, Any] | None:
        name = str(row.get("name") or "").strip()
        fetch_start_ymd = str(row.get("fetch_start_ymd") or "").strip()
        fetch_end_ymd = str(row.get("fetch_end_ymd") or "").strip()
        event_created_at = row.get("event_created_at")
        if not name or not fetch_start_ymd or event_created_at is None:
            return None
        if not fetch_end_ymd and isinstance(event_created_at, datetime):
            event_date = event_created_at.date()
            fetch_end_ymd = (event_date + timedelta(days=7)).strftime("%Y%m%d")
        if not fetch_end_ymd:
            return None
        try:
            meta_id = int(row.get("id"))
        except (TypeError, ValueError):
            return None
        return {
            "id": meta_id,
            "name": name,
            "event_created_at": event_created_at,
            "fetch_start_ymd": fetch_start_ymd,
            "fetch_end_ymd": fetch_end_ymd,
        }

    def _ensure_wxindex_words_table(self) -> None:
        sql = """
            CREATE TABLE IF NOT EXISTS hot_content_wxindex_words (
                id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
                name VARCHAR(256) NOT NULL COMMENT '词',
                dt VARCHAR(8) NOT NULL COMMENT '日期 yyyymmdd',
                total_score DOUBLE NOT NULL COMMENT '微信指数',
                created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
                PRIMARY KEY (id),
                UNIQUE KEY uk_name_dt (name, dt),
                KEY idx_name (name),
                KEY idx_dt (dt)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql)

    def _ensure_odps_sync_log_table(self) -> None:
        sql = """
            CREATE TABLE IF NOT EXISTS hot_content_odps_sync_log (
                id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
                partition_dt VARCHAR(8) NOT NULL COMMENT 'ODPS 分区 dt',
                strategy VARCHAR(128) NOT NULL COMMENT '需求 strategy',
                demand_id CHAR(32) NOT NULL COMMENT 'ODPS demand_id',
                demand_name VARCHAR(1024) NOT NULL DEFAULT '' COMMENT '需求名',
                demand_type VARCHAR(32) NOT NULL DEFAULT '' COMMENT '特征点/短语',
                record_id BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '来源 hot_content_records.id',
                weight DOUBLE NULL DEFAULT NULL COMMENT 'ODPS 需求权重（记录 wxindex 最高分 / 1000000）',
                extend TEXT NULL DEFAULT NULL COMMENT 'ODPS extend 扩展字段 JSON',
                synced_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '写入 ODPS 时间',
                PRIMARY KEY (id),
                UNIQUE KEY uk_odps_sync (partition_dt, strategy, demand_id),
                KEY idx_record_partition (record_id, partition_dt),
                KEY idx_partition_strategy (partition_dt, strategy)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
        """
        with self.conn.cursor() as cursor:
            cursor.execute(sql)
            self._ensure_odps_sync_log_weight_column(cursor)
            self._ensure_odps_sync_log_extend_column(cursor)