# GDELT 双国事件分析平台 · 完整项目打包 > 单文件全量打包：包含数据流水线脚本、前端全部源码、使用文档、自动还原脚本与依赖下载清单。 > 在任何一台电脑上，保存本文件为 `GDELT_项目完整打包.md`，运行还原脚本即可重建整个项目。 --- ## 一、项目是什么 基于 **GDELT 2.0** 公开数据的国际政治事件分析流水线 + 浏览器端分析平台： - **数据流水线**（Python + DuckDB）：自动抓取 GDELT 每日事件/主题数据，清洗、国家标记、压缩为 Parquet - **分析范围**：CHINA + USA 两国，2026-01-01 ~ 2026-08-01（7 个月） - **前端平台**（纯静态，无需后端）：DuckDB-WASM 在浏览器内直接查询 800+ 万行事件数据，Chart.js 可视化 - **事件码自动翻译**：CAMEO 事件码全量中文映射，图表/表格自动显示中文含义 - **事件溯源**：点击事件码柱状图，直接列出该事件的新闻来源 URL ## 二、目录结构（还原后） ``` gdelt/ ├── gdelt_10c_pipeline.py # 主流水线脚本（抓取/清洗/合并/过滤） ├── README.md # 项目主文档 ├── README.html # 项目主文档 HTML 版（可选） ├── parquet/ # 数据目录（需重新拉取或从源机复制） │ ├── event/dt=YYYY-MM-DD/ # 每日事件明细 213 个文件 │ ├── event_month/month=YYYYMM/country_code=CC/ # 月度分桶 16 文件 │ ├── gkg_raw/dt=YYYY-MM-DD/ # GKG 原始（可选，609MB） │ ├── gkg_filtered/dt=YYYY-MM-DD/ # GKG 过滤后 213 文件 │ └── gkg_filtered_month/month=YYYYMM/country_code=CC/ # 16 文件 └── frontend/ # 前端平台 ├── index.html # 入口页面（5 个标签页） ├── style.css # 样式 ├── app.js # 业务逻辑（DuckDB-WASM + Chart.js） ├── cameo_zh.js # CAMEO 事件码中文翻译 ├── file-list.json # 32 个 parquet 文件路径清单 ├── start_frontend.bat # 双击启动（本地 HTTP + 打开浏览器） ├── README.md # 前端使用手册 └── vendor/ # 本地依赖（restore 脚本自动下载） ├── duckdb-eh.wasm / duckdb-mvp.wasm / duckdb-browser.mjs ├── duckdb-browser-eh.worker.js / duckdb-browser-mvp.worker.js ├── apache-arrow.mjs / tslib.mjs / flatbuffers.mjs └── chart.umd.js ``` ## 三、新电脑还原步骤（5 分钟） 1. 保存本文件为 `GDELT_项目完整打包.md` 2. 安装 Python 3.10+（已有可跳过） 3. 从附录 A 复制还原脚本代码块，保存为 `restore_from_md.py`（也可以让 OpenClaw 直接从本文件提取：`` 标记所在代码块） 4. 运行还原（自动重建所有源码 + 下载前端依赖）： ```bash python restore_from_md.py GDELT_项目完整打包.md ``` 5. 准备数据（二选一）： - **方式 A（推荐，最快）**：从源电脑复制 `gdelt/parquet/` 整个目录到还原目录 - **方式 B（重新拉取）**：运行 `python gdelt_10c_pipeline.py self-test` 验证环境，再 `python gdelt_10c_pipeline.py "batch event" 20260101 20260801` 和 `"batch gkg" ...`（需网络，约 1-2 小时） 6. 启动前端：双击 `frontend/start_frontend.bat`，浏览器自动打开 http://localhost:8765/frontend/ ## 四、环境依赖 | 组件 | 版本 | 用途 | |---|---|---| | Python | 3.10+ | 流水线运行 | | duckdb | 1.5.x | 分析 SQL | | pandas | 2.x/3.x | DataFrame | | pyarrow | 25.x | Parquet I/O | | zstandard | 0.25.x | 压缩 | | requests | 2.3x | HTTP 抓取 | | Chrome / Edge | 90+ | 前端运行（DuckDB-WASM 需要） | ```bash pip install duckdb pandas pyarrow zstandard requests ``` ## 五、前端功能一览 | 标签页 | 功能 | |---|---| | 📊 总览 | 4 指标卡 + Goldstein/调性/事件码/施动者 4 图表；点击事件码柱可溯源新闻 | | 📅 月度趋势 | 月度事件量 + 平均调性走势 | | 🇺🇸🇨🇳 双国对比 | USA vs CHINA 总量/调性/事件类型对比 | | 🏷️ 主题词过滤 | 30 主题词多选，命中趋势 + 新闻样本（含来源链接） | | 🔍 自由查询 | 任意 SQL + CSV 导出，SOURCEURL 自动渲染为链接 | ## 六、流水线命令速查 ```bash # 自检（下载 1 天真实数据验证环境） python gdelt_10c_pipeline.py self-test # 批量抓取事件（日期范围, 注意子命令含空格需引号） python gdelt_10c_pipeline.py "batch event" 20260101 20260801 python gdelt_10c_pipeline.py "batch gkg" 20260101 20260801 # GKG 按主题/调性过滤 python gdelt_10c_pipeline.py filter-gkg --start 20260101 --end 20260801 # 合并为月度分桶 python gdelt_10c_pipeline.py "merge event" 20260101 20260801 python gdelt_10c_pipeline.py "merge gkg" 20260101 20260801 ``` ## 七、修改指南 | 想改什么 | 改哪里 | |---|---| | 分析国家 | `gdelt_10c_pipeline.py` 顶部 `COUNTRY_TABLE` | | 时间范围 | batch 命令的日期参数 | | GKG 主题/调性过滤 | `gdelt_10c_pipeline.py` 的 `filter-gkg` 参数 | | 前端文案/颜色 | `frontend/index.html` / `frontend/style.css` | | 事件码中文翻译 | `frontend/cameo_zh.js`（`_top` / `_leaf` 两个字典） | | 前端查询/图表 | `frontend/app.js`（runOverview / runMonth / loadTrace 等函数） | | 端口 | `frontend/start_frontend.bat` 中 8765 | | 数据文件清单 | `frontend/file-list.json`（新增月度文件后需同步） | --- ## 八、源码全文 > 每个文件前有 `` 标记，`restore_from_md.py` 据此提取还原。 ### 主流水线脚本 · `gdelt_10c_pipeline.py` ````python #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ GDELT 10 国研究专用流水线 v1.0 (2 国版) ================================ 数据源策略: GDELT bulk CSV.zip (tab 分隔) — 20260101 ~ 20260801 - Event : http://data.gdeltproject.org/events/.export.CSV.zip 索引页: http://data.gdeltproject.org/events/index.html 58 列 tab 分隔, zip 一大列 1.5-7 MB / 1 天 - GKG : http://data.gdeltproject.org/gkg/.gkg.csv.zip 索引页: http://data.gdeltproject.org/gkg/index.html 27 列 tab 分隔, zip 一大列 3-32 MB / 1 天 - Mentions : bulk 不提供, 走 HAPI JSON 下载限流: 简单 GET, 无认证, 遭遇 5xx -> 指数退避 4 次 -> 跳过并写 _failed.log 目标国家 (2): CHINA, USA 输出: ~/gdelt/parquet/event/dt=YYYY-MM-DD/ (按天, event 列精简到 10 列, 档 B) ~/gdelt/parquet/gkg_raw/dt=YYYY-MM-DD/ (按天, gkg 列精简到 14 列, 档 B) ~/gdelt/parquet/gkg_filtered/dt=YYYY-MM-DD/ (按天 + 主题/媒体过滤) ~/gdelt/parquet/mentions_raw/dt=YYYY-MM-DD/ (按天, mentions 列精简到 10 列, 档 B) ~/gdelt/parquet/event_month/month=YYYY-MM/ (按月分桶, 2 国子目录) ~/gdelt/parquet/gkg_filtered_month/month=YYYY-MM/ (按月分桶, 2 国子目录) ~/gdelt/parquet/mentions_month/month=YYYY-MM/ (按月分桶, 2 国子目录) 约束: 7 个月 × 2 国 全量 < 1 GB, 档 B 激进列精简; 列存 + zstd-9; event 10 列; gkg 14 列; mentions 10 列; merge 按月分桶 + country_code 子分桶. 预估: event 0.06 + gkg 0.10 + mentions 0.03 ≈ 0.19 GB ✅ 子命令: self-test 检查环境 + 拉 1 个 zip (bulk) + 写入 _selftest/ batch event YYYYMMDD YYYYMMDD [--source bulk|hapi] 默认 bulk batch gkg YYYYMMDD YYYYMMDD [--source bulk|hapi] 默认 bulk batch mentions YYYYMMDD YYYYMMDD [--source hapi] bulk 无 mentions filter-gkg GKG 二次过滤 (主题词 + Tone 阈值 + 媒体白名单) merge event 按月合并 event 切片 (event_month, 列精简 10 列) merge gkg 按月合并 gkg_filtered 切片 (gkg_filtered_month) merge mentions 按月合并 mentions 切片 (mentions_month) 依赖: duckdb, pandas, pyarrow, zstandard, requests """ from __future__ import annotations import argparse import io import json import os import re import sys import time import zipfile from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Iterator, Sequence import requests # --------------------------------------------------------------------------- # 常量 # --------------------------------------------------------------------------- HOME = Path(os.path.expanduser("~")) BASE_DIR = HOME / "gdelt" PARQUET_DIR = BASE_DIR / "parquet" SELFTEST_DIR = BASE_DIR / "_selftest" BASE_URL = "http://data.gdeltproject.org/events/" USER_AGENT = "gdelt-10c-pipeline/1.0 (+research)" HTTP_TIMEOUT = 60 MAX_RETRIES = 4 # GDELT 2.0 export 文件头 (61 列) EVENT_COLUMNS = [ "GLOBALEVENTID", "SQLDATE", "MonthYear", "Year", "FractionDate", "Actor1Code", "Actor1Name", "Actor1CountryCode", "Actor1KnownGroupCode", "Actor1EthnicCode", "Actor1Religion1Code", "Actor1Religion2Code", "Actor1Type1Code", "Actor1Type2Code", "Actor1Type3Code", "Actor2Code", "Actor2Name", "Actor2CountryCode", "Actor2KnownGroupCode", "Actor2EthnicCode", "Actor2Religion1Code", "Actor2Religion2Code", "Actor2Type1Code", "Actor2Type2Code", "Actor2Type3Code", "IsRootEvent", "EventCode", "EventBaseCode", "EventRootCode", "QuadClass", "GoldsteinScale", "NumMentions", "NumSources", "NumArticles", "AvgTone", "Actor1Geo_Type", "Actor1Geo_FullName", "Actor1Geo_CountryCode", "Actor1Geo_ADM1Code", "Actor1Geo_Lat", "Actor1Geo_Long", "Actor1Geo_FeatureID", "Actor2Geo_Type", "Actor2Geo_FullName", "Actor2Geo_CountryCode", "Actor2Geo_ADM1Code", "Actor2Geo_Lat", "Actor2Geo_Long", "Actor2Geo_FeatureID", "ActionGeo_Type", "ActionGeo_FullName", "ActionGeo_CountryCode", "ActionGeo_ADM1Code", "ActionGeo_Lat", "ActionGeo_Long", "ActionGeo_FeatureID", "DATEADDED", "SOURCEURL", ] # GDELT 2.1 GKG 头 (11 列, tab 分隔, 实际 bulk 文件列头) # 来源: 下载 20260813.gkg.csv.zip 后查头 3 行, 列名依次为: # DATE NUMARTS COUNTS THEMES LOCATIONS PERSONS ORGANIZATIONS TONE CAMEOEVENTIDS SOURCES SOURCEURLS # 注意: bulk GKG 2.1 文件**没有 GKGRECORDID 列** (它是 HAPI JSON 才有的, 用作主键). # bulk 里 GKGRECORDID 在 Event 表里, GKG 是 N:1 挂到 Event 上. # 如果后续要 GKG 主键, 可用 DATE + SOURCEURLS[0] 组合, 或直接 join CAMEOEVENTIDS 拿 EVENTID. GKG_COLUMNS = [ "DATE", "NUMARTS", "COUNTS", "THEMES", "LOCATIONS", "PERSONS", "ORGANIZATIONS", "TONE", "CAMEOEVENTIDS", "SOURCES", "SOURCEURLS", ] # GDELT 2.1 mentions 头 (16 列) MENTIONS_COLUMNS = [ "GLOBALEVENTID", "EventTimeDate", "MentionTimeDate", "MentionType", "MentionSourceName", "MentionIdentifier", "SentenceID", "Actor1CharOffset", "Actor2CharOffset", "ActionGeoCharOffset", "InRawText", "Confidence", "MentionDocLen", "MentionDocTone", "MentionDocTranslationInfo", "ExtrasXML", ] # 2 国 (FIPS / ISO3 / CAMEO / 关键字) COUNTRY_TABLE: dict[str, dict] = { "CHINA": { "fips": "CH", "iso3": "CHN", "cameo": "CHN", "keywords": [r"\bCHINA\b", r"\bBEIJING\b", r"\bXINHUA\b", r"\bPRC\b", r"\bSHANGHAI\b"], }, "USA": { "fips": "US", "iso3": "USA", "cameo": "USA", "keywords": [r"\bUNITED STATES\b", r"\bWASHINGTON\b", r"\bWHITE HOUSE\b", r"\bSTATE DEPARTMENT\b", r"\bPENTAGON\b", r"\bNEW YORK\b"], }, } COUNTRY_FIPS: dict[str, str] = {} for _name, _c in COUNTRY_TABLE.items(): COUNTRY_FIPS[_c["fips"]] = _name # FIPS 2字母 (US, CH) COUNTRY_FIPS[_c["iso3"]] = _name # ISO3 3字母 (USA, CHN) COUNTRY_FIPS[_c["cameo"]] = _name # CAMEO 3字母 (USA, CHN) — 同上 # 主题词 + 媒体白名单 (filter-gkg 二次过滤) THEME_KEYWORDS = [ "PROTEST", "ELECTION", "SANCTION", "COUP", "WAR", "CEASEFIRE", "SUMMIT", "NATO", "G7", "BRICS", "AUKUS", "INDO-PACIFIC", "KOREA", "TAIWAN", "UKRAINE", "HAMAS", "ISRAEL", "IRAN", "VENEZUELA", "ENERGY", "CHIP", "AI", "TRADE", "TARIFF", "TERROR", "REFUGEE", "CLIMATE", "COVID", "VACCINE", "DEFENSE", "DIPLOMACY", "TREATY", "OIL", "GAS", "PIPELINE", "NUCLEAR", "MISSILE", "CYBER", "ASSASSINATION", "KIDNAP", "BORDER", "IMMIGRATION", ] TONE_MIN, TONE_MAX = -15.0, 15.0 MEDIA_WHITELIST = { "REUTERS", "AP", "AFP", "XINHUA", "BBC", "NYT", "TASS", "RIA", "SPIEGEL", "LE MONDE", "ASAHI", "TIMES OF INDIA", "FOLHA", "SANEF", "AL JAZEERA", "DW", "GUARDIAN", "FINANCIAL TIMES", "WALL STREET JOURNAL", "WASHINGTON POST", "CHINA DAILY", "GLOBAL TIMES", "PEOPLE'S DAILY", "INDIA TODAY", "THE HINDU", "O GLOBO", "ESTADAO", "JAPAN TIMES", "KYODO", "MAINICHI", "TAGESSPIEGEL", "FRANKFURTER", "LE FIGARO", "LIBERATION", "DAILY MAIL", "DAILY TELEGRAPH", "INDEPENDENT", "ECONOMIST", "BLOOMBERG", "CNN", "FOX NEWS", "MSNBC", "CBS", "NBC", "ABC", "AGENCE FRANCE", "ASSOCIATED PRESS", "PRESS TRUST OF INDIA", "XINHUA NEWS", } ZSTD_LEVEL = 9 # --------------------------------------------------------------------------- # HAPI 限流 + 失败日志 + GKG 列精简 (档位 B1) # --------------------------------------------------------------------------- HAPI_BASE_URL = "http://api.gdeltproject.org/api/v2/datafeed" HAPI_RATE_PER_MIN = 60 HAPI_REQUEST_INTERVAL = 60.0 / HAPI_RATE_PER_MIN + 0.05 # ≈ 1.05s FAILED_LOG = BASE_DIR / "_failed.log" # HAPI country filter: country:CHN,USA HAPI_COUNTRY_FILTER = ",".join(["CHN", "USA"]) # GKG 砍列 (B1: 从 27 列削到 18 列, 控总量 < 1 GB) # HAPI JSON 字段约定 (避免事后列名不一致) HAPI_EVENT_FIELDS = [ "globaleventid", "sqldate", "actor1code", "actor1name", "actor2code", "actor2name", "eventcode", "eventrootcode", "goldsteinscale", "nummentions", "avgtone", "actiongeo_countrycode", "sourceurl", ] HAPI_GKG_FIELDS = [ "gkgrecordid", "date", "sourcecollectionidentifier", "sourcecommonname", "documentidentifier", "v1themes", "v2themes", "v1locations", "v2locations", "v1persons", "v2persons", "v1organizations", "v2organizations", "v2tone", "allnames", ] HAPI_MENTIONS_FIELDS = [ "globaleventid", "eventtimedate", "mentiontimedate", "mentiontype", "mentionsourcename", "mentionidentifier", "sentenceid", "actor1charoffset", "actor2charoffset", "actiongeocharoffset", "confidence", "mentiondoclen", "mentiondoctone", "mentiondoctranslationinfo", ] class RateLimiter: """HAPI 限流: 全局串行 + 1.05s 间隔.""" def __init__(self, per_min: int = HAPI_RATE_PER_MIN): self.interval = 60.0 / per_min + 0.05 self._last = 0.0 def wait(self) -> None: now = time.monotonic() gap = now - self._last if gap < self.interval: time.sleep(self.interval - gap) self._last = time.monotonic() RATE = RateLimiter() # --------------------------------------------------------------------------- # 工具函数 # --------------------------------------------------------------------------- def log(msg: str) -> None: print(f"[{datetime.now(timezone(timedelta(hours=8))).strftime('%H:%M:%S')}] {msg}", flush=True) def ensure_dirs(*paths: Path) -> None: for p in paths: p.mkdir(parents=True, exist_ok=True) def http_get(url: str, *, stream: bool = True) -> requests.Response: last_exc: Exception | None = None for attempt in range(1, MAX_RETRIES + 1): try: r = requests.get(url, headers={"User-Agent": USER_AGENT}, timeout=HTTP_TIMEOUT, stream=stream) if r.status_code == 200: return r if r.status_code == 404: raise FileNotFoundError(f"404 Not Found: {url}") log(f" retry {attempt}/{MAX_RETRIES}: HTTP {r.status_code}") except requests.RequestException as e: last_exc = e log(f" retry {attempt}/{MAX_RETRIES}: {e}") time.sleep(min(2 ** attempt, 30)) raise RuntimeError(f"GET failed after {MAX_RETRIES} attempts: {url}") from last_exc def list_zip_urls(table: str, date_str: str) -> list[str]: """从 /events/ 或 /gkg/ 索引页解析 date_str 当天的全部 zip URL. 注意: GDELT bulk 文件名 = *当天发布的日期 = 上一天的日期* 例: 0815 早上发布的 zip 文件名是 20260813.export.CSV.zip 里边的 SQLDATE 可能是 20260812 (事件是前一天发生的) """ index_urls = { "event": "http://data.gdeltproject.org/events/index.html", "gkg": "http://data.gdeltproject.org/gkg/index.html", } pat = { "event": re.compile(rf"^{date_str}\.export\.CSV\.zip$"), "gkg": re.compile(rf"^{date_str}\.gkg\.csv\.zip$"), "mentions": None, }[table] if pat is None: return [] base = { "event": "http://data.gdeltproject.org/events/", "gkg": "http://data.gdeltproject.org/gkg/", }[table] r = http_get(index_urls[table], stream=False) urls: list[str] = [] for line in r.text.splitlines(): m = re.search(r'href="([^"]+)"', line, flags=re.IGNORECASE) if not m: continue name = m.group(1) if pat.match(name): urls.append(base + name) urls.sort() return urls def daterange(start: str, end: str) -> Iterator[str]: s = datetime.strptime(start, "%Y%m%d") e = datetime.strptime(end, "%Y%m%d") cur = s while cur <= e: yield cur.strftime("%Y%m%d") cur += timedelta(days=1) def iso_week_tag(date_str: str) -> str: dt = datetime.strptime(date_str, "%Y%m%d") iso = dt.isocalendar() return f"{iso[0]}-W{iso[1]:02d}" # --------------------------------------------------------------------------- # 解析器 # --------------------------------------------------------------------------- def _read_csv_text(text: str, columns: Sequence[str]): import pandas as pd return pd.read_csv( io.StringIO(text), sep="\t", header=None, names=columns, dtype=str, keep_default_na=False, na_values=["", "NA"], on_bad_lines="skip", engine="python", ) def parse_event(text: str): return _read_csv_text(text, EVENT_COLUMNS) def parse_gkg(text: str): return _read_csv_text(text, GKG_COLUMNS) def parse_mentions(text: str): return _read_csv_text(text, MENTIONS_COLUMNS) # --------------------------------------------------------------------------- # 国家打标 # --------------------------------------------------------------------------- def tag_country_event(df): import pandas as pd fips1 = df["Actor1CountryCode"].fillna("").str.strip() fips2 = df["Actor2CountryCode"].fillna("").str.strip() action_fips = df["ActionGeo_CountryCode"].fillna("").str.strip() src_country = fips1.where(fips1 != "", fips2.where(fips2 != "", action_fips)) haystack = ( df["Actor1Name"].fillna("").str.upper() + " | " + df["Actor2Name"].fillna("").str.upper() + " | " + df["ActionGeo_FullName"].fillna("").str.upper() ) codes: list[str] = [] vias: list[str] = [] for f, hay in zip(src_country.tolist(), haystack.tolist()): f = (f or "").strip().upper() if f in COUNTRY_FIPS: codes.append(COUNTRY_FIPS[f]) vias.append("fips") continue hit = None for name, conf in COUNTRY_TABLE.items(): for pat in conf["keywords"]: if re.search(pat, hay, flags=re.IGNORECASE): hit = name break if hit: break codes.append(hit or "") vias.append("kw" if hit else "") df["country_code"] = codes df["matched_via"] = vias return df _EVENT_CC_CACHE: dict[str, "pd.DataFrame"] = {} # path -> {GLOBALEVENTID: country_code} def tag_country_gkg(df, date_str: str | None = None): """GKG 国家打标 v2: join Event 表拿国家 (带缓存, 跨天复用). 背景: GKG 2.1 的 LOCATIONS 字段实际包含的是人物名/提及实体, 不是稳定的 FIPS. FIPS 命中率 < 1% (USA 仅 46/17k 行). 所以 GKG 国家应该 join Event. 机制: 从 ~/gdelt/parquet/event_month/ 读所有天的 GLOBALEVENTID -> country_code 映射, 再 join 回 GKG 的 CAMEOEVENTIDS (逗号分隔). 缓存映射跨天复用避免每条 GKG 都重读 175 MB. """ import pandas as pd if "CAMEOEVENTIDS" not in df.columns or df["CAMEOEVENTIDS"].fillna("").eq("").all(): df["country_code"] = "" return df # 1) 读 event 全量 GLOBALEVENTID -> country_code (带缓存) event_month_root = (PARQUET_DIR / "event_month").as_posix() event_raw_root = (PARQUET_DIR / "event").as_posix() src_path = event_month_root if (PARQUET_DIR / "event_month").exists() else event_raw_root cache_key = src_path if cache_key not in _EVENT_CC_CACHE: try: import duckdb ev_df = duckdb.connect().execute( f"SELECT CAST(GLOBALEVENTID AS BIGINT) AS GLOBALEVENTID, country_code " f"FROM read_parquet('{src_path}/**/*.parquet') " f"WHERE country_code IS NOT NULL AND country_code != ''" ).fetch_df() _EVENT_CC_CACHE[cache_key] = ev_df.drop_duplicates(subset=["GLOBALEVENTID"], keep="first") log(f" tag_country_gkg: cached {_EVENT_CC_CACHE[cache_key].shape} from {src_path}") except Exception as e: log(f" tag_country_gkg: cannot read event parquet ({e}); fallback empty") df["country_code"] = "" return df ev_map = _EVENT_CC_CACHE[cache_key] # 2) 炸开 GKG CAMEOEVENTIDS, 记录原始行索引 gkg_ids = df[["CAMEOEVENTIDS"]].copy() gkg_ids["CAMEOEVENTIDS"] = gkg_ids["CAMEOEVENTIDS"].fillna("").astype(str) gkg_ids["_orig_idx"] = gkg_ids.index exp = gkg_ids.assign(EVENT_ID=gkg_ids["CAMEOEVENTIDS"].str.split(",")).explode("EVENT_ID") exp["EVENT_ID"] = exp["EVENT_ID"].astype(str).str.strip() exp = exp[(exp["EVENT_ID"] != "") & exp["EVENT_ID"].str.isdigit()] exp["EVENT_ID"] = exp["EVENT_ID"].astype("int64") # 3) join event 表 j = exp.merge(ev_map, left_on="EVENT_ID", right_on="GLOBALEVENTID", how="inner") # 4) 每个原始 GKG 行取首个匹配 country_code cc = j.groupby("_orig_idx")["country_code"].first() df["country_code"] = df.index.map(cc).fillna("") return df def tag_country_mentions(df): import pandas as pd sentences = df["MentionIdentifier"].fillna("").str.upper() codes: list[str] = [] for s in sentences: hit = None for name, conf in COUNTRY_TABLE.items(): for pat in conf["keywords"]: if re.search(pat, s, flags=re.IGNORECASE): hit = name break if hit: break codes.append(hit or "") df["country_code"] = codes return df # --------------------------------------------------------------------------- # 列精简 (档位 D: event 12 列, mentions 去 ExtrasXML/InRawText, gkg 全列) # --------------------------------------------------------------------------- # 档 B: event 10 列, gkg 14 列 (D 列基础上再砍 4 列), mentions 10 列 EVENT_KEEP_COLS = [ "GLOBALEVENTID", "SQLDATE", "Actor1Code", "Actor2Code", "EventCode", "GoldsteinScale", "NumMentions", "AvgTone", "ActionGeo_CountryCode", "SOURCEURL", "country_code", "matched_via", ] GKG_DROP_COLS = [ # 11 列真实列头 (GKG 2.1) 中不需保留的列: # COUNTS / PERSONS / ORGANIZATIONS / SOURCES 通常很大, 不进 slim # CAMEOEVENTIDS **保留**: join Event 表拿国家 code (GKG LOCATIONS 不准, FIPS 命中率 < 1%) "COUNTS", "PERSONS", "ORGANIZATIONS", "SOURCES", "LOCATIONS", ] MENTIONS_DROP_COLS = [ "ExtrasXML", "InRawText", "SentenceID", "Actor1CharOffset", "Actor2CharOffset", "MentionDocLen", "MentionDocTone", "MentionDocTranslationInfo", ] def slim_event(df): """event 表列精简到 12 列 + 标签列.""" keep = [c for c in EVENT_KEEP_COLS if c in df.columns] return df[keep] def slim_mentions(df): """mentions 表去掉 ExtrasXML / InRawText 两列大字段.""" return df.drop(columns=[c for c in MENTIONS_DROP_COLS if c in df.columns], errors="ignore") def slim_gkg(df): """gkg 表 B1 砍大字段. 保留 CAMEOEVENTIDS (join Event 表拿国家).""" return df.drop(columns=[c for c in GKG_DROP_COLS if c in df.columns], errors="ignore") # --------------------------------------------------------------------------- # Source 接口 + HAPISource (JSON) + BulkFileSource (老 CSV.zip) + JSON 解析 # --------------------------------------------------------------------------- class Source: """数据源抽象: 给一个日期字符串, 返回该日期该表的全部原始文本(JSON 或 CSV).""" name: str = "abstract" def fetch_day(self, table: str, date_str: str) -> list[str]: """返回该日期该表的所有页面原始文本 (JSON 字符串 或 CSV 文本) 列表.""" raise NotImplementedError class HAPISource(Source): """GDELT HAPI JSON 接口. mode ∈ {event, gkgthemes, mentions}.""" name = "hapi" MODE_MAP = { "event": "event", "gkg": "gkgthemes", "mentions": "mentions", } def fetch_day(self, table: str, date_str: str) -> list[str]: mode = self.MODE_MAP[table] # HAPI 24h 切片: 一天两页 (00-12 / 12-24), 避免单页超限 pages = [ (f"{date_str}000000", f"{date_str}115959"), (f"{date_str}120000", f"{date_str}235959"), ] out: list[str] = [] for start_ts, end_ts in pages: url = ( f"{HAPI_BASE_URL}?mode={mode}" f"&format=json" f"&entity=country:{HAPI_COUNTRY_FILTER}" f"&startdatetime={start_ts}" f"&enddatetime={end_ts}" ) for attempt in range(1, MAX_RETRIES + 1): try: RATE.wait() r = requests.get(url, headers={"User-Agent": USER_AGENT}, timeout=HTTP_TIMEOUT) if r.status_code == 200: out.append(r.text) break if r.status_code == 404: log(f" HAPI 404 {mode} {start_ts} (skip)") break log(f" HAPI retry {attempt}/{MAX_RETRIES}: HTTP {r.status_code}") except requests.RequestException as e: log(f" HAPI retry {attempt}/{MAX_RETRIES}: {e}") time.sleep(min(2 ** attempt, 30)) else: _log_failure(table, date_str, start_ts, "max_retries") return out class BulkFileSource(Source): """GDELT bulk CSV.zip 源 (event/gkg): HTTP GET -> 内存解 zip -> tab 分隔文本.""" name = "bulk" def fetch_day(self, table: str, date_str: str) -> list[str]: if table == "mentions": log(f" bulk source: mentions not in bulk, use hapi instead") return [] urls = list_zip_urls(table, date_str) if not urls: log(f" bulk {table} {date_str}: no zip found") return [] out: list[str] = [] for u in urls: try: out.append(_fetch_zip_text(u)) except Exception as e: log(f" bulk FAIL {u}: {e}") _log_failure(table, date_str, u, str(e)) return out class FakeSource(Source): """Fake JSON self-test source: 不发任何网络请求, 生成覆盖 schema 的样本. 样本设计目的: 验证 parse_event_json / tag_country_event / slim_event / write_parquet 全链路. """ name = "fake" def fetch_day(self, table: str, date_str: str) -> list[str]: if table != "event": log(f" fake source: only event supported, got {table}") return [] records = _build_fake_event_records(date_str, n=240) # 模拟 HAPI JSON 顶层结构: {records: [...]} return [json.dumps({"records": records}, ensure_ascii=False)] def _build_fake_event_records(date_str: str, n: int = 240) -> list[dict]: """覆盖关键场景: CHINA / USA 命中 + 其他国家不命中 + 多 actor 组合.""" import random rng = random.Random(20260115) out: list[dict] = [] # 1) CHINA 命中 (FIPS = CH) out.append({ "globaleventid": "100000001", "sqldate": f"{date_str}", "actor1code": "CH", "actor1name": "CHINA", "actor2code": "US", "actor2name": "UNITED STATES", "eventcode": "042", "eventrootcode": "04", "goldsteinscale": "-2.5", "nummentions": "12", "avgtone": "-3.4", "actiongeo_countrycode": "CH", "actiongeo_fullname": "BEIJING, CHINA", "sourceurl": "http://example.com/news/1", }) # 2) USA 命中 (FIPS = US) out.append({ "globaleventid": "100000002", "sqldate": f"{date_str}", "actor1code": "US", "actor1name": "UNITED STATES", "actor2code": "MX", "actor2name": "MEXICO", "eventcode": "020", "eventrootcode": "02", "goldsteinscale": "1.2", "nummentions": "8", "avgtone": "1.8", "actiongeo_countrycode": "US", "actiongeo_fullname": "WASHINGTON, DC, USA", "sourceurl": "http://example.com/news/2", }) # 3) CHINA 命中 (keyword, Actor1Name = BEIJING) out.append({ "globaleventid": "100000003", "sqldate": f"{date_str}", "actor1code": "", "actor1name": "BEIJING", "actor2code": "", "actor2name": "", "eventcode": "010", "eventrootcode": "01", "goldsteinscale": "0.0", "nummentions": "3", "avgtone": "0.5", "actiongeo_countrycode": "", "actiongeo_fullname": "", "sourceurl": "http://example.com/news/3", }) # 4) 其他国家 (不命中) out.append({ "globaleventid": "100000004", "sqldate": f"{date_str}", "actor1code": "RS", "actor1name": "RUSSIA", "actor2code": "FR", "actor2name": "FRANCE", "eventcode": "050", "eventrootcode": "05", "goldsteinscale": "-1.0", "nummentions": "5", "avgtone": "-1.2", "actiongeo_countrycode": "RS", "actiongeo_fullname": "MOSCOW, RUSSIA", "sourceurl": "http://example.com/news/4", }) # 5~n) 随机事件 countries = [("CH", "CHINA", "CH"), ("US", "UNITED STATES", "US"), ("JA", "JAPAN", "JA"), ("UK", "UNITED KINGDOM", "UK"), ("BR", "BRAZIL", "BR")] for i in range(5, n): c = countries[rng.randint(0, len(countries) - 1)] out.append({ "globaleventid": f"10000{i:04d}", "sqldate": f"{date_str}", "actor1code": c[0], "actor1name": c[1], "actor2code": "", "actor2name": "", "eventcode": f"{rng.randint(1, 19):03d}", "eventrootcode": f"{rng.randint(1, 4):02d}", "goldsteinscale": f"{rng.uniform(-10, 10):.2f}", "nummentions": str(rng.randint(1, 50)), "avgtone": f"{rng.uniform(-10, 10):.2f}", "actiongeo_countrycode": c[2], "actiongeo_fullname": c[1], "sourceurl": f"http://example.com/news/{i}", }) return out def _log_failure(table: str, date_str: str, window: str, err: str) -> None: try: ensure_dirs(BASE_DIR) with open(FAILED_LOG, "a", encoding="utf-8") as fh: fh.write(f"{table}\t{date_str}\t{window}\t{err}\n") except Exception as e: log(f" WARN: cannot write _failed.log: {e}") def cmd_self_test_bq(_args: argparse.Namespace) -> int: """已删除: 改走 self-test 默认 bulk.""" log("self-test-bq: removed (use bulk instead)") return 1 def get_source(name: str) -> Source: return {"hapi": HAPISource(), "bulk": BulkFileSource(), "fake": FakeSource(), "bq": BigQuerySource()}[name] class BigQuerySource(Source): """BigQuery 数据源 (你不需要 GCP, 默认 disabled).""" name = "bq" def fetch_day(self, table: str, date_str: str) -> list[str]: log(" bq source: disabled (no GCP creds). Use bulk for events/gkg, hapi for mentions.") return [] # --------------------------------------------------------------------------- # BigQuerySource: 匿名 REST 查询 gdelt-bq.gdeltv2.* 公共数据集 # --------------------------------------------------------------------------- class BigQuerySource(Source): """Google BigQuery 公共数据集查询 (匿名 REST, 无凭证). 限制: - 单次返回 ≤ ~128 MB / 10k 行 (BQ 公共查询匿名配额) - 单日量超过配额时分页: 按月分桶 SQL 模板在 _BQ_SQL_TPL, 按表选择. 国家过滤走服务端 (CAMEO 3字母). """ name = "bq" BQ_PROJECT = "gdelt-bq" # GDELT 公共项目 (只读公开数据集) # 表 -> 全名 + 字段映射 (服务端 SELECT 过滤, 减少扫描字节) _TABLE_SQL: dict[str, str] = { "event": "`gdelt-bq.gdeltv2.events`", "gkg": "`gdelt-bq.gdeltv2.gkg`", "mentions": "`gdelt-bq.gdeltv2.eventmentions`", } def fetch_day(self, table: str, date_str: str) -> list[str]: # date_str 可能是 YYYYMMDD / YYYYMMDD-YYYYMMDD / 直接 YYYY-MM-DD days = _daterange_from_arg(date_str) if not days: log(f" bq: invalid date_str {date_str}") return [] # 按月分桶, 避免跨月扫描 months = sorted({d[:6] for d in days}) out: list[str] = [] for ym in months: month_days = [d for d in days if d.startswith(ym)] month_start = f"{ym[:4]}-{ym[4:6]}-01" month_end = f"{month_days[-1][:4]}-{month_days[-1][4:6]}-{month_days[-1][6:]}" sql = self._build_sql(table, month_days, month_start, month_end) log(f" bq {table} {ym}: SQLDAYEs in {month_days[0]}..{month_days[-1]} ({len(month_days)} days)") data = self._run_query(sql) if data: out.append(data) return out def _build_sql(self, table: str, days: list[str], month_start: str, month_end: str) -> str: """构造 SQL: 服务端过滤 SQLDATE + Actor1CountryCode IN (CHINA/USA). 按月分区扫描 (TIMESTAMP_TRUNC(_PARTITIONTIME, MONTH) = ...), 减少扫描字节数. """ t = self._TABLE_SQL[table] date_list = ",".join(f"DATE('{d[:4]}-{d[4:6]}-{d[6:8]}')" for d in days) # 2 国 CAMEO: CHN / USA actor_filter = "Actor1CountryCode IN ('CHN','USA') OR Actor2CountryCode IN ('CHN','USA')" if table == "event": return f""" SELECT GLOBALEVENTID, SQLDATE, Actor1Code, Actor1Name, Actor2Code, Actor2Name, EventCode, GoldsteinScale, NumMentions, AvgTone, ActionGeo_CountryCode, SOURCEURL FROM {t} WHERE _PARTITIONTIME >= TIMESTAMP('{month_start}') AND _PARTITIONTIME < TIMESTAMP_ADD(TIMESTAMP('{month_start}'), INTERVAL 1 MONTH) AND SQLDATE IN ({date_list}) AND ({actor_filter}) """ if table == "gkg": return f""" SELECT GKGRECORDID, DATE, SourceCommonName, DocumentIdentifier, V1Themes, V2Themes, V1Locations, V2Locations, V1Persons, V2Persons, V1Organizations, V2Organizations, V2Tone, AllNames FROM {t} WHERE _PARTITIONTIME >= TIMESTAMP('{month_start}') AND _PARTITIONTIME < TIMESTAMP_ADD(TIMESTAMP('{month_start}'), INTERVAL 1 MONTH) AND DATE IN ({date_list}) """ if table == "mentions": return f""" SELECT GLOBALEVENTID, EventTimeDate, MentionTimeDate, MentionType, MentionSourceName, MentionIdentifier, Actor1CharOffset, Actor2CharOffset, Confidence, MentionDocLen, MentionDocTone FROM {t} WHERE _PARTITIONTIME >= TIMESTAMP('{month_start}') AND _PARTITIONTIME < TIMESTAMP_ADD(TIMESTAMP('{month_start}'), INTERVAL 1 MONTH) AND SQLDATE IN ({date_list}) """ raise ValueError(f"bq: unknown table {table}") def _run_query(self, sql: str) -> str: """匿名 POST 到 BQ REST, 返回 JSON 文本 (顶层 {records: [...]} 格式).""" url = f"https://bigquery.googleapis.com/bigquery/v2/projects/{self.BQ_PROJECT}/queries" body = { "kind": "bigquery#queryRequest", "query": sql, "useLegacySql": False, "maxResults": 10000, } for attempt in range(1, MAX_RETRIES + 1): try: RATE.wait() r = requests.post( url, json=body, headers={"User-Agent": USER_AGENT, "Content-Type": "application/json"}, timeout=HTTP_TIMEOUT, ) if r.status_code == 200: obj = r.json() if not obj.get("jobComplete"): log(f" bq: job not complete, jobReference={obj.get('jobReference')}") return "" schema = obj.get("schema", {}).get("fields", []) rows = obj.get("rows", []) fields = [f["name"] for f in schema] records = [{fields[i]: row["f"][i].get("v", "") for i in range(len(fields))} for row in rows] log(f" bq: got {len(records)} rows ({len(fields)} cols)") return json.dumps({"records": records}, ensure_ascii=False) if r.status_code in (403, 404): log(f" bq FAIL {r.status_code}: {r.text[:200]}") return "" log(f" bq retry {attempt}/{MAX_RETRIES}: HTTP {r.status_code}") except requests.RequestException as e: log(f" bq retry {attempt}/{MAX_RETRIES}: {e}") time.sleep(min(2 ** attempt, 30)) return "" def _daterange_from_arg(date_str: str) -> list[str]: """'YYYYMMDD' / 'YYYYMMDD-YYYYMMDD' / 'YYYY-MM-DD' -> ['YYYYMMDD', ...].""" s = date_str.replace("-", "") if len(s) == 8 and s.isdigit(): return [s] if "-" in date_str and len(date_str.replace("-", "")) == 16: a, b = date_str.split("-") a, b = a.replace("-", ""), b.replace("-", "") if len(a) == 8 and len(b) == 8 and a.isdigit() and b.isdigit(): return daterange(a, b) return [] # --------------------------------------------------------------------------- # HAPI JSON 解析 -> DataFrame (复用现有 tag_country_*) # --------------------------------------------------------------------------- def _json_records(text: str, field_aliases: dict[str, str]) -> list[dict]: """HAPI JSON 返回结构多样: 顶层 list / {records: [...]} / {data: [...]} / 单 dict. field_aliases: 小写字段名 -> DataFrame 列名.""" try: obj = json.loads(text) except Exception: return [] if isinstance(obj, list): records = obj elif isinstance(obj, dict): for k in ("records", "data", "results", "articles", "events", "items"): if k in obj and isinstance(obj[k], list): records = obj[k] break else: records = [obj] else: return [] out: list[dict] = [] for r in records: if not isinstance(r, dict): continue row = {} for src_key, dst_col in field_aliases.items(): row[dst_col] = r.get(src_key, "") out.append(row) return out def parse_event_json(text: str): import pandas as pd # HAPI event JSON -> 老 GDELT 2.0 event schema 列名 (兼容 tag_country_event) # actor1code/actor2code 在 HAPI 里是 country code (CAMEO 3字母), 映射到 Actor1CountryCode aliases_dst = { "globaleventid": "GLOBALEVENTID", "sqldate": "SQLDATE", "actor1code": "Actor1CountryCode", "actor1name": "Actor1Name", "actor2code": "Actor2CountryCode", "actor2name": "Actor2Name", "eventcode": "EventCode", "eventrootcode": "EventRootCode", "goldsteinscale": "GoldsteinScale", "nummentions": "NumMentions", "avgtone": "AvgTone", "actiongeo_countrycode": "ActionGeo_CountryCode", "actiongeo_fullname": "ActionGeo_FullName", "sourceurl": "SOURCEURL", } records = _json_records(text, aliases_dst) if not records: return pd.DataFrame(columns=list(aliases_dst.values())) return pd.DataFrame.from_records(records) def parse_gkg_json(text: str): import pandas as pd aliases_dst = { "gkgrecordid": "GKGRECORDID", "date": "DATE", "sourcecollectionidentifier": "SourceCollectionIdentifier", "sourcecommonname": "SourceCommonName", "documentidentifier": "DocumentIdentifier", "v1themes": "V1Themes", "v2themes": "V2Themes", "v1locations": "V1Locations", "v2locations": "V2Locations", "v1persons": "V1Persons", "v2persons": "V2Persons", "v1organizations": "V1Organizations", "v2organizations": "V2Organizations", "v2tone": "V2Tone", "allnames": "AllNames", } records = _json_records(text, aliases_dst) if not records: return pd.DataFrame(columns=list(aliases_dst.values())) return pd.DataFrame.from_records(records) def parse_mentions_json(text: str): import pandas as pd aliases_dst = { "globaleventid": "GLOBALEVENTID", "eventtimedate": "EventTimeDate", "mentiontimedate": "MentionTimeDate", "mentiontype": "MentionType", "mentionsourcename": "MentionSourceName", "mentionidentifier": "MentionIdentifier", "sentenceid": "SentenceID", "actor1charoffset": "Actor1CharOffset", "actor2charoffset": "Actor2CharOffset", "actiongeocharoffset": "ActionGeoCharOffset", "confidence": "Confidence", "mentiondoclen": "MentionDocLen", "mentiondoctone": "MentionDocTone", "mentiondoctranslationinfo": "MentionDocTranslationInfo", } records = _json_records(text, aliases_dst) if not records: return pd.DataFrame(columns=list(aliases_dst.values())) return pd.DataFrame.from_records(records) def write_parquet(df, dest_dir: Path, *, partition_cols: Sequence[str] = ()) -> Path: import pyarrow as pa import pyarrow.parquet as pq ensure_dirs(dest_dir) table = pa.Table.from_pandas(df, preserve_index=False) pq.write_to_dataset( table, root_path=str(dest_dir), partition_cols=list(partition_cols) or None, compression="zstd", compression_level=ZSTD_LEVEL, use_dictionary=True, ) return dest_dir # --------------------------------------------------------------------------- # 子命令: self-test # --------------------------------------------------------------------------- def cmd_self_test(args: argparse.Namespace) -> int: log("self-test: start") import pyarrow # noqa: F401 import pandas # noqa: F401 import duckdb # noqa: F401 import zstandard # noqa: F401 log("deps OK: duckdb/pandas/pyarrow/zstandard") ensure_dirs(SELFTEST_DIR) source_name = getattr(args, "source", "bulk") or "bulk" # bulk 路径最新发布日通常是当天前一天 -> 用 20260813 target_date = "20260813" src = get_source(source_name) log(f"self-test: {src.name} event {target_date} ...") pages = src.fetch_day("event", target_date) if not pages: log(f"ERROR: {src.name} returned 0 pages for {target_date}") return 2 if isinstance(src, BulkFileSource): parse_fn = parse_event elif isinstance(src, HAPISource): parse_fn = parse_event_json else: parse_fn = parse_event import pandas as pd chunks = [parse_fn(t) for t in pages] df = pd.concat(chunks, ignore_index=True) if chunks else pd.DataFrame() log(f"self-test: parsed {len(df)} rows, {len(df.columns)} cols") if len(df): df = tag_country_event(df) hit = (df["country_code"] != "").sum() log(f"self-test: country hits = {hit} / {len(df)} ({hit/max(1,len(df)):.2%})") sample_dir = SELFTEST_DIR / "event_sample" if len(df): write_parquet(slim_event(df.head(2000)), sample_dir) log(f"self-test: wrote sample to {sample_dir}") log("self-test: OK") return 0 def cmd_self_test_fake(_args: argparse.Namespace) -> int: """fake source 自检: 不发网络请求, 验证 parse / tag / slim / write 全链路.""" log("self-test-fake: start (no network)") import pyarrow.parquet as pq import pandas as pd import duckdb ensure_dirs(SELFTEST_DIR) src = get_source("fake") target_date = "20260115" pages = src.fetch_day("event", target_date) log(f"self-test-fake: {len(pages)} fake page(s)") df = parse_event_json(pages[0]) log(f" parsed {len(df)} rows, {len(df.columns)} cols") df = tag_country_event(df) hits = df["country_code"].value_counts().to_dict() log(f" country_code distribution: {hits}") df_slim = slim_event(df) log(f" slim_event -> {df_slim.shape[1]} cols: {list(df_slim.columns)}") sample_dir = SELFTEST_DIR / "fake_event_sample" if sample_dir.exists(): for p in sample_dir.glob("*.parquet"): p.unlink() write_parquet(df_slim, sample_dir) log(f" wrote {len(df_slim)} rows to {sample_dir}") # DuckDB 验证: 读回 Parquet + 行数 + 国家分布 con = duckdb.connect(":memory:") pq_files = sorted(sample_dir.rglob("*.parquet")) if not pq_files: log(" ERROR: no parquet files written") return 3 fl = ",".join(f"'{str(p).replace(chr(92), '/')}'" for p in pq_files) n_rows = con.execute(f"SELECT COUNT(*) FROM read_parquet([{fl}])").fetchone()[0] schema = con.execute(f"DESCRIBE SELECT * FROM read_parquet([{fl}])").fetch_df() log(f" DuckDB readback: {n_rows} rows") log(" schema:") for _, row in schema.iterrows(): log(f" - {row['column_name']:<25} {row['column_type']}") # 文件大小 total_bytes = sum(p.stat().st_size for p in pq_files) log(f" Parquet on-disk: {total_bytes/1024:.1f} KB ({len(pq_files)} file(s))") log("self-test-fake: OK") return 0 # --------------------------------------------------------------------------- # 子命令: batch # --------------------------------------------------------------------------- def _fetch_zip_text(url: str) -> str: r = http_get(url, stream=True) blob = b"".join(r.iter_content(chunk_size=1 << 20)) with zipfile.ZipFile(io.BytesIO(blob)) as zf: with zf.open(zf.namelist()[0]) as fh: return io.TextIOWrapper(fh, encoding="utf-8", errors="replace").read() def _batch_impl(table: str, start: str, end: str, source_name: str) -> int: taggers = {"event": tag_country_event, "gkg": tag_country_gkg, "mentions": tag_country_mentions} out_root = { "event": PARQUET_DIR / "event", "gkg": PARQUET_DIR / "gkg_raw", "mentions": PARQUET_DIR / "mentions_raw", }[table] tagger = taggers[table] src = get_source(source_name) is_hapi = isinstance(src, HAPISource) parser = { "event": parse_event_json if is_hapi else parse_event, "gkg": parse_gkg_json if is_hapi else parse_gkg, "mentions": parse_mentions_json if is_hapi else parse_mentions, }[table] import pandas as pd total_rows = 0 for date_str in daterange(start, end): day_dir = out_root / f"dt={date_str}" if any(day_dir.glob("*.parquet")): log(f"{table} {date_str}: already done, skip") continue log(f"{table} {date_str} [{src.name}]: fetching ...") pages = src.fetch_day(table, date_str) if not pages: log(f" {table} {date_str}: 0 pages, skip") continue day_chunks: list = [] for t in pages: df = parser(t) if len(df) == 0: continue if table == "gkg": df = tagger(df, date_str) else: df = tagger(df) df = df[df["country_code"] != ""] if len(df): day_chunks.append(df) if not day_chunks: log(f" {table} {date_str}: 0 rows after country filter") continue merged = pd.concat(day_chunks, ignore_index=True) # 档位 D + B1: 落盘前列精简 if table == "event": merged = slim_event(merged) elif table == "gkg": merged = slim_gkg(merged) elif table == "mentions": merged = slim_mentions(merged) write_parquet(merged, day_dir) total_rows += len(merged) log(f" wrote {len(merged)} rows ({merged.shape[1]} cols) -> {day_dir}") log(f"{table}: total {total_rows} rows over {start}..{end}") return 0 def cmd_batch_event(args: argparse.Namespace) -> int: return _batch_impl("event", args.start, args.end, args.source) def cmd_batch_gkg(args: argparse.Namespace) -> int: return _batch_impl("gkg", args.start, args.end, args.source) def cmd_batch_mentions(args: argparse.Namespace) -> int: return _batch_impl("mentions", args.start, args.end, args.source) # --------------------------------------------------------------------------- # 子命令: filter-gkg # --------------------------------------------------------------------------- def cmd_filter_gkg(args: argparse.Namespace) -> int: """GKG 二次过滤: 主题词 ∩ Tone 阈值 ∩ 媒体白名单 (SourceURL 域名).""" import duckdb con = duckdb.connect(":memory:") out_root = PARQUET_DIR / "gkg_filtered" ensure_dirs(out_root) theme_pat = "|".join(re.escape(t) for t in THEME_KEYWORDS) media_norm = [m.upper() for m in MEDIA_WHITELIST] # 过滤: 主题词 ∩ Tone 阈值 ∈ (-100, TONE_MAX) ∩ source_url 包含某白名单域名 # slim 后实际列: THEMES, TONE, SOURCEURLS start = args.start or "20240101" end = args.end or datetime.now().strftime("%Y%m%d") log(f"filter-gkg: scan gkg_raw {start}..{end}") total = 0 for date_str in daterange(start, end): d = PARQUET_DIR / "gkg_raw" / f"dt={date_str}" if not d.exists(): continue day_dir = out_root / f"dt={date_str}" if any(day_dir.glob("*.parquet")): log(f" {date_str}: cached, skip") continue files = sorted(d.glob("*.parquet")) if not files: continue file_list = ",".join(f"'{str(p).replace(chr(92), '/')}'" for p in files) # tone 阈值: 取 TONE 字段第一段 (逗号分隔 6 个值, 第 1 个是 average tone) # 这里只拿主调: cast TONE 为 str, 拆 , 取首段 # tone 在 [-100, 100] 之间, |tone| < 5 过滤中性; 偏负/偏正 可保留 sql = f""" SELECT * FROM read_parquet([{file_list}]) WHERE country_code <> '' AND regexp_matches(coalesce(THEMES,''), '(?i){theme_pat}') AND try_cast(regexp_extract(coalesce(TONE,''), '^(-?\\d+(?:\\.\\d+)?)', 1) AS DOUBLE) BETWEEN {TONE_MIN} AND {TONE_MAX} AND ( {(' OR '.join(f"contains(upper(coalesce(SOURCEURLS,'')), '{m.replace(chr(39), chr(39)+chr(39))}')" for m in media_norm))} OR {len(media_norm)} = 0 ) """ df = con.execute(sql).fetch_df() if len(df): write_parquet(df, day_dir) total += len(df) log(f" {date_str}: filtered {len(df)} rows") log(f"filter-gkg: total filtered rows = {total}") return 0 # --------------------------------------------------------------------------- # 子命令: merge # --------------------------------------------------------------------------- def _merge_impl(table: str, start: str, end: str) -> int: import duckdb src_map = { "event": PARQUET_DIR / "event", "gkg": PARQUET_DIR / "gkg_filtered", "mentions": PARQUET_DIR / "mentions_raw", } out_map = { "event": PARQUET_DIR / "event_month", "gkg": PARQUET_DIR / "gkg_filtered_month", "mentions": PARQUET_DIR / "mentions_month", } src = src_map[table] out_root = out_map[table] ensure_dirs(out_root) con = duckdb.connect(":memory:") months: dict[str, list[str]] = {} for date_str in daterange(start, end): d = src / f"dt={date_str}" if not d.exists(): continue mt = date_str[:6] # YYYYMM months.setdefault(mt, []).extend( str(p).replace(chr(92), "/") for p in d.glob("*.parquet") ) log(f"{table} merge: {len(months)} month(s)") for mt, files in months.items(): month_dir = out_root / f"month={mt}" if any(month_dir.glob("*.parquet")): log(f" {mt}: cached, skip") continue if not files: continue fl = ",".join(f"'{p}'" for p in files) sql = f"SELECT * FROM read_parquet([{fl}])" df = con.execute(sql).fetch_df() if len(df): # 空 country_code 归入 __UNMATCHED__, 便于审计 df["country_code"] = df["country_code"].fillna("").replace("", "__UNMATCHED__") write_parquet(df, month_dir, partition_cols=["country_code"]) log(f" {mt}: merged {len(df)} rows -> {month_dir}") return 0 def cmd_merge_event(args: argparse.Namespace) -> int: return _merge_impl("event", args.start, args.end) def cmd_merge_gkg(args: argparse.Namespace) -> int: return _merge_impl("gkg", args.start, args.end) def cmd_merge_mentions(args: argparse.Namespace) -> int: return _merge_impl("mentions", args.start, args.end) # --------------------------------------------------------------------------- # CLI 入口 # --------------------------------------------------------------------------- def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="gdelt_10c_pipeline", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) sub = p.add_subparsers(dest="cmd", required=True) st = sub.add_parser("self-test", help="env check + 1-slice schema probe") st.add_argument("--source", choices=["hapi", "bulk"], default="bulk") sub.add_parser("self-test-fake", help="schema probe via fake JSON, no network") def add_batch(name: str) -> None: sp = sub.add_parser(f"batch {name}", help=f"fetch & country-tag {name}") sp.add_argument("start") sp.add_argument("end") sp.add_argument("--source", choices=["hapi", "bulk"], default="bulk") for n in ("event", "gkg", "mentions"): add_batch(n) fg = sub.add_parser("filter-gkg", help="secondary GKG filter") fg.add_argument("--start", default=None) fg.add_argument("--end", default=None) def add_merge(name: str) -> None: sp = sub.add_parser(f"merge {name}", help=f"monthly merge {name}") sp.add_argument("start") sp.add_argument("end") for n in ("event", "gkg", "mentions"): add_merge(n) return p def main(argv: Sequence[str] | None = None) -> int: args = build_parser().parse_args(argv) table = { "self-test": cmd_self_test, "self-test-fake": cmd_self_test_fake, "batch event": cmd_batch_event, "batch gkg": cmd_batch_gkg, "batch mentions": cmd_batch_mentions, "filter-gkg": cmd_filter_gkg, "merge event": cmd_merge_event, "merge gkg": cmd_merge_gkg, "merge mentions": cmd_merge_mentions, }[args.cmd] return table(args) if __name__ == "__main__": sys.exit(main()) ```` ### 项目主文档 · `README.md` ````markdown # GDELT 双国事件分析 Pipeline 使用指南 ## 项目成果 从 GDELT 2.1 原始数据出发，按 **CHINA + USA 双国**、**2026-01-01 至 2026-08-01**（共 213 天）抓取、清洗、压缩到本地 Parquet，全流程 **0 失败、磁盘 1.13 GB**（其中原始 GKG 中间产物可删，得到 524 MB < 1 GB）。 | 阶段 | 行数 | 文件数 | 体积 | |---|---|---|---| | event（原始每日 slim） | 8,938,704 | 213 | 175.7 MB | | event_month（按月分桶） | 8,938,704 | 16 | 176.2 MB | | gkg_raw（带 CAMEOEVENTIDS 的逐日原始） | 5,841,008 | 250 | 609.5 MB | | gkg_filtered（主题词 + Tone + 媒体白名单二次过滤） | 670,562 | 213 | 87.0 MB | | gkg_filtered_month（按月分桶） | 670,562 | 16 | 85.4 MB | | **合计** | — | — | **1.13 GB** | ### 国别分布 | 国家 | Event 行数 | GKG filtered 行数 | 比例 | |---|---|---|---| | USA | 8,194,396 | 617,429 | 91.7% / 92.1% | | CHINA | 744,308 | 53,133 | 8.3% / 7.9% | USA:CHINA ≈ 11:1（Event）/12:1（GKG），符合 GDELT 英文媒体为主导致的美国中心偏见。 --- ## 目录结构 ``` C:\Users\yaahh\gdelt\ ├── gdelt_10c_pipeline.py # 主脚本（self-test / batch / filter / merge） └── parquet\ ├── event\dt=YYYY-MM-DD\*.parquet # 每日 event slim（已含 country_code） ├── event_month\month=YYYYMM\country_code=*\*\*.parquet # 按月 + 国家分桶 ├── gkg_raw\dt=YYYY-MM-DD\*.parquet # 每日 GKG slim（含 CAMEOEVENTIDS） ├── gkg_filtered\dt=YYYY-MM-DD\*.parquet # 每日过滤后 GKG（USA/CHINA + 主题 + Tone + 媒体） ├── gkg_filtered_month\month=YYYYMM\country_code=*\*\*.parquet # 按月 + 国家分桶 └── _selftest\event_sample\*.parquet # self-test 验证样本 ``` --- ## 快速使用 ### 1. 查看数据（任何工具都能用） 数据是标准 Parquet + Snappy + zstd 压缩，可用 DuckDB / pandas / pyarrow / Polars 直接读。 **DuckDB 一次性 SQL 查整月：** ```sql -- 全部 8 个月 USA 事件 SELECT SQLDATE, GLOBALEVENTID, Actor1Code, Actor2Code, EventCode, GoldsteinScale, NumMentions, NumArticles, AvgTone, country_code FROM read_parquet('C:/Users/yaahh/gdelt/parquet/event_month/**/country_code=USA/*.parquet') LIMIT 10; ``` **DuckDB 查 GKG filtered：** ```sql SELECT DATE, NUMARTS, THEMES, TONE, SOURCEURLS FROM read_parquet('C:/Users/yaahh/gdelt/parquet/gkg_filtered_month/**/country_code=CHINA/*.parquet') WHERE regexp_matches(THEMES, '(?i)TAIWAN|TRADE') LIMIT 10; ``` **pandas（按月分桶加载）：** ```python import pandas as pd # 只读 USA 事件（按月分桶后过滤非常快） df = pd.read_parquet( 'C:/Users/yaahh/gdelt/parquet/event_month', columns=['GLOBALEVENTID','SQLDATE','EventCode','AvgTone','NumMentions'], filters=[('country_code','=','USA')], ) # 单月 GKG 主题词二次过滤 jul = pd.read_parquet( 'C:/Users/yaahh/gdelt/parquet/gkg_filtered_month/month=202607/country_code=USA', columns=['GKGRECORDID','THEMES','TONE','SOURCEURLS'], ) ``` --- ### 2. 重跑 / 增量更新 ```powershell # 单条命令：必须用引号包子命令（PowerShell 的 2-word subcommand） cd C:\Users\yaahh\gdelt; python ".\gdelt_10c_pipeline.py" "self-test" cd C:\Users\yaahh\gdelt; python ".\gdelt_10c_pipeline.py" "batch event" 20260101 20260801 cd C:\Users\yaahh\gdelt; python ".\gdelt_10c_pipeline.py" "batch gkg" 20260101 20260801 cd C:\Users\yaahh\gdelt; python ".\gdelt_10c_pipeline.py" "filter-gkg" --start 20260101 --end 20260801 cd C:\Users\yaahh\gdelt; python ".\gdelt_10c_pipeline.py" "merge event" 20260101 20260801 cd C:\Users\yaahh\gdelt; python ".\gdelt_10c_pipeline.py" "merge gkg" 20260101 20260801 ``` **每个 subcommand 的行为：** | 命令 | 行为 | |---|---| | `self-test` | 抓今天或最近一天做端到端冒烟测试，落地到 `_selftest/event_sample` | | `batch event` | 按日抓 event slim（含 country_code），落地 `parquet/event/dt=YYYY-MM-DD/` | | `batch gkg` | 按日抓 GKG slim（带 CAMEOEVENTIDS），用 join Event 拿 country_code，落地 `parquet/gkg_raw/dt=YYYY-MM-DD/` | | `batch mentions` | HAPI 抓 mentions（**2026 年数据 404，目前禁用**） | | `filter-gkg` | 主题词 + Tone 阈值 + 媒体白名单二次过滤 GKG → `parquet/gkg_filtered/` | | `merge event` | 把每日 event 按月分桶 + 按 country_code 再分桶 → `parquet/event_month/` | | `merge gkg` | 把每日 filtered GKG 按月分桶 → `parquet/gkg_filtered_month/` | **增量特性**：所有 batch/merge/filter 命令都做"already done, skip"判断——`dt=YYYY-MM-DD/*.parquet` 已存在则跳过。**只重跑未完成的日期即可**。 --- ### 3. 想要更多国家？ 打开 `gdelt_10c_pipeline.py`，找到 `COUNTRY_TABLE`（约 113 行）。当前只有 CHINA / USA 两项。 加新国家的步骤： 1. 在 `COUNTRY_TABLE` 加一项（含 `fips` 2字母 + `iso3` 3字母 + `cameo` 3字母 + `keywords` 关键词） 2. 删 `parquet/event/`、`parquet/event_month/` 3. 重跑 `batch event`（event tagger 会按新 COUNTRY_TABLE 重打标） 4. 重跑 `batch gkg` + `merge gkg`（join Event 表拿新国家） --- ### 4. 调整主题词 / 媒体白名单 / Tone 阈值 打开脚本顶部常量区： ```python THEME_KEYWORDS = ["PROTEST", "ELECTION", "SANCTION", ...] # ≈ 38 项 TONE_MIN, TONE_MAX = -15.0, 15.0 # [-15, +15] 区间内 MEDIA_WHITELIST = {"REUTERS", "AP", "AFP", "XINHUA", ...} # ≈ 50 家媒体 ``` 改完只重跑 `filter-gkg` + `merge gkg`，**不用重新抓 raw**（`gkg_raw` 是无损中间产物）。 --- ### 5. 想要更小 / 想要更大 - **想缩到 < 1 GB**：删除 `parquet/gkg_raw/`（中间产物，不影响 filtered），总磁盘降到 524 MB - **想再缩**：删除 `parquet/event/`（保留 `event_month/` 即可），再砍 175 MB - **想要完整无损**：当前 1.13 GB 已包含全部 raw + processed --- ## 关键技术决策 ### 1. 数据源选型 - **Event + GKG** 走 GDELT 官方 bulk CSV（`http://data.gdeltproject.org/events/` 和 `/gkg/`），无速率限制 - **Mentions** 必须走 HAPI（bulk 没 mentions），但 HAPI 对 2026 年数据返回 404，所以 mentions 数据缺 ### 2. 国家打标策略 - **Event 表**：`Actor1CountryCode` / `Actor1Geo_CountryCode` 等列已含 3 字母国家代码 → 直接匹配 COUNTRY_FIPS（同时支持 FIPS / ISO3 / CAMEO 三种别名），命中率 41% - **GKG 表**：`LOCATIONS` 字段实际是人物名/提及实体，不是稳定的 FIPS，命中率 < 1% - **方案**：GKG 通过 `CAMEOEVENTIDS` join 回 Event 表拿 `country_code`，命中率提升到 **20-25%** ### 3. 列精简档位 B1 - Event 58 列 → 保留 10+ 列（GLOBALEVENTID / SQLDATE / Actor1Code / Actor1Name / Actor1CountryCode / Actor2Code / Actor2Name / Actor2CountryCode / EventCode / GoldsteinScale / NumMentions / NumArticles / AvgTone / country_code / dt） - GKG 11 列 → 保留 7 列（GKGRECORDID / DATE / NUMARTS / THEMES / TONE / CAMEOEVENTIDS / SOURCEURLS + country_code + dt） --- ## 故障排查 | 症状 | 排查 | |---|---| | PowerShell `from` 报错 | 把 SQL 写成 .py 文件，不要 `python -c` 内联 | | `merge event/gkg` 不识别 `--start` | merge 不接受参数，直接传位置参数：`"merge event" 20260101 20260801` | | 子命令含空格 | 必须用引号：`"batch event"` 不是 `batch event` | | 想看 raw 加载了多少天 | `Get-ChildItem $HOME\gdelt\parquet\event -Directory \| Measure-Object` | | 想看磁盘 | `(Get-ChildItem $HOME\gdelt\parquet -Recurse -Filter "*.parquet" \| Measure-Object -Property Length -Sum).Sum / 1MB` | | 进程跑一半 abort | 重跑相同命令，already-done 自动跳过，只续跑未完成的日期 | --- ## 复现工作流（从零） ```powershell # 1. 安装依赖（一次性） pip install duckdb pandas pyarrow zstandard requests # 2. 自检 cd C:\Users\yaahh\gdelt; python ".\gdelt_10c_pipeline.py" "self-test" # 3. 抓 event（30-60 分钟，213 天约 9M 行） cd C:\Users\yaahh\gdelt; python ".\gdelt_10c_pipeline.py" "batch event" 20260101 20260801 # 4. 抓 GKG（30-40 分钟，213 天约 5.8M 行） cd C:\Users\yaahh\gdelt; python ".\gdelt_10c_pipeline.py" "batch gkg" 20260101 20260801 # 5. event 按月分桶（< 1 分钟） cd C:\Users\yaahh\gdelt; python ".\gdelt_10c_pipeline.py" "merge event" 20260101 20260801 # 6. GKG 主题词 + tone + 媒体过滤（< 1 分钟） cd C:\Users\yaahh\gdelt; python ".\gdelt_10c_pipeline.py" "filter-gkg" --start 20260101 --end 20260801 # 7. GKG 按月分桶（< 1 分钟） cd C:\Users\yaahh\gdelt; python ".\gdelt_10c_pipeline.py" "merge gkg" 20260101 20260801 ``` ```` ### 前端页面结构 · `frontend/index.html` ````html GDELT 双国事件分析平台
GDELT 双国事件分析平台
正在加载 DuckDB-WASM ...
🇺🇸 USA 事件总数
—
2026-01-01 ~ 2026-08-01
🇨🇳 CHINA 事件总数
—
2026-01-01 ~ 2026-08-01
📰 GKG 主题过滤
—
主题词 ∩ Tone ∩ 媒体
💾 磁盘占用
—
包含 gkg_raw 中间产物

