GraphRAG 入門 #4 Neptune Database に Bulk Loader でナレッジグラフを構築して Neptune Analytics と連携してみた【Amazon Neptune】

import csv
import os

OUT = "coffee_data"
os.makedirs(OUT, exist_ok=True)

def write_csv(filename, header, rows):
    with open(os.path.join(OUT, filename), "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(header)
        w.writerows(rows)

# ---------- Farm ----------
farms = [
    ("F01", "エスペランサ農園", "Colombia", "Huila", 1800),
    ("F02", "モンテベルデ農園", "Costa Rica", "Tarrazu", 1700),
    ("F03", "サンライズヒル農園", "Ethiopia", "Yirgacheffe", 2000),
    ("F04", "北風高地農園", "Kenya", "Nyeri", 1900),
    ("F05", "リオドラド農園", "Brazil", "Cerrado", 1100),
]
write_csv("farms.csv",
    [":ID", ":LABEL", "farm_id:String", "name:String", "country:String", "region:String", "altitude_m:Int"],
    [(fid, "Farm", fid, name, country, region, alt) for fid, name, country, region, alt in farms])

# ---------- Certification ----------
certifications = [
    ("C01", "Organic"),
    ("C02", "FairTrade"),
    ("C03", "RainforestAlliance"),
]
write_csv("certifications.csv",
    [":ID", ":LABEL", "cert_id:String", "type:String"],
    [(cid, "Certification", cid, ctype) for cid, ctype in certifications])

# ---------- Farm -[:HAS_CERTIFICATION]-> Certification ----------
has_certification = [
    ("F01", "C01"), ("F01", "C02"),
    ("F02", "C03"),
    ("F03", "C01"),
    ("F04", "C02"), ("F04", "C03"),
    # F05は認証なし（意図的な比較対象）
]
write_csv("has_certification.csv",
    [":ID", ":START_ID", ":END_ID", ":TYPE"],
    [(f"hc-{i+1}", f, c, "HAS_CERTIFICATION") for i, (f, c) in enumerate(has_certification)])

# ---------- Exporter ----------
exporters = [
    ("EX01", "Andes Export Co.", "Colombia"),
    ("EX02", "Pacific Bean Logistics", "Costa Rica"),
    ("EX03", "African Highlands Trading", "Ethiopia"),
]
write_csv("exporters.csv",
    [":ID", ":LABEL", "exporter_id:String", "name:String", "country:String"],
    [(eid, "Exporter", eid, name, country) for eid, name, country in exporters])

farm_exporter = {"F01": "EX01", "F05": "EX01", "F02": "EX02", "F03": "EX03", "F04": "EX03"}

# ---------- HarvestLot: 農園ごとに3ロット = 15 ----------
harvest_dates = ["2025-11-05", "2025-12-10", "2026-01-15"]
grades = ["AA", "A", "Specialty"]
harvest_lots, farm_harvested, lot_exported_by = [], [], []
lot_id_counter = 1
for farm_id, *_ in farms:
    for i in range(3):
        lot_id = f"L{lot_id_counter:02d}"
        qty = 500 + lot_id_counter * 37
        harvest_lots.append((lot_id, harvest_dates[i], qty, grades[i]))
        farm_harvested.append((farm_id, lot_id))
        lot_exported_by.append((lot_id, farm_exporter[farm_id]))
        lot_id_counter += 1

write_csv("harvest_lots.csv",
    [":ID", ":LABEL", "lot_id:String", "harvest_date:Date", "quantity_kg:Int", "grade:String"],
    [(lid, "HarvestLot", lid, d, q, g) for lid, d, q, g in harvest_lots])
write_csv("farm_harvested.csv",
    [":ID", ":START_ID", ":END_ID", ":TYPE"],
    [(f"fh-{i+1}", f, l, "HARVESTED") for i, (f, l) in enumerate(farm_harvested)])
write_csv("lot_exported_by.csv",
    [":ID", ":START_ID", ":END_ID", ":TYPE"],
    [(f"le-{i+1}", l, e, "EXPORTED_BY") for i, (l, e) in enumerate(lot_exported_by)])

# ---------- RoastBatch (10) + Product (7) ----------
products = [
    ("P01", "エスペランサシングルオリジン", "single_origin"),
    ("P02", "モンテベルデシングルオリジン", "single_origin"),
    ("P03", "サンライズヒルシングルオリジン", "single_origin"),
    ("P04", "北風高地シングルオリジン", "single_origin"),
    ("P05", "リオドラドシングルオリジン", "single_origin"),
    ("P06", "サンライズブレンド", "blend"),
    ("P07", "モーニングブレンド", "blend"),
]
write_csv("products.csv",
    [":ID", ":LABEL", "product_id:String", "name:String", "category:String"],
    [(pid, "Product", pid, name, cat) for pid, name, cat in products])

roast_batches = [
    ("RB01", "2026-01-05", "medium", 40), ("RB02", "2026-01-06", "light", 35),
    ("RB03", "2026-01-07", "medium", 45), ("RB04", "2026-01-08", "medium", 38),
    ("RB05", "2026-01-09", "dark", 42),   ("RB06", "2026-01-10", "medium", 40),
    ("RB07", "2026-01-11", "light", 30),  ("RB08", "2026-01-12", "medium", 33),
    ("RB09", "2026-01-13", "medium", 36), ("RB10", "2026-01-14", "dark", 28),
]
write_csv("roast_batches.csv",
    [":ID", ":LABEL", "batch_id:String", "roast_date:Date", "roast_level:String", "weight_kg:Int"],
    [(bid, "RoastBatch", bid, d, lvl, w) for bid, d, lvl, w in roast_batches])

# RB03とRB08は2つのロットをブレンドしている（多対多の例）
lot_used_in = [
    ("L01", "RB01"), ("L04", "RB02"),
    ("L02", "RB03"), ("L07", "RB03"),   # ブレンド → P06 サンライズブレンド
    ("L10", "RB04"), ("L13", "RB05"),
    ("L01", "RB06"), ("L08", "RB07"),
    ("L10", "RB08"), ("L05", "RB08"),   # ブレンド → P07 モーニングブレンド
    ("L14", "RB09"), ("L11", "RB10"),
]
write_csv("lot_used_in.csv",
    [":ID", ":START_ID", ":END_ID", ":TYPE"],
    [(f"ui-{i+1}", l, b, "USED_IN") for i, (l, b) in enumerate(lot_used_in)])

batch_packaged_as = [
    ("RB01", "P01"), ("RB02", "P02"), ("RB03", "P06"), ("RB04", "P04"),
    ("RB05", "P05"), ("RB06", "P01"), ("RB07", "P03"), ("RB08", "P07"),
    ("RB09", "P05"), ("RB10", "P04"),
]
write_csv("batch_packaged_as.csv",
    [":ID", ":START_ID", ":END_ID", ":TYPE"],
    [(f"pa-{i+1}", b, p, "PACKAGED_AS") for i, (b, p) in enumerate(batch_packaged_as)])

# ---------- Customer (10), Order (25), Shipment (25) ----------
customers = [(f"CUST{i:02d}", f"サンプルカフェ{i}号店", "cafe" if i % 2 == 0 else "retail") for i in range(1, 11)]
write_csv("customers.csv",
    [":ID", ":LABEL", "customer_id:String", "name:String", "type:String"],
    [(cid, "Customer", cid, name, ctype) for cid, name, ctype in customers])

order_dates = ["2026-02-01", "2026-02-08", "2026-02-15", "2026-02-22", "2026-03-01"]
orders, shipments, customer_placed, order_for_product, order_fulfilled_by = [], [], [], [], []
product_ids = [p[0] for p in products]
for i in range(1, 26):
    order_id, ship_id = f"O{i:02d}", f"S{i:02d}"
    cust_id = f"CUST{((i - 1) % 10) + 1:02d}"
    product_id = product_ids[(i - 1) % len(product_ids)]
    qty = 5 + (i % 8) * 2
    odate = order_dates[(i - 1) % len(order_dates)]
    orders.append((order_id, odate, qty))
    shipments.append((ship_id, odate, f"TRK-{1000+i}"))
    customer_placed.append((cust_id, order_id))
    order_for_product.append((order_id, product_id))
    order_fulfilled_by.append((order_id, ship_id))

write_csv("orders.csv",
    [":ID", ":LABEL", "order_id:String", "order_date:Date", "quantity_kg:Int"],
    [(oid, "Order", oid, d, q) for oid, d, q in orders])
write_csv("shipments.csv",
    [":ID", ":LABEL", "shipment_id:String", "ship_date:Date", "tracking_no:String"],
    [(sid, "Shipment", sid, d, trk) for sid, d, trk in shipments])
write_csv("customer_placed.csv",
    [":ID", ":START_ID", ":END_ID", ":TYPE"],
    [(f"cp-{i+1}", c, o, "PLACED") for i, (c, o) in enumerate(customer_placed)])
write_csv("order_for_product.csv",
    [":ID", ":START_ID", ":END_ID", ":TYPE"],
    [(f"fp-{i+1}", o, p, "FOR_PRODUCT") for i, (o, p) in enumerate(order_for_product)])
write_csv("order_fulfilled_by.csv",
    [":ID", ":START_ID", ":END_ID", ":TYPE"],
    [(f"fb-{i+1}", o, s, "FULFILLED_BY") for i, (o, s) in enumerate(order_fulfilled_by)])

# ---------- QualityIncident (4) ----------
quality_incidents = [
    ("QI01", "2026-02-10", "中", "焙煎ムラの報告", "S04"),          # 北風高地シングルオリジンの出荷
    ("QI02", "2026-02-17", "低", "香りが弱いとの顧客指摘", "S06"),   # サンライズブレンドの出荷
    ("QI03", "2026-02-24", "高", "異物混入の報告", "S05"),          # リオドラドシングルオリジンの出荷
    ("QI04", "2026-03-03", "低", "賞味期限表示の誤り", "S01"),       # エスペランサシングルオリジンの出荷
]
write_csv("quality_incidents.csv",
    [":ID", ":LABEL", "incident_id:String", "report_date:Date", "severity:String", "description:String"],
    [(qid, "QualityIncident", qid, d, sev, desc) for qid, d, sev, desc, _ in quality_incidents])
write_csv("incident_reported_for.csv",
    [":ID", ":START_ID", ":END_ID", ":TYPE"],
    [(f"rf-{i+1}", qid, ship, "REPORTED_FOR") for i, (qid, _, _, _, ship) in enumerate(quality_incidents)])

print("done:", os.listdir(OUT))

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "CreateBucket",
      "Effect": "Allow",
      "Action": "s3:CreateBucket",
      "Resource": "arn:aws:s3:::example-neptune-coffee-demo"
    },
    {
      "Sid": "UploadObjects",
      "Effect": "Allow",
      "Action": ["s3:PutObject", "s3:ListBucket"],
      "Resource": [
        "arn:aws:s3:::example-neptune-coffee-demo",
        "arn:aws:s3:::example-neptune-coffee-demo/*"
      ]
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": { "Service": "rds.amazonaws.com" },
      "Action": "sts:AssumeRole"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject", "s3:ListBucket"],
      "Resource": [
        "arn:aws:s3:::example-neptune-coffee-demo",
        "arn:aws:s3:::example-neptune-coffee-demo/*"
      ]
    }
  ]
}

