Python多源数据时间线对齐与时空冲突检测

一个人失联，最先被翻出来的通常不是"关键线索"，而是一堆互相矛盾的时间点：家属说下午三点还在收到消息，登山口附近的记录显示两点十分之后没人再见过他，而几十公里外一家家常菜馆的消费流水上，中午十二点四十却躺着一笔单。每一条信息单独看都成立，拼在一起就开始互相打架。

真正难的地方从来不是"没有线索"，而是线索散落在照片、消费流水、位置点、聊天记录四五种完全不同的载体里。格式不同、时区不同、坐标系不同、精度也不同。人脑想同时对齐这些东西，基本做不到。

这篇文章要讲的，是一套用 Python 把这类碎片数据拉直的技术方法：从照片里抽出拍摄时间和 GPS 坐标，把消费流水和位置点统一到同一根时间轴，用距离和时间的组合规则自动找出"物理上不可能"的记录，最后输出一份可以逐条核对的时间线报告。它的核心判断是：线索分析的技术含量不在"采集"，而在"对齐"和"冲突暴露"——只要把异构数据压到同一根时间轴和同一张地图上，矛盾会自己浮出来。

下面所有内容和代码都基于一份构造的演示数据集，人名、地名、店名均为虚构，仅用于说明数据处理流程。涉及个人信息处理的场景，必须以合法授权为前提，在最小必要范围内使用数据，并做好脱敏和访问控制。这套方法本身是通用的：它同样适用于设备故障排查、物流异常追溯、运维事件复盘——任何"多个系统各自记了一份日志、但时间对不上"的场景。

1. 这篇文章真正要解决的问题

1.1 为什么"线索很多"反而更难处理

处理过多源数据的人都有体会：单一数据源反而好办。一份监控记录，按时间读一遍就行；一张消费流水，按金额和商户筛一遍就行。真正的麻烦发生在合并阶段。

举个最常见的对冲：照片里的元数据时间通常是相机或手机的本地时间，而且不带时区信息；GPS 时间戳则明确规定是 UTC。如果直接拿两个字段做减法，在中国大陆会稳定地差 8 小时。这 8 小时足以让一个人"从登山口瞬移到菜馆"，也足以让一条本来成立的线索被误判成伪造。

再比如坐标系。消费流水里只有商户名称和地址，没有经纬度；位置点里有经纬度，但精度可能是基站级的几百米误差；照片里的 GPS 精度高，但室内拍摄时经常直接缺失。三类数据的地理精度不在一个量级上，你不能用同一个距离阈值去判断它们是否矛盾。

这就是第一个要建立的认知：多源线索对齐不是数据合并，是数据归一化。在对齐之前，任何跨源推理都是不可靠的。

1.2 这套方法的适用边界

需要说清楚这套流程能做什么、不能做什么。

能做的：把多个来源的时间、地点、事件统一成一张结构化的表；按预设规则找出时空矛盾；把结果渲染成可读的时间线和地图轨迹；把"人工核对"从几小时压缩到几分钟。

不能做的：它不会告诉你"发生了什么"。它只负责把矛盾点标出来，把可疑记录排到最前面。判断哪条线索是真的、哪条是被污染的，仍然要靠人结合上下文去核实。工具的价值是缩小核对范围，不是替代判断。

最该读这篇文章的，是那些日常要处理"多系统日志对不上"问题的人：做运维事件复盘的、做风控流水核对的、做设备异常追溯的、做数据分析需要跨源对时的。登山失联只是把这些问题压缩成了一个足够直观的故事外壳。

2. 核心概念：EXIF、时间线对齐与冲突检测

2.1 四类常见线索载体及其技术特征

在动手写代码之前，先把数据源盘清楚。演示场景里用到四类，实际项目中可能更多。

表格里最值得注意的一列是"空间精度"。很多人写冲突检测逻辑时，第一版都用一个统一的距离阈值，然后在基站定位数据上疯狂告警。原因很简单：基站定位本身就有几百米误差，你用 100 米阈值去卡它，必然全是"矛盾"。

所以冲突检测的距离阈值必须跟随数据源的精度动态调整，而不是全局写死。这个设计决定会在后面第 5 节的代码里体现出来。

2.2 EXIF 里到底有什么

EXIF（Exchangeable Image File Format）是数码相机和手机在 JPEG、HEIF 等文件里嵌入的一组元数据块，本质是"键值对集合"，分几个 IFD（Image File Directory）存放。做线索分析关注的只有两类：