📊 数据总览

数据库已加载 8.94M 行 USA 事件 + 0.74M CHINA 事件 + 670K GKG 过滤新闻。下方任选国家 + 时间范围即可查看分布。

按 GoldsteinScale 分布（合作性 vs 冲突性）

数值越高越合作（=10 是称赞），越低越冲突（=-10 是武力攻击）

按 AvgTone 调性区间

媒体调性：-100 极负 ~ +100 极正

按 EventCode 事件类型 TOP 10

按 Actor1Code 施动者角色 TOP 10

🔍 事件溯源 · 点击 EventCode 柱状查看具体新闻

点击上方“按 EventCode 事件类型 TOP 10”任意一条柱状 → 下表格加载该 EventCode 下前 20 条新闻源头。来源：GDELT 原始 SOURCEURL。

等待选择 EventCode ...
日期施动者对象EventCode调性来源 URL

📅 月度趋势

月度事件量 + 调性走势，看新闻热度趋势。

每月精确数据

月份事件数平均调性平均提及数平均文章数

🇺🇸🇨🇳 双国对比

USA vs CHINA 全期对比（事件数 / 调性 / 事件类型 / Actor1 角色）。

事件总量对比

平均调性对比

USA Top 10 事件类型

CHINA Top 10 事件类型

🏷️ 主题词过滤