!aws s3 mb s3://example-neptune-coffee-demo --region ap-northeast-1
!aws s3 cp coffee_data/ s3://example-neptune-coffee-demo/coffee_data/ --recursive

%load

%%oc
MATCH (n) RETURN count(n) AS node_count

%%oc
MATCH ()-[r]->() RETURN count(r) AS edge_count

%%oc -d {"Farm":"name","Certification":"type","HarvestLot":"lot_id","Exporter":"name","RoastBatch":"batch_id","Product":"name","Customer":"name","Order":"order_id","Shipment":"shipment_id","QualityIncident":"incident_id"}
MATCH p=(n)-[r]->(m)
RETURN p

%%oc
MATCH (p:Product {name: '北風高地シングルオリジン'})<-[:PACKAGED_AS]-(rb:RoastBatch)
RETURN rb.batch_id, rb.roast_date, rb.roast_level

%%oc -d {"Product":"name","RoastBatch":"batch_id"}
MATCH p=(:Product {name: '北風高地シングルオリジン'})<-[:PACKAGED_AS]-(:RoastBatch)
RETURN p

%%oc
MATCH (f:Farm)
WHERE NOT (f)-[:HAS_CERTIFICATION]->(:Certification)
MATCH (f)-[:HARVESTED]->(:HarvestLot)-[:USED_IN]->(:RoastBatch)-[:PACKAGED_AS]->(p:Product)
RETURN DISTINCT f.name, p.name