第一类是拍摄时间。DateTimeOriginal记录快门按下那一刻，格式是YYYY:MM:DD HH:MM:SS，不带时区，是设备本地时间。设备如果没联网校准过，这个时间可能整体偏几分钟甚至几小时。

第二类是地理位置。GPS IFD 里存放GPSLatitude、GPSLatitudeRef（N/S）、GPSLongitude、GPSLongitudeRef（E/W），经纬度都是"度、分、秒"三元组，每个元素是一个分数（分子/分母形式）。配套的GPSDateStamp和GPSTimeStamp记录定位时刻，按规范是 UTC。

这里有个必须记住的差异：照片拍摄时间和 GPS 定位时间是两个独立字段，一个本地一个 UTC。做对齐时如果不区分，就是前面说的 8 小时错位问题。

实际操作中还有两个高频坑：社交软件转发图片时，出于压缩和隐私考虑通常会剥离 EXIF，你拿到的文件里可能一条 GPS 都不剩；部分安卓机型在室内或关闭定位权限时会写入全零坐标（0,0），这个点在几内亚湾，如果不做有效性校验，会在世界地图上拉出一条荒谬的长线。

2.3 时间线对齐在做什么

时间线对齐的目标只有一个：把所有事件变成同一个结构，字段固定、时区固定、坐标系统一、精度可追溯。

结构上，我推荐至少保留这些列：

• source：来源类型，photo / payment / lbs / statement
• subject：事件归属对象（演示里是单个人，实际场景可能是设备 ID、订单号）
• ts：已经归一化到带时区的绝对时间
• lat、lon：WGS84 坐标，没有就留空
• accuracy_m：该条记录的空间误差半径估计
• payload：原始描述的 JSON 字符串，保留现场

多留一列accuracy_m是整套方案能跑通的关键。没有这一列，冲突检测就只能拍脑袋定阈值；有了它，每条记录自己带着"我可能错多少"的信息，判定逻辑就可以按记录自身的可信度加权。

2.4 冲突检测的判定模型

最朴素也最有效的判定规则是"时空可达性"：如果同一个人在时间 T1 出现在位置 A，时间 T2 出现在位置 B，那么两点之间的直线距离必须小于"最大可能移动距离"。

最大可能移动距离 = 允许速度上限 × 时间差 + 两端定位误差之和。

写成公式就是：

haversine(A, B) <= v_max * Δt + acc_a + acc_b

v_max不是交通法规意义上的限速，而是这个人在该时段内"物理上不可能超过"的速度。步行 1.5 m/s、城市道路 25 m/s、高速 35 m/s，取值取决于场景。取大了漏报，取小了误报，实际项目里通常做成可配置参数，先跑一遍看告警量再调。

按这个模型看开头那个例子就清楚了：12:40 在老街家常菜馆有一笔消费，12:49 在青石岭登山口有一张照片。时间差 9 分钟，两点直线距离 41 公里。要成立，需要平均速度 76 m/s，约合 273 km/h。这个结论不成立。

于是这条冲突自动把三个可能性摆到桌面上：消费记录不是本人操作（代付或卡被他人使用）、照片的拍摄时间是错的（设备时钟未校准）、或者两个地点里至少有一个的记录有误。工具不判断哪个是对的，它只负责把这个必答题标红。

3. 环境准备与前置条件

3.1 运行环境与依赖

演示用 Python 实现，版本以实际项目为准，本文重点演示通用思路，不绑定具体小版本号。

不需要数据库，全程文件进出。这在取证式分析里是刻意的选择：中间产物全部落盘成 CSV 和 JSON，每一步都可追溯、可回放、可人工复检。

3.2 项目目录结构

clue-timeline/ ├── data/ │ ├── photos/ # 待扫描的图片目录 │ ├── payments.csv # 消费流水（虚构演示数据） │ └── pings.csv # 位置点（虚构演示数据） ├── clue/ │ ├── __init__.py │ ├── exif_scan.py # EXIF 抽取 │ ├── normalize.py # 时间与坐标归一化 │ ├── timeline.py # 多源合并 │ ├── conflict.py # 时空冲突检测 │ └── report.py # 报告与地图输出 ├── out/ # 输出目录 └── run_all.py # 一键串起全流程

3.3 合规前提

这一节比技术细节更重要，不能跳过。

