Sansheng-Intelligence/scripts/update_textile_intelligence.py

#!/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())