"""新热事件需求导出明细:按热点记录创建时间筛选 hot_content_demand_exports。""" from __future__ import annotations import re from datetime import date, datetime, timedelta from typing import Any from zoneinfo import ZoneInfo from sqlalchemy import text from app.db.hot_content_mysql import HotContentSessionLocal SHANGHAI_TZ = ZoneInfo("Asia/Shanghai") DATE_RE = re.compile(r"^\d{8}$") ITEM_TYPE_WORD = "词" ITEM_TYPE_POINT = "点" MAX_EXPORT_ROWS = 50_000 _EXPORT_SELECT = """ SELECT e.id, e.source, e.hot_title, e.item_text, e.point_category, e.matched_demand, e.is_as_demand, e.contribution_score, e.wxindex_keyword, e.all_hot_keywords, e.wxindex_latest_score, e.wxindex_trend, e.event_sense_score, e.senior_fit_score, e.item_type, r.created_at AS record_created_at FROM hot_content_demand_exports e INNER JOIN hot_content_records r ON r.id = e.record_id """ def _normalize_date(date_value: str | None) -> str | None: if not date_value: return None normalized = str(date_value).replace("-", "").strip() if not normalized: return None if not DATE_RE.match(normalized): raise ValueError("date must be yyyymmdd or yyyy-mm-dd") return normalized def _parse_yyyymmdd(value: str) -> date: return datetime.strptime(value, "%Y%m%d").date() def _resolve_date_range( start_dt: str | None, end_dt: str | None, ) -> tuple[datetime, datetime]: """返回 [start_at, end_at_exclusive) 上海时区半开区间。""" today = datetime.now(SHANGHAI_TZ).date() normalized_start = _normalize_date(start_dt) normalized_end = _normalize_date(end_dt) start_day = _parse_yyyymmdd(normalized_start) if normalized_start else today end_day = _parse_yyyymmdd(normalized_end) if normalized_end else today if start_day > end_day: raise ValueError("开始日期不能晚于结束日期") start_at = datetime.combine(start_day, datetime.min.time(), tzinfo=SHANGHAI_TZ) end_at_exclusive = datetime.combine( end_day + timedelta(days=1), datetime.min.time(), tzinfo=SHANGHAI_TZ, ) return start_at, end_at_exclusive def _item_type_label(item_type: str | None) -> str: value = str(item_type or "").strip() if value == ITEM_TYPE_WORD: return "特征点" if value == ITEM_TYPE_POINT: return "短语" return value or "-" def _normalize_item_type(item_type: str | None) -> str | None: if item_type is None: return None value = str(item_type).strip() if not value or value in {"全部", "all"}: return None if value in {ITEM_TYPE_WORD, "特征点"}: return ITEM_TYPE_WORD if value in {ITEM_TYPE_POINT, "短语"}: return ITEM_TYPE_POINT raise ValueError("item_type 须为 词、点、特征点、短语,或留空表示全部") def _build_filters( *, start_dt: str | None, end_dt: str | None, is_as_demand: int | None, has_matched_demand: int | None, item_type: str | None, min_wxindex_latest_score: float | None, min_event_sense_score: float | None, min_senior_fit_score: float | None, ) -> tuple[str, dict[str, object], datetime, datetime]: start_at, end_at_exclusive = _resolve_date_range(start_dt, end_dt) where_parts = [ "r.created_at >= :start_at", "r.created_at < :end_at_exclusive", ] params: dict[str, object] = { "start_at": start_at.replace(tzinfo=None), "end_at_exclusive": end_at_exclusive.replace(tzinfo=None), } if is_as_demand is not None: if is_as_demand not in (0, 1): raise ValueError("is_as_demand 须为 0 或 1") where_parts.append("e.is_as_demand = :is_as_demand") params["is_as_demand"] = is_as_demand if has_matched_demand is not None: if has_matched_demand not in (0, 1): raise ValueError("has_matched_demand 须为 0 或 1") if has_matched_demand == 1: where_parts.append( "e.matched_demand IS NOT NULL AND TRIM(e.matched_demand) <> ''" ) else: where_parts.append( "(e.matched_demand IS NULL OR TRIM(e.matched_demand) = '')" ) normalized_item_type = _normalize_item_type(item_type) if normalized_item_type: where_parts.append("e.item_type = :item_type") params["item_type"] = normalized_item_type if min_wxindex_latest_score is not None: if min_wxindex_latest_score < 0: raise ValueError("min_wxindex_latest_score 不能为负数") where_parts.append("e.wxindex_latest_score >= :min_wxindex_latest_score") params["min_wxindex_latest_score"] = min_wxindex_latest_score if min_event_sense_score is not None: if min_event_sense_score < 0: raise ValueError("min_event_sense_score 不能为负数") where_parts.append("e.event_sense_score >= :min_event_sense_score") params["min_event_sense_score"] = min_event_sense_score if min_senior_fit_score is not None: if min_senior_fit_score < 0: raise ValueError("min_senior_fit_score 不能为负数") where_parts.append("e.senior_fit_score >= :min_senior_fit_score") params["min_senior_fit_score"] = min_senior_fit_score where_sql = f"WHERE {' AND '.join(where_parts)}" return where_sql, params, start_at, end_at_exclusive def _row_to_dict(row: dict[str, Any]) -> dict[str, object]: created_at = row.get("record_created_at") if isinstance(created_at, datetime): record_created_at = created_at.strftime("%Y-%m-%d %H:%M:%S") else: record_created_at = str(created_at) if created_at is not None else "" is_as_demand_raw = row.get("is_as_demand") is_as_demand_int = int(is_as_demand_raw) if is_as_demand_raw is not None else 0 contribution = row.get("contribution_score") return { "id": int(row["id"]), "source": str(row.get("source") or ""), "hot_title": str(row.get("hot_title") or ""), "item_text": str(row.get("item_text") or ""), "point_category": str(row.get("point_category") or ""), "item_type": str(row.get("item_type") or ""), "item_type_label": _item_type_label(row.get("item_type")), "matched_demand": str(row.get("matched_demand") or ""), "is_as_demand": is_as_demand_int, "is_as_demand_label": "是" if is_as_demand_int == 1 else "否", "contribution_score": float(contribution) if contribution is not None else None, "wxindex_keyword": str(row.get("wxindex_keyword") or ""), "all_hot_keywords": str(row.get("all_hot_keywords") or ""), "wxindex_latest_score": float(row.get("wxindex_latest_score") or 0), "wxindex_trend": str(row.get("wxindex_trend") or ""), "event_sense_score": float(row["event_sense_score"]) if row.get("event_sense_score") is not None else None, "senior_fit_score": float(row["senior_fit_score"]) if row.get("senior_fit_score") is not None else None, "item_type": str(row.get("item_type") or ""), "record_created_at": record_created_at, } def query_hot_content_demand_exports( *, start_dt: str | None = None, end_dt: str | None = None, is_as_demand: int | None = None, has_matched_demand: int | None = None, item_type: str | None = None, min_wxindex_latest_score: float | None = None, min_event_sense_score: float | None = None, min_senior_fit_score: float | None = None, page: int = 1, page_size: int = 20, ) -> dict[str, object]: where_sql, params, _, _ = _build_filters( start_dt=start_dt, end_dt=end_dt, is_as_demand=is_as_demand, has_matched_demand=has_matched_demand, item_type=item_type, min_wxindex_latest_score=min_wxindex_latest_score, min_event_sense_score=min_event_sense_score, min_senior_fit_score=min_senior_fit_score, ) offset = (page - 1) * page_size count_sql = text( f""" SELECT COUNT(*) AS cnt FROM hot_content_demand_exports e INNER JOIN hot_content_records r ON r.id = e.record_id {where_sql} """ ) list_sql = text( f""" {_EXPORT_SELECT} {where_sql} ORDER BY r.created_at ASC, e.id ASC LIMIT :limit OFFSET :offset """ ) list_params = {**params, "limit": page_size, "offset": offset} with HotContentSessionLocal() as session: total = int(session.execute(count_sql, params).scalar_one()) rows = session.execute(list_sql, list_params).mappings().all() return { "total": total, "page": page, "page_size": page_size, "items": [_row_to_dict(dict(row)) for row in rows], } def export_hot_content_demand_exports( *, start_dt: str | None = None, end_dt: str | None = None, is_as_demand: int | None = None, has_matched_demand: int | None = None, item_type: str | None = None, min_wxindex_latest_score: float | None = None, min_event_sense_score: float | None = None, min_senior_fit_score: float | None = None, ) -> list[dict[str, object]]: where_sql, params, _, _ = _build_filters( start_dt=start_dt, end_dt=end_dt, is_as_demand=is_as_demand, has_matched_demand=has_matched_demand, item_type=item_type, min_wxindex_latest_score=min_wxindex_latest_score, min_event_sense_score=min_event_sense_score, min_senior_fit_score=min_senior_fit_score, ) export_sql = text( f""" {_EXPORT_SELECT} {where_sql} ORDER BY r.created_at ASC, e.id ASC LIMIT :limit """ ) export_params = {**params, "limit": MAX_EXPORT_ROWS + 1} with HotContentSessionLocal() as session: rows = session.execute(export_sql, export_params).mappings().all() if len(rows) > MAX_EXPORT_ROWS: raise ValueError(f"导出条数超过上限 {MAX_EXPORT_ROWS},请缩小日期或筛选范围") return [_row_to_dict(dict(row)) for row in rows]