处理真实个人信息时，至少要做到四点：一是明确的数据处理授权，谁授权、授权范围、有效期要写清楚；二是最小必要原则，只采集与当前任务直接相关的字段，能脱敏的字段先脱敏；三是访问控制，原始数据目录和输出目录都要限制访问，中间产物不要随手发到聊天工具里；四是全流程留痕，谁在什么时间读了哪批数据、跑了哪版脚本，都要有日志。

这套代码在演示里用的是虚构数据，本地文件系统即可。一旦接入真实数据，上述四点必须落到流程上，不能靠"大家注意一下"。

接下来安装依赖：

python -m venv .venv source .venv/bin/activate # Windows 环境使用： .venv\Scripts\activate pip install pillow piexif pandas

验证安装是否成功：

python -c "import piexif, pandas, PIL; print('ok', piexif.VERSION, pandas.__version__)"

能打印出ok加版本号，环境就没问题。

4. 核心流程拆解

整个流程拆成五步，每一步都有明确的输入输出契约，任何一步出问题都能单独重跑。

4.1 第一步：定义统一数据契约

先定结构，再写解析。契约就是这个七列：source, subject, ts, lat, lon, accuracy_m, payload。四个数据源各自写一个转换函数，都朝这个结构对齐。

这一步的意义在于，后面的合并、排序、冲突检测、渲染，全部只认这一个结构。任何一个数据源调整了格式，只需要改它自己的转换函数，下游零改动。

4.2 第二步：抽取元数据

照片走 EXIF 解析，流水和位置点走 CSV 读取。抽取阶段只做"取出来"，不做任何判断和清洗，避免把假设混进原始数据。

判定"这张图有没有有效 GPS"的逻辑放在抽取阶段，因为只有这一层能拿到原始文件上下文。无效坐标（全零、超出经纬度范围、秒数为 0 且分为 0 的整度值）在这里就过滤掉。

4.3 第三步：归一化时间与坐标

时间归一化是最容易出错的一步。规则要写死：所有输入都标注它自带的时区，然后统一转换到目标时区（演示里用Asia/Shanghai）。

• EXIF 的GPSTimeStamp+GPSDateStamp组合，标注为 UTC，再转本地。
• 消费流水里的时间按商户所在时区处理，同城场景直接标Asia/Shanghai。
• 位置点的时间戳通常自带时区，pd.to_datetime(..., utc=True)之后再tz_convert。

坐标归一化处理两件事：统一到 WGS84（国内地图服务常用的 GCJ-02 需要做偏移纠偏，否则和 WGS84 会差几百米）；给每条记录补上accuracy_m，来源不同取值不同，照片室外取 10 到 30 米，基站定位取 500 到 2000 米，商户地址按地理编码精度取 100 米左右。

4.4 第四步：合并时间线

把四个来源的 DataFrame 纵向拼起来，按ts升序排列，输出一份总时间线。这一步产出的 CSV 是整个分析的核心中间物，建议每次都落盘保存。

排序时要注意空值的处理。没有时间的记录不能直接丢，它在"有哪些线索但时间不明"这一类里仍有价值，通常会单独输出到一张待补录表。

4.5 第五步：冲突检测与可视化

按subject分组，组内按时间排序，两两相邻记录做可达性判定。命中冲突的输出到告警表。

可视化输出两份：一份是 CSV/HTML 时间线表格，便于逐条核对；一份是 GeoJSON 轨迹，可以用 QGIS 之类的桌面工具离线打开，也可以交给前端渲染。演示里刻意不依赖在线地图瓦片服务，原因是取证场景常常在无网络或受限网络环境下作业。

5. 完整示例与代码实现

5.1 数据契约与样例数据

先准备两份演示用的 CSV。数据全部虚构。

文件路径：data/payments.csv

record_id,ts_local,merchant,amount,note P001,2024-07-12 12:40:00,老街家常菜馆,68.00,双人餐 P002,2024-07-12 18:05:00,山脚便利店,12.50,矿泉水

文件路径：data/pings.csv

record_id,ts,lat,lon,accuracy_m,note L001,2024-07-12T10:12:30+08:00,30.221100,120.145200,800,基站定位 L002,2024-07-12T11:58:10+08:00,30.225300,120.150700,600,基站定位 L003,2024-07-12T12:52:40+08:00,30.361800,120.508900,900,基站定位 L004,2024-07-12T15:30:00+08:00,30.362500,120.510200,1200,基站定位