%%oc -d {"Farm":"name","HarvestLot":"lot_id","RoastBatch":"batch_id","Product":"name"}
MATCH (f:Farm)
WHERE NOT (f)-[:HAS_CERTIFICATION]->(:Certification)
MATCH p=(f)-[:HARVESTED]->(:HarvestLot)-[:USED_IN]->(:RoastBatch)-[:PACKAGED_AS]->(:Product)
RETURN p

%%oc -d {"Farm":"name","HarvestLot":"lot_id","RoastBatch":"batch_id","Product":"name"}
MATCH p=(:Product {name: 'サンライズブレンド'})<-[:PACKAGED_AS]-(:RoastBatch)<-[:USED_IN]-(:HarvestLot)<-[:HARVESTED]-(:Farm)
RETURN p

%%oc
MATCH (f:Farm {name: 'エスペランサ農園'})-[:HARVESTED]->(:HarvestLot)-[:USED_IN]->(:RoastBatch)-[:PACKAGED_AS]->(:Product)<-[:FOR_PRODUCT]-(:Order)<-[:PLACED]-(c:Customer)
RETURN DISTINCT c.name

%%oc -d {"Farm":"name","HarvestLot":"lot_id","RoastBatch":"batch_id","Product":"name","Order":"order_id","Customer":"name"}
MATCH p=(:Farm {name: 'エスペランサ農園'})-[:HARVESTED]->(:HarvestLot)-[:USED_IN]->(:RoastBatch)-[:PACKAGED_AS]->(:Product)<-[:FOR_PRODUCT]-(:Order)<-[:PLACED]-(:Customer)
RETURN p

