2. 欺诈数据源接入与统一字段映射

任务目标

把交易、用户、设备、IP 四类异构数据（CSV/JSON）接入，通过一张统一字段映射表，转换为标准化的「实体-关系」记录，供后续图谱构建使用。

环境与依赖

• Python 3.8+，仅用标准库 csv、json、os，无需第三方包。
• 目录结构：data/ 下放 4 个源文件，脚本输出 data/normalized.jsonl。

源数据样例

交易 CSV（data/transactions.csv）：

txn_id,user_id,device_id,ip,amount,ts
T1,U100,D9,1.2.3.4,880.5,2024-01-01T10:00:00
T2,U101,D9,1.2.3.4,1200,2024-01-01T10:05:00

用户 JSON（data/users.json）：

[{"uid":"U100","name":"alice","phone":"13800000001"},
 {"uid":"U101","name":"bob","phone":"13800000002"}]

设备 JSON（data/devices.json）：

[{"did":"D9","model":"Pixel","os":"Android"},
 {"did":"D10","model":"iPhone","os":"iOS"}]

IP CSV（data/ips.csv）：

ip_addr,country,city
1.2.3.4,CN,Beijing
5.6.7.8,US,NYC

统一字段映射表

每类源数据声明：源字段 → 标准字段，以及该源产生的实体类型与关系类型。标准字段统一为 id、type、attrs。

核心实现

import csv, json, os

# 统一字段映射表：源类型 -> (实体类型, 字段映射, 关系定义)
MAPPING = {
    "transactions": {
        "entity": "Transaction",
        "fields": {"txn_id": "id", "amount": "amount", "ts": "ts"},
        "relations": [
            ("user_id", "User", "PERFORMED"),
            ("device_id", "Device", "USED_DEVICE"),
            ("ip", "IP", "FROM_IP"),
        ],
    },
    "users": {
        "entity": "User",
        "fields": {"uid": "id", "name": "name", "phone": "phone"},
        "relations": [],
    },
    "devices": {
        "entity": "Device",
        "fields": {"did": "id", "model": "model", "os": "os"},
        "relations": [],
    },
    "ips": {
        "entity": "IP",
        "fields": {"ip_addr": "id", "country": "country", "city": "city"},
        "relations": [],
    },
}

def load_source(path):
    if path.endswith(".csv"):
        with open(path, newline="", encoding="utf-8") as f:
            return list(csv.DictReader(f))
    with open(path, encoding="utf-8") as f:
        return json.load(f)

def normalize(source_type, rows):
    cfg = MAPPING[source_type]
    entities, relations = [], []
    for row in rows:
        attrs = {std: row.get(src) for src, std in cfg["fields"].items() if src != "id"}
        eid = row.get(next(s for s, t in cfg["fields"].items() if t == "id"))
        entities.append({"id": eid, "type": cfg["entity"], "attrs": attrs})
        for src_field, target_type, rel_type in cfg["relations"]:
            if row.get(src_field):
                relations.append({
                    "from": eid, "from_type": cfg["entity"],
                    "to": row[src_field], "to_type": target_type,
                    "rel": rel_type,
                })
    return entities, relations

def main():
    sources = {
        "transactions": "data/transactions.csv",
        "users": "data/users.json",
        "devices": "data/devices.json",
        "ips": "data/ips.csv",
    }
    all_entities, all_relations = [], []
    for stype, path in sources.items():
        rows = load_source(path)
        ents, rels = normalize(stype, rows)
        all_entities += ents
        all_relations += rels
        print(f"{stype}: {len(ents)} entities, {len(rels)} relations")

    os.makedirs("data", exist_ok=True)
    with open("data/normalized.jsonl", "w", encoding="utf-8") as f:
        for e in all_entities:
            f.write(json.dumps({"kind": "entity", **e}, ensure_ascii=False) + "\n")
        for r in all_relations:
            f.write(json.dumps({"kind": "relation", **r}, ensure_ascii=False) + "\n")
    print(f"total: {len(all_entities)} entities, {len(all_relations)} relations")

if __name__ == "__main__":
    main()

运行与验证

python normalize.py

预期输出（由样例数据唯一确定）：

transactions: 2 entities, 6 relations
users: 2 entities, 0 relations
devices: 2 entities, 0 relations
ips: 2 entities, 0 relations
total: 8 entities, 6 relations

通过条件：data/normalized.jsonl 每行是合法 JSON，kind 为 entity 或 relation；实体 id 与源数据一致；交易产生的 6 条关系分别指向 User/Device/IP 三类目标。

常见问题

• CSV 编码非 UTF-8 时，open 需显式指定 encoding。
• 新增数据源只需在 MAPPING 中追加一项，无需改动 normalize 逻辑。
• 字段缺失时 row.get 返回 None，如需过滤可在 normalize 中加判空。