"""Cross-source recipe demo (acceptance A9 + A8): World Bank GDP (current US$, country-year) × Frankfurter USD→CAD daily rates
(aggregated to yearly means) → GDP in CAD, with validated identifiers, units, cardinality, row counts and lineage; then a
verified research export. Also demonstrates the two refusals (granularity mismatch without aggregation; unit mismatch in +).

    AKOS_URL=https://akos.spboucher.ai AKOS_TOKEN=akos_… python demo_research.py [--project research-demo] [--out /tmp/export]
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import time
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
from akos.sdk import Akos, AkosError  # noqa: E402

ap = argparse.ArgumentParser(); ap.add_argument("--project", default="research-demo"); ap.add_argument("--out", default="/tmp/akos-export"); ap.add_argument("--skip-discover", action="store_true")
a = ap.parse_args()
A = Akos()


def step(msg):
    print(f"\n### {msg}")


def wait(job_id, timeout=600):
    j = A.wait_job(job_id, timeout=timeout)
    if j["state"] != "done":
        print("JOB FAILED:", j.get("error", "")[:800]); sys.exit(1)
    return j["result"]


step("project in research mode")
try:
    A._req("POST", "/projects", json={"slug": a.project, "name": "Research demo (GDP in CAD)", "policy_mode": "research"})
except AkosError as e:
    if e.status != 409: raise
A._req("PATCH", f"/projects/{a.project}", json={"policy_mode": "research"})

if not a.skip_discover:
    step("discover: Frankfurter 2015–2023 range endpoint (probe, merged into the existing API) ")
    r = A.discover(project_id=a.project, probe_url="https://api.frankfurter.dev/v1/2015-01-01..2023-12-31?base=USD&symbols=CAD", name="Frankfurter exchange rates", provider_slug="frankfurter",
                   provider_name="Frankfurter (ECB reference rates)", license="ECB reference rates — free reuse with attribution; Frankfurter MIT", records_path="$.rates")
    res = wait(r["job_id"]); print(json.dumps(res, indent=1)[:600]); vres = wait(res["verify_job_id"]); print("verify:", vres.get("promoted_to"), [(s["step"], s["result"]) for s in vres["steps"]])

pk_fx = A.package("frankfurter.frankfurter-exchange-rates"); pk_gdp = A.package("worldbank.world-bank-indicators")
op_fx = next(e["operation_id"] for e in pk_fx["endpoints"] if "2015" in e["path"]); op_gdp = pk_gdp["endpoints"][0]["operation_id"]
print("packages:", pk_fx["akp_id"], f"v{pk_fx['version']}", pk_fx["lifecycle"], "|", pk_gdp["akp_id"], f"v{pk_gdp['version']}", pk_gdp["lifecycle"])

step("governed acquisitions in research mode (policy requires tested AKP + license)")
ex_gdp = A.execute(project_id=a.project, package_id=pk_gdp["id"], operation_id=op_gdp, query={"format": "json", "date": "2015:2023", "per_page": "100"}, purpose="GDP current US$ Canada 2015-2023", caller="demo")
print("gdp:", ex_gdp["state"], ex_gdp["status_code"], ex_gdp["response_bytes"], "bytes", ex_gdp["response_hash"][:16], "policy:", ex_gdp["policy_decision"]["reasons"])
ex_fx = A.execute(project_id=a.project, package_id=pk_fx["id"], operation_id=op_fx, query={"base": "USD", "symbols": "CAD"}, purpose="USD/CAD daily reference rates 2015-2023", caller="demo")
print("fx:", ex_fx["state"], ex_fx["status_code"], ex_fx["response_bytes"], "bytes", ex_fx["response_hash"][:16], "policy:", ex_fx["policy_decision"]["reasons"])

step("datasets with identifiers, units, granularity")
ds_gdp = A.create_dataset(project_id=a.project, name="GDP current US$ — Canada 2015-2023", execution_ids=[ex_gdp["id"]], records_path="$[1]", identifiers=["countryiso3code", "date"], units={"value": "USD"}, granularity="country-year")
print("gdp dataset:", ds_gdp["id"], ds_gdp["row_count"], "rows", ds_gdp["verification"], "columns:", list(ds_gdp["schema_json"]["columns"])[:8])
ds_fx = A.create_dataset(project_id=a.project, name="USD/CAD daily rates 2015-2023", execution_ids=[ex_fx["id"]], records_path="$.rates", identifiers=["key"], units={"CAD": "CAD/USD"}, granularity="day", normalize={"derive": [{"name": "year", "expr": "substr(key,0,4)"}]})
print("fx dataset:", ds_fx["id"], ds_fx["row_count"], "rows", ds_fx["verification"], "columns:", list(ds_fx["schema_json"]["columns"]))
for d in (ds_gdp, ds_fx):
    bad = [c for c in d["assessment"]["checks"] if not c["ok"]]
    print("  assessment", d["name"][:30], "verified" if d["assessment"]["verified"] else f"NOT verified: {bad}")

step("refusal 1: join day-level FX with country-year GDP without aggregation → granularity mismatch")
bad1 = A.compose(project_id=a.project, name="gdp-cad-bad-granularity", left_dataset_id=ds_gdp["id"], right_dataset_id=ds_fx["id"], on=[{"left": "date", "right": "year"}], how="inner")
print("status:", bad1["status"], "errors:", bad1["validation"]["errors"])
step("refusal 2: unit mismatch in addition (USD + CAD/USD)")
bad2 = A.compose(project_id=a.project, name="gdp-cad-bad-units", left_dataset_id=ds_gdp["id"], right_dataset_id=ds_fx["id"], on=[{"left": "date", "right": "year"}], aggregate={"side": "right", "by": ["year"], "agg": {"CAD": "mean"}}, derive=[{"name": "nonsense", "expr": "left.value + right.CAD"}])
print("status:", bad2["status"], "errors:", bad2["validation"]["errors"])

step("valid recipe: yearly mean USD→CAD × GDP USD = GDP CAD")
rec = A.compose(project_id=a.project, name="gdp-cad", left_dataset_id=ds_gdp["id"], right_dataset_id=ds_fx["id"], on=[{"left": "date", "right": "year"}], how="inner",
                aggregate={"side": "right", "by": ["year"], "agg": {"CAD": "mean"}}, derive=[{"name": "gdp_cad", "expr": "left.value * right.CAD", "unit": "CAD"}], output_name="GDP Canada in CAD (yearly mean FX) 2015-2023")
v = rec["validation"]
print("status:", rec["status"], "cardinality:", v.get("cardinality"), "matched:", v.get("matched_keys"), "left/right rows:", v.get("left_rows"), v.get("right_rows"), "unmatched:", v.get("unmatched_left"), v.get("unmatched_right"), "units:", v.get("units"), "warnings:", v.get("warnings"))
if rec["status"] != "validated":
    print("errors:", v.get("errors")); sys.exit(1)
run = A.run_recipe(rec["id"])
out = run["dataset"]
print("output dataset:", out["id"], out["row_count"], "rows, verification:", out["verification"], "hash:", out["hash"][:16])
print("sample:", json.dumps(A.dataset(out["id"])["preview"][:2], default=str)[:400])
lin = A.lineage(out["id"]); print("lineage:", len(lin["nodes"]), "nodes", len(lin["edges"]), "edges; types:", sorted({n["type"] for n in lin["nodes"]}))

step("verified research export")
r = A.export(out["id"], verified=True); res = wait(r["job_id"])
if res.get("blocked"):
    print("BLOCKED:", res["reasons"]); sys.exit(2)
os.makedirs(a.out, exist_ok=True)
path = os.path.join(a.out, f"akos-export-{out['id'][:8]}.zip")
open(path, "wb").write(A.artifact(res["artifact_id"]))
print("export:", res["verification_label"], path, res["bytes"], "bytes; expected output hash", res["expected_output_hash"][:16])
json.dump({"gdp_execution": ex_gdp["id"], "fx_execution": ex_fx["id"], "ds_gdp": ds_gdp["id"], "ds_fx": ds_fx["id"], "recipe": rec["id"], "output_dataset": out["id"], "export_zip": path, "export_sha256": res["sha256"], "expected_output_hash": res["expected_output_hash"]}, open(os.path.join(a.out, "demo-ids.json"), "w"), indent=1)
print("ids →", os.path.join(a.out, "demo-ids.json"))