"""生成经营者首屏使用的今日必看混排。""" from __future__ import annotations import json from datetime import date from pathlib import Path from typing import Any TODAY_LIMIT = 24 SECTION_LIMITS = { "calendar": 5, "exhibition": 5, "policy": 4, "standard": 4, "activity": 6, } TREND_CATEGORIES = ("trend_blogs", "fashion_trends", "textile_gallery") TREND_KEYWORDS = ( "家纺", "家居", "床品", "床垫", "被子", "毛毯", "毯", "枕", "窗帘", "地毯", "面料", "纺织", "织物", "羊毛", "棉花", "亚麻", "丝绸", "灯芯绒", "材质", "材料", "色彩", "颜色", "图案", "纹样", "印花", "刺绣", "织造", "可持续", "home textile", "fabric", "material", "colour", "color", "pattern", "weave", "wool", "cotton", "linen", "silk", "rug", "carpet", "curtain", "bedding", "blanket", "pillow", "interior", ) TREND_MATERIAL_KEYWORDS = ( "家纺", "床品", "床垫", "被子", "毛毯", "窗帘", "地毯", "面料", "纺织", "织物", "羊毛", "棉花", "亚麻", "丝绸", "灯芯绒", "织造", "可持续", "home textile", "fabric", "textile", "wool", "cotton", "linen", "silk", "rug", "carpet", "curtain", "bedding", "blanket", "pillow", ) def load_trends(path: Path) -> dict[str, Any]: try: payload = json.loads(path.read_text(encoding="utf-8")) return payload if isinstance(payload, dict) else {} except (OSError, json.JSONDecodeError): return {} def _date_value(value: Any) -> date | None: try: return date.fromisoformat(str(value or "")[:10]) except ValueError: return None def _trend_candidates(now: date, trends: dict[str, Any] | None) -> list[tuple[int, str, dict[str, Any]]]: candidates: list[tuple[int, str, dict[str, Any]]] = [] categories = (trends or {}).get("categories") or {} verified_at = (trends or {}).get("generated_at") or "" actions = { "trend_blogs": "判断其对家纺面料、成本或消费方向的影响，并记录需要验证的选品假设", "fashion_trends": "提取可迁移到家纺的色彩、材质和图案要点，形成打样任务", "textile_gallery": "从纹样、配色和织造细节中提取一组可落地的产品灵感", } targets = { "trend_blogs": ["市场变化", "面料与成本", "消费方向"], "fashion_trends": ["色彩", "材质", "图案"], "textile_gallery": ["纹样", "配色", "织造细节"], } bases = {"trend_blogs": 840, "fashion_trends": 830, "textile_gallery": 790} for category_key in TREND_CATEGORIES: category = categories.get(category_key) or {} items = category.get("items") if isinstance(category, dict) else [] for raw in items or []: raw_id = str(raw.get("id") or "") if not raw_id: continue title_text = " ".join(str(raw.get(key) or "") for key in ("title", "title_original")).lower() summary_text = " ".join(str(raw.get(key) or "") for key in ("summary", "summary_original")).lower() title_hits = sum(keyword in title_text for keyword in TREND_KEYWORDS) summary_hits = sum(keyword in summary_text for keyword in TREND_KEYWORDS) if category_key != "textile_gallery" and not title_hits and not summary_hits: continue if category_key == "trend_blogs" and not any(keyword in f"{title_text} {summary_text}" for keyword in TREND_MATERIAL_KEYWORDS): continue published = _date_value(raw.get("published_at")) age = max(0, (now - published).days) if published else 30 if category_key != "textile_gallery" and age > 45: continue recency = max(0, 30 - min(age, 30)) score = bases[category_key] + min(title_hits, 3) * 30 + min(summary_hits, 4) * 5 + recency + (20 if raw.get("thumbnail") else 0) item = dict(raw) item.update({ "id": f"trend-{raw_id}", "section": "standard", "source_id": f"trend_{category_key}", "trend_category_key": category_key, "image": raw.get("thumbnail") or "", "region": "全球", "authority": "公开趋势来源", "verified_at": verified_at, "product_targets": targets[category_key], "today_action": actions[category_key], }) candidates.append((score, "standard", item)) return candidates def build_today( now: date, calendar: list[dict[str, Any]], exhibitions: list[dict[str, Any]], policies: list[dict[str, Any]], standards: list[dict[str, Any]], activities: list[dict[str, Any]] | None = None, trends: dict[str, Any] | None = None, ) -> list[dict[str, Any]]: candidates: list[tuple[int, str, dict[str, Any]]] = [] for item in exhibitions: start = _date_value(item.get("start_date")) if start is None: continue days = (start - now).days if 0 <= days <= 21: item["today_action"] = item.get("attendance_recommendation") or "核对参展或观展安排" candidates.append((900 - days, "exhibition", item)) for item in calendar: start = _date_value(item.get("start_date")) stocking_start = _date_value(item.get("stocking_start")) stocking_end = _date_value(item.get("stocking_end") or item.get("start_date")) if not start or not stocking_start or not stocking_end: continue days = (start - now).days actionable = ( item.get("business_relevance") in ("high", "medium") and item.get("affected_countries") and item.get("platforms") and item.get("product_suggestions") and stocking_start <= now <= stocking_end ) if 0 <= days <= 180 and actionable: item["today_action"] = item.get("next_action") or "核对备货与入仓计划" candidates.append((850 + int(item.get("priority_score") or 0) - min(days, 180), "calendar", item)) for item in policies: if item.get("policy_type") == "产品安全与召回" and item.get("change_status") != "新发布召回": continue if item.get("change_status") not in ("新发布召回", "检测到近期更新"): continue item["today_action"] = item.get("action") or "核对官方规则与受影响商品" candidates.append((900 if item.get("risk_level") == "高" else 780, "policy", item)) for item in standards: item["today_action"] = item.get("application") or "转成打样或质检任务" candidates.append((520, "standard", item)) for item in activities or []: start = _date_value(item.get("start_date")) if start is None: continue days = (start - now).days relevance = item.get("relevance") max_days = 30 if relevance == "direct" else 14 if 0 <= days <= max_days and relevance in ("direct", "must"): item["today_action"] = item.get("next_action") or "核对议程、报名截止与适用市场" relevance_score = 70 if relevance == "direct" else 45 official_score = 25 if item.get("official_activity") else 0 candidates.append((820 + relevance_score + official_score - days, "activity", item)) candidates.extend(_trend_candidates(now, trends)) candidates.sort(key=lambda row: (-row[0], str(row[2].get("title") or ""))) selected: list[tuple[int, str, dict[str, Any]]] = [] selected_ids: set[str] = set() section_counts = {section: 0 for section in SECTION_LIMITS} def add(row: tuple[int, str, dict[str, Any]]) -> None: _score, section, item = row item_id = str(item.get("id") or "") if not item_id or item_id in selected_ids or len(selected) >= TODAY_LIMIT: return if section_counts[section] >= SECTION_LIMITS[section]: return selected.append(row) selected_ids.add(item_id) section_counts[section] += 1 for category_key in TREND_CATEGORIES: row = next((candidate for candidate in candidates if candidate[2].get("trend_category_key") == category_key), None) if row: add(row) for section in ("policy", "calendar", "exhibition", "activity"): row = next((candidate for candidate in candidates if candidate[1] == section), None) if row: add(row) row = next((candidate for candidate in candidates if candidate[1] == "standard" and not candidate[2].get("trend_category_key")), None) if row: add(row) for row in candidates: add(row) selected.sort(key=lambda row: (-row[0], str(row[2].get("title") or ""))) return [item for _score, _section, item in selected]