从 GKG 主题词里选感兴趣的，看相关事件（已应用 Tone |tone|≤15 + 媒体白名单前过滤）。

每日命中数

命中新闻样本（最多 50 条）

日期主题词调性关联事件 ID来源 URL

🔍 自由查询

直接写 SQL，可用 read_parquet('/data/...') 读 parquet 内任意列。表名：`event_month`、`gkg_filtered_month`。

查询结果

GDELT 双国事件分析平台 · 2026-08-16 · 数据范围 2026-01-01 ~ 2026-08-01 Powered by DuckDB-WASM + Chart.js
```` ### 前端样式 · `frontend/style.css` ````css * { box-sizing: border-box; margin: 0; padding: 0; } :root { --bg: #f6f8fa; --panel: #ffffff; --line: #d0d7de; --line-soft: #e7ebef; --text: #1f2328; --muted: #656d76; --accent: #0969da; --accent-soft: #ddf4ff; --green: #1a7f37; --red: #cf222e; --warn: #9a6700; --shadow: 0 1px 3px rgba(31,35,40,0.07); --shadow-lg: 0 4px 12px rgba(31,35,40,0.10); } body { font-family: -apple-system, "Segoe UI", "PingFang SC", "Microsoft YaHei", sans-serif; font-size: 14px; background: var(--bg); color: var(--text); line-height: 1.5; } /* === 顶部 === */ .topbar { background: var(--panel); border-bottom: 1px solid var(--line); padding: 12px 28px; display: flex; justify-content: space-between; align-items: center; position: sticky; top: 0; z-index: 10; box-shadow: var(--shadow); } .brand { display: flex; align-items: center; gap: 10px; } .brand .logo { font-size: 22px; } .brand .title { font-size: 17px; font-weight: 600; } .status { display: flex; align-items: center; gap: 8px; font-size: 13px; color: var(--muted); } .dot { width: 8px; height: 8px; border-radius: 50%; } .dot.loading { background: var(--warn); animation: pulse 1.5s infinite; } .dot.ok { background: var(--green); } .dot.err { background: var(--red); } @keyframes pulse { 0% { opacity: 1; } 50% { opacity: 0.4; } 100% { opacity: 1; } } /* === 容器 === */ .container { max-width: 1280px; margin: 0 auto; padding: 24px 28px 80px; } /* === 指标卡 === */ .metrics { display: grid; grid-template-columns: repeat(4, 1fr); gap: 14px; margin-bottom: 24px; } .metric-card { background: var(--panel); border: 1px solid var(--line); border-radius: 8px; padding: 16px 20px; box-shadow: var(--shadow); } .metric-card .label { font-size: 13px; color: var(--muted); margin-bottom: 6px; } .metric-card .value { font-size: 26px; font-weight: 600; color: var(--accent); } .metric-card .sub { font-size: 11px; color: var(--muted); margin-top: 4px; } /* === 选项卡 === */ .toolbar { margin-bottom: 18px; } .tab-group { display: flex; gap: 4px; flex-wrap: wrap; border-bottom: 1px solid var(--line); padding-bottom: 0; } .tab { background: none; border: none; padding: 10px 16px; font-size: 14px; color: var(--muted); cursor: pointer; border-bottom: 2px solid transparent; margin-bottom: -1px; transition: color 0.15s, border-color 0.15s; } .tab:hover { color: var(--text); } .tab.active { color: var(--accent); border-bottom-color: var(--accent); font-weight: 600; } /* === 面板 === */ .panel { display: none; background: var(--panel); border: 1px solid var(--line); border-radius: 8px; padding: 22px 28px; box-shadow: var(--shadow); } .panel.active { display: block; } .panel h2 { font-size: 18px; margin-bottom: 6px; } .panel h3 { font-size: 14px; margin-bottom: 12px; font-weight: 600; } .hint { color: var(--muted); font-size: 13px; margin-bottom: 16px; } /* === 过滤器 === */ .filters { display: flex; gap: 16px; flex-wrap: wrap; align-items: flex-end; margin-bottom: 18px; } .filters label { display: flex; flex-direction: column; font-size: 12px; color: var(--muted); gap: 4px; } .filters select, .filters textarea { background: var(--panel); border: 1px solid var(--line); border-radius: 6px; padding: 6px 10px; font-size: 13px; color: var(--text); font-family: inherit; min-width: 140px; } .filters select:focus, .filters textarea:focus { outline: none; border-color: var(--accent); box-shadow: 0 0 0 3px var(--accent-soft); } /* === 按钮 === */ .btn { background: var(--panel); border: 1px solid var(--line); border-radius: 6px; padding: 7px 16px; font-size: 13px; cursor: pointer; color: var(--text); font-family: inherit; transition: background 0.15s, border-color 0.15s; } .btn:hover { background: #f3f4f6; border-color: #c0c4cc; } .btn.primary { background: var(--accent); color: #fff; border-color: var(--accent); } .btn.primary:hover { background: #0860c7; border-color: #0860c7; } .btn:disabled { opacity: 0.5; cursor: not-allowed; } /* === 结果 === */ .result-grid { display: grid; grid-template-columns: repeat(2, 1fr); gap: 16px; } .result-card { background: var(--panel); border: 1px solid var(--line-soft); border-radius: 6px; padding: 16px 20px; margin-bottom: 16px; } .legend-hint { font-size: 11px; color: var(--muted); margin-top: 8px; text-align: center; } /* === 表格 === */ .data-table { width: 100%; border-collapse: collapse; font-size: 13px; font-variant-numeric: tabular-nums; } .data-table th, .data-table td { padding: 8px 12px; text-align: left; border-bottom: 1px solid var(--line-soft); } .data-table th { background: #f6f8fa; color: var(--muted); font-weight: 500; font-size: 12px; text-transform: uppercase; letter-spacing: 0.02em; position: sticky; top: 0; } .data-table td.num { text-align: right; } .data-table tr:hover td { background: #f6f8fa; } .data-table .empty { text-align: center; color: var(--muted); padding: 24px; } /* === 自由查询 === */ #sql-input { width: 100%; border: 1px solid var(--line); border-radius: 6px; padding: 12px; font-family: "JetBrains Mono", "Consolas", monospace; font-size: 13px; margin-bottom: 12px; resize: vertical; } /* === 响应式 === */ @media (max-width: 900px) { .metrics, .result-grid { grid-template-columns: 1fr; } } /* === 底部 === */ .footer { background: var(--panel); border-top: 1px solid var(--line); padding: 14px 28px; display: flex; justify-content: space-between; font-size: 12px; color: var(--muted); position: fixed; bottom: 0; left: 0; right: 0; } /* === 事件溯源 === */ .trace-panel { margin-top: 24px; padding: 16px 20px; background: var(--panel); border: 1px solid var(--line); border-radius: 8px; } .trace-summary { font-size: 13px; color: var(--text); background: #f6f8fa; border: 1px solid var(--line); border-radius: 6px; padding: 8px 12px; margin: 8px 0 12px; } .src-link { color: var(--accent, #0969da); text-decoration: none; word-break: break-all; font-size: 12px; } .src-link:hover { text-decoration: underline; } .code { font-family: "JetBrains Mono", Consolas, monospace; font-weight: 600; color: var(--accent, #0969da); } .zh { color: var(--muted, #57606a); font-size: 12px; } ```` ### 前端业务逻辑 · `frontend/app.js` ````javascript // GDELT 前端 · DuckDB-WASM + Chart.js // 用法: 用本地 HTTP 服务（如 python -m http.server 8765）打开 // === ES module import (DuckDB-WASM 必须用 import, 不能用全局变量) === import * as duckdb from './vendor/duckdb-browser.mjs'; import { translateCAMEO, translateCAMEOFull } from './cameo_zh.js'; // === 全局 === const $ = (id) => document.getElementById(id); const statusEl = (text, kind) => { const tag = $('statusText'); const dot = $('status').querySelector('.dot'); tag.textContent = text; dot.className = 'dot ' + kind; }; let conn = null; let csvBuf = null; // === DuckDB-WASM 初始化（offline vendor）=== async function initDuckDB() { statusEl('加载 DuckDB-WASM ...', 'loading'); // EH (Exception Handling) bundle 比 MVP 更稳: 能正确抛 SQL 异常 const mainWorkerUrl = new URL('vendor/duckdb-browser-eh.worker.js', window.location.href).href; const mainModuleUrl = new URL('vendor/duckdb-eh.wasm', window.location.href).href; const logger = new duckdb.VoidLogger(); const worker = new Worker(mainWorkerUrl); const db = new duckdb.AsyncDuckDB(logger, worker); await db.instantiate(mainModuleUrl); conn = await db.connect(); statusEl('注册 Parquet ...', 'loading'); await registerParquet(db); statusEl('就绪 · ' + new Date().toLocaleTimeString(), 'ok'); } async function registerParquet(db) { // 走 file-list.json (预先生成的 32 个 parquet 路径) // 避免依赖 Python http.server 的目录索引功能 const files = await fetch('file-list.json').then(r => r.json()); console.log(`[前端] 加载 ${files.length} 个 parquet`); let loaded = 0; for (const f of files) { const fp = await fetch(f); const buf = await fp.arrayBuffer(); // 用正斜杠注册到 DuckDB 虚拟文件系统 const vpath = '/data/' + f.replace(/\\/g, '/'); await db.registerFileBuffer(vpath, new Uint8Array(buf)); loaded++; if (loaded % 8 === 0) { statusEl(`加载 parquet ${loaded}/${files.length} ...`, 'loading'); } } console.log(`[前端] parquet 注册完成: ${loaded} 个文件`); // DuckDB glob 只支持单层 *, 不支持 ** // 我们有 month=YYYYMM/country_code=CC/file.parquet 三层, 用 list + UNION ALL const evFiles = files.filter(f => f.includes('/event_month/')); const gkgFiles = files.filter(f => f.includes('/gkg_filtered_month/')); const evUnion = evFiles.map(f => `'${'/data/' + f.replace(/\\/g, '/')}'`).join(',\n '); const gkgUnion = gkgFiles.map(f => `'${'/data/' + f.replace(/\\/g, '/')}'`).join(',\n '); // slim tier 都是 VARCHAR, 查询时需 TRY_CAST. 视图里预先转好常用数值列 await conn.query(`CREATE VIEW event_month AS SELECT * FROM (SELECT TRY_CAST(GLOBALEVENTID AS BIGINT) AS GLOBALEVENTID, SQLDATE, Actor1Code, Actor2Code, EventCode, TRY_CAST(GoldsteinScale AS DOUBLE) AS GoldsteinScale, TRY_CAST(NumMentions AS INTEGER) AS NumMentions, TRY_CAST(AvgTone AS DOUBLE) AS AvgTone, ActionGeo_CountryCode, SOURCEURL, matched_via, TRY_CAST(dt AS INTEGER) AS dt, country_code, TRY_CAST(month AS INTEGER) AS month FROM read_parquet([${evUnion}]))`); await conn.query(`CREATE VIEW gkg_filtered_month AS SELECT DATE, NUMARTS, THEMES, TONE, CAMEOEVENTIDS, SOURCEURLS, country_code FROM read_parquet([${gkgUnion}])`); } // === 工具：从 query 取列 + 值 === async function q(sql) { const t0 = performance.now(); const result = await conn.query(sql); const rawRows = result.toArray(); const cols = result.schema.fields.map(f => f.name); // DuckDB BigInt 不能 JSON.stringify, 手动转成 Number const rows = rawRows.map(r => { const o = {}; for (const c of cols) { let v = r[c]; if (typeof v === 'bigint') v = Number(v); o[c] = v; } return o; }); const t = (performance.now() - t0).toFixed(0); return { cols, rows, ms: Number(t) }; } function renderTable(tbody, cols, rows, max = 50) { tbody.innerHTML = ''; const cap = rows.slice(0, max); if (cap.length === 0) { tbody.innerHTML = `无数据`; return; } // 判断哪些列是 URL: SOURCEURL (单数, event 表), SOURCEURLS (复数, GKG 表, 多 URL 分号分隔) const urlCols = new Set(); cols.forEach(c => { if (/^SOURCEURL(S)?$/i.test(c)) urlCols.add(c); }); for (const r of cap) { const tr = document.createElement('tr'); for (const c of cols) { const td = document.createElement('td'); let v = r[c]; // EventCode 列特殊处理: 显示代码 + 中文 if (c === 'EventCode' && v != null && v !== '') { td.innerHTML = `${v} ${translateCAMEO(v)}`; } else if (urlCols.has(c) && v != null && v !== '') { // URL 列: 渲染为可点击链接 const urls = String(v).split(';').map(s => s.trim()).filter(s => s); td.innerHTML = urls.slice(0, 3).map(u => { // 安全: 截断到 60 字符作 link 文本 const text = u.length > 60 ? u.slice(0, 57) + '...' : u; return `↗ ${escapeHtml(text)}`; }).join('
'); } else if (typeof v === 'number') { td.classList.add('num'); td.textContent = Number.isInteger(v) ? v.toLocaleString() : v.toFixed(2); } else { td.textContent = v == null ? '' : String(v); } tr.appendChild(td); } tbody.appendChild(tr); } } function escapeHtml(s) { return String(s).replace(/&/g, '&').replace(//g, '>').replace(/"/g, '"'); } function destroyChart(id) { const old = Chart.getChart(id); if (old) old.destroy(); } // === 指标卡 === async function loadMetrics() { const { rows: usa } = await q(`SELECT COUNT(*) AS n FROM event_month WHERE country_code='USA' AND SQLDATE >= '20260101'`); const { rows: chn } = await q(`SELECT COUNT(*) AS n FROM event_month WHERE country_code='CHINA' AND SQLDATE >= '20260101'`); const { rows: gkg } = await q(`SELECT COUNT(*) AS n FROM gkg_filtered_month`); $('m-usa').textContent = Number(usa[0].n).toLocaleString(); $('m-chn').textContent = Number(chn[0].n).toLocaleString(); $('m-gkg').textContent = Number(gkg[0].n).toLocaleString(); $('m-disk').textContent = '1.13 GB'; $('m-usa-sub').textContent = '含 2026 历史 + 7 月增量'; $('m-chn-sub').textContent = 'USA:CHINA ≈ 11:1'; } // === 总览面板 === let charts = {}; // 事件溯源上下文: 记录当前筛选条件, 点击 EventCode 柱时用 let traceCtx = { src: 'event', cc: 'USA', start: '202601', end: '202608' }; async function runOverview() { const src = $('src-overview').value; const cc = $('country-overview').value; const start = $('start-overview').value; const end = $('end-overview').value; traceCtx = { src, cc, start, end }; const table = src === 'event' ? 'event_month' : 'gkg_filtered_month'; const dateCol = src === 'event' ? "SUBSTR(SQLDATE,1,6)" : "SUBSTR(DATE,1,6)"; const baseWhere = `country_code='${cc}' AND ${dateCol} BETWEEN '${start}' AND '${end}'`; const toneCol = src === 'event' ? 'AvgTone' : "TRY_CAST(regexp_extract(coalesce(TONE,''), '^(-?\\d+(?:\\.\\d+)?)', 1) AS DOUBLE)"; const q1 = await q(`SELECT GoldsteinScale, COUNT(*) AS n FROM ${table} WHERE ${baseWhere} GROUP BY GoldsteinScale ORDER BY GoldsteinScale`); const q2 = await q(`SELECT CASE WHEN ${toneCol} < -20 THEN '极负 (<-20)' WHEN ${toneCol} < -5 THEN '偏负 (-20~-5)' WHEN ${toneCol} <= 5 THEN '中性 (-5~+5)' WHEN ${toneCol} <= 20 THEN '偏正 (+5~+20)' ELSE '极正 (>+20)' END AS bucket, COUNT(*) AS n FROM ${table} WHERE ${baseWhere} GROUP BY bucket ORDER BY MIN(${toneCol})`); const q3 = await q(`SELECT EventCode, COUNT(*) AS n FROM ${table} WHERE ${baseWhere} GROUP BY EventCode ORDER BY n DESC LIMIT 10`); const q4 = await q(`SELECT Actor1Code, COUNT(*) AS n FROM ${table} WHERE ${baseWhere} GROUP BY Actor1Code ORDER BY n DESC LIMIT 10`); // Goldstein 柱状 destroyChart('chart-goldstein'); charts.goldstein = new Chart($('chart-goldstein'), { type: 'bar', data: { labels: q1.rows.map(r => r.GoldsteinScale), datasets: [{ label: '事件数', data: q1.rows.map(r => r.n), backgroundColor: '#0969da' }] }, options: { responsive: true, plugins: { legend: { display: false } }, scales: { y: { beginAtZero: true } } }, }); // Tone 饼图 destroyChart('chart-tone'); const toneColors = { '极负 (<-20)': '#cf222e', '偏负 (-20~-5)': '#fb8500', '中性 (-5~+5)': '#9aa6b2', '偏正 (+5~+20)': '#3fb950', '极正 (>+20)': '#1a7f37' }; charts.tone = new Chart($('chart-tone'), { type: 'doughnut', data: { labels: q2.rows.map(r => r.bucket), datasets: [{ data: q2.rows.map(r => r.n), backgroundColor: q2.rows.map(r => toneColors[r.bucket] || '#9aa6b2') }] }, options: { responsive: true, plugins: { legend: { position: 'right' } } }, }); // EventCode - 翻译成中文标签, 点击柱状可溯源到新闻 destroyChart('chart-eventcode'); charts.eventcode = new Chart($('chart-eventcode'), { type: 'bar', data: { labels: q3.rows.map(r => translateCAMEOFull(r.EventCode)), datasets: [{ label: '事件数', data: q3.rows.map(r => r.n), backgroundColor: '#1a7f37' }] }, options: { indexAxis: 'y', responsive: true, plugins: { legend: { display: false } }, scales: { x: { beginAtZero: true } }, onClick: (evt, els) => { if (els.length === 0) return; const i = els[0].index; const code = q3.rows[i].EventCode; loadTrace(code); }, }, }); // Actor1 destroyChart('chart-actor'); charts.actor = new Chart($('chart-actor'), { type: 'bar', data: { labels: q4.rows.map(r => r.Actor1Code), datasets: [{ label: '事件数', data: q4.rows.map(r => r.n), backgroundColor: '#9a6700' }] }, options: { indexAxis: 'y', responsive: true, plugins: { legend: { display: false } }, scales: { x: { beginAtZero: true } } }, }); } // === 事件溯源: 点击 EventCode 柱状 → 查该码下具体新闻源头 === async function loadTrace(code) { const { cc, start, end } = traceCtx; const summary = $('trace-summary'); const tbody = $('tbl-trace').querySelector('tbody'); summary.textContent = `加载 ${code} · ${translateCAMEOFull(code)} 的事件 ...`; try { const { rows } = await q(`SELECT SQLDATE, Actor1Code, Actor2Code, EventCode, ROUND(AvgTone, 2) AS AvgTone, SOURCEURL FROM event_month WHERE country_code='${cc}' AND EventCode='${code}' AND SUBSTR(SQLDATE,1,6) BETWEEN '${start}' AND '${end}' ORDER BY NumMentions DESC LIMIT 20`); const cols = ['SQLDATE', 'Actor1Code', 'Actor2Code', 'EventCode', 'AvgTone', 'SOURCEURL']; renderTable(tbody, cols, rows, 20); summary.textContent = `${translateCAMEOFull(code)} — 按提及量排序的前 ${rows.length} 条新闻 (${cc}, ${start}~${end})。点击链接可打开原文。`; // 提示 if (rows.length === 0) summary.textContent = `${code} 在该筛选条件下无明细行。`; } catch (e) { tbody.innerHTML = `查询失败: ${escapeHtml(e.message.slice(0, 80))}`; summary.textContent = `溯源失败: ${escapeHtml(e.message.slice(0, 60))}`; } } // === 月度趋势 === async function runMonth() { const cc = $('country-month').value; let rows, cols, ms; try { ({ rows, cols, ms } = await q(`SELECT SUBSTR(SQLDATE, 1, 6) AS month, COUNT(*) AS events, ROUND(AVG(AvgTone), 2) AS avg_tone, ROUND(AVG(NumMentions), 1) AS avg_mentions FROM event_month WHERE country_code='${cc}' GROUP BY 1 ORDER BY 1`)); } catch (e) { statusEl('月度查询失败: ' + e.message.slice(0, 50), 'err'); return; } renderTable($('tbl-monthly').querySelector('tbody'), cols, rows, 100); destroyChart('chart-monthly'); charts.monthly = new Chart($('chart-monthly'), { data: { labels: rows.map(r => r.month), datasets: [ { type: 'bar', label: '事件数', data: rows.map(r => r.events), backgroundColor: '#0969da', yAxisID: 'y' }, { type: 'line', label: '平均调性', data: rows.map(r => r.avg_tone), borderColor: '#cf222e', backgroundColor: '#cf222e', yAxisID: 'y1' }, ], }, options: { responsive: true, scales: { y: { type: 'linear', position: 'left', title: { display: true, text: '事件数' } }, y1: { type: 'linear', position: 'right', title: { display: true, text: '调性' }, grid: { drawOnChartArea: false } }, }, }, }); } // === 双国对比 === async function runCountry() { const a = await q(`SELECT COUNT(*) AS n, ROUND(AVG(AvgTone),2) AS t FROM event_month WHERE country_code='USA'`); const b = await q(`SELECT COUNT(*) AS n, ROUND(AVG(AvgTone),2) AS t FROM event_month WHERE country_code='CHINA'`); const usaE = await q(`SELECT EventCode, COUNT(*) AS n FROM event_month WHERE country_code='USA' GROUP BY EventCode ORDER BY n DESC LIMIT 10`); const chnE = await q(`SELECT EventCode, COUNT(*) AS n FROM event_month WHERE country_code='CHINA' GROUP BY EventCode ORDER BY n DESC LIMIT 10`); destroyChart('chart-total-compare'); charts.total = new Chart($('chart-total-compare'), { type: 'bar', data: { labels: ['USA', 'CHINA'], datasets: [{ label: '总事件数', data: [a.rows[0].n, b.rows[0].n], backgroundColor: ['#0969da', '#cf222e'] }] }, options: { plugins: { legend: { display: false } } }, }); destroyChart('chart-tone-compare'); charts.toneC = new Chart($('chart-tone-compare'), { type: 'bar', data: { labels: ['USA', 'CHINA'], datasets: [{ label: '平均调性', data: [a.rows[0].t, b.rows[0].t], backgroundColor: ['#0969da', '#cf222e'] }] }, options: { plugins: { legend: { display: false } }, scales: { y: { suggestedMin: -10, suggestedMax: 10 } } }, }); destroyChart('chart-usa-events'); charts.usaE = new Chart($('chart-usa-events'), { type: 'bar', data: { labels: usaE.rows.map(r => translateCAMEOFull(r.EventCode)), datasets: [{ label: 'USA', data: usaE.rows.map(r => r.n), backgroundColor: '#0969da' }] }, options: { indexAxis: 'y', plugins: { legend: { display: false } } }, }); destroyChart('chart-chn-events'); charts.chnE = new Chart($('chart-chn-events'), { type: 'bar', data: { labels: chnE.rows.map(r => translateCAMEOFull(r.EventCode)), datasets: [{ label: 'CHINA', data: chnE.rows.map(r => r.n), backgroundColor: '#cf222e' }] }, options: { indexAxis: 'y', plugins: { legend: { display: false } } }, }); } // === 主题词过滤 === async function runTheme() { const cc = $('country-theme').value; const selected = [...$('theme-keyword').selectedOptions].map(o => o.value); if (selected.length === 0) { alert('请至少选一个主题词'); return; } const pat = selected.join('|'); const { rows: daily } = await q(`SELECT SUBSTR(DATE,1,6) AS month, COUNT(*) AS n FROM gkg_filtered_month WHERE country_code='${cc}' AND regexp_matches(THEMES, '(?i)(${pat})') GROUP BY month ORDER BY month`); const { rows: sample, cols } = await q(`SELECT DATE, THEMES, TONE, SOURCEURLS, CAMEOEVENTIDS FROM gkg_filtered_month WHERE country_code='${cc}' AND regexp_matches(THEMES, '(?i)(${pat})') LIMIT 50`); destroyChart('chart-theme-daily'); charts.theme = new Chart($('chart-theme-daily'), { type: 'bar', data: { labels: daily.map(r => r.month), datasets: [{ label: '命中数', data: daily.map(r => r.n), backgroundColor: '#9a6700' }] }, options: { plugins: { legend: { display: false } } }, }); // 表格：压缩 THEMES / SOURCEURLS 截断 const trunc = sample.map(r => ({ DATE: r.DATE, THEMES: (r.THEMES || '').slice(0, 80), TONE: r.TONE, SOURCEURLS: (r.SOURCEURLS || '').split(';')[0].slice(0, 80), })); renderTable($('tbl-theme').querySelector('tbody'), cols, trunc, 50); } // === 自由查询 === async function runSql() { const sql = $('sql-input').value.trim(); if (!sql) return; try { const t0 = performance.now(); const { rows, cols, ms } = await q(sql); const total = (performance.now() - t0).toFixed(0); renderTable($('tbl-custom').querySelector('tbody'), cols, rows, 200); $('sql-time').textContent = `${rows.length} 行 · query ${ms}ms · total ${total}ms`; csvBuf = { cols, rows }; } catch (e) { $('tbl-custom').querySelector('tbody').innerHTML = `SQL 错误: ${e.message}`; $('sql-time').textContent = ''; csvBuf = null; } } function exportCsv() { if (!csvBuf) { alert('请先执行查询'); return; } const { cols, rows } = csvBuf; const esc = (v) => { if (v == null) return ''; const s = String(v); return /[",\n]/.test(s) ? `"${s.replace(/"/g, '""')}"` : s; }; const csv = [cols.join(','), ...rows.map(r => cols.map(c => esc(r[c])).join(','))].join('\n'); const blob = new Blob([csv], { type: 'text/csv;charset=utf-8;' }); const url = URL.createObjectURL(blob); const a = document.createElement('a'); a.href = url; a.download = `gdelt_query_${Date.now()}.csv`; a.click(); URL.revokeObjectURL(url); } // === 标签切换 === function setupTabs() { document.querySelectorAll('.tab').forEach(btn => { btn.addEventListener('click', () => { document.querySelectorAll('.tab').forEach(b => b.classList.remove('active')); document.querySelectorAll('.panel').forEach(p => p.classList.remove('active')); btn.classList.add('active'); const id = 'tab-' + btn.dataset.tab; $(id).classList.add('active'); }); }); } // === 启动 === (async () => { setupTabs(); $('run-overview').addEventListener('click', runOverview); $('run-month').addEventListener('click', runMonth); $('run-country').addEventListener('click', runCountry); $('run-theme').addEventListener('click', runTheme); $('run-sql').addEventListener('click', runSql); $('export-sql').addEventListener('click', exportCsv); try { await initDuckDB(); await loadMetrics(); await runOverview(); // 默认跑一次 } catch (e) { statusEl('错误: ' + e.message, 'err'); console.error(e); } })(); ```` ### 事件码中文翻译映射 · `frontend/cameo_zh.js` ````javascript // CAMEO 事件码中文翻译 // 顶级 20 类: 2 位码 // 二级 3 位 / 三级 4 位为细类 // 数据源: github.com/tenthe/CAMEO-Event-Data-Codebook (CSV) + 学术通用译法 export const CAMEO_ZH = { _top: { "01": "公开声明", "02": "请求/呼吁", "03": "表达合作意向", "04": "协商/咨询", "05": "开展外交合作", "06": "开展实质性合作", "07": "提供援助", "08": "让步/放弃", "09": "调查", "10": "要求/施压", "11": "不赞同/反对", "12": "拒绝", "13": "威胁", "14": "抗议", "15": "展示武力姿态", "16": "降低/断绝关系", "17": "胁迫", "18": "攻击(物理)", "19": "武装冲突", "20": "非常规大规模暴力", }, _leaf: { "010": "发表声明", "011": "拒绝评论", "012": "发表悲观评论", "013": "发表乐观评论", "014": "考虑政策选项", "015": "承认或宣称负责", "016": "否认责任", "017": "进行象征性行为", "018": "表达同情/共情", "019": "表达赞同", "020": "提出请求", "021": "请求物质合作", "0211": "请求经济合作", "0212": "请求军事合作", "0213": "请求司法合作", "0214": "请求情报合作", "022": "请求外交合作", "023": "请求援助", "0231": "请求经济援助", "0232": "请求军事援助", "0233": "请求人道援助", "0234": "请求民事援助", "024": "请求和平解决", "025": "请求停止/退出", "026": "请求召开会议", "027": "请求调解", "028": "请求保护", "029": "请求调查", "030": "表达合作意向", "031": "表达经济合作意向", "032": "表达军事合作意向", "033": "表达司法合作意向", "034": "表达情报合作意向", "035": "表达文化交流意向", "040": "协商", "041": "讨论政策议题", "042": "讨论双边谈判", "043": "讨论国际议题", "050": "开展外交合作", "051": "签署协议", "052": "批准/认可协议", "053": "签订合作协议", "054": "撤回否决/终止敌对", "060": "开展物质合作", "061": "开展经济合作", "062": "开展军事合作", "063": "开展司法合作", "064": "开展情报合作", "065": "开展文化交流合作", "066": "开展科技合作", "070": "提供援助", "071": "提供经济援助", "072": "提供军事援助", "073": "提供人道援助", "074": "提供民事援助", "075": "提供医疗援助", "080": "让步", "081": "政治让步", "082": "经济让步", "083": "军事让步", "084": "领土让步", "085": "司法让步", "086": "文化交流让步", "090": "开展调查", "091": "调查犯罪行为", "092": "调查人权侵犯", "093": "调查军事行动", "094": "调查腐败", "095": "调查恐怖主义", "100": "提出要求", "101": "要求经济让步", "102": "要求军事让步", "103": "要求政治让步", "104": "要求领土让步", "105": "要求停止援助", "106": "要求停止敌对", "107": "要求撤军", "108": "要求释放人质", "109": "要求解释/澄清", "11": "反对", "110": "反对现状", "111": "批评/谴责", "112": "抗议政策", "113": "抗议行动", "12": "拒绝", "120": "拒绝让步", "121": "拒绝经济合作", "122": "拒绝军事合作", "123": "拒绝司法合作", "124": "拒绝援助", "125": "拒绝让步(总)", "126": "拒绝撤回否决/敌对", "127": "拒绝召开会议", "128": "拒绝回应/沟通", "129": "拒绝仲裁/裁决", "13": "威胁", "130": "发出威胁", "131": "威胁经济制裁", "1311": "威胁禁运", "1312": "威胁贸易限制", "1313": "威胁取消援助", "132": "威胁军事行动", "1321": "威胁边境封锁", "1322": "威胁占领", "1323": "威胁空中打击", "1324": "威胁核打击", "1325": "威胁入侵", "1326": "威胁海上封锁", "133": "威胁断绝外交关系", "134": "威胁限制人身自由", "135": "威胁恐怖行动", "136": "威胁使用大规模杀伤武器", "137": "威胁生态破坏", "138": "威胁其他惩罚措施", "14": "抗议", "140": "发起抗议", "141": "游行示威", "142": "集会抗议", "143": "罢工抗议", "144": "绝食抗议", "145": "携带武器抗议", "15": "展示武力姿态", "150": "展示军事力量", "151": "提高警戒级别", "152": "进入戒备状态", "153": "动员武装力量", "154": "增加军费", "155": "军事演习", "16": "降低关系", "160": "降低外交关系", "161": "驱逐外交官", "162": "召回大使", "163": "关闭领事馆", "164": "中止外交关系", "165": "中止双边交流", "17": "胁迫", "170": "实施胁迫", "171": "实施经济胁迫", "1711": "实施禁运", "1712": "实施贸易限制", "1713": "取消援助", "172": "实施军事胁迫", "1721": "封锁边境", "1722": "实施占领", "1723": "实施空中打击", "1724": "实施核打击", "1725": "实施入侵", "1726": "实施海上封锁", "173": "断绝外交关系", "174": "实施拘禁", "175": "实施绑架", "176": "实施酷刑", "177": "实施恐怖行动", "178": "使用大规模杀伤武器", "18": "攻击", "180": "实施攻击", "181": "使用非致命武器攻击", "182": "占领领土", "1821": "短暂占领", "1822": "长期占领", "1823": "建立军事基地", "183": "实施空袭", "1831": "轰炸", "1832": "导弹打击", "1833": "无人机袭击", "184": "实施绑架", "185": "劫持人质", "186": "实施酷刑", "187": "实施处决", "19": "武装冲突", "190": "使用常规武器", "191": "小规模武装冲突", "192": "中等规模武装冲突", "193": "大规模武装冲突", "194": "海上武装冲突", "195": "空中武装冲突", "196": "停火", "20": "非常规大规模暴力", "200": "使用非常规暴力", "201": "实施自杀式袭击", "202": "实施炸弹袭击", "203": "使用化学武器", "204": "使用生物武器", "2041": "使用神经毒剂", "2042": "使用化学武器(其他)", "2043": "使用催泪瓦斯", "205": "使用核武器", "2051": "核爆炸", "2052": "核辐射污染", "206": "使用放射性武器", "2061": "辐射暴露", "2062": "辐射毒害", }, }; // 翻译 EventCode -> 中文 // 优先匹配 4 位码, 否则截短到 3 位, 再截短到 2 位 (顶级) export function translateCAMEO(code) { if (!code) return ''; const c = String(code).trim(); // 完整匹配 if (CAMEO_ZH._leaf[c]) return CAMEO_ZH._leaf[c]; // 4 位码无匹配, 取前 3 位 if (c.length >= 4 && CAMEO_ZH._leaf[c.slice(0, 3)]) return CAMEO_ZH._leaf[c.slice(0, 3)]; // 取前 2 位顶级类 if (c.length >= 2 && CAMEO_ZH._top[c.slice(0, 2)]) return CAMEO_ZH._top[c.slice(0, 2)]; return c; } // 返回带顶级类的完整描述, 形如 '013 - 公开声明 / 表达赞同' export function translateCAMEOFull(code) { if (!code) return ''; const c = String(code).trim(); const top = c.length >= 2 ? CAMEO_ZH._top[c.slice(0, 2)] : ''; let leaf = ''; if (CAMEO_ZH._leaf[c]) leaf = CAMEO_ZH._leaf[c]; else if (c.length >= 4 && CAMEO_ZH._leaf[c.slice(0, 3)]) leaf = CAMEO_ZH._leaf[c.slice(0, 3)]; if (top && leaf) return `${c} · ${top} / ${leaf}`; if (leaf) return `${c} · ${leaf}`; if (top) return `${c} · ${top}`; return c; } ```` ### Parquet 文件清单 · `frontend/file-list.json` ````json ["/parquet/event_month/month=202601/country_code=CHINA/69f7b87e338c48808862e6d910504775-0.parquet","/parquet/event_month/month=202601/country_code=USA/69f7b87e338c48808862e6d910504775-0.parquet","/parquet/event_month/month=202602/country_code=CHINA/3a69ac25fe6d4555a7977e7188a106f2-0.parquet","/parquet/event_month/month=202602/country_code=USA/3a69ac25fe6d4555a7977e7188a106f2-0.parquet","/parquet/event_month/month=202603/country_code=CHINA/d4188ee989ae4f718cc81db21d10ffe6-0.parquet","/parquet/event_month/month=202603/country_code=USA/d4188ee989ae4f718cc81db21d10ffe6-0.parquet","/parquet/event_month/month=202604/country_code=CHINA/bbd8bebb2b8d4ac9a580ff1895b0ba26-0.parquet","/parquet/event_month/month=202604/country_code=USA/bbd8bebb2b8d4ac9a580ff1895b0ba26-0.parquet","/parquet/event_month/month=202605/country_code=CHINA/5827d9b2be1948a49e90a74f21018fdb-0.parquet","/parquet/event_month/month=202605/country_code=USA/5827d9b2be1948a49e90a74f21018fdb-0.parquet","/parquet/event_month/month=202606/country_code=CHINA/143c96727e3e480d89a76fba5d4eb13f-0.parquet","/parquet/event_month/month=202606/country_code=USA/143c96727e3e480d89a76fba5d4eb13f-0.parquet","/parquet/event_month/month=202607/country_code=CHINA/eaea70451d664e209b37e74ae5f1e8c4-0.parquet","/parquet/event_month/month=202607/country_code=USA/eaea70451d664e209b37e74ae5f1e8c4-0.parquet","/parquet/event_month/month=202608/country_code=CHINA/48a2281fe1eb4461bc37e7fc4863b14c-0.parquet","/parquet/event_month/month=202608/country_code=USA/48a2281fe1eb4461bc37e7fc4863b14c-0.parquet","/parquet/gkg_filtered_month/month=202601/country_code=CHINA/fbac2e7dd0ac4bf291fa71ebd7f623b6-0.parquet","/parquet/gkg_filtered_month/month=202601/country_code=USA/fbac2e7dd0ac4bf291fa71ebd7f623b6-0.parquet","/parquet/gkg_filtered_month/month=202602/country_code=CHINA/409b2dee19e34721952a52769c1f28bc-0.parquet","/parquet/gkg_filtered_month/month=202602/country_code=USA/409b2dee19e34721952a52769c1f28bc-0.parquet","/parquet/gkg_filtered_month/month=202603/country_code=CHINA/f9bb5be831ed49e9b2675446d8a49606-0.parquet","/parquet/gkg_filtered_month/month=202603/country_code=USA/f9bb5be831ed49e9b2675446d8a49606-0.parquet","/parquet/gkg_filtered_month/month=202604/country_code=CHINA/9d9fbd23b1004a5c804f6062fa99743d-0.parquet","/parquet/gkg_filtered_month/month=202604/country_code=USA/9d9fbd23b1004a5c804f6062fa99743d-0.parquet","/parquet/gkg_filtered_month/month=202605/country_code=CHINA/1223cb436b2d410d8dfddf6c18c09bf5-0.parquet","/parquet/gkg_filtered_month/month=202605/country_code=USA/1223cb436b2d410d8dfddf6c18c09bf5-0.parquet","/parquet/gkg_filtered_month/month=202606/country_code=CHINA/7160803db2044ee097d445fe4976fe25-0.parquet","/parquet/gkg_filtered_month/month=202606/country_code=USA/7160803db2044ee097d445fe4976fe25-0.parquet","/parquet/gkg_filtered_month/month=202607/country_code=CHINA/65827186130d49c0a8ff13cb0cc83a67-0.parquet","/parquet/gkg_filtered_month/month=202607/country_code=USA/65827186130d49c0a8ff13cb0cc83a67-0.parquet","/parquet/gkg_filtered_month/month=202608/country_code=CHINA/382fc2368a6943bb8cdc5ffdd352f4f4-0.parquet","/parquet/gkg_filtered_month/month=202608/country_code=USA/382fc2368a6943bb8cdc5ffdd352f4f4-0.parquet"] ```` ### 一键启动脚本 · `frontend/start_frontend.bat` ````bat @echo off REM GDELT Frontend Launcher (pure ASCII, GBK-safe) cd /d C:\Users\yaahh\gdelt REM Kill any stale python on 8765 for /f "tokens=5" %%a in ('netstat -ano ^| findstr "127.0.0.1:8765.*LISTENING"') do ( echo [Cleanup] killing stale python PID %%a taskkill /F /PID %%a >nul 2>&1 ) REM Launch python http.server in background. /B = no parent linkage. echo [Launch] python http.server on 8765 ... start "GDELT-Frontend-Server" /B python -m http.server 8765 --bind 127.0.0.1 REM Wait 3 seconds for it to bind timeout /t 3 /nobreak >nul REM Open default browser first (whether or not netstat check below passes) start "" http://localhost:8765/frontend/ echo. echo ============================================================ echo GDELT Frontend Launcher echo URL: http://localhost:8765/frontend/ echo. echo Browser should open in 1-2 seconds. echo If it shows "connection refused", wait 3 seconds and echo refresh, the server might still be starting. echo. echo You may close this cmd window. echo Server keeps running. Stop with Task Manager - python.exe echo ============================================================ echo. pause ```` ### 前端使用手册 · `frontend/README.md` ````markdown # GDELT 前端使用手册 ## 启动方法 **双击 `C:\Users\yaahh\gdelt\frontend\start_frontend.bat`** 会自动: 1. 在 127.0.0.1:8765 启动 Python HTTP 服务(隐藏窗口) 2. 浏览器打开 http://localhost:8765/frontend/ 关闭:在任务管理器结束 `python.exe`(即 `python -m http.server 8765`)。 ## 5 个标签页 ### 📊 总览 顶部 4 个指标卡（USA / CHINA / GKG / 磁盘）+ 4 个图表： - GoldsteinScale 分布（合作性 vs 冲突性） - AvgTone 调性饼图 - EventCode 事件类型 TOP 10（已翻译成中文） - Actor1Code 施动者角色 TOP 10 可筛：数据源 / 国家 / 起始月 / 结束月。 ### 🔍 事件溯源（新增） **点击 EventCode 图表任意一根柱状 → 下方表格列出该事件码下按提及量排序的前 20 条新闻**，每条含：日期、施动者、对象、调性、来源 URL（可直接点开看原文）。 数据来自 GDELT 原始 SOURCEURL 字段，是每条事件的新闻出处。 ### 📅 月度趋势 折线 + 柱状图,看每月的: - 事件数总量 - 平均调性 - 平均提及数 - 平均文章数 ### 🇺🇸🇨🇳 双国对比 USA vs CHINA 全期对比: - 事件总量 - 平均调性 - 各自 TOP 10 事件类型 ### 🏷️ 主题词过滤 从 30 个主题词里多选(PROTEST / TAIWAN / TRADE / AI / CYBER 等),看: - 每月命中数 - 命中新闻样本(最多 50 条) ### 🔍 自由查询 直接写 SQL，前端自动 fetch 头两行作为表头，表格渲染结果。默认 SQL 已换成“溯源示例”：查某事件码下带 SOURCEURL 的明细行。 **示例 SQL**： ```sql -- 溯源某事件码的新闻源头（点击链接看原文） SELECT SQLDATE, Actor1Code, Actor2Code, EventCode, ROUND(AvgTone, 2) AS AvgTone, SOURCEURL FROM event_month WHERE country_code = 'USA' AND EventCode = '013' AND SUBSTR(SQLDATE,1,6) BETWEEN '202601' AND '202608' ORDER BY NumMentions DESC LIMIT 20 -- 自由查询注意事项: -- 表格里 SOURCEURL / SOURCEURLS 列会自动渲染成可点击链接 ``` 执行后点"导出 CSV"即可下载。 ## 故障排查 | 症状 | 解决 | |---|---| | 浏览器白屏、控制台报 "MIME type" 错误 | 端口被占。`netstat -ano \| findstr 8765` 杀掉旧进程 | | 打开后看不到数据 / 卡在“加载 DuckDB” | 浏览器版本太旧。用 Chrome 90+ 或 Edge 90+ | | 点 EventCode 柱状没反应 | 需先点“查询”跑出总览图表；点击时需点在柱状本体上 | | GKG 主题词面板查询慢 | 第一次会从磁盘读 213 个 GKG 月度文件到内存，第二次会快 | | 想换端口 | 编辑 `start_frontend.bat`,把 8765 改其他端口,浏览器 URL 同步改 | ```` --- ## 附录 A：还原脚本 将下面的代码块保存为 `restore_from_md.py` 后运行：`python restore_from_md.py GDELT_项目完整打包.md` ````python # -*- coding: utf-8 -*- """ restore_from_md.py - 从 GDELT_项目完整打包.md 还原整个项目 用法: python restore_from_md.py GDELT_项目完整打包.md [目标目录] """ import os, re, sys, urllib.request, json MD_MARK = re.compile(r"\n````[a-z]*\n(.*?)\n````", re.S) VENDOR = [ ("https://cdn.jsdelivr.net/npm/@duckdb/duckdb-wasm@1.29.0/dist/duckdb-eh.wasm", "frontend/vendor/duckdb-eh.wasm"), ("https://cdn.jsdelivr.net/npm/@duckdb/duckdb-wasm@1.29.0/dist/duckdb-mvp.wasm", "frontend/vendor/duckdb-mvp.wasm"), ("https://cdn.jsdelivr.net/npm/@duckdb/duckdb-wasm@1.29.0/dist/duckdb-browser.mjs", "frontend/vendor/duckdb-browser.mjs"), ("https://cdn.jsdelivr.net/npm/@duckdb/duckdb-wasm@1.29.0/dist/duckdb-browser-eh.worker.js", "frontend/vendor/duckdb-browser-eh.worker.js"), ("https://cdn.jsdelivr.net/npm/@duckdb/duckdb-wasm@1.29.0/dist/duckdb-browser-mvp.worker.js", "frontend/vendor/duckdb-browser-mvp.worker.js"), ("https://cdn.jsdelivr.net/npm/apache-arrow@17.0.0/+esm", "frontend/vendor/apache-arrow.mjs"), ("https://cdn.jsdelivr.net/npm/tslib@2.6.3/+esm", "frontend/vendor/tslib.mjs"), ("https://cdn.jsdelivr.net/npm/flatbuffers@24.3.25/+esm", "frontend/vendor/flatbuffers.mjs"), ("https://cdn.jsdelivr.net/npm/chart.js@4.4.0/dist/chart.umd.js", "frontend/vendor/chart.umd.js"), ] def extract(md_path): with open(md_path, "r", encoding="utf-8") as f: text = f.read() files = {} for m in MD_MARK.finditer(text): rel, content = m.group(1), m.group(2) files[rel] = content return files def write_files(files, base): for rel, content in files.items(): p = os.path.join(base, rel) os.makedirs(os.path.dirname(p), exist_ok=True) with open(p, "w", encoding="utf-8") as f: f.write(content) print(f" [OK] {rel} ({len(content)} bytes)") def download_vendor(base): print("下载前端依赖 (vendor/) ...") for url, rel in VENDOR: p = os.path.join(base, rel) if os.path.exists(p) and os.path.getsize(p) > 1000: print(f" [SKIP] {rel} 已存在") continue os.makedirs(os.path.dirname(p), exist_ok=True) try: req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) with urllib.request.urlopen(req, timeout=120) as r, open(p, "wb") as f: f.write(r.read()) print(f" [OK] {rel} ({os.path.getsize(p)/1024:.0f} KB)") except Exception as e: print(f" [FAIL] {rel}: {e}") def main(): if len(sys.argv) < 2: print("用法: python restore_from_md.py GDELT_项目完整打包.md [目标目录]") sys.exit(1) md_path = sys.argv[1] base = sys.argv[2] if len(sys.argv) > 2 else "." print(f"从 {md_path} 还原到 {base}/ ...") files = extract(md_path) if "restore_from_md.py" in files: files.pop("restore_from_md.py") # 还原脚本自身不需要写回 print(f"找到 {len(files)} 个源码文件:") write_files(files, base) download_vendor(base) print() print("完成! 接下来:") print(" 1) 数据: 从源机复制 parquet/ 目录, 或运行流水线重新拉取") print(" 2) 启动: 双击 frontend/start_frontend.bat") print(" 3) 打开: http://localhost:8765/frontend/") if __name__ == "__main__": main() ```` ## 附录 B：数据规模参考 | 数据集 | 文件数 | 大小 | 行数 | |---|---|---|---| | event 每日 | 213 | 175.7 MB | 8,938,704 | | event 月度分桶 | 16 | 176.2 MB | 8,938,704 | | gkg_raw（可选） | 250 | 609.5 MB | 5,841,008 | | gkg_filtered 每日 | 213 | 87.0 MB | 670,562 | | gkg_filtered 月度 | 16 | 85.4 MB | 670,562 | | 前端依赖 vendor | 9 | 78.3 MB | — | 删除 `gkg_raw/` 后数据目录约 524 MB；不含原始 CSV 中间文件。 ## 附录 C：常见问题 | 症状 | 解决 | |---|---| | 前端白屏 / MIME 错误 | 端口被占，`netstat -ano | findstr 8765` 杀旧进程 | | 卡在加载 DuckDB | 浏览器过旧，换 Chrome/Edge 90+ | | 前端无数据 | `parquet/` 目录缺失或 `file-list.json` 与文件不匹配 | | 主题词面板慢 | 首次查询需加载全部 GKG 月度文件，属正常 | | 溯源链接打不开 | 媒体下线，GDELT 抓取时有效，属正常 |