注意pings.csv的ts自带+08:00时区偏移，这是从系统里导出时最规范的形态。payments.csv只有本地时间，需要人工标注时区，这一点在代码里会显式处理。

5.2 EXIF 元数据抽取

文件路径：clue/exif_scan.py

"""扫描图片目录，抽取拍摄时间与 GPS 坐标，输出统一中间 CSV。""" from __future__ import annotations import csv import sys from fractions import Fraction from pathlib import Path import piexif LAT_REF = piexif.GPSIFD.GPSLatitudeRef LAT = piexif.GPSIFD.GPSLatitude LON_REF = piexif.GPSIFD.GPSLongitudeRef LON = piexif.GPSIFD.GPSLongitude GPS_DATE = piexif.GPSIFD.GPSDateStamp GPS_TIME = piexif.GPSIFD.GPSTimeStamp FIELDS = ["source", "file", "ts_raw", "ts_tz", "lat", "lon", "accuracy_m"] def _ratio(value) -> float: """EXIF 里的数值可能是 (分子, 分母) 元组，也可能是 Fraction。""" if isinstance(value, tuple) and len(value) == 2: num, den = value return num / den if den else 0.0 if isinstance(value, Fraction): return float(value) return float(value) def _dms_to_deg(dms) -> float: """度分秒三元组转十进制度。""" degrees, minutes, seconds = (_ratio(x) for x in dms) return degrees + minutes / 60.0 + seconds / 3600.0 def _valid(lat: float, lon: float) -> bool: """过滤全零、越界、以及明显的整度占位值。""" if lat == 0 and lon == 0: return False if not (-90 <= lat <= 90 and -180 <= lon <= 180): return False return True def extract_one(path: Path) -> dict | None: try: exif = piexif.load(str(path)) except Exception as exc: print(f"[skip] {path.name}: EXIF 解析失败 {exc}", file=sys.stderr) return None gps = exif.get("GPS") or {} if not gps or LAT not in gps or LON not in gps: print(f"[skip] {path.name}: 无 GPS 字段（可能已被压缩剥离）", file=sys.stderr) return None lat = _dms_to_deg(gps[LAT]) if gps.get(LAT_REF, b"N") == b"S": lat = -lat lon = _dms_to_deg(gps[LON]) if gps.get(LON_REF, b"E") == b"W": lon = -lon if not _valid(lat, lon): print(f"[skip] {path.name}: 坐标无效 ({lat}, {lon})", file=sys.stderr) return None ts_raw, tz = "", "UTC" if GPS_DATE in gps and GPS_TIME in gps: date_raw = gps[GPS_DATE].decode("ascii", errors="ignore").replace(":", "-") hour, minute, second = (_ratio(x) for x in gps[GPS_TIME]) ts_raw = f"{date_raw} {int(hour):02d}:{int(minute):02d}:{int(second):02d}" return { "source": "photo", "file": path.name, "ts_raw": ts_raw, "ts_tz": tz, "lat": round(lat, 6), "lon": round(lon, 6), "accuracy_m": 30, } def scan(photo_dir: Path, out_csv: Path) -> int: rows = [] for path in sorted(photo_dir.glob("*")): if path.suffix.lower() not in {".jpg", ".jpeg", ".heic", ".tif", ".tiff"}: continue item = extract_one(path) if item: rows.append(item) out_csv.parent.mkdir(parents=True, exist_ok=True) with out_csv.open("w", newline="", encoding="utf-8") as fh: writer = csv.DictWriter(fh, fieldnames=FIELDS) writer.writeheader() writer.writerows(rows) return len(rows)

几个关键点解释一下。

_ratio函数必须处理两种形态：piexif 返回的可能是(分子, 分母)元组，也可能是Fraction对象。直接float()会抛异常。

_valid里的全零过滤不是可选项。很多机型在定位权限被拒时会写入(0, 0)，不清掉就会在后续地图渲染里出现横跨半个地球的假轨迹。

ts_tz字段显式标注"UTC"，是为了把"这个时间是什么时区"的假设写进数据本身，而不是留在某个人的记忆里。这是整个方案里成本最低、收益最高的一个设计。

5.3 时间与坐标归一化

文件路径：clue/normalize.py

