JayeshSuryavanshi/GeoSocialX

# GeoSocialX [](https://github.com/JayeshSuryavanshi/GeoSocialX/actions/workflows/ci.yml) [](https://pypi.org/project/geosocialx/) [](https://pypi.org/project/geosocialx/) [](https://github.com/JayeshSuryavanshi/GeoSocialX/blob/main/LICENSE) GeoSocialX 是一个 Python 包，旨在让推文的地理空间分析变得更简单。无论您是社会科学家、数据分析师，还是仅仅对推文的地理空间分布规律感到好奇，GeoSocialX 都能为您所用。  _{输出示例：旧金山带地理标签帖子的密度热力图（来自内置的示例数据），由 `MapVisualizer.to_html_map` 渲染。请参阅[快速入门](#quickstart)来重现此结果。} ## 概述 GeoSocialX 将社交媒体数据与地理空间分析连接起来。它提供了一个便捷的封装，围绕 X API v2（通过 [Tweepy](https://www.tweepy.org/)）来获取特定地理区域内的带地理标签的推文，并将其保存以供进一步分析。 ## 功能特性 - **X API v2 集成：** 使用 `point_radius` 搜索运算符获取特定地理区域（纬度、经度和半径）内的最新推文。地理编码在本地进行验证，因此超出范围的坐标或超过容量的半径会快速失败，从而避免浪费付费的 API 调用。 - **简单的持久化存储：** 将获取的推文保存到换行符分隔的 JSON 文件中，以便进行下游处理。 - **地理空间提取：** 从 v2 推文字典中提取精确的 `(经度, 纬度)` 坐标点，并附带覆盖率报告，显示有多少推文包含精确坐标，而多少推文仅包含地点引用。仅包含地点的推文可以选择性地解析为其地点的边界框中心，并且每个点都会记录其 `source`（`"exact"` 或 `"place"`）。 - **地理空间分析：** 边界框、中心点、大圆（haversine）距离、半径过滤以及轻量级的基于网格的热点查找器——全部纯粹使用标准库实现。 - **可视化：** 将点导出为 GeoJSON（使用标准库），或者渲染带有标记和密度热力图的交互式 Leaflet 地图（可选的 `folium` 额外组件）。 ## 环境要求 - Python 3.10 或更高版本 - 处于**付费层级**（Basic 或更高）的 X Developer 账户——免费层级不支持最近搜索 - X API **bearer token**（对于最近搜索，仅应用级身份验证即可满足要求） ## 依赖项 在 `pyproject.toml` 中声明： - [`tweepy`](https://www.tweepy.org/) `>=4.10` — X API v2 客户端 可选的额外组件： - `example` — [`python-dotenv`](https://pypi.org/project/python-dotenv/)，用于从 `.env` 文件加载凭证（`pip install "geosocialx[example]"`）。 - `maps` — [`folium`](https://python-visualization.github.io/folium/)，用于渲染交互式 Leaflet 地图（`pip install "geosocialx[maps]"`）。导出 GeoJSON 时不需要。 - `test` — [`coverage`](https://pypi.org/project/coverage/)，用于在带有覆盖率测量的情况下运行测试套件（`pip install "geosocialx[test]"`）。 ## 安装 ### 从 PyPI 安装 GeoSocialX 已发布在 PyPI 上，其分发名称为 **`geosocialx`**（导入名也是 `geosocialx`）： ``` pip install geosocialx ``` 包含可选的额外组件（以逗号分隔）： ``` pip install "geosocialx[example,maps,test]" ``` ### 从源码安装 ``` git clone https://github.com/JayeshSuryavanshi/GeoSocialX.git cd GeoSocialX pip install . # or: pip install ".[example,maps,test]" ``` ## 快速入门 分析部分**不需要 API 密钥**——只需将其指向任何以换行符分隔的 X API v2 推文数据转储，即可探索它们发生的*地点*（以及*时间*）： ``` from geosocialx import GeospatialExtractor, GeospatialAnalyzer, MapVisualizer ex = GeospatialExtractor() tweets = ex.load_tweets("tweets.json") print(ex.coverage(tweets)) # how much of the data is geotagged points = ex.extract_points(tweets) # (lon, lat) points; place-only tweets included analyzer = GeospatialAnalyzer(points) print(analyzer.summary()) # count, bounding box, centroid, span_km, time range print(analyzer.densest_cells(top=3)) # busiest ~1 km grid cells (hotspots) print(analyzer.time_bins("day")) # activity over time, e.g. {'2024-01-01': 12} MapVisualizer(points).save_geojson("tweets.geojson") # open in any GIS or geojson.io ``` 手头没有数据？在 [`examples/analyze.py`](https://github.com/JayeshSuryavanshi/GeoSocialX/blob/main/examples/analyze.py) 中提供了一个基于内置示例的完整、可运行的离线演示。 要**获取您自己的**带地理标签的推文（需要付费层级上的 X API bearer token——请参阅[配置](#configuration)）： ``` from geosocialx import XDataFetcher fetcher = XDataFetcher(bearer_token="YOUR_BEARER_TOKEN") tweets = fetcher.fetch_tweets("37.7749,-122.4194,10mi", count=100) # 10 mi around SF fetcher.save_tweets_to_file(tweets, "tweets.json") ``` …或者直接在 shell 中执行： ``` geosocialx --geocode "37.7749,-122.4194,10mi" --count 100 --output tweets.json ``` ## 配置 GeoSocialX 需要一个 X API bearer token，该 token 从 `X_BEARER_TOKEN` 环境变量中读取（当安装了 `example` 额外组件时，可从 `.env` 文件加载）： ``` X_BEARER_TOKEN=your_bearer_token ``` `XDataFetcher` 也可以在用户上下文中进行身份验证，只需将四个 OAuth 1.0a 凭证（`api_key`、`api_key_secret`、`access_token`、`access_token_secret`）作为关键字参数传递即可，但对于最近搜索来说，仅应用级别的 bearer token 身份验证是最简单的途径。（`TwitterDataFetcher` 保留作为 `XDataFetcher` 的已弃用别名。） ## 用法 ### 命令行 安装此包会注册一个 `geosocialx` 控制台命令： ``` geosocialx --geocode "37.7749,-122.4194,10mi" --count 100 --output tweets.json ``` 这会将获取到的推文写入 `tweets.json`（每行一个 JSON 对象），并且当有推文引用了某个地点时，会将收集到的地点边界框写入 `tweets.json.places.json`。同样的入口点也可以通过 `python main.py` 使用（使用 `--help` 运行这两者之一以查看所有选项）。 ### 编程方式 ``` from geosocialx import XDataFetcher fetcher = XDataFetcher(bearer_token="your_bearer_token") # 获取 San Francisco 10 英里范围内最多 100 条推文。 tweets = fetcher.fetch_tweets("37.7749,-122.4194,10mi", count=100) if tweets is not None: # None means the API/network call failed fetcher.save_tweets_to_file(tweets, "tweets.json") fetcher.save_places_to_file("tweets.json.places.json") # place bboxes, if any ``` 如果调用失败，`fetch_tweets` 会返回 `None`（并记录错误），因此请使用 `if tweets is not None` 进行保护。如果传入 `None`，`save_tweets_to_file` 会引发 `ValueError`，因此它绝不会因为获取失败而截断现有文件。 ### 分析并可视化已保存的推文 一旦您获得了 `tweets.json`（来自上述步骤，或任何以换行符分隔的 v2 推文转储），流水线的其余部分就是离线的，不需要 API 访问权限： ``` from geosocialx import GeospatialExtractor, GeospatialAnalyzer, MapVisualizer extractor = GeospatialExtractor() tweets = extractor.load_tweets("tweets.json") print(extractor.coverage(tweets)) # e.g. {'total': 100, 'with_point': 12, ...} # 传入一个 {place_id: bbox} 映射（来自 save_places_to_file），以将 # 仅地点推文解析至其 bounding-box 质心；省略它则仅获取精确点。 places = extractor.load_places("tweets.json.places.json") points = extractor.extract_points(tweets, places=places) analyzer = GeospatialAnalyzer(points) print(analyzer.summary()) # count, bounding_box, centroid, span_km print(analyzer.densest_cells(top=3)) # busiest grid cells viz = MapVisualizer(points) viz.save_geojson("tweets.geojson") # standard library, always available viz.to_html_map("tweets_map.html") # interactive map — needs the `maps` extra ``` 在 [`examples/analyze.py`](https://github.com/JayeshSuryavanshi/GeoSocialX/blob/main/examples/analyze.py) 中有一个完整、**离线**的版本，它不需要 API 访问权限或付费层级；它针对提交的样本转储运行： ``` python examples/analyze.py ``` ## 核心模块与函数 ### `geosocialx.data_fetcher` **`XDataFetcher(bearer_token=None, *, api_key=None, api_key_secret=None, access_token=None, access_token_secret=None, wait_on_rate_limit=True)`** 构建一个 Tweepy v2 `Client`。传入 `bearer_token` 用于仅应用验证，或者将所有四个 OAuth 1.0a 凭证作为关键字参数传入用于用户上下文验证。如果两者均未提供，则引发 `ValueError`。默认情况下，它会等待速率限制结束（429 错误）；传入 `wait_on_rate_limit=False` 则会快速失败。 - **`fetch_tweets(geocode, count=100, extra_query="-is:retweet", tweet_fields=(...), start_time=None, end_time=None)`** — 返回给定区域内的推文字典 `list`，如果 API 调用或网络传输失败，则返回 `None`（并记录错误）。`geocode` 格式为 `"纬度,经度,半径"`（例如 `"37.7749,-122.4194,10mi"`，半径 ≤ 25mi/40km），在本地进行验证并转换为 v2 的 `point_radius:[经度 纬度 半径]` 运算符（注意：v2 将**经度放在前面**）。`start_time`/`end_time`（`datetime`）可选择性地缩小大约 7 天的最近搜索窗口内的搜索范围。结果中引用的地点边界框会被收集到 `places` 属性中。 - **`save_tweets_to_file(tweets, file_name)`** — 将每个推文字典以换行符分隔的 JSON 格式写入。在传入 `None` 时会引发 `ValueError`，而不是截断目标文件。 - **`save_places_to_file(file_name)`** — 将收集到的 `{place_id: bbox}` 映射以 JSON 格式写入。 ### `geosocialx.geospatial_extractor` **`GeospatialExtractor`** 将原始的 v2 推文字典转换为地理坐标点。 - **`extract_points(tweets, places=None)`** — 为每个具有可用位置的推文返回一个 `GeoPoint(tweet_id, longitude, latitude, text, created_at, author_id, source)` 列表。具有精确坐标的推文产生 `source="exact"`；如果提供了 `places` 映射，则仅有地点信息的推文会额外被解析为其边界框中心点，并标记为 `source="place"`。格式错误或超出范围的坐标将被跳过。 - **`coverage(tweets)`** — 返回 `{total, with_point, place_only, no_geo}`，以便您查看地理数据的稀疏程度（`with_point` 正好计算了 `extract_points` 作为精确点保留的内容）。 - **`load_tweets(path)`** / **`load_places(path)`** — 加载由获取器写入的以换行符分隔的推文 / `{place_id: bbox}` 映射。 ### `geosocialx.geospatial_analyzer` **`GeospatialAnalyzer(points)`** 在没有第三方依赖的情况下计算空间统计数据。 - **`count()`**, **`bounding_box()`** → `(最小经度, 最小纬度, 最大经度, 最大纬度)`, **`centroid()`** → `(经度, 纬度)`。 - **`haversine_km(lon1, lat1, lon2, lat2)`** — 以公里为单位的大圆距离（静态方法）。 - **`points_within(lon, lat, radius_km)`** — 半径范围内的点。 - **`densest_cells(cell_size_deg=0.01, top=5)`** — 最繁忙的网格单元。单元格在*度数*上是相等的，但在面积上并不相等（在 `0.01°` 时，高度约为 1.1 公里，但宽度约为 1.1·cos(纬度) 公里），因此这是一个城市内部的热点启发式方法，而不是跨纬度的密度估计。 - **`time_bins(freq="day"|"hour")`** — 根据点的 `created_at` 时间戳将其划分到对应的时间桶中，结果为 `{时间桶: 计数}`（按时间桶排序），跳过任何没有可解析时间戳的点。 - **`summary()`** — `count`、`bounding_box`、`centroid`、边界框对角线 `span_km`，以及当点包含时间戳时的 `earliest`/`latest`（UTC ISO-8601）。 ### `geosocialx.data_visualization` **`MapVisualizer(points)`** 将点渲染到磁盘。 - **`to_geojson()`** / **`save_geojson(path)`** — 一个 GeoJSON `FeatureCollection`（仅使用标准库；每个 feature 都包含该点的 `source`）。 - **`to_html_map(path, zoom_start=12, heatmap=True)`** — 一个带有标记和可选热力图层级的交互式 Leaflet 地图。需要 `maps` 额外组件（`folium`）；如果缺失，会引发 `ImportError` 并提供安装提示。 ## 运行测试 测试套件无需网络（Tweepy 被模拟）： ``` python -m unittest discover -s tests ``` 带有覆盖率（首先安装 `test` 额外组件）并执行基于 folium 的地图测试（安装 `maps` 额外组件）： ``` pip install ".[maps,test]" coverage run -m unittest discover -s tests coverage report ``` ## 项目结构 ``` GeoSocialX/ ├── geosocialx/ │ ├── __init__.py # exports the pipeline classes + __version__ │ ├── data_fetcher.py # XDataFetcher (X API v2) │ ├── geospatial_extractor.py # GeospatialExtractor, GeoPoint │ ├── geospatial_analyzer.py # GeospatialAnalyzer (pure stdlib) │ ├── data_visualization.py # MapVisualizer (GeoJSON + optional folium) │ ├── cli.py # `geosocialx` console entry point │ └── py.typed # PEP 561 marker (ships the type hints) ├── examples/ │ ├── analyze.py # offline analyze/visualize demo │ ├── sample_tweets.json # committed sample dump │ └── sample_places.json # committed sample place bboxes ├── tests/ # network-free unit tests ├── .github/workflows/ci.yml # test matrix (3.10–3.13) + lint ├── main.py # thin shim over geosocialx.cli ├── pyproject.toml └── README.md ``` ## 许可证 基于 MIT 许可证发布（请参阅 [`LICENSE`](https://github.com/JayeshSuryavanshi/GeoSocialX/blob/main/LICENSE) 文件）。 ## 作者 Jayesh Kishor Suryavanshi

标签：ESC4, OSINT, Python, Twitter API, 地理空间分析, 无后门, 热力图, 社交媒体分析, 逆向工具