Jack
f1ce583fd1
同步家纺情报 v2 数据、59 个严格权威源、趋势图库、移动端交互、Google/Yandex 翻译门控及静态 SEO/GEO 构建。 补充面向 C# 研发的 README，明确当前无 LLM、多模态与生图模型，记录翻译请求参数、缓存与失败策略、数据处理边界、前端职责和无 GitHub Pages 部署流程。 新增按日期倒序维护的 CHANGELOG，记录影响、验证与代码/数据/dist 回滚方法。 验证：11 个 Python 契约测试、13 个 Node 前端测试通过；静态构建生成 1096 个 Sitemap URL。
629 lines
27 KiB
Python
629 lines
27 KiB
Python
#!/usr/bin/env python3
|
||
"""抓取与家纺出海直接相关或跨境经营必须关注的活动。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import re
|
||
from datetime import date, datetime
|
||
from typing import Any
|
||
from urllib.parse import urljoin
|
||
|
||
import requests
|
||
from bs4 import BeautifulSoup
|
||
|
||
try:
|
||
from .intelligence_web_translation import translation_fields
|
||
except ImportError:
|
||
from intelligence_web_translation import translation_fields
|
||
|
||
|
||
ACTIVITY_SOURCES: tuple[dict[str, str], ...] = (
|
||
{
|
||
"id": "activity_cifnews",
|
||
"name": "雨果跨境活动",
|
||
"url": "https://www.cifnews.com/activity",
|
||
"authority": "行业媒体发现",
|
||
"level": "行业媒体",
|
||
},
|
||
{
|
||
"id": "activity_amazon",
|
||
"name": "Amazon 全球开店官方直播",
|
||
"url": "https://gs.amazon.cn/resources/live-broadcast",
|
||
"authority": "平台官方",
|
||
"level": "官方",
|
||
},
|
||
{
|
||
"id": "activity_wipo",
|
||
"name": "WIPO 官方 Webinar",
|
||
"url": "https://www.wipo.int/meetings/en/topic.jsp?group_id=7",
|
||
"authority": "国际组织官方",
|
||
"level": "官方",
|
||
},
|
||
{
|
||
"id": "activity_cbp",
|
||
"name": "美国海关 CBP Trade Webinar",
|
||
"url": "https://www.cbp.gov/trade/stakeholder-engagement/webinars",
|
||
"authority": "美国政府官方",
|
||
"level": "官方",
|
||
},
|
||
{
|
||
"id": "activity_ccpit",
|
||
"name": "中国贸促会经贸活动云",
|
||
"url": "https://bizevent.ccpit.org/",
|
||
"authority": "中国贸促会官方",
|
||
"level": "政府/贸促机构",
|
||
},
|
||
{
|
||
"id": "activity_amazon_global",
|
||
"name": "Amazon 全球卖家活动",
|
||
"url": "https://sell.amazon.com/events",
|
||
"authority": "Amazon 平台官方",
|
||
"level": "官方",
|
||
},
|
||
{
|
||
"id": "activity_ebay",
|
||
"name": "eBay Seller Events",
|
||
"url": "https://community.ebay.com/ebay-seller-events/",
|
||
"authority": "eBay 平台官方",
|
||
"level": "官方",
|
||
},
|
||
{
|
||
"id": "activity_eu_ip",
|
||
"name": "欧盟知识产权帮助中心活动",
|
||
"url": "https://intellectual-property-helpdesk.ec.europa.eu/news-events/upcoming-trainings-events_en",
|
||
"authority": "欧盟委员会官方",
|
||
"level": "国际/区域官方",
|
||
},
|
||
{
|
||
"id": "activity_uspto",
|
||
"name": "USPTO 商标与品牌保护活动",
|
||
"url": "https://www.uspto.gov/about-us/events",
|
||
"authority": "美国专利商标局官方",
|
||
"level": "政府官方",
|
||
},
|
||
{
|
||
"id": "activity_eu_trade",
|
||
"name": "欧盟贸易政策活动",
|
||
"url": "https://policy.trade.ec.europa.eu/events_en",
|
||
"authority": "欧盟委员会贸易总司官方",
|
||
"level": "国际/区域官方",
|
||
},
|
||
)
|
||
|
||
TEXTILE_TERMS = re.compile(
|
||
r"家纺|家居|床垫|床品|床单|寝具|被子|被套|棉被|枕|靠垫|窗帘|毛巾|纺织|面料|Wayfair|"
|
||
r"textile|bedding|mattress|cushion",
|
||
re.I,
|
||
)
|
||
HOME_TEXTILE_TERMS = re.compile(
|
||
r"家纺|家居纺织|床垫|床品|床单|寝具|被子|被套|棉被|枕头|枕芯|靠垫|窗帘|毛巾|浴巾|"
|
||
r"home textile|bedding|mattress|pillow|cushion|duvet|bed linen|curtain|towel",
|
||
re.I,
|
||
)
|
||
APPAREL_ONLY_TERMS = re.compile(r"女装|男装|西装|礼服|鞋服|美妆|women'?s suit|menswear|womenswear|apparel only", re.I)
|
||
COMPLIANCE_TERMS = re.compile(
|
||
r"合规|法规|税则|海关|关税|商标|知识产权|品牌保护|侵权|认证|EPR|VAT|GPSR|EAPA|INFORM|"
|
||
r"trademark|Madrid System|classification|heading\s*\d|trade violations?",
|
||
re.I,
|
||
)
|
||
LOGISTICS_TERMS = re.compile(r"物流|仓储|FBA|配送|退货|海外仓|供应链|履约", re.I)
|
||
PLATFORM_ACTION_TERMS = re.compile(
|
||
r"入驻|招商|平台规则|账户|账号|认证|INFORM|合规|费用|佣金|物流|仓储|FBA|退货|履约|关税|税务|召回|"
|
||
r"onboarding|seller verification|account|compliance|fee|fulfillment|warehouse|returns?|tariff|recall",
|
||
re.I,
|
||
)
|
||
PLATFORM_TERMS: tuple[tuple[str, str], ...] = (
|
||
("Amazon", r"Amazon|亚马逊"),
|
||
("Walmart", r"Walmart|沃尔玛"),
|
||
("TikTok Shop", r"TikTok|TK\b"),
|
||
("Temu", r"Temu"),
|
||
("Shopee", r"Shopee"),
|
||
("Lazada", r"Lazada"),
|
||
("Mercado Libre", r"Mercado Libre|美客多"),
|
||
("Ozon", r"Ozon"),
|
||
("eBay", r"eBay"),
|
||
("Noon", r"Noon"),
|
||
("Wayfair", r"Wayfair"),
|
||
("Rakuten", r"Rakuten|乐天"),
|
||
("Coupang", r"Coupang"),
|
||
("Jumia", r"Jumia"),
|
||
)
|
||
MONTHS = {name: index for index, name in enumerate(
|
||
("January", "February", "March", "April", "May", "June", "July", "August", "September", "October", "November", "December"),
|
||
start=1,
|
||
)}
|
||
MONTHS.update({name[:3]: index for name, index in MONTHS.copy().items()})
|
||
|
||
|
||
def _text(value: str) -> str:
|
||
return " ".join((value or "").split())
|
||
|
||
|
||
def _id(source_id: str, start_date: str, url: str) -> str:
|
||
digest = hashlib.sha256(f"{source_id}|{start_date}|{url}".encode("utf-8")).hexdigest()[:18]
|
||
return f"activity-{digest}"
|
||
|
||
|
||
def _platforms(title: str) -> list[str]:
|
||
return [name for name, pattern in PLATFORM_TERMS if re.search(pattern, title, re.I)]
|
||
|
||
|
||
def _region(title: str, location: str, platforms: list[str]) -> str:
|
||
value = f"{title} {location}"
|
||
if re.search(r"欧洲|英国|德国|法国|EU\b|Europe", value, re.I):
|
||
return "欧洲"
|
||
if re.search(r"美国|加拿大|北美|US\b|USA|Canada", value, re.I):
|
||
return "北美"
|
||
if re.search(r"中东|沙特|阿联酋|Noon", value, re.I):
|
||
return "中东"
|
||
if re.search(r"日本|韩国|Rakuten|Coupang", value, re.I):
|
||
return "东亚"
|
||
if re.search(r"东南亚|Shopee|Lazada|TikTok", value, re.I):
|
||
return "东南亚"
|
||
if re.search(r"拉美|巴西|墨西哥|美客多|Mercado", value, re.I):
|
||
return "拉丁美洲"
|
||
if location and location not in ("线上", "Virtual"):
|
||
return "中国"
|
||
return "全球"
|
||
|
||
|
||
def _profile(title: str, source_id: str) -> tuple[str, str, str]:
|
||
if HOME_TEXTILE_TERMS.search(title) and not APPAREL_ONLY_TERMS.search(title):
|
||
return (
|
||
"家纺/纺织直接相关",
|
||
"direct",
|
||
"与家纺、纺织品或家居品类直接相关，优先核对产品规格、税则归类、平台机会与报名截止时间。",
|
||
)
|
||
if COMPLIANCE_TERMS.search(title):
|
||
return (
|
||
"跨境合规必看",
|
||
"must",
|
||
"涉及关税、海关、知识产权或平台法规，可能直接影响商品上架、清关、账户与品牌安全。",
|
||
)
|
||
if LOGISTICS_TERMS.search(title):
|
||
return (
|
||
"物流履约必看",
|
||
"must",
|
||
"涉及入仓、配送、退货或供应链成本，适合运营、跟单和物流负责人参加。",
|
||
)
|
||
if _platforms(title) and PLATFORM_ACTION_TERMS.search(title):
|
||
return (
|
||
"平台招商与运营",
|
||
"must",
|
||
"由平台或行业机构发布，适合核对最新招商条件、站点机会、旺季节奏和运营规则。",
|
||
)
|
||
if source_id in {"activity_ccpit", "activity_eu_trade"} and re.search(r"出海|跨境|贸易|关税|海关|Trade Policy|Export Control|digital trade|supply chain", title, re.I):
|
||
return (
|
||
"跨境经营必看",
|
||
"must",
|
||
"由政府或区域官方发布，适合核对贸易政策、出海服务、关税、供应链与市场准入变化。",
|
||
)
|
||
return ("经营增长", "observe", "属于一般跨境经营活动，家纺团队可按目标市场和当前问题决定是否参加。")
|
||
|
||
|
||
def _localized_title(title: str) -> str:
|
||
exact = {
|
||
"Mattresses and Cushions: Hybrids and More of Heading 9404": "美国海关：床垫与靠垫混合产品（税则号 9404）归类讲座",
|
||
"Wadding, Gauze, Bandages, and Similar Articles": "美国海关：絮胎、纱布、绷带及类似制品归类讲座",
|
||
"Enforce and Protect Act (EAPA) Q3": "美国海关：EAPA 反逃避执法与举报机制讲座",
|
||
"Madrid System Webinar: Managing International Trademark Registration – Subsequent Designation and Renewal (in Chinese)": "WIPO 中文讲座：国际商标注册的后期指定与续展管理",
|
||
}
|
||
return exact.get(title, title)
|
||
|
||
|
||
def _item(
|
||
*, source: dict[str, str], title: str, start_date: date, url: str, checked_at: str,
|
||
location: str = "线上", image: str = "", source_language: str = "zh-CN", profile_text: str = "",
|
||
) -> dict[str, Any] | None:
|
||
title = _text(title)
|
||
category, relevance, summary = _profile(_text(profile_text) or title, source["id"])
|
||
if relevance == "observe":
|
||
return None
|
||
platforms = _platforms(title)
|
||
region = _region(title, location, platforms)
|
||
display_title = _localized_title(title)
|
||
item = {
|
||
"id": _id(source["id"], start_date.isoformat(), url),
|
||
"section": "activity",
|
||
"title": display_title,
|
||
"title_original": title if display_title != title else "",
|
||
"summary": summary,
|
||
"start_date": start_date.isoformat(),
|
||
"end_date": start_date.isoformat(),
|
||
"days_until": 0,
|
||
"location": location or "线上",
|
||
"venue": location or "线上",
|
||
"format": "线上" if location in ("线上", "Virtual") else "线下",
|
||
"region": region,
|
||
"platforms": platforms,
|
||
"activity_type": category,
|
||
"relevance": relevance,
|
||
"source": source["name"],
|
||
"source_id": source["id"],
|
||
"source_url": url,
|
||
"url": url,
|
||
"registration_url": url,
|
||
"authority": source["authority"],
|
||
"source_level": source["level"],
|
||
"confidence": "中" if source["level"] == "行业媒体" else "高",
|
||
"official_activity": source["level"] != "行业媒体",
|
||
"image": image,
|
||
"verified_at": checked_at,
|
||
"checked_at": checked_at,
|
||
"next_action": "先核对议程、适用站点和报名截止时间，再决定由老板、运营、合规或跟单人员参加。",
|
||
"poster": {"eyebrow": "跨境活动", "accent": "#2f7d6d", "ink": "#10251f", "paper": "#eef8f4"},
|
||
}
|
||
translation = translation_fields(url, source_language, title)
|
||
item.update(translation)
|
||
item.update({
|
||
"source_needs_translation": translation["needs_translation"],
|
||
"source_translation_url": translation["translation_url"],
|
||
"source_translation_fallback_url": translation["translation_fallback_url"],
|
||
})
|
||
return item
|
||
|
||
|
||
def _parse_cifnews(html: str, source: dict[str, str], now: date, checked_at: str) -> list[dict[str, Any]]:
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
rows: list[dict[str, Any]] = []
|
||
for card in soup.select(".cif-flexbox__wrap"):
|
||
title_node = card.select_one(".title-text")
|
||
image_node = card.select_one(".cif-flexbox__imgbox img")
|
||
title = _text(title_node.get_text(" ", strip=True) if title_node else "")
|
||
image = str(image_node.get("src") or "") if image_node else ""
|
||
date_node = card.select_one(".desc.time")
|
||
if date_node:
|
||
match = re.search(r"(20\d{2})[.-](\d{1,2})[.-](\d{1,2})", date_node.get_text(" ", strip=True))
|
||
link = card.select_one(".signup-active-btn[href]") or card.select_one("a.title[href]")
|
||
if not match or not link:
|
||
continue
|
||
start = date(int(match.group(1)), int(match.group(2)), int(match.group(3)))
|
||
if start < now:
|
||
continue
|
||
location_node = card.select_one(".bottom .desc")
|
||
item = _item(
|
||
source=source, title=title, start_date=start,
|
||
url=urljoin(source["url"], str(link.get("href") or "")), checked_at=checked_at,
|
||
location=_text(location_node.get_text(" ", strip=True) if location_node else "线下"),
|
||
image=image,
|
||
)
|
||
if item:
|
||
rows.append(item)
|
||
continue
|
||
for event in card.select(".series-item.is-active[href]"):
|
||
match = re.search(r"(\d{1,2})[.-](\d{1,2})", _text(event.get_text(" ", strip=True)))
|
||
if not match:
|
||
continue
|
||
start = date(now.year, int(match.group(1)), int(match.group(2)))
|
||
if start < now:
|
||
continue
|
||
city_node = event.select_one(".series-item-city")
|
||
city = _text(city_node.get_text(" ", strip=True) if city_node else "线下")
|
||
item = _item(
|
||
source=source, title=f"{title} · {city}", start_date=start,
|
||
url=urljoin(source["url"], str(event.get("href") or "")), checked_at=checked_at,
|
||
location=city, image=image,
|
||
)
|
||
if item:
|
||
rows.append(item)
|
||
return rows
|
||
|
||
|
||
def _parse_amazon(html: str, source: dict[str, str], now: date, checked_at: str) -> list[dict[str, Any]]:
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
rows: list[dict[str, Any]] = []
|
||
for card in soup.select(".tile-wrap"):
|
||
value = _text(card.get_text(" ", strip=True))
|
||
match = re.search(r"(20\d{2}-\d{2}-\d{2})", value)
|
||
link = card.select_one("a[href]")
|
||
if not match or not link:
|
||
continue
|
||
start = date.fromisoformat(match.group(1))
|
||
if start < now:
|
||
continue
|
||
title = value[:match.start()].strip(" -|：")
|
||
item = _item(
|
||
source=source, title=title, start_date=start,
|
||
url=urljoin(source["url"], str(link.get("href") or "")), checked_at=checked_at,
|
||
location="线上", source_language="zh-CN",
|
||
)
|
||
if item:
|
||
item["platforms"] = ["Amazon"]
|
||
rows.append(item)
|
||
return rows
|
||
|
||
|
||
def _english_date(value: str) -> date | None:
|
||
match = re.search(
|
||
r"(January|February|March|April|May|June|July|August|September|October|November|December)\s+(\d{1,2}),\s+(20\d{2})",
|
||
value,
|
||
)
|
||
if not match:
|
||
return None
|
||
return date(int(match.group(3)), MONTHS[match.group(1)], int(match.group(2)))
|
||
|
||
|
||
def _parse_wipo(html: str, source: dict[str, str], now: date, checked_at: str) -> list[dict[str, Any]]:
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
rows: list[dict[str, Any]] = []
|
||
for card in soup.select("article.listing--event"):
|
||
link = card.select_one("h3 a[href]")
|
||
if not link:
|
||
continue
|
||
title = _text(link.get_text(" ", strip=True))
|
||
if not re.search(r"Madrid System|Trademark|Industrial Design|Hague System|Goods and Services", title, re.I):
|
||
continue
|
||
start = _english_date(_text(card.get_text(" ", strip=True)))
|
||
if not start or start < now:
|
||
continue
|
||
item = _item(
|
||
source=source, title=title, start_date=start,
|
||
url=urljoin(source["url"], str(link.get("href") or "")), checked_at=checked_at,
|
||
location="线上", source_language="en",
|
||
)
|
||
if item:
|
||
rows.append(item)
|
||
return rows
|
||
|
||
|
||
def _parse_cbp(html: str, source: dict[str, str], now: date, checked_at: str) -> list[dict[str, Any]]:
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
rows: list[dict[str, Any]] = []
|
||
for card in soup.select("li.usa-icon-list__item"):
|
||
link = card.select_one("a[href]")
|
||
if not link:
|
||
continue
|
||
title = _text(link.get_text(" ", strip=True))
|
||
if not (TEXTILE_TERMS.search(title) or re.search(r"EAPA|Trade Violations?", title, re.I)):
|
||
continue
|
||
start = _english_date(_text(card.get_text(" ", strip=True)))
|
||
if not start or start < now:
|
||
continue
|
||
item = _item(
|
||
source=source, title=title, start_date=start,
|
||
url=urljoin(source["url"], str(link.get("href") or "")), checked_at=checked_at,
|
||
location="线上", source_language="en",
|
||
)
|
||
if item:
|
||
rows.append(item)
|
||
return rows
|
||
|
||
|
||
def _parse_ccpit(html: str, source: dict[str, str], now: date, checked_at: str) -> list[dict[str, Any]]:
|
||
match = re.search(r"var\s+meetingList\s*=\s*(\{[\s\S]*?\});", html)
|
||
if not match:
|
||
return []
|
||
payload = json.loads(match.group(1))
|
||
rows: list[dict[str, Any]] = []
|
||
for meeting in payload.get("content") or []:
|
||
raw_title = _text(str(meeting.get("title") or ""))
|
||
profile_text = _text(" ".join(str(meeting.get(key) or "") for key in ("title", "briefIntroduction", "agenda", "targetWebsite")))
|
||
if not re.search(r"出海|跨境|外贸|数字贸易|知识产权|商标|海关|关税|纺织|家纺|服装|平台", profile_text, re.I):
|
||
continue
|
||
start_value = str(meeting.get("startDate") or "")[:10]
|
||
if not re.match(r"20\d{2}-\d{2}-\d{2}", start_value):
|
||
continue
|
||
start = date.fromisoformat(start_value)
|
||
if start < now or int(meeting.get("status") or 0) not in (2, 3):
|
||
continue
|
||
location = "线上" if meeting.get("onLineMeeting") else _text(" ".join(str(meeting.get(key) or "") for key in ("addressProvince", "addressCity", "addressDetail"))).strip() or "中国"
|
||
item = _item(
|
||
source=source, title=raw_title, start_date=start,
|
||
url=urljoin(source["url"], f"/meetings/{meeting.get('id')}"), checked_at=checked_at,
|
||
location=location, source_language="zh-CN", profile_text=profile_text,
|
||
)
|
||
if item:
|
||
item["end_date"] = str(meeting.get("endDate") or start_value)[:10]
|
||
item["registration_deadline"] = str((meeting.get("signSetting") or {}).get("signEndDate") or "")[:10]
|
||
item["free"] = bool(meeting.get("free"))
|
||
rows.append(item)
|
||
return rows
|
||
|
||
|
||
def _parse_amazon_global(html: str, source: dict[str, str], now: date, checked_at: str) -> list[dict[str, Any]]:
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
rows: list[dict[str, Any]] = []
|
||
for card in soup.select(".background-color-white.has-box-shadow"):
|
||
value = _text(card.get_text(" ", strip=True))
|
||
date_match = re.search(r"(?:When:\s*)?([A-Z][a-z]+\s+\d{1,2}(?:-\d{1,2})?,\s+20\d{2})", value)
|
||
if not date_match:
|
||
continue
|
||
parts = date_match.group(1).replace("–", "-").split(",")
|
||
day_match = re.search(r"([A-Z][a-z]+)\s+(\d{1,2})(?:-\d{1,2})?", parts[0])
|
||
if not day_match:
|
||
continue
|
||
start = date(int(parts[1].strip()), MONTHS[day_match.group(1)], int(day_match.group(2)))
|
||
if start < now:
|
||
continue
|
||
title_node = card.select_one("h2.heading.font-size-xlarge")
|
||
title = _text(title_node.get_text(" ", strip=True) if title_node else value[:date_match.start()])
|
||
link = next((node for node in card.select("a[href]") if "register" in _text(node.get_text(" ", strip=True)).lower() or "注册" in node.get_text()), card.select_one("a[href]"))
|
||
if not link:
|
||
continue
|
||
item = _item(source=source, title=title, start_date=start, url=urljoin(source["url"], str(link.get("href") or "")), checked_at=checked_at, location="线上", source_language="zh-CN", profile_text=value)
|
||
if item:
|
||
item["platforms"] = ["Amazon"]
|
||
rows.append(item)
|
||
return rows
|
||
|
||
|
||
def _parse_ebay(html: str, source: dict[str, str], now: date, checked_at: str) -> list[dict[str, Any]]:
|
||
rows: list[dict[str, Any]] = []
|
||
seen: set[str] = set()
|
||
for script in BeautifulSoup(html, "html.parser").select("script"):
|
||
text = script.string or script.get_text()
|
||
if "start_date" not in text or "events/details" not in text:
|
||
continue
|
||
try:
|
||
payload = json.loads(text)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
stack: list[Any] = [payload]
|
||
while stack:
|
||
value = stack.pop()
|
||
if isinstance(value, dict):
|
||
if value.get("start_date") and value.get("title") and value.get("url") and "/events/details/" in str(value["url"]):
|
||
start = date.fromisoformat(str(value["start_date"])[:10])
|
||
if start >= now and str(value["url"]) not in seen:
|
||
seen.add(str(value["url"]))
|
||
item = _item(source=source, title=_text(str(value["title"])), start_date=start, url=str(value["url"]), checked_at=checked_at, location="线上", source_language="en", profile_text=str(value["title"]))
|
||
if item:
|
||
item["platforms"] = ["eBay"]
|
||
rows.append(item)
|
||
stack.extend(value.values())
|
||
elif isinstance(value, list):
|
||
stack.extend(value)
|
||
return rows
|
||
|
||
|
||
def _ecl_date(card: Any) -> date | None:
|
||
text = _text(card.get_text(" ", strip=True))
|
||
match = re.search(r"\b(\d{1,2})(?:-\d{1,2})?\s+([A-Za-z]+)\s+(20\d{2})\b", text)
|
||
if not match:
|
||
return None
|
||
month = MONTHS.get(match.group(2)) or MONTHS.get(match.group(2)[:3])
|
||
return date(int(match.group(3)), month, int(match.group(1))) if month else None
|
||
|
||
|
||
def _parse_eu_cards(html: str, source: dict[str, str], now: date, checked_at: str, *, trade: bool = False) -> list[dict[str, Any]]:
|
||
rows: list[dict[str, Any]] = []
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
for card in soup.select("article.ecl-content-item.ecl-content-item--inline"):
|
||
link = card.select_one("a[data-ecl-title-link][href]") or card.select_one("a[href]")
|
||
start = _ecl_date(card)
|
||
if not link or not start or start < now:
|
||
continue
|
||
title = _text(link.get_text(" ", strip=True))
|
||
profile_text = _text(card.get_text(" ", strip=True))
|
||
if trade and not re.search(r"custom|tariff|trade|export|digital|supply|textile|policy|关税|贸易|海关", profile_text, re.I):
|
||
continue
|
||
if not trade and not re.search(r"trademark|design|copyright|brand|IP|知识产权|商标|版权", profile_text, re.I):
|
||
continue
|
||
location = "线上" if re.search(r"Live streaming available|Webinar", profile_text, re.I) else "欧洲"
|
||
item = _item(source=source, title=title, start_date=start, url=urljoin(source["url"], str(link.get("href") or "")), checked_at=checked_at, location=location, source_language="en", profile_text=profile_text)
|
||
if item:
|
||
rows.append(item)
|
||
return rows
|
||
|
||
|
||
def _parse_eu_ip(html: str, source: dict[str, str], now: date, checked_at: str) -> list[dict[str, Any]]:
|
||
return _parse_eu_cards(html, source, now, checked_at, trade=False)
|
||
|
||
|
||
def _parse_eu_trade(html: str, source: dict[str, str], now: date, checked_at: str) -> list[dict[str, Any]]:
|
||
return _parse_eu_cards(html, source, now, checked_at, trade=True)
|
||
|
||
|
||
def _parse_uspto(html: str, source: dict[str, str], now: date, checked_at: str) -> list[dict[str, Any]]:
|
||
rows: list[dict[str, Any]] = []
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
for row in soup.select("table.views-table.views-view-table.cols-4 tbody tr[date]"):
|
||
raw_date = str(row.get("date") or "")
|
||
if not re.fullmatch(r"20\d{6}", raw_date):
|
||
continue
|
||
start = date(int(raw_date[:4]), int(raw_date[4:6]), int(raw_date[6:]))
|
||
if start < now:
|
||
continue
|
||
title_node = row.select_one("td.name a[href]")
|
||
if not title_node:
|
||
continue
|
||
title = _text(title_node.get_text(" ", strip=True))
|
||
profile_text = _text(row.get_text(" ", strip=True))
|
||
if re.search(r"CANCELED|POSTPONED", profile_text, re.I) or not re.search(r"trademark|brand|counterfeit|design|intellectual property|商标|品牌|假冒|外观", profile_text, re.I):
|
||
continue
|
||
location = _text((row.select_one("td.location") or row).get_text(" ", strip=True))
|
||
item = _item(source=source, title=title, start_date=start, url=urljoin(source["url"], str(title_node.get("href") or "")), checked_at=checked_at, location=location or "线上", source_language="en", profile_text=profile_text)
|
||
if item:
|
||
rows.append(item)
|
||
return rows
|
||
|
||
|
||
PARSERS = {
|
||
"activity_cifnews": _parse_cifnews,
|
||
"activity_amazon": _parse_amazon,
|
||
"activity_wipo": _parse_wipo,
|
||
"activity_cbp": _parse_cbp,
|
||
"activity_ccpit": _parse_ccpit,
|
||
"activity_amazon_global": _parse_amazon_global,
|
||
"activity_ebay": _parse_ebay,
|
||
"activity_eu_ip": _parse_eu_ip,
|
||
"activity_uspto": _parse_uspto,
|
||
"activity_eu_trade": _parse_eu_trade,
|
||
}
|
||
|
||
|
||
def build_activities(now: date, checked_at: str, session: requests.Session) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
||
"""返回未来活动与逐源抓取状态；活动源不冒充原 52 个权威资讯源。"""
|
||
activities: list[dict[str, Any]] = []
|
||
statuses: list[dict[str, Any]] = []
|
||
for source in ACTIVITY_SOURCES:
|
||
try:
|
||
response = session.get(source["url"], timeout=45, allow_redirects=True)
|
||
response.raise_for_status()
|
||
parsed = PARSERS[source["id"]](response.text, source, now, checked_at)
|
||
activities.extend(parsed)
|
||
statuses.append({
|
||
**source,
|
||
"ok": True,
|
||
"status_code": response.status_code,
|
||
"final_url": response.url,
|
||
"items": len(parsed),
|
||
"checked_at": checked_at,
|
||
"error": "",
|
||
})
|
||
except (requests.RequestException, ValueError) as error:
|
||
statuses.append({
|
||
**source,
|
||
"ok": False,
|
||
"status_code": 0,
|
||
"final_url": source["url"],
|
||
"items": 0,
|
||
"checked_at": checked_at,
|
||
"error": str(error),
|
||
})
|
||
unique: dict[tuple[str, str], dict[str, Any]] = {}
|
||
for item in activities:
|
||
item["days_until"] = (date.fromisoformat(item["start_date"]) - now).days
|
||
key = (str(item["start_date"]), re.sub(r"\W+", "", str(item["title"]).lower()))
|
||
existing = unique.get(key)
|
||
if not existing or (item["official_activity"] and not existing["official_activity"]):
|
||
unique[key] = item
|
||
ordered = sorted(
|
||
unique.values(),
|
||
key=lambda item: (item["start_date"], 0 if item["official_activity"] else 1, item["title"]),
|
||
)
|
||
return ordered, statuses
|
||
|
||
|
||
def merge_previous_activities(current: list[dict[str, Any]], previous: list[dict[str, Any]], now: date) -> list[dict[str, Any]]:
|
||
"""当前抓取优先，补回上一快照中仍未过期的活动，避免源站窗口滚动造成数据闪断。"""
|
||
merged: dict[tuple[str, str], dict[str, Any]] = {}
|
||
for item in current:
|
||
key = (str(item.get("start_date") or ""), re.sub(r"\W+", "", str(item.get("title") or "").lower()))
|
||
if key[0] and key[1]:
|
||
item["days_until"] = (date.fromisoformat(key[0]) - now).days
|
||
merged[key] = item
|
||
# ponytail: 仅按结束日期保留上一快照，活动撤回或改期需下一次抓取覆盖。
|
||
for item in previous:
|
||
try:
|
||
end_date = date.fromisoformat(str(item.get("end_date") or item.get("start_date") or "")[:10])
|
||
except ValueError:
|
||
continue
|
||
if end_date < now:
|
||
continue
|
||
key = (str(item.get("start_date") or ""), re.sub(r"\W+", "", str(item.get("title") or "").lower()))
|
||
if key[0] and key[1] and key not in merged:
|
||
item = dict(item)
|
||
item["days_until"] = (date.fromisoformat(key[0]) - now).days
|
||
merged[key] = item
|
||
return sorted(merged.values(), key=lambda item: (item["start_date"], 0 if item.get("official_activity") else 1, item.get("title") or ""))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
session = requests.Session()
|
||
session.headers["User-Agent"] = "Mozilla/5.0"
|
||
rows, source_statuses = build_activities(date.today(), datetime.now().isoformat(), session)
|
||
print(f"活动 {len(rows)} 条，来源 {sum(item['ok'] for item in source_statuses)}/{len(source_statuses)}")
|