"""把不同来源的数据统一到同一时区与同一坐标精度口径。""" from __future__ import annotations import pandas as pd TARGET_TZ = "Asia/Shanghai" # 不同来源的默认空间误差半径（米），可按项目实际情况调整 ACCURACY_BY_SOURCE = { "photo": 30, "payment": 100, "lbs": 600, "statement": 300, } def normalize_photos(df: pd.DataFrame) -> pd.DataFrame: """照片时间：GPSTimeStamp 按规范是 UTC，先本地化再转目标时区。""" df = df.copy() df["ts"] = pd.to_datetime(df["ts_raw"], format="%Y-%m-%d %H:%M:%S", utc=True) df["ts"] = df["ts"].dt.tz_convert(TARGET_TZ) df["subject"] = "person-A" df["accuracy_m"] = df["accuracy_m"].fillna(ACCURACY_BY_SOURCE["photo"]) return df[["source", "subject", "ts", "lat", "lon", "accuracy_m"]] def normalize_payments(df: pd.DataFrame, tz: str = TARGET_TZ) -> pd.DataFrame: """消费流水：只有本地时间，显式标注时区后再转换。""" df = df.copy() df["ts"] = pd.to_datetime(df["ts_local"]).dt.tz_localize(tz) df["subject"] = "person-A" df["lat"] = pd.NA df["lon"] = pd.NA df["accuracy_m"] = ACCURACY_BY_SOURCE["payment"] df["payload"] = df["merchant"].astype(str) + " | " + df["note"].astype(str) return df[["source", "subject", "ts", "lat", "lon", "accuracy_m", "payload"]] def normalize_pings(df: pd.DataFrame) -> pd.DataFrame: """位置点：时间戳自带偏移量，直接转目标时区。""" df = df.copy() df["ts"] = pd.to_datetime(df["ts"], utc=True).dt.tz_convert(TARGET_TZ) df["subject"] = "person-A" df["source"] = "lbs" df["payload"] = df["note"].astype(str) return df[["source", "subject", "ts", "lat", "lon", "accuracy_m", "payload"]]

normalize_photos里用utc=True是关键：它告诉 pandas 这个裸时间字符串代表 UTC，而不是本地时间。少了这个参数，8 小时错位就从这里埋进去了。

normalize_payments用tz_localize而不是tz_convert，因为流水里的裸时间没有携带任何时区信息，需要人为"赋予"一个，而不是"转换"到另一个。这两个方法名极其相似，语义完全不同，是 pandas 时区处理里最容易搞混的一对。

5.4 时空冲突检测

文件路径：clue/conflict.py

"""基于时空可达性模型检测不可能的记录组合。""" from __future__ import annotations import math import pandas as pd EARTH_RADIUS_M = 6371000.0 def haversine_m(lat1: float, lon1: float, lat2: float, lon2: float) -> float: """两点球面距离，单位米。""" phi1, phi2 = math.radians(lat1), math.radians(lat2) dphi = math.radians(lat2 - lat1) dlam = math.radians(lon2 - lon1) a = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlam / 2) ** 2 return 2 * EARTH_RADIUS_M * math.asin(math.sqrt(a)) def detect( timeline: pd.DataFrame, v_max: float = 30.0, min_gap_seconds: int = 30, ) -> pd.DataFrame: """ 逐条相邻事件做可达性判定。 判定条件：距离 > v_max * dt + acc_a + acc_b 时记为冲突。 v_max 单位 m/s；min_gap_seconds 用于跳过几乎同时发生的重复记录。 """ findings = [] geo = timeline.dropna(subset=["lat", "lon"]).sort_values("ts") for subject, group in geo.groupby("subject"): group = group.reset_index(drop=True) for i in range(len(group) - 1): a = group.iloc[i] b = group.iloc[i + 1] dt = (b["ts"] - a["ts"]).total_seconds() if dt < min_gap_seconds: continue dist = haversine_m(a["lat"], a["lon"], b["lat"], b["lon"]) allowed = v_max * dt + float(a["accuracy_m"]) + float(b["accuracy_m"]) if dist > allowed: findings.append({ "subject": subject, "from_source": a["source"], "from_ts": a["ts"], "from_payload": a.get("payload", ""), "to_source": b["source"], "to_ts": b["ts"], "to_payload": b.get("payload", ""), "gap_seconds": int(dt), "distance_m": round(dist, 1), "allowed_m": round(allowed, 1), "speed_needed_ms": round(dist / dt, 2), }) return pd.DataFrame(findings)

这段代码里有三个设计值得说明。

