feat: 发布三笙AI情报局独立版公司研发基线
同步家纺情报 v2 数据、59 个严格权威源、趋势图库、移动端交互、Google/Yandex 翻译门控及静态 SEO/GEO 构建。 补充面向 C# 研发的 README，明确当前无 LLM、多模态与生图模型，记录翻译请求参数、缓存与失败策略、数据处理边界、前端职责和无 GitHub Pages 部署流程。 新增按日期倒序维护的 CHANGELOG，记录影响、验证与代码/数据/dist 回滚方法。 验证：11 个 Python 契约测试、13 个 Node 前端测试通过；静态构建生成 1096 个 Sitemap URL。
This commit is contained in:
Jack
@@ -0,0 +1,201 @@
|
||||
"""生成经营者首屏使用的今日必看混排。"""
|
||||
|
||||
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]
|
||||
Reference in New Issue
Block a user