922 lines
52 KiB
Python
922 lines
52 KiB
Python
#!/usr/bin/env python3
|
||
"""抓取家纺情报权威源并生成前端数据。"""
|
||
from __future__ import annotations
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import re
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
from dataclasses import asdict
|
||
from datetime import date, datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from urllib.parse import unquote, urlparse
|
||
from zoneinfo import ZoneInfo
|
||
import requests
|
||
if __package__:
|
||
from .intelligence_calendar import (
|
||
COUNTRY_ENGLISH_NAMES,
|
||
aggregate_calendar_insights,
|
||
calendar_opportunity as market_calendar_opportunity,
|
||
calendar_scope,
|
||
enrich_calendar_item as enrich_market_calendar_item,
|
||
holiday_title as market_holiday_title,
|
||
)
|
||
from .intelligence_activities import build_activities, merge_previous_activities
|
||
from .intelligence_exhibitions import build_exhibitions, translate_exhibitions
|
||
from .intelligence_fetch import FetchResult, fetch_all_sources, fetch_json, iso_now, make_session, normalize_text
|
||
from .intelligence_platforms import build_platforms
|
||
from .intelligence_sources import SOURCES, SourceSpec, validate_registry
|
||
from .intelligence_today import build_today, load_trends
|
||
from .intelligence_translation import translate_calendar_titles
|
||
from .intelligence_web_translation import translation_fields
|
||
from .update_textile_trends import load_zh_cache, save_zh_cache, translate_to_zh_cn, validate_payload
|
||
else:
|
||
from intelligence_calendar import (
|
||
COUNTRY_ENGLISH_NAMES,
|
||
aggregate_calendar_insights,
|
||
calendar_opportunity as market_calendar_opportunity,
|
||
calendar_scope,
|
||
enrich_calendar_item as enrich_market_calendar_item,
|
||
holiday_title as market_holiday_title,
|
||
)
|
||
from intelligence_activities import build_activities, merge_previous_activities
|
||
from intelligence_exhibitions import build_exhibitions, translate_exhibitions
|
||
from intelligence_fetch import FetchResult, fetch_all_sources, fetch_json, iso_now, make_session, normalize_text
|
||
from intelligence_platforms import build_platforms
|
||
from intelligence_sources import SOURCES, SourceSpec, validate_registry
|
||
from intelligence_today import build_today, load_trends
|
||
from intelligence_translation import translate_calendar_titles
|
||
from intelligence_web_translation import translation_fields
|
||
from update_textile_trends import load_zh_cache, save_zh_cache, translate_to_zh_cn, validate_payload
|
||
UTC = timezone.utc
|
||
COUNTRY_CODES = (
|
||
"CN", "US", "GB", "DE", "FR", "IT", "ES", "NL", "CA", "AU", "JP", "KR", "BR", "MX",
|
||
"NZ", "SG", "ID", "VN", "PH", "ZA", "PL", "SE", "DK", "NO", "FI", "BE", "AT", "CH",
|
||
"PT", "IE", "CZ", "TR", "RU",
|
||
)
|
||
COUNTRY_NAMES = {
|
||
"CN": "中国", "US": "美国", "GB": "英国", "DE": "德国", "FR": "法国",
|
||
"IT": "意大利", "ES": "西班牙", "NL": "荷兰", "CA": "加拿大", "AU": "澳大利亚",
|
||
"JP": "日本", "KR": "韩国", "BR": "巴西", "MX": "墨西哥", "NZ": "新西兰",
|
||
"SG": "新加坡", "ID": "印度尼西亚", "VN": "越南", "PH": "菲律宾", "ZA": "南非",
|
||
"PL": "波兰", "SE": "瑞典", "DK": "丹麦", "NO": "挪威", "FI": "芬兰",
|
||
"BE": "比利时", "AT": "奥地利", "CH": "瑞士", "PT": "葡萄牙", "IE": "爱尔兰",
|
||
"CZ": "捷克", "TR": "土耳其", "AE": "阿联酋", "RU": "俄罗斯",
|
||
}
|
||
UK_AREA_NAMES = {
|
||
"england-and-wales": "英格兰和威尔士",
|
||
"scotland": "苏格兰",
|
||
"northern-ireland": "北爱尔兰",
|
||
}
|
||
TEXTILE_RECALL_KEYWORDS = (
|
||
"bedding", "blanket", "comforter", "duvet", "sheet", "pillow", "curtain", "rug",
|
||
"carpet", "towel", "sleep sack", "mattress", "mattress cover", "crib liner",
|
||
)
|
||
POSTER_PALETTES = (
|
||
("#0051d6", "#102a43", "#eef7ff"),
|
||
("#2563eb", "#0b1324", "#eef5ff"),
|
||
("#16a085", "#071b18", "#eafbf7"),
|
||
("#5b6ee1", "#111a3c", "#f1f3ff"),
|
||
("#2f8f83", "#0f302c", "#eefaf7"),
|
||
)
|
||
CULTURE_PALETTES = {
|
||
"china": ("#0051d6", "#102a43", "#eef7ff"),
|
||
"japan": ("#4776d0", "#172b4d", "#f2f6ff"),
|
||
"korea": ("#315d9c", "#151b25", "#f1f6fb"),
|
||
"north-america": ("#1677a6", "#12334a", "#edf8fc"),
|
||
"british-isles": ("#365b9d", "#153561", "#f1f5fb"),
|
||
"central-europe": ("#2f6b63", "#182722", "#edf8f5"),
|
||
"southern-europe": ("#4a78a8", "#18384f", "#f1f7fb"),
|
||
"nordic": ("#3d7f91", "#183d49", "#eef7f8"),
|
||
"oceania": ("#168b83", "#164f4a", "#edf9f7"),
|
||
"southeast-asia": ("#0f8f78", "#105d5b", "#ecfaf6"),
|
||
"latin-america": ("#5568c9", "#273266", "#f1f2ff"),
|
||
"southern-africa": ("#3f7c69", "#123d32", "#eff8f3"),
|
||
"turkiye": ("#3e7194", "#17384a", "#eef6fa"),
|
||
"islamic": ("#16836f", "#0e534a", "#edf9f5"),
|
||
"sport": ("#1976a8", "#17324d", "#eef7fb"),
|
||
"retail": ("#0051d6", "#102a43", "#eef7ff"),
|
||
}
|
||
def poster_for(key: str, section: str, culture_key: str = "") -> dict[str, str]:
|
||
digest = hashlib.sha1(key.encode("utf-8")).hexdigest()
|
||
palette = CULTURE_PALETTES.get(culture_key) or POSTER_PALETTES[int(digest[:2], 16) % len(POSTER_PALETTES)]
|
||
labels = {"calendar": "备货日历", "exhibition": "展会通", "policy": "政策预警", "standard": "趋势雷达"}
|
||
return {"eyebrow": labels.get(section, "家纺跨境情报"), "accent": palette[0], "ink": palette[1], "paper": palette[2]}
|
||
def make_id(prefix: str, *parts: str) -> str:
|
||
raw = "||".join(parts).lower().encode("utf-8")
|
||
return f"{prefix}-{hashlib.sha1(raw).hexdigest()[:14]}"
|
||
def date_value(value: str) -> date:
|
||
return date.fromisoformat(value[:10])
|
||
def base_item(source: SourceSpec, result: FetchResult, section: str) -> dict[str, Any]:
|
||
item_id = make_id(section, source.id, source.title_zh or result.title, source.url)
|
||
return {
|
||
"id": item_id,
|
||
"section": section,
|
||
"source_id": source.id,
|
||
"source": source.name,
|
||
"authority": source.authority,
|
||
"region": source.region,
|
||
"title": source.title_zh or result.title,
|
||
"title_original": result.title,
|
||
"summary": source.summary_zh or result.description,
|
||
"summary_original": result.description,
|
||
"url": result.final_url,
|
||
"source_url": source.url,
|
||
"image": result.image,
|
||
"poster": poster_for(item_id, section),
|
||
"verified_at": result.checked_at,
|
||
"content_hash": result.content_hash,
|
||
"change_status": "检测到更新" if result.changed else "持续监测",
|
||
"confidence": "高" if source.authority in ("官方", "平台官方") else "中高",
|
||
}
|
||
def build_nager_events(now: date, horizon: date, source: SourceSpec, result: FetchResult) -> list[dict[str, Any]]:
|
||
years = range(now.year, horizon.year + 1)
|
||
requests_list = [(code, year) for code in COUNTRY_CODES for year in years]
|
||
payloads: dict[tuple[str, int], list[dict[str, Any]]] = {}
|
||
with ThreadPoolExecutor(max_workers=8) as executor:
|
||
futures = {
|
||
executor.submit(fetch_json, f"https://date.nager.at/api/v3/PublicHolidays/{year}/{code}"): (code, year)
|
||
for code, year in requests_list
|
||
}
|
||
for future in as_completed(futures):
|
||
code_year = futures[future]
|
||
try:
|
||
data = future.result()
|
||
payloads[code_year] = data if isinstance(data, list) else []
|
||
except Exception as exc:
|
||
print(f"[日历警告] Nager {code_year[0]} {code_year[1]}: {exc}")
|
||
items: list[dict[str, Any]] = []
|
||
for (code, _year), rows in payloads.items():
|
||
for row in rows:
|
||
event_date = str(row.get("date", ""))
|
||
if not event_date:
|
||
continue
|
||
day = date_value(event_date)
|
||
if day < now or day > horizon:
|
||
continue
|
||
local_title = normalize_text(str(row.get("localName") or ""))
|
||
english_title = normalize_text(str(row.get("name") or local_title or "公共假日"))
|
||
if english_title.lower().replace("’", "'") == "new year's day" and day.month == 12 and day.day == 31:
|
||
continue
|
||
title = market_holiday_title(local_title, english_title, code)
|
||
subdivision_codes = [normalize_text(str(value)) for value in row.get("counties") or [] if value]
|
||
applicable_areas, country_scope = calendar_scope(bool(row.get("global", True)), subdivision_codes)
|
||
item = base_item(source, result, "calendar")
|
||
country_slug = re.sub(r"[^a-z0-9]+", "-", COUNTRY_ENGLISH_NAMES.get(code, code).lower().replace("ü", "u")).strip("-")
|
||
event_source_url = f"https://date.nager.at/PublicHoliday/{country_slug}/{day.year}"
|
||
item.update({
|
||
"id": make_id("calendar", source.id, code, event_date, title),
|
||
"title": title,
|
||
"title_original": local_title if local_title and local_title != title else english_title,
|
||
"title_english": english_title,
|
||
"summary": f"{COUNTRY_NAMES.get(code, code)}公共假日。{market_calendar_opportunity(f'{title} {english_title}', source.id)}",
|
||
"region": COUNTRY_NAMES.get(code, code),
|
||
"start_date": event_date,
|
||
"end_date": event_date,
|
||
"impact": "中",
|
||
"opportunity": market_calendar_opportunity(f"{title} {english_title}", source.id),
|
||
"market_area": code,
|
||
"is_national_holiday": bool(row.get("global", True)),
|
||
"subdivision_codes": subdivision_codes,
|
||
"applicable_areas": applicable_areas,
|
||
"country_scope": country_scope,
|
||
"url": event_source_url,
|
||
"source_url": event_source_url,
|
||
"authority": "结构化公共假日数据",
|
||
})
|
||
item.update(translation_fields(
|
||
event_source_url,
|
||
"zh-CN" if code == "CN" else "en",
|
||
local_title or english_title,
|
||
))
|
||
items.append(enrich_market_calendar_item(item, now, code, poster_for))
|
||
return items
|
||
def build_govuk_events(now: date, horizon: date, source: SourceSpec, result: FetchResult) -> list[dict[str, Any]]:
|
||
try:
|
||
payload = json.loads(result.raw.decode("utf-8"))
|
||
except Exception:
|
||
return []
|
||
items: list[dict[str, Any]] = []
|
||
for area_key, area in payload.items():
|
||
area_name = UK_AREA_NAMES.get(area_key, normalize_text(str(area.get("division") or area_key)))
|
||
for row in area.get("events", []):
|
||
event_date = str(row.get("date", ""))
|
||
if not event_date:
|
||
continue
|
||
day = date_value(event_date)
|
||
if day < now or day > horizon:
|
||
continue
|
||
title = normalize_text(str(row.get("title") or "Bank holiday"))
|
||
item = base_item(source, result, "calendar")
|
||
item.update({
|
||
"id": make_id("calendar", source.id, area_key, event_date, title),
|
||
"title": title,
|
||
"title_english": title,
|
||
"region": "英国",
|
||
"start_date": event_date,
|
||
"end_date": event_date,
|
||
"impact": "中",
|
||
"opportunity": market_calendar_opportunity(title, source.id),
|
||
"summary": f"英国政府公布的 {area_name} 银行假日。{market_calendar_opportunity(title, source.id)}",
|
||
"market_area": f"GB:{area_key}",
|
||
"is_national_holiday": False,
|
||
"applicable_areas": [area_name],
|
||
"country_scope": f"仅适用于{area_name}",
|
||
})
|
||
items.append(enrich_market_calendar_item(item, now, "GB", poster_for))
|
||
return items
|
||
def build_aladhan_events(now: date, horizon: date, source: SourceSpec, result: FetchResult) -> list[dict[str, Any]]:
|
||
"""从逐月公历转换结果中提取主要伊斯兰节庆节点。"""
|
||
requests_list = [(month, year) for year in range(now.year, horizon.year + 1) for month in range(1, 13)]
|
||
rows: list[dict[str, Any]] = []
|
||
with ThreadPoolExecutor(max_workers=8) as executor:
|
||
futures = {
|
||
executor.submit(fetch_json, f"https://api.aladhan.com/v1/gToHCalendar/{month}/{year}"): (month, year)
|
||
for month, year in requests_list
|
||
}
|
||
for future in as_completed(futures):
|
||
month_year = futures[future]
|
||
try:
|
||
payload = future.result()
|
||
if isinstance(payload, dict) and isinstance(payload.get("data"), list):
|
||
rows.extend(payload["data"])
|
||
except Exception as exc:
|
||
print(f"[日历警告] AlAdhan {month_year[1]}-{month_year[0]:02d}: {exc}")
|
||
markers = {
|
||
(9, 1): ("斋月开始", "高", "提前 120 天准备中东礼赠床品、祈祷毯、餐桌布艺与夜间居家场景产品。"),
|
||
(10, 1): ("开斋节 Eid al-Fitr", "高", "围绕家庭团聚与礼赠准备高品质床品套装、毛巾、桌布和节庆包装。"),
|
||
(12, 10): ("宰牲节 Eid al-Adha", "高", "提前规划礼赠套装、客房床品、餐桌布艺和适配家庭聚会的耐用产品。"),
|
||
(1, 1): ("伊斯兰新年", "中", "针对穆斯林市场准备焕新、礼赠和节庆主题软装内容。"),
|
||
}
|
||
items: list[dict[str, Any]] = []
|
||
for row in rows:
|
||
hijri = row.get("hijri") if isinstance(row, dict) else None
|
||
gregorian = row.get("gregorian") if isinstance(row, dict) else None
|
||
if not isinstance(hijri, dict) or not isinstance(gregorian, dict):
|
||
continue
|
||
month_info = hijri.get("month") if isinstance(hijri.get("month"), dict) else {}
|
||
marker = (int(month_info.get("number", 0)), int(hijri.get("day", 0)))
|
||
if marker not in markers:
|
||
continue
|
||
raw_date = str(gregorian.get("date", ""))
|
||
try:
|
||
day = datetime.strptime(raw_date, "%d-%m-%Y").date()
|
||
except ValueError:
|
||
continue
|
||
if day < now or day > horizon:
|
||
continue
|
||
title, impact, opportunity = markers[marker]
|
||
item = base_item(source, result, "calendar")
|
||
item.update({
|
||
"id": make_id("calendar", source.id, day.isoformat(), title),
|
||
"title": title,
|
||
"title_original": normalize_text(str(month_info.get("en") or title)),
|
||
"region": "中东及穆斯林市场",
|
||
"start_date": day.isoformat(),
|
||
"end_date": day.isoformat(),
|
||
"impact": impact,
|
||
"opportunity": opportunity,
|
||
"summary": f"基于 AlAdhan 公历与伊斯兰历转换结果。{opportunity}",
|
||
"market_area": "MENA",
|
||
})
|
||
items.append(enrich_market_calendar_item(item, now, "", poster_for))
|
||
return items
|
||
def first_weekday_after(day: date, weekday: int) -> date:
|
||
offset = (weekday - day.weekday()) % 7
|
||
return day + timedelta(days=offset)
|
||
def fourth_thursday(year: int, month: int) -> date:
|
||
first = date(year, month, 1)
|
||
first_thursday = first_weekday_after(first, 3)
|
||
return first_thursday + timedelta(days=21)
|
||
def fixed_calendar_items(now: date, horizon: date, results: dict[str, FetchResult]) -> list[dict[str, Any]]:
|
||
items: list[dict[str, Any]] = []
|
||
templates: list[tuple[str, date, date, str, str, str, str]] = []
|
||
for year in range(now.year, horizon.year + 1):
|
||
thanksgiving = fourth_thursday(year, 11)
|
||
black_friday = thanksgiving + timedelta(days=1)
|
||
cyber_monday = thanksgiving + timedelta(days=4)
|
||
templates.extend((
|
||
("calendar_nrf", date(year, 7, 15), date(year, 9, 15), "美国开学季与大学返校季", "高", "围绕 Twin XL 宿舍床品、浴巾、床垫保护套和收纳软品准备一站式套装。", "US"),
|
||
("calendar_uk_school_terms", date(year, 9, 1), date(year, 9, 30), "英国秋季返校与学生住宿采购季", "高", "按目标地区校历核对开学时间，准备耐洗单人床品、床垫保护套、浴巾与宿舍收纳软品。", "GB"),
|
||
("calendar_de_school_holidays", date(year, 8, 10), date(year, 9, 15), "德国夏末返校与学生住宿采购季", "高", "按目标州校历核对开学时间，准备耐用单人被套、床垫保护套、毛巾和中性色小空间软装。", "DE"),
|
||
("calendar_nager", date(year, 8, 20), date(year, 9, 15), "法国新学年返校与学生住房采购季", "高", "按目标大区校历核对返校日，准备单人床品、被套、浴巾和小户型软装。", "FR"),
|
||
("calendar_japan", date(year, 3, 1), date(year, 4, 15), "日本新学期与独居搬家采购季", "高", "围绕四月新学期、大学入学和独居搬家准备日式单人寝具、毛巾、窗帘与收纳软品。", "JP"),
|
||
("calendar_nager", date(year, 8, 1), date(year, 9, 15), "加拿大返校与大学宿舍采购季", "高", "按目标省校历核对返校时间，准备保暖单人床品、床垫保护套、浴巾和宿舍收纳。", "CA"),
|
||
("calendar_nager", date(year, 1, 5), date(year, 2, 15), "澳大利亚新学年与学生住宿采购季", "高", "按目标州校历核对第一学期开学日，准备透气床品、浴巾、防晒窗帘和学生公寓软装。", "AU"),
|
||
("calendar_tmall", date(year, 11, 11), date(year, 11, 11), "双十一全球购物节", "高", "提前 120 至 180 天规划冬季床品、礼盒套装、智能温控与高客单家居软装。", "CN"),
|
||
("calendar_nrf", black_friday, black_friday, "Black Friday 黑色星期五", "高", "提前 120 至 180 天完成北美海外仓入仓，主推套装、礼赠和强折扣款。", "US"),
|
||
("calendar_nrf", cyber_monday, cyber_monday, "Cyber Monday 网络星期一", "高", "围绕线上转化准备轻小件、组合装、快递友好包装和补货库存。", "US"),
|
||
("calendar_nrf", date(year, 11, 25), date(year, 12, 24), "北美冬季假日消费季", "高", "以圣诞配色、保暖床品、盖毯、餐桌布艺和礼赠包装形成系列化供给。", "US"),
|
||
))
|
||
for source_id, start, end, title, impact, opportunity, country_code in templates:
|
||
if end < now or start > horizon:
|
||
continue
|
||
result = results[source_id]
|
||
source = result.source
|
||
item = base_item(source, result, "calendar")
|
||
item.update({
|
||
"id": make_id("calendar", source_id, start.isoformat(), title),
|
||
"title": title,
|
||
"start_date": start.isoformat(),
|
||
"end_date": end.isoformat(),
|
||
"impact": impact,
|
||
"opportunity": opportunity,
|
||
"summary": f"{source.summary_zh}{opportunity}",
|
||
})
|
||
school_review = {
|
||
"US": ("National Retail Federation Back-to-School", "美国零售联合会专题数据", "https://nrf.com/research-insights/holiday-data-and-trends/back-to-school", "en"),
|
||
"FR": ("法国教育部学校日历", "法国教育部官方校历（按大区复核）", "https://www.education.gouv.fr/calendrier-scolaire-100148", "fr"),
|
||
"JP": ("日本文部科学省教育概览", "日本政府教育信息（具体日期按学校复核）", "https://www.mext.go.jp/en/policy/education/overview/index.htm", "en"),
|
||
"CA": ("Ontario School Year Calendars", "加拿大安大略省政府校历（其他省另行复核）", "https://www.ontario.ca/page/school-year-calendars", "en"),
|
||
"AU": ("Australian school term dates", "澳大利亚政府学期日期入口（按州复核）", "https://www.education.gov.au/about-department/resources/term-dates-across-australia", "en"),
|
||
}
|
||
translation_language = str(result.page_language or "")
|
||
translation_text = f"{result.title} {result.description}"
|
||
if country_code in school_review and ("开学" in title or "返校" in title or "新学期" in title or "新学年" in title):
|
||
source_name, authority, review_url, translation_language = school_review[country_code]
|
||
translation_text = source_name
|
||
item.update({
|
||
"url": review_url,
|
||
"source_url": review_url,
|
||
"source": source_name,
|
||
"authority": authority,
|
||
})
|
||
item.update(translation_fields(
|
||
str(item.get("source_url") or item.get("url") or ""),
|
||
translation_language,
|
||
translation_text,
|
||
))
|
||
item["market_area"] = country_code
|
||
items.append(enrich_market_calendar_item(item, now, country_code, poster_for))
|
||
for source_id in ("calendar_nfl", "calendar_la28"):
|
||
result = results[source_id]
|
||
source = result.source
|
||
if not source.start_date or date_value(source.end_date) < now or date_value(source.start_date) > horizon:
|
||
continue
|
||
item = base_item(source, result, "calendar")
|
||
item.update({
|
||
"start_date": source.start_date,
|
||
"end_date": source.end_date,
|
||
"venue": source.venue,
|
||
"impact": "高",
|
||
"opportunity": source.opportunity,
|
||
"market_area": "US",
|
||
})
|
||
items.append(enrich_market_calendar_item(item, now, "US", poster_for))
|
||
return items
|
||
def dedupe_calendar(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
uk_government_keys = {
|
||
(str(item.get("start_date")), normalize_text(str(item.get("title_english") or item.get("title"))).lower())
|
||
for item in items
|
||
if item.get("source_id") == "calendar_uk"
|
||
}
|
||
seen: dict[tuple[str, str, str], dict[str, Any]] = {}
|
||
output: list[dict[str, Any]] = []
|
||
ordered = sorted(
|
||
items,
|
||
key=lambda row: (
|
||
row["start_date"],
|
||
-int(row.get("priority_score") or 0),
|
||
row["region"],
|
||
row["title"],
|
||
),
|
||
)
|
||
for item in ordered:
|
||
english_key = normalize_text(str(item.get("title_english") or item.get("title"))).lower()
|
||
if item.get("source_id") == "calendar_nager" and item.get("country_code") == "GB":
|
||
if (str(item.get("start_date")), english_key) in uk_government_keys:
|
||
continue
|
||
market_area = normalize_text(str(item.get("market_area") or item.get("region"))).lower()
|
||
key = (str(item.get("start_date")), market_area, english_key)
|
||
if key in seen:
|
||
existing = seen[key]
|
||
codes = list(dict.fromkeys([*(existing.get("subdivision_codes") or []), *(item.get("subdivision_codes") or [])]))
|
||
areas, scope = calendar_scope(False, codes)
|
||
existing.update({"subdivision_codes": codes, "applicable_areas": areas, "country_scope": scope})
|
||
continue
|
||
seen[key] = item
|
||
output.append(item)
|
||
return output
|
||
def build_calendar(now: date, horizon: date, results: dict[str, FetchResult]) -> tuple[list[dict[str, Any]], int]:
|
||
items = build_nager_events(now, horizon, results["calendar_nager"].source, results["calendar_nager"])
|
||
items.extend(build_govuk_events(now, horizon, results["calendar_uk"].source, results["calendar_uk"]))
|
||
items.extend(build_aladhan_events(now, horizon, results["calendar_aladhan"].source, results["calendar_aladhan"]))
|
||
items.extend(fixed_calendar_items(now, horizon, results))
|
||
return dedupe_calendar(items), len(items)
|
||
def cpsc_title_from_url(url: str, fallback: str) -> str:
|
||
"""CPSC API 偶尔会串接相邻召回标题，官方详情 URL 的 slug 更稳定。"""
|
||
path = unquote(urlparse(url).path)
|
||
if "/Recalls/" not in path:
|
||
return fallback
|
||
slug = path.rstrip("/").split("/")[-1]
|
||
if len(slug) < 12:
|
||
return fallback
|
||
return normalize_text(slug.replace("-", " "))
|
||
def parse_cpsc_recalls(source: SourceSpec, result: FetchResult, now: date, limit: int = 24) -> list[dict[str, Any]]:
|
||
try:
|
||
payload = json.loads(result.raw.decode("utf-8"))
|
||
except Exception:
|
||
return []
|
||
matches: list[dict[str, Any]] = []
|
||
for row in payload if isinstance(payload, list) else []:
|
||
products = row.get("Products") or []
|
||
product_names = " ".join(str(product.get("Name", "")) for product in products if isinstance(product, dict))
|
||
haystack = f"{row.get('Title', '')} {product_names}".lower()
|
||
if not any(
|
||
re.search(rf"(?<![a-z]){re.escape(keyword)}(?:s|es)?(?![a-z])", haystack)
|
||
for keyword in TEXTILE_RECALL_KEYWORDS
|
||
):
|
||
continue
|
||
recall_date = normalize_text(str(row.get("RecallDate") or ""))
|
||
recall_url = normalize_text(str(row.get("URL") or result.final_url))
|
||
api_title = normalize_text(str(row.get("Title") or product_names or "CPSC Recall"))
|
||
title_original = cpsc_title_from_url(recall_url, api_title)
|
||
description = normalize_text(str(row.get("Description") or source.summary_zh))[:600]
|
||
combined = f"{title_original} {description}".lower()
|
||
platforms = [name for token, name in (
|
||
("amazon", "Amazon"), ("walmart", "Walmart"), ("target", "Target"),
|
||
("temu", "Temu"), ("shein", "SHEIN"), ("ebay", "eBay"),
|
||
) if token in combined]
|
||
severe = any(token in combined for token in ("death", "fire", "suffocation", "entrapment", "strangulation"))
|
||
try:
|
||
published_day = date.fromisoformat(recall_date[:10])
|
||
except ValueError:
|
||
published_day = date.min
|
||
change_status = "新发布召回" if published_day >= now - timedelta(days=45) else "历史有效召回"
|
||
item_id = make_id("policy", source.id, str(row.get("RecallID", "")), title_original)
|
||
item = {
|
||
"id": item_id,
|
||
"section": "policy",
|
||
"source_id": source.id,
|
||
"source": source.name,
|
||
"authority": source.authority,
|
||
"region": source.region,
|
||
"title": f"美国消费品召回：{title_original}",
|
||
"title_original": title_original,
|
||
"summary": description,
|
||
"summary_original": normalize_text(str(row.get("Description") or ""))[:600],
|
||
"url": recall_url,
|
||
"source_url": source.url,
|
||
"image": "",
|
||
"poster": poster_for(item_id, "policy"),
|
||
"published_at": recall_date,
|
||
"verified_at": result.checked_at,
|
||
"content_hash": result.content_hash,
|
||
"change_status": change_status,
|
||
"confidence": "高",
|
||
"risk_level": "高" if severe else "中",
|
||
"policy_type": "产品安全与召回",
|
||
"scope_type": "region",
|
||
"affected_markets": ["美国"],
|
||
"platforms": platforms,
|
||
"affected_products": [name for name in product_names.split(" | ") if name] or [product_names],
|
||
"effective_at": recall_date,
|
||
"change_summary": f"CPSC 发布产品召回，涉及{product_names or '相关纺织与家居产品'}。",
|
||
"consequence": "继续销售受召回商品可能导致平台下架、退款、监管处罚和人身安全责任。",
|
||
"action": "立即核对在售与在途商品的型号、批次和风险描述，暂停受影响商品并按官方召回要求处理。",
|
||
}
|
||
item["action_priority"] = policy_action_priority(
|
||
str(item["policy_type"]), str(item["change_status"]), str(item["risk_level"])
|
||
)
|
||
item.update(translation_fields(recall_url))
|
||
matches.append(item)
|
||
matches.sort(key=lambda row: row.get("published_at", ""), reverse=True)
|
||
return matches[:limit]
|
||
def policy_consequence(policy_type: str) -> str:
|
||
if "平台" in policy_type:
|
||
return "未按平台规则调整，可能导致商品限流、下架、扣款或店铺权限受限。"
|
||
if "知识产权" in policy_type:
|
||
return "未完成检索和留档，可能引发投诉、下架、冻结资金或跨境诉讼。"
|
||
if any(token in policy_type for token in ("关税", "清关", "进口")):
|
||
return "税费、申报或主体责任判断错误，会直接影响落地成本、清关时效和利润。"
|
||
return "未及时核对规则，可能影响商品准入、标签、材料证明、物流或平台经营。"
|
||
def policy_action_priority(policy_type: str, change_status: str, risk_level: str) -> int:
|
||
score = 60 if risk_level == "高" else 35
|
||
if change_status in ("新发布召回", "检测到近期更新"):
|
||
score += 80
|
||
if any(token in policy_type for token in ("关税", "清关", "进口", "平台")):
|
||
score += 35
|
||
elif "知识产权" in policy_type:
|
||
score += 20
|
||
elif policy_type == "产品安全与召回":
|
||
score += 10
|
||
return score
|
||
def build_policies(results: dict[str, FetchResult], now: date) -> list[dict[str, Any]]:
|
||
items: list[dict[str, Any]] = []
|
||
for source in SOURCES:
|
||
if source.section != "policy":
|
||
continue
|
||
result = results[source.id]
|
||
if not result.ok:
|
||
continue
|
||
if source.extractor == "cpsc":
|
||
recalls = parse_cpsc_recalls(source, result, now)
|
||
if recalls:
|
||
items.extend(recalls)
|
||
continue
|
||
item = base_item(source, result, "policy")
|
||
policy_type, markets, platforms, action = policy_profile(source)
|
||
try:
|
||
published_day = date.fromisoformat(str(result.published_at or "")[:10])
|
||
except ValueError:
|
||
published_day = date.min
|
||
verified_change = result.changed and published_day >= now - timedelta(days=45)
|
||
item.update({
|
||
"published_at": result.published_at,
|
||
"risk_level": source.risk_level,
|
||
"policy_type": policy_type,
|
||
"scope_type": "platform" if platforms else ("ip" if "知识产权" in policy_type else "region"),
|
||
"affected_markets": markets,
|
||
"platforms": platforms,
|
||
"change_status": "检测到近期更新" if verified_change else "官方规则持续监测",
|
||
"change_summary": source.summary_zh or result.description,
|
||
"consequence": policy_consequence(policy_type),
|
||
"action": action,
|
||
})
|
||
item["action_priority"] = policy_action_priority(
|
||
policy_type, str(item["change_status"]), str(item.get("risk_level") or "")
|
||
)
|
||
item.update(translation_fields(
|
||
str(item.get("url") or source.url),
|
||
str(result.page_language or ""),
|
||
f"{result.title} {result.description}",
|
||
))
|
||
items.append(item)
|
||
return sorted(
|
||
items,
|
||
key=lambda row: (
|
||
-int(row.get("action_priority") or 0),
|
||
str(row.get("published_at") or ""),
|
||
row["source"],
|
||
),
|
||
)
|
||
def refine_textile_translation(value: str) -> str:
|
||
replacements = {
|
||
"游戏庭院": "游戏围栏",
|
||
"游戏床床垫": "游戏围栏垫",
|
||
"夏尔巴人的毯子和毯子": "羊羔绒盖毯",
|
||
"电动毯子和毯子": "电热盖毯和毛毯",
|
||
"夏尔巴人、加热的毯子和毯子": "羊羔绒电热盖毯和毛毯",
|
||
"休闲毯和毯子": "盖毯和毛毯",
|
||
"亚马逊": "Amazon",
|
||
"在Amazon": "在 Amazon",
|
||
"漂亮的生活": "Pretty-Life",
|
||
}
|
||
output = value
|
||
for source, target in replacements.items():
|
||
output = output.replace(source, target)
|
||
return output
|
||
def translate_policy_titles(policies: list[dict[str, Any]], output_dir: Path, max_new: int = 72) -> None:
|
||
cache = load_zh_cache(output_dir)
|
||
session = make_session()
|
||
translated_count = 0
|
||
cache_changed = False
|
||
for item in policies:
|
||
if item.get("source_id") != "policy_cpsc":
|
||
continue
|
||
original = normalize_text(str(item.get("title_original") or ""))
|
||
if not original:
|
||
continue
|
||
translated = cache.get(original)
|
||
if not translated and translated_count < max_new:
|
||
translated = translate_to_zh_cn(session, original)
|
||
if translated:
|
||
cache[original] = translated
|
||
translated_count += 1
|
||
cache_changed = True
|
||
if translated:
|
||
item["title"] = f"美国消费品召回：{refine_textile_translation(translated)}"
|
||
item["translation_status"] = "已翻译并保留原文"
|
||
else:
|
||
item["translation_status"] = "保留官方原文"
|
||
summary_original = normalize_text(str(item.get("summary_original") or ""))
|
||
if not summary_original:
|
||
continue
|
||
summary_translated = cache.get(summary_original)
|
||
if not summary_translated and translated_count < max_new:
|
||
summary_translated = translate_to_zh_cn(session, summary_original)
|
||
if summary_translated:
|
||
cache[summary_original] = summary_translated
|
||
translated_count += 1
|
||
cache_changed = True
|
||
if summary_translated:
|
||
item["summary"] = refine_textile_translation(summary_translated)
|
||
if cache_changed:
|
||
save_zh_cache(output_dir, cache)
|
||
def translate_activity_titles(activities: list[dict[str, Any]], output_dir: Path, max_new: int = 72) -> None:
|
||
cache = load_zh_cache(output_dir)
|
||
session = make_session()
|
||
translated_count = 0
|
||
cache_changed = False
|
||
for item in activities:
|
||
original = normalize_text(str(item.get("title") or ""))
|
||
if not original or re.search(r"[\u4e00-\u9fff]", original):
|
||
continue
|
||
translated = cache.get(original)
|
||
if not translated and translated_count < max_new:
|
||
translated = translate_to_zh_cn(session, original)
|
||
if translated:
|
||
cache[original] = translated
|
||
translated_count += 1
|
||
cache_changed = True
|
||
if translated:
|
||
item["title_original"] = original
|
||
item["title"] = refine_textile_translation(translated)
|
||
item["translation_status"] = "已翻译并保留原文"
|
||
else:
|
||
item["translation_status"] = "保留官方原文"
|
||
if cache_changed:
|
||
save_zh_cache(output_dir, cache)
|
||
def policy_profile(source: SourceSpec) -> tuple[str, list[str], list[str], str]:
|
||
if source.policy_type:
|
||
return source.policy_type, list(source.markets or (source.region,)), list(source.platforms), source.action_zh
|
||
profiles = {
|
||
"policy_ustr": ("关税与贸易救济", ["美国", "中国出口"], [], "核对家纺税号与附加关税清单，重算落地成本、报价和供应链方案。"),
|
||
"policy_cbp": ("进口与清关", ["美国"], [], "核对申报、原产地、小额包裹、强迫劳动和承运商资料要求。"),
|
||
"policy_eu_customs": ("关税与清关", ["欧盟"], [], "按欧盟目的国复核 IOSS、低价值包裹、税费、商品编码和进口主体责任。"),
|
||
"policy_eu_textiles": ("纺织可持续合规", ["欧盟"], [], "评估生态设计、数字产品护照、生产者责任和绿色声明对产品资料的影响。"),
|
||
"policy_echa_chem": ("化学品与材料合规", ["欧盟"], [], "核查染料、涂层、阻燃和功能整理所涉物质的限制、授权与供应链文件。"),
|
||
"policy_ftc_textile": ("纺织标签", ["美国"], [], "逐项核对纤维成分、原产地、责任主体和广告披露，并留存标签审核版本。"),
|
||
"policy_uk_tariff": ("关税与商品编码", ["英国"], [], "查询商品编码、关税、增值税和进出口限制，确认进口主体及税费展示方式。"),
|
||
"policy_mofcom": ("外贸与跨境电商", ["中国"], [], "确认出口、跨境电商、市场准入和管制政策是否影响订单、结算与物流。"),
|
||
"policy_wipo_branddb": ("知识产权检索", ["全球"], [], "在命名、图案和包装定稿前检索近似商标与图形要素，并保存检索记录。"),
|
||
"policy_euipo_search": ("知识产权检索", ["欧盟"], [], "在欧盟上新前检索商标与外观设计，核对权利状态、类别和权利人。"),
|
||
}
|
||
return profiles.get(
|
||
source.id,
|
||
("综合合规", [source.region], [], "打开官方来源核查适用商品、地区、生效日期与申报要求，并由业务负责人确认执行。"),
|
||
)
|
||
def build_standards(results: dict[str, FetchResult]) -> list[dict[str, Any]]:
|
||
items: list[dict[str, Any]] = []
|
||
for source in SOURCES:
|
||
if source.section != "standard":
|
||
continue
|
||
result = results[source.id]
|
||
if not result.ok:
|
||
continue
|
||
item = base_item(source, result, "standard")
|
||
profile = standard_profile(source.id)
|
||
item.update({
|
||
"published_at": result.published_at,
|
||
"signal": "权威源已验证",
|
||
**profile,
|
||
})
|
||
items.append(item)
|
||
return items
|
||
def standard_profile(source_id: str) -> dict[str, Any]:
|
||
profiles = {
|
||
"standard_pantone": {
|
||
"trend_dimension": "色彩与设计",
|
||
"application": "将 Cloud Dancer 作为中性色方向，与目的国热销色、材质肌理和季节需求交叉验证后，再转成床品、窗帘和软装打样方案。",
|
||
"product_targets": ["床品配色", "窗帘色组", "软装组合", "包装视觉"],
|
||
"task_type": "生成配色打样单",
|
||
"task_output": "A4 HTML / 浏览器打印 PDF",
|
||
"decision_note": "年度代表色是设计信号，不等于所有市场都会热卖；先用小样和平台数据验证。",
|
||
"official_color": {
|
||
"name_zh": "云舞白（本站译名）",
|
||
"name_en": "Cloud Dancer",
|
||
"pantone_code": "PANTONE 11-4201",
|
||
"pantone_system": "具体色卡后缀以 Pantone 实体色卡和授权数据为准",
|
||
"hex": "#F2F0E8",
|
||
"rgb": "242, 240, 232",
|
||
"value_notice": "Pantone 名称与编号来自官方年度色页面；HEX 与 RGB 是本站屏幕近似值，不代替实体色卡、Lab 值或大货对色。",
|
||
},
|
||
"swatches": [
|
||
{"name_zh": "云舞白（本站译名）", "name_en": "Cloud Dancer", "hex": "#F2F0E8", "rgb": "242, 240, 232", "role": "60% 主色", "source_type": "官方英文名与编号 / 本站屏幕近似值"},
|
||
{"name_zh": "植物灰绿", "name_en": "Botanical Sage", "hex": "#738075", "rgb": "115, 128, 117", "role": "25% 辅色", "source_type": "本站设计建议"},
|
||
{"name_zh": "深海蓝", "name_en": "Deep Ocean Blue", "hex": "#25465A", "rgb": "37, 70, 90", "role": "10% 结构色", "source_type": "本站设计建议"},
|
||
{"name_zh": "珊瑚红", "name_en": "Coral Red", "hex": "#C85D4B", "rgb": "200, 93, 75", "role": "5% 点缀", "source_type": "本站设计建议"},
|
||
],
|
||
"sample_templates": [
|
||
{"product": "床品四件套", "material": "60支精梳棉或棉莱赛尔", "weight": "按面料规格核算，不以成品克重代替支数", "craft": "活性印染 / 纯色拼接", "pattern_scale": "主色大面积，结构色用于滚边或细线条", "market": "欧美中高客单与酒店风市场", "platform": "Amazon / 独立站", "price_band": "中高价带"},
|
||
{"product": "窗帘与靠垫", "material": "棉麻混纺或遮光复合面料", "weight": "窗帘建议先按遮光率与垂感选样", "craft": "提花 / 包边 / 同色异材质", "pattern_scale": "辅色用于中尺度植物纹，点缀色不超过 5%", "market": "欧洲与北美现代自然风", "platform": "Etsy / Amazon / Wayfair", "price_band": "中价至中高价"},
|
||
{"product": "盖毯与软装", "material": "棉、再生纤维或羊羔绒混合方案", "weight": "按季节分别打轻量与保暖样", "craft": "绗缝 / 流苏 / 纹理织造", "pattern_scale": "保持低饱和，点缀色用于小面积标签和包边", "market": "秋冬居家与礼赠场景", "platform": "Walmart / Amazon / TikTok Shop", "price_band": "大众至中价"},
|
||
],
|
||
"brief_items": ["目标国家与季节", "主色/辅色/点缀色比例", "适用材质与克重", "床品/窗帘/包装三套效果", "平台主图 A/B 测试计划"],
|
||
"measurement_note": "HEX 仅为屏幕打样参考，不替代 Pantone 实物色卡、供应商 Lab 值或大货对色。",
|
||
},
|
||
"standard_textile_exchange": {
|
||
"trend_dimension": "材料与可持续",
|
||
"application": "筛选再生纤维、优选材料和可追溯供应链方案，要求供应商同时提供材料声明、认证范围、批次证据与环境主张依据。",
|
||
"product_targets": ["纱线与面料", "填充材料", "再生纤维", "供应链证据"],
|
||
"task_type": "下载材料证据清单",
|
||
"decision_note": "可持续卖点必须能追溯到具体工厂、批次、产品范围和声明依据。",
|
||
"brief_items": ["材料名称与供应商", "认证机构与证书编号", "证书覆盖工厂/产品/有效期", "批次与交易凭证", "再生成分比例", "宣传文案证据来源"],
|
||
},
|
||
"standard_oeko_tex": {
|
||
"trend_dimension": "认证与限值",
|
||
"application": "按目标市场和产品接触等级核对面料、染整、辅料和成品的有害物质限值，确认认证证书覆盖具体工厂、产品类别和有效期。",
|
||
"product_targets": ["婴童床品", "贴身纺织", "染整助剂", "辅料与成品"],
|
||
"task_type": "下载认证核对单",
|
||
"decision_note": "证书存在不代表你的 SKU 自动被覆盖，必须核对证书主体、范围、有效期和产品等级。",
|
||
"brief_items": ["证书编号与查询结果", "持证主体与实际工厂", "产品级别与接触类型", "覆盖材料/颜色/辅料", "有效期", "大货批次与检测报告关联"],
|
||
},
|
||
"standard_aatcc": {
|
||
"trend_dimension": "性能与测试",
|
||
"application": "把色牢度、耐磨、起球、尺寸稳定、吸湿快干与凉感等指标转成采购技术规格，并在大货前完成同批次实验室验证。",
|
||
"product_targets": ["接触凉感", "色牢度", "耐磨起球", "尺寸稳定"],
|
||
"task_type": "下载性能测试单",
|
||
"decision_note": "Q-MAX 等指标只有和测试方法、调湿条件、单位及合格阈值一起写明才有采购意义。",
|
||
"brief_items": ["产品与使用场景", "测试标准/方法版本", "试样调湿与测试条件", "单位与目标值", "洗前/洗后要求", "抽样规则", "第三方实验室与报告编号"],
|
||
},
|
||
}
|
||
return profiles.get(source_id, {
|
||
"trend_dimension": "综合趋势",
|
||
"application": "与家纺图文趋势交叉验证后用于产品决策。",
|
||
"product_targets": ["床品", "窗帘", "软装"],
|
||
"task_type": "生成产品任务",
|
||
"decision_note": "先核对来源与适用范围，再用于打样或采购。",
|
||
"brief_items": ["目标市场", "适用品类", "证据来源", "打样或采购动作"],
|
||
})
|
||
def public_source_status(result: FetchResult) -> dict[str, Any]:
|
||
payload = asdict(result.source)
|
||
payload.update({
|
||
"ok": result.ok,
|
||
"status_code": result.status_code,
|
||
"final_url": result.final_url,
|
||
"content_type": result.content_type,
|
||
"elapsed_ms": result.elapsed_ms,
|
||
"size_bytes": result.size_bytes,
|
||
"content_hash": result.content_hash,
|
||
"hash_version": 3,
|
||
"changed": result.changed,
|
||
"checked_at": result.checked_at,
|
||
"page_title": result.title,
|
||
"page_description": result.description,
|
||
"page_language": result.page_language,
|
||
"image": result.image,
|
||
"evidence": result.evidence or [],
|
||
"error": result.error,
|
||
})
|
||
return payload
|
||
def write_json(path: Path, payload: Any) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
temp = path.with_suffix(path.suffix + ".tmp")
|
||
temp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8", newline="\n")
|
||
temp.replace(path)
|
||
def load_previous_hashes(path: Path) -> dict[str, str]:
|
||
if not path.exists():
|
||
return {}
|
||
try:
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
except (OSError, json.JSONDecodeError):
|
||
return {}
|
||
return {
|
||
str(row.get("id")): str(row.get("content_hash"))
|
||
for row in payload.get("sources", [])
|
||
if isinstance(row, dict) and row.get("id") and row.get("content_hash") and row.get("hash_version") == 3
|
||
}
|
||
def existing_trend_source_count(output_dir: Path) -> int:
|
||
path = output_dir / "textile-trends.json"
|
||
if not path.exists():
|
||
return 0
|
||
try:
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
return max(0, int(payload.get("source_count") or 0))
|
||
except (OSError, ValueError, TypeError, json.JSONDecodeError):
|
||
return 0
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="更新家纺跨境情报数据")
|
||
parser.add_argument("--output-dir", default="data", help="JSON 输出目录")
|
||
parser.add_argument("--now", default="", help="验收基准日期 YYYY-MM-DD，默认使用北京时间日期")
|
||
parser.add_argument("--horizon", default="2028-12-31", help="日历最远展示日期")
|
||
parser.add_argument("--max-workers", type=int, default=8, help="源站并发数")
|
||
parser.add_argument("--strict", action="store_true", help="任一权威源失败即退出，不覆盖正式数据")
|
||
args = parser.parse_args()
|
||
validate_registry()
|
||
now = date.fromisoformat(args.now) if args.now else datetime.now(tz=ZoneInfo("Asia/Shanghai")).date()
|
||
horizon = date.fromisoformat(args.horizon)
|
||
if horizon < now:
|
||
raise ValueError("horizon 不能早于 now")
|
||
output_dir = Path(args.output_dir)
|
||
previous_payload: dict[str, Any] = {}
|
||
previous_path = output_dir / "textile-intelligence.json"
|
||
try:
|
||
previous_payload = json.loads(previous_path.read_text(encoding="utf-8"))
|
||
except (OSError, ValueError, TypeError, json.JSONDecodeError):
|
||
previous_payload = {}
|
||
trend_source_count = existing_trend_source_count(output_dir)
|
||
previous_hashes = load_previous_hashes(output_dir / "textile-intelligence-source-status.json")
|
||
results = fetch_all_sources(max(1, min(args.max_workers, 12)))
|
||
for source_id, result in results.items():
|
||
previous_hash = previous_hashes.get(source_id)
|
||
result.changed = bool(previous_hash and result.content_hash and previous_hash != result.content_hash)
|
||
success_count = sum(1 for result in results.values() if result.ok)
|
||
failed = [result for result in results.values() if not result.ok]
|
||
print(f"权威源验证：{success_count}/{len(SOURCES)} 通过")
|
||
status_payload = {
|
||
"version": 4,
|
||
"generated_at": iso_now(),
|
||
"target_sources": len(SOURCES),
|
||
"success_sources": success_count,
|
||
"failed_sources": len(failed),
|
||
"existing_trend_sources": trend_source_count,
|
||
"total_source_baseline": trend_source_count + success_count,
|
||
"sources": [public_source_status(results[source.id]) for source in SOURCES],
|
||
}
|
||
write_json(output_dir / "textile-intelligence-source-status.json", status_payload)
|
||
if failed and args.strict:
|
||
for result in failed:
|
||
print(f"[严格模式失败] {result.source.name}: {result.error}")
|
||
return 1
|
||
valid_results = {source_id: result for source_id, result in results.items() if result.ok}
|
||
required_calendar = (
|
||
"calendar_nager", "calendar_uk", "calendar_uk_school_terms", "calendar_de_school_holidays",
|
||
"calendar_aladhan", "calendar_nrf",
|
||
"calendar_tmall", "calendar_nfl", "calendar_la28",
|
||
)
|
||
if any(source_id not in valid_results for source_id in required_calendar):
|
||
print("关键日历源未全部通过，停止生成正式数据")
|
||
return 1
|
||
activities, activity_sources = build_activities(now, iso_now(), make_session())
|
||
activities = merge_previous_activities(activities, list(previous_payload.get("activities") or []), now)
|
||
translate_activity_titles(activities, output_dir)
|
||
activity_success = sum(bool(source.get("ok")) for source in activity_sources)
|
||
status_payload.update({
|
||
"activity_target_sources": len(activity_sources),
|
||
"activity_success_sources": activity_success,
|
||
"activity_failed_sources": len(activity_sources) - activity_success,
|
||
"activity_sources": activity_sources,
|
||
})
|
||
write_json(output_dir / "textile-intelligence-source-status.json", status_payload)
|
||
if args.strict and activity_success != len(activity_sources):
|
||
for source in activity_sources:
|
||
if not source.get("ok"):
|
||
print(f"[活动源严格模式失败] {source['name']}: {source.get('error')}")
|
||
return 1
|
||
calendar, raw_calendar_count = build_calendar(now, horizon, valid_results)
|
||
translate_calendar_titles(calendar, output_dir)
|
||
calendar_insights = aggregate_calendar_insights(calendar, now, poster_for)
|
||
exhibitions = build_exhibitions(now, horizon, SOURCES, valid_results, base_item, make_id, poster_for)
|
||
translate_exhibitions(exhibitions, output_dir)
|
||
policies = build_policies(valid_results, now)
|
||
translate_policy_titles(policies, output_dir)
|
||
standards = build_standards(valid_results)
|
||
for item in standards:
|
||
source_result = valid_results.get(str(item.get("source_id") or ""))
|
||
if source_result:
|
||
item.update(translation_fields(
|
||
str(item.get("url") or item.get("source_url") or ""),
|
||
str(source_result.page_language or ""),
|
||
f"{source_result.title} {source_result.description}",
|
||
))
|
||
platforms = build_platforms(iso_now(), make_session())
|
||
trends = load_trends(output_dir / "textile-trends.json")
|
||
today = build_today(now, calendar_insights, exhibitions, policies, standards, activities, trends)
|
||
payload = {
|
||
"version": 2,
|
||
"generated_at": iso_now(),
|
||
"as_of_date": now.isoformat(),
|
||
"horizon_date": horizon.isoformat(),
|
||
"source_target": len(SOURCES),
|
||
"source_success": success_count,
|
||
"existing_trend_sources": trend_source_count,
|
||
"total_source_baseline": trend_source_count + success_count,
|
||
"summary": {
|
||
"calendar_events": len(calendar),
|
||
"calendar_insights": len(calendar_insights),
|
||
"calendar_featured": sum(
|
||
item.get("business_relevance") in ("high", "medium")
|
||
and item.get("affected_countries")
|
||
and item.get("platforms")
|
||
and item.get("product_suggestions")
|
||
and item.get("stocking_start")
|
||
and date_value(item.get("stocking_end") or item["start_date"]) >= now
|
||
for item in calendar_insights
|
||
),
|
||
"calendar_raw_events": raw_calendar_count,
|
||
"calendar_duplicates_removed": raw_calendar_count - len(calendar),
|
||
"calendar_markets": len({item.get("country_code") or item.get("market_group") for item in calendar}),
|
||
"calendar_years": sorted({date_value(item["start_date"]).year for item in calendar}),
|
||
"calendar_event_types": len({item.get("event_type") for item in calendar}),
|
||
"exhibitions": len(exhibitions),
|
||
"policies": len(policies),
|
||
"standards": len(standards),
|
||
"platforms": len(platforms),
|
||
"activities": len(activities),
|
||
"activity_sources": activity_success,
|
||
"today_items": len(today),
|
||
},
|
||
"today": today,
|
||
"calendar": calendar,
|
||
"calendar_insights": calendar_insights,
|
||
"exhibitions": exhibitions,
|
||
"policies": policies,
|
||
"standards": standards,
|
||
"platforms": platforms,
|
||
"activities": activities,
|
||
"activity_sources": activity_sources,
|
||
"sources": [public_source_status(results[source.id]) for source in SOURCES if results[source.id].ok],
|
||
}
|
||
validate_payload(payload)
|
||
write_json(output_dir / "textile-intelligence.json", payload)
|
||
print(
|
||
"情报数据生成完成："
|
||
f"日历原始节点 {len(calendar)} / 聚合节点 {len(calendar_insights)} / 精准机会 {payload['summary']['calendar_featured']} / "
|
||
f"展会 {len(exhibitions)} / 活动 {len(activities)} / 政策 {len(policies)} / 标准 {len(standards)}"
|
||
)
|
||
return 0
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|