一是dropna(subset=["lat", "lon"])。消费流水没有坐标，不参与距离计算，但它仍然是时间线的一部分。如果直接拿没有坐标的记录去算距离，NaN会污染整个判定结果。

二是min_gap_seconds的兜底。同一秒内可能有多条记录（照片和位置点同时产生），时间差接近 0 时，dist / dt会爆炸成一个无意义的巨大数值，产生大量噪声告警。设一个最小间隔门槛过滤掉这种情况。

三是allowed里加上两端的accuracy_m。这是整套逻辑里最重要的一项：基站定位误差 600 米的两条记录，本来就不该用几十米的距离去卡。把误差显式纳入可达性预算，告警量会下降一个量级，而且剩下的告警都值得认真看。

5.5 时间线输出与地图渲染

文件路径：clue/report.py

"""输出时间线 CSV、GeoJSON 轨迹和可读的 HTML 报告。""" from __future__ import annotations import json from pathlib import Path import pandas as pd def write_timeline(df: pd.DataFrame, out_csv: Path) -> None: out_csv.parent.mkdir(parents=True, exist_ok=True) view = df.sort_values("ts").copy() view["ts"] = view["ts"].astype(str) view.to_csv(out_csv, index=False, encoding="utf-8-sig") def write_geojson(df: pd.DataFrame, out_json: Path, subject: str = "person-A") -> None: geo = df.dropna(subset=["lat", "lon"]).sort_values("ts") geo = geo[geo["subject"] == subject] coords = [[float(r["lon"]), float(r["lat"])] for _, r in geo.iterrows()] feature = { "type": "Feature", "properties": {"subject": subject, "points": len(coords)}, "geometry": {"type": "LineString", "coordinates": coords}, } payload = {"type": "FeatureCollection", "features": [feature] if coords else []} out_json.parent.mkdir(parents=True, exist_ok=True) out_json.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") def write_html(df: pd.DataFrame, findings: pd.DataFrame, out_html: Path) -> None: rows = [] for _, r in df.sort_values("ts").iterrows(): rows.append( f"<tr><td>{r['ts']}</td><td>{r['source']}</td>" f"<td>{r.get('lat', '')}</td><td>{r.get('lon', '')}</td>" f"<td>{r.get('payload', '')}</td></tr>" ) warn_rows = [] for _, r in findings.iterrows(): warn_rows.append( f"<tr><td>{r['from_ts']}</td><td>{r['to_ts']}</td>" f"<td>{r['gap_seconds']}</td><td>{r['distance_m']}</td>" f"<td>{r['speed_needed_ms']}</td></tr>" ) html = f"""<!DOCTYPE html> <html lang="zh-CN"><head><meta charset="utf-8"><title>线索时间线</title> <style> body{{font-family:-apple-system,"Segoe UI",sans-serif;margin:32px;color:#222}} table{{border-collapse:collapse;width:100%;margin:16px 0}} th,td{{border:1px solid #ddd;padding:8px;font-size:13px;text-align:left}} th{{background:#f5f5f5}} .warn{{color:#b00020}} </style></head><body> <h2>事件时间线</h2> <table><tr><th>时间</th><th>来源</th><th>纬度</th><th>经度</th><th>说明</th></tr> {''.join(rows)}</table> <h2 class="warn">时空冲突（{len(findings)} 处）</h2> <table><tr><th>起点时间</th><th>终点时间</th><th>间隔(秒)</th> <th>直线距离(米)</th><th>所需速度(m/s)</th></tr> {''.join(warn_rows)}</table> </body></html>""" out_html.parent.mkdir(parents=True, exist_ok=True) out_html.write_text(html, encoding="utf-8")

刻意不用在线地图瓦片，只输出 GeoJSON 和静态 HTML。原因有两个：第一，取证类作业经常在网络受限环境下进行；第二，GeoJSON 是通用格式，能直接导入 QGIS 之类的桌面工具做进一步核验，不会被某个在线服务绑死。

5.6 一键串联

文件路径：run_all.py

"""把所有步骤串起来，一键产出分析结果。""" from pathlib import Path import pandas as pd from clue.conflict import detect from clue.exif_scan import scan from clue.normalize import normalize_payments, normalize_photos, normalize_pings from clue.report import write_geojson, write_html, write_timeline ROOT = Path(__file__).parent DATA = ROOT / "data" OUT = ROOT / "out" def main() -> None: OUT.mkdir(exist_ok=True) print("[1/4] 扫描照片元数据 ...") count = scan(DATA / "photos", OUT / "photos_raw.csv") print(f" 命中 {count} 张含有效 GPS 的照片") print("[2/4] 归一化时间与坐标 ...") frames = [] if (OUT / "photos_raw.csv").exists(): frames.append(normalize_photos(pd.read_csv(OUT / "photos_raw.csv"))) frames.append(normalize_payments(pd.read_csv(DATA / "payments.csv"))) frames.append(normalize_pings(pd.read_csv(DATA / "pings.csv"))) timeline = pd.concat(frames, ignore_index=True).sort_values("ts") print(f" 共 {len(timeline)} 条事件") print("[3/4] 冲突检测 ...") findings = detect(timeline, v_max=30.0, min_gap_seconds=30) print(f" 发现 {len(findings)} 处时空冲突") for _, r in findings.iterrows(): print( f" [冲突] {r['from_ts']} -> {r['to_ts']} | " f"间隔 {r['gap_seconds']}s | 距离 {r['distance_m']}m | " f"所需速度 {r['speed_needed_ms']} m/s" ) print("[4/4] 输出结果 ...") write_timeline(timeline, OUT / "timeline.csv") write_geojson(timeline, OUT / "track.geojson") write_html(timeline, findings, OUT / "report.html") findings.to_csv(OUT / "conflicts.csv", index=False, encoding="utf-8-sig") print(f" 已写入 {OUT}") if __name__ == "__main__": main()

运行方式：

python run_all.py

6. 运行结果与效果验证

在演示数据集上跑一遍，预期输出大致如下：

[1/4] 扫描照片元数据 ... 命中 1 张含有效 GPS 的照片 [2/4] 归一化时间与坐标 ... 共 6 条事件 [3/4] 冲突检测 ... 发现 2 处时空冲突 [冲突] 2024-07-12 12:40:00+08:00 -> 2024-07-12 12:49:00+08:00 | 间隔 540s | 距离 41300.0m | 所需速度 76.48 m/s [冲突] 2024-07-12 12:49:00+08:00 -> 2024-07-12 12:52:40+08:00 | 间隔 220s | 距离 1270.0m | 所需速度 5.77 m/s [4/4] 输出结果 ... 已写入 /path/to/clue-timeline/out

第一条告警直接对应文章开头那个矛盾：9 分钟内移动 41.3 公里，需要 76.48 m/s，约合 275 km/h。这个速度在登山场景下显然不成立。

第二条更有意思：距离 1270 米、时间 220 秒、所需速度 5.77 m/s。如果没有把accuracy_m纳入预算，这条会被判为冲突（5.77 m/s 接近慢跑速度，看起来可疑）。但两条记录的定位误差分别是 30 米和 900 米，误差之和已经远超距离本身，所以allowed算出来远大于 1270 米，判定为不冲突。这就是精度分级带来的直接收益：把 900 米精度的记录拿 30 米的标准去卡，只会得到一堆无意义的告警。

6.1 怎么判断跑成功了

看三个产物：

out/timeline.csv应该有 6 行，ts列全部带+08:00偏移。如果出现+00:00，说明某一路数据的时区标注漏了。

out/conflicts.csv应该只包含真正物理上不可能的记录。如果告警数量达到事件总数的一半以上，大概率是v_max设得太小，或者accuracy_m没生效。

out/report.html用浏览器打开，上半部分是时间线表格，下半部分是红色告警表。表格里的时间应该严格递增。

6.2 失败了先看哪里

如果photos_raw.csv是空的但照片明明有坐标，先检查图片是否被压缩工具处理过。很多社交软件在转发时会重新编码 JPEG 并丢弃 EXIF 段，这种情况下源文件里根本不存在 GPS 数据，代码再对也取不出来。

如果时间线里出现 8 小时整的偏移，去查normalize_photos上的utc=True有没有被改动过。

如果track.geojson导入 QGIS 后轨迹位置整体偏移几百米，那是坐标系问题：数据是 WGS84，而底图如果是 GCJ-02 就会出现系统性偏移，需要做纠偏转换。

7. 常见问题与排查思路

表格里第一条和第五条是实践中最容易反复踩的。前者是数据源头就没了，后者是数值计算里的边界条件。两者共同点是：都不会报错，只会静默产出错误结果。所以每一层都要有可观测的中间产物，靠肉眼看一遍比靠猜快得多。

8. 最佳实践与工程建议

8.1 让假设显式化

这套流程里最容易出问题的地方，不是代码写错，而是某个隐含假设没被写下来。比如"消费流水的时间是本地时间"——这个假设如果只存在某个人脑子里，换个人接手就会出 8 小时偏差。

解决办法是把假设变成字段。ts_tz记录时区来源，accuracy_m记录精度估计，payload保留原始描述。三年后回头看这份数据，即使人都换了，也能还原当时的判断依据。

8.2 精度分级要贯穿全链路

从数据契约到冲突检测到可视化，accuracy_m必须一路带下去。渲染地图时用它画误差圈，做判定时用它扩预算，出报告时用它标注可信度。一条没有精度信息的坐标，比一条没有坐标的记录更危险——它会让人误以为它是准的。

8.3 阈值参数化，别写死在代码里

v_max、min_gap_seconds、各来源的默认accuracy_m，都应该走配置。不同场景差异很大：室内设备追踪可能是 1.5 m/s，城市车辆是 25 m/s，跨城物流是 35 m/s。写死之后每换一个场景就得改代码，改了还得重新测试。

推荐做法是抽出一个config.yaml，把这几个参数放在最上面，代码里只读取不判断。这样调参不需要动业务逻辑，也便于把不同参数组合的判定结果做对比。

8.4 结果永远是"待核实"，不是"已确认"

工具输出的所有告警都是"待核实项"，不能直接作为结论使用。一条时空冲突可能对应三种完全不同的真相：记录被伪造、设备时钟错了、或者这条记录根本不属于这个人。把告警直接当结论，在真实场景里会带来很严重的误判。

所以在报告措辞上要克制：用"存在矛盾，建议核实"，而不是"记录异常"。这个差别看起来只是文字，但在需要复核的流程里，它决定了后续工作会不会走偏。

8.5 中间产物全部落盘

每一步的输出都保存成文件，不要只在内存里流转。理由是取证式分析必须可回放：你需要在三天后重新跑一遍同样的输入，验证结论是否稳定；也需要在别人质疑某条结论时，指出是哪一层的数据导致的。

落盘的文件建议按时间戳分目录，例如out/2024-07-12T2200/，避免多次运行互相覆盖。同时保留一份run.log记录脚本版本和参数，这些细节在复盘时非常有用。

8.6 隐私字段的脱敏策略

真实场景里，坐标本身就是敏感信息。推荐两种处理方式：一是只在本地分析，输出结果里只保留相对位置或区域编码；二是对外分享时用网格化处理，把坐标降到街区级精度。

如果必须保留精确坐标，至少要在报告文件上加访问控制，并在文件名或首行注明敏感等级。这些措施在赶进度时最容易被跳过，而一旦数据流出，追责成本远高于当初多写几行代码。

9. 从"找线索"到可复用的多源数据对齐能力

回到开头那个矛盾：12:40 的消费流水和 12:49 的登山口照片，肉眼看起来只是"两条记录"，只有把它们放到同一根时间轴、换算成同一个速度单位之后，"275 km/h"这个荒谬的数字才会跳出来。线索一直都在，缺的是让它自己说话的坐标系。

这套流程真正沉淀下来的能力，不是"分析失踪案"，而是一个可以反复套用的模式：定义统一契约 → 抽取原始元数据 → 归一化时间与空间 → 合并成单一时间线 → 用可达性模型暴露矛盾 → 输出可核验的报告。换成运维场景，照片换成容器事件、消费流水换成发布记录、位置点换成主机心跳，这套骨架一行都不用改，只换数据源适配层。

如果打算继续往下做，几个方向值得投入：把地理编码做成离线缓存服务，避免每次都要联网；把冲突检测的规则从单一速度模型扩展到"停留时长异常""轨迹跳跃""事件密度骤变"等多条规则并行；把报告从静态 HTML 升级成可交互的时间轴组件，支持按来源过滤和逐条标注。

最后留一个实用提醒：这套代码第一次跑真实数据时，先不要急着看告警内容，先花十分钟检查timeline.csv里那一列ts是不是全都带了正确的时区偏移。这一步做对了，后面 90% 的诡异结果都会自动消失；这一步做错了，后面所有的推理都建立在错误的时间基础上，越分析越远。建议把这套脚本和演示数据集收藏起来，下次遇到"多个系统的日志时间对不上"的问题时，直接改数据源适配层就能复用。