%%oc
MATCH (qi:QualityIncident {incident_id: 'QI01'})-[:REPORTED_FOR]->(:Shipment)<-[:FULFILLED_BY]-(:Order)-[:FOR_PRODUCT]->(:Product)<-[:PACKAGED_AS]-(:RoastBatch)<-[:USED_IN]-(:HarvestLot)<-[:HARVESTED]-(f:Farm)-[:HAS_CERTIFICATION]->(cert:Certification)
RETURN DISTINCT f.name, cert.type

%%oc -d {"Farm":"name","Product":"name","QualityIncident":"description","Certification":"type"}
MATCH p=(qi:QualityIncident {incident_id: 'QI01'})-[:REPORTED_FOR]->(:Shipment)<-[:FULFILLED_BY]-(:Order)-[:FOR_PRODUCT]->(:Product)<-[:PACKAGED_AS]-(:RoastBatch)<-[:USED_IN]-(:HarvestLot)<-[:HARVESTED]-(:Farm)-[:HAS_CERTIFICATION]->(:Certification)
RETURN p

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": ["rds.amazonaws.com", "neptune-graph.amazonaws.com"]
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

%%oc
MATCH (n) RETURN count(n) AS node_count

%%oc
MATCH (qi:QualityIncident {incident_id: 'QI01'})-[:REPORTED_FOR]->(:Shipment)<-[:FULFILLED_BY]-(:Order)-[:FOR_PRODUCT]->(:Product)<-[:PACKAGED_AS]-(:RoastBatch)<-[:USED_IN]-(:HarvestLot)<-[:HARVESTED]-(f:Farm)-[:HAS_CERTIFICATION]->(cert:Certification)
RETURN DISTINCT f.name, cert.type