"""Photo selection engine and auditor.

Selection is a SET problem, not a per-image score. Taking the top N by any
quality measure returns N near-identical frames of the best-lit moment. Every
finding this module reports came out of a real selection that shipped, and the
rules below are the ones that were learned the expensive way.

Four rules this module enforces, each one a bug that happened:

1. NAMED, NOT SORTED-AND-SLICED. The first selection here took the first six
   files in name order. It silently dropped the only frame with two subjects in
   it and kept a near-copy of the cover. Sorting by filename is not selection.

2. THE DENY LIST IS FILENAMES, NOT A RULE. A "hold everything in folder X or
   with prefix Y" rule broke the day a held photograph turned up filed under a
   different prefix, outside that folder. Rules about where a file lives are
   guesses about the file. A flat list of names is not.

3. ONE SOURCE FOR COPY AND RENDER. When the list of what gets copied out and
   the list of what gets rendered are two lists, they drift, and the drift
   ships. They are one list here.

4. THE TOOL NEVER DROPS ANYTHING. Rejecting one great frame costs more than
   keeping ten mediocre ones: the person who finds their best shot in the bin
   stops trusting the tool that day, and never gets that trust back. This
   module reports and flags. The human picks. Every threshold below is tuned
   for recall.

And one rule about honesty: an undecidable measurement returns None, and None
is reported as "undecidable", never coerced to a number. A plausible-looking
number for an unmeasurable property is how these systems lose people.

The decision log is not a by-product. It is the point. Every audit writes what
was chosen, what was rejected, what beat what inside a cluster, and why. Those
within-cluster pairs are the highest-value record this produces: scene, light
and subject are held constant, so the only variable left is the thing the
person actually cared about.

No dependency beyond Pillow, which the builders already use.

CLI:
    python scripts/photo_select.py audit --pool DIR --select a.jpg,b.jpg [--deny x.jpg]
    python scripts/photo_select.py cluster --pool DIR
    python scripts/photo_select.py history
"""

from __future__ import annotations

import argparse
import json
import sys
from dataclasses import dataclass, field, asdict
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[1]
RUN_LOG = REPO_ROOT / "state" / "photo-select-runs.jsonl"

IMG_EXT = {".jpg", ".jpeg", ".png", ".webp"}

# RAW formats the engine KNOWS EXIST but cannot analyse yet (embedded-preview
# extraction is the first Phase 1 item). Professionals shoot RAW, so a pool
# walk that silently skips these lies about the pool - the same silent-drop
# failure class this module exists to prevent. They are counted and reported
# as not-analysable, never ignored.
RAW_EXT = {".cr2", ".cr3", ".nef", ".nrw", ".arw", ".raf", ".orf",
           ".rw2", ".dng", ".pef", ".srw", ".x3f"}

PARKED_PREFIX = "x-"

# Working and superseded directories, by the repo's own naming convention. A
# pool walk that does not skip these resurrects frames somebody deliberately
# retired, which is the same class of mistake as ignoring the parked prefix.
SKIP_DIR_PREFIXES = ("_", ".")

# Thresholds. Deliberately loose: this module flags for a human, it does not
# drop, so a false flag costs a glance and a missed duplicate costs a reprint.
# dHash distance is over 64 bits.
SAME_FRAMING = 4        # at or under this, dHash alone settles it (see compare)
DUP_HAMMING = 6         # at or under this, plus histogram agreement -> duplicate
DUP_HIST = 0.85
MAYBE_HAMMING = 12      # at or under this -> worth a human look
MAYBE_HIST = 0.92


# --------------------------------------------------------------------------
# fingerprints
# --------------------------------------------------------------------------

@dataclass
class Fingerprint:
    name: str
    dhash: int | None            # None = could not be read
    histogram: tuple | None
    width: int | None
    height: int | None
    error: str | None = None

    @property
    def readable(self) -> bool:
        return self.dhash is not None and self.histogram is not None


def fingerprint(path: Path) -> Fingerprint:
    """dHash plus a coarse RGB histogram. Returns an unreadable Fingerprint
    rather than raising, so one corrupt file cannot take down a whole audit."""
    from PIL import Image, ImageOps

    try:
        im = ImageOps.exif_transpose(Image.open(path))
        w, h = im.size

        # dHash: 9x8 grey, compare each pixel to its right neighbour.
        g = im.convert("L").resize((9, 8), Image.LANCZOS)
        px = list(g.getdata())
        bits = 0
        for row in range(8):
            for col in range(8):
                left = px[row * 9 + col]
                right = px[row * 9 + col + 1]
                bits = (bits << 1) | (1 if left > right else 0)

        # 4x4x4 RGB histogram, normalised. Catches recolours and exposure
        # shifts that dHash is deliberately blind to.
        small = im.convert("RGB").resize((64, 64), Image.LANCZOS)
        buckets = [0] * 64
        for r, gg, b in small.getdata():
            buckets[(r // 64) * 16 + (gg // 64) * 4 + (b // 64)] += 1
        total = sum(buckets) or 1
        hist = tuple(round(v / total, 6) for v in buckets)

        return Fingerprint(path.name, bits, hist, w, h)
    except Exception as exc:                      # noqa: BLE001 - report, never crash
        return Fingerprint(path.name, None, None, None, None, error=str(exc))


def hamming(a: int, b: int) -> int:
    return bin(a ^ b).count("1")


def hist_similarity(a: tuple, b: tuple) -> float:
    """Histogram intersection. 1.0 = identical distribution."""
    return round(sum(min(x, y) for x, y in zip(a, b)), 6)


def compare(f1: Fingerprint, f2: Fingerprint) -> dict:
    """Verdict on one pair. verdict is one of:
    duplicate / possible / distinct / undecidable."""
    if not (f1.readable and f2.readable):
        return {"a": f1.name, "b": f2.name, "verdict": "undecidable",
                "hamming": None, "hist": None,
                "why": "one or both frames could not be read"}

    ham = hamming(f1.dhash, f2.dhash)
    hs = hist_similarity(f1.histogram, f2.histogram)

    # dHash compares each pixel to its neighbour, so a uniform exposure or
    # contrast lift leaves it untouched by construction. Demanding histogram
    # agreement on top of a near-identical hash throws that away and calls a
    # brightened copy a different photograph. Caught by
    # test_exposure_only_change_is_still_the_same_scene: hamming 1, histogram
    # intersection 0.17, and the pair sailed through. Below SAME_FRAMING the
    # hash settles it alone; the histogram only corroborates the middle band.
    if ham <= SAME_FRAMING:
        verdict, why = "duplicate", (
            f"same framing (hamming {ham}) - exposure or colour may differ, "
            f"the photograph does not (intersection {hs:.2f})")
    elif ham <= DUP_HAMMING and hs >= DUP_HIST:
        verdict, why = "duplicate", (
            f"same framing (hamming {ham}) and same colour distribution "
            f"(intersection {hs:.2f})")
    elif ham <= MAYBE_HAMMING or hs >= MAYBE_HIST:
        verdict, why = "possible", (
            f"close on one axis only (hamming {ham}, intersection {hs:.2f}) "
            f"- a human should look")
    else:
        verdict, why = "distinct", f"hamming {ham}, intersection {hs:.2f}"

    return {"a": f1.name, "b": f2.name, "verdict": verdict,
            "hamming": ham, "hist": hs, "why": why}


def cluster(fps: list[Fingerprint]) -> list[list[str]]:
    """Group frames that are duplicates of each other. Single-link: if A is a
    duplicate of B and B of C, all three are one scene. Bursts are contiguous
    in practice, but nothing here assumes that - EXIF timestamps are wrong,
    missing or in the wrong timezone often enough to not build on them."""
    readable = [f for f in fps if f.readable]
    parent = {f.name: f.name for f in readable}

    def find(x):
        while parent[x] != x:
            parent[x] = parent[parent[x]]
            x = parent[x]
        return x

    def union(x, y):
        rx, ry = find(x), find(y)
        if rx != ry:
            parent[ry] = rx

    for i, f1 in enumerate(readable):
        for f2 in readable[i + 1:]:
            if compare(f1, f2)["verdict"] == "duplicate":
                union(f1.name, f2.name)

    groups: dict[str, list[str]] = {}
    for f in readable:
        groups.setdefault(find(f.name), []).append(f.name)
    return sorted((sorted(v) for v in groups.values()), key=lambda g: g[0])


# --------------------------------------------------------------------------
# the pool
# --------------------------------------------------------------------------

def collect_pool(dirs) -> tuple[dict, list[dict]]:
    """Walk one or more directories into {filename: path}.

    A real pool is a tree, not a folder: the first live audit blocked on a
    frame that was present the whole time, one directory down. Recursion is
    the default because that is what the selections actually look like.

    Filenames are the key, because the deny list is filenames (rule 2). If two
    directories hold the same name that key is ambiguous, so the collision is
    reported rather than silently resolved, and a denied name holds EVERY path
    that carries it. Fail safe, not fail convenient.
    """
    if isinstance(dirs, (str, Path)):
        dirs = [dirs]

    found: dict[str, list[Path]] = {}
    for root in (Path(d) for d in dirs):
        if not root.exists():
            continue
        for p in sorted(root.rglob("*")):
            if p.suffix.lower() not in IMG_EXT or not p.is_file():
                continue
            rel = p.relative_to(root).parts[:-1]
            if any(part.startswith(SKIP_DIR_PREFIXES) for part in rel):
                continue
            found.setdefault(p.name, []).append(p)

    collisions = [{"name": n, "paths": [str(x) for x in ps_]}
                  for n, ps_ in sorted(found.items()) if len(ps_) > 1]
    return {n: ps_[0] for n, ps_ in found.items()}, collisions


def scan_raw(dirs) -> list[str]:
    """RAW filenames present under the pool dirs, same skip rules as
    collect_pool. Reported by audit() so RAW is never silently invisible."""
    if isinstance(dirs, (str, Path)):
        dirs = [dirs]
    names = []
    for root in (Path(d) for d in dirs):
        if not root.exists():
            continue
        for p in sorted(root.rglob("*")):
            if p.suffix.lower() not in RAW_EXT or not p.is_file():
                continue
            rel = p.relative_to(root).parts[:-1]
            if any(part.startswith(SKIP_DIR_PREFIXES) for part in rel):
                continue
            names.append(p.name)
    return sorted(set(names))


# --------------------------------------------------------------------------
# the audit
# --------------------------------------------------------------------------

@dataclass
class Report:
    pool_dirs: list[str]
    pool_size: int
    selected: list[str]
    denied: list[str] = field(default_factory=list)
    findings: list[dict] = field(default_factory=list)
    pairs: list[dict] = field(default_factory=list)
    clusters: list[list[str]] = field(default_factory=list)
    unrepresented: list[list[str]] = field(default_factory=list)
    undecidable: list[str] = field(default_factory=list)

    @property
    def blocking(self) -> list[dict]:
        return [f for f in self.findings if f["level"] == "BLOCK"]

    @property
    def ok(self) -> bool:
        return not self.blocking


def audit(pool_dirs, selected: list[str], denylist: list[str] | None = None) -> Report:
    """Audit a selection against its pool. Never mutates the selection.

    `pool_dirs` is a directory or a list of them; each is walked recursively.
    """
    denylist = denylist or []
    if isinstance(pool_dirs, (str, Path)):
        pool_dirs = [pool_dirs]

    by_name, collisions = collect_pool(pool_dirs)
    raw_names = set(scan_raw(pool_dirs))
    pool_paths = [by_name[n] for n in sorted(by_name)]
    rep = Report([str(d) for d in pool_dirs], len(pool_paths),
                 list(selected), list(denylist))

    if raw_names:
        rep.findings.append({
            "level": "WARN", "kind": "raw-not-analysable",
            "subject": f"{len(raw_names)} RAW file(s) in the pool",
            "detail": "present but not analysed - the engine reads processed "
                      "formats only until embedded-preview support lands. "
                      "These frames are invisible to every check below."})

    for c in collisions:
        rep.findings.append({
            "level": "WARN", "kind": "name-collision",
            "subject": c["name"],
            "detail": "same filename in more than one directory (" +
                      "; ".join(c["paths"]) + ") - a held name holds all of them"})

    # -- rule 2: the deny list is filenames, matched flat, wherever they live.
    for name in sorted(set(selected) & set(denylist)):
        rep.findings.append({
            "level": "BLOCK", "kind": "denylisted",
            "subject": name,
            "detail": "a held frame is in the selection - matched by filename, "
                      "not by folder or prefix"})

    # -- parked files
    for name in sorted(n for n in selected if n.startswith(PARKED_PREFIX)):
        rep.findings.append({
            "level": "BLOCK", "kind": "parked",
            "subject": name,
            "detail": f"parked frame ({PARKED_PREFIX} prefix) is in the selection"})

    # -- missing from pool. A selected RAW that exists on disk is a different
    #    finding from a frame that is not there at all: it will not render a
    #    hole, but nothing about it was checked, and undecidable must never
    #    silently pass as fine.
    for name in sorted(set(selected) - set(by_name)):
        if name in raw_names:
            rep.undecidable.append(name)
            rep.findings.append({
                "level": "WARN", "kind": "raw-selected",
                "subject": name,
                "detail": "selected frame is a RAW file the engine cannot "
                          "analyse yet - present on disk, checked by nothing"})
        else:
            rep.findings.append({
                "level": "BLOCK", "kind": "missing",
                "subject": name,
                "detail": "selected frame is not in the pool - the selection "
                          "would render a hole"})

    present = [n for n in selected if n in by_name]
    sel_fps = [fingerprint(by_name[n]) for n in present]
    rep.undecidable = sorted(set(rep.undecidable) |
                             {f.name for f in sel_fps if not f.readable})
    for f in sel_fps:
        if not f.readable:
            rep.findings.append({
                "level": "WARN", "kind": "undecidable", "subject": f.name,
                "detail": f"could not be read, so it was not checked: {f.error}"})

    # -- rule 1: near-duplicates INSIDE the selection. This is the bug that
    #    shipped: a near-copy of the cover kept, a unique frame dropped.
    for i, f1 in enumerate(sel_fps):
        for f2 in sel_fps[i + 1:]:
            v = compare(f1, f2)
            if v["verdict"] in ("duplicate", "possible"):
                rep.pairs.append(v)
                rep.findings.append({
                    "level": "BLOCK" if v["verdict"] == "duplicate" else "WARN",
                    "kind": f"near-duplicate:{v['verdict']}",
                    "subject": f"{v['a']} + {v['b']}",
                    "detail": v["why"]})

    # -- scene coverage: what the pool offered that the selection never used.
    pool_fps = [fingerprint(p) for p in pool_paths
                if p.name not in denylist and not p.name.startswith(PARKED_PREFIX)]
    rep.clusters = cluster(pool_fps)
    chosen = set(present)
    for group in rep.clusters:
        if not (set(group) & chosen):
            rep.unrepresented.append(group)

    covered = sum(1 for g in rep.clusters if set(g) & chosen)
    if rep.clusters and len(present) > covered:
        rep.findings.append({
            "level": "WARN", "kind": "diversity",
            "subject": f"{len(present)} frames over {covered} scenes",
            "detail": "the selection spends more frames than it shows scenes - "
                      "some of these are angles on the same moment"})

    return rep


# --------------------------------------------------------------------------
# the log - the reason this module exists
# --------------------------------------------------------------------------

def manifest_markdown(rep: Report, label: str) -> str:
    L = [f"# Selection audit - {label}", ""]
    L.append("Pool: " + ", ".join(f"`{d}`" for d in rep.pool_dirs) +
             f" ({rep.pool_size} frames)")
    L.append(f"Selected: {len(rep.selected)}  |  Held: {len(rep.denied)}")
    L.append(f"Verdict: {'PASS' if rep.ok else 'BLOCKED'}")
    L.append("")

    if rep.findings:
        L += ["## Findings", "", "| Level | Kind | Subject | Detail |",
              "|---|---|---|---|"]
        for f in rep.findings:
            L.append(f"| {f['level']} | {f['kind']} | {f['subject']} | {f['detail']} |")
        L.append("")
    else:
        L += ["## Findings", "", "None. Every selected frame is present, "
              "distinct, and not held.", ""]

    if rep.pairs:
        L += ["## Pairs a human should decide", "",
              "Scene, light and subject are near-identical inside each pair, so "
              "whichever one is kept records a preference with everything else "
              "held constant. This is the part worth keeping.", "",
              "| A | B | Verdict | Hamming | Colour |", "|---|---|---|---|---|"]
        for p in rep.pairs:
            L.append(f"| {p['a']} | {p['b']} | {p['verdict']} | "
                     f"{p['hamming']} | {p['hist']} |")
        L.append("")

    if rep.unrepresented:
        L += ["## Scenes available and not used", "",
              "Not a criticism. A record of what the selection passed over, so "
              "a later reader can tell a deliberate omission from an oversight.", ""]
        for g in rep.unrepresented:
            L.append(f"- {', '.join(g)}")
        L.append("")

    if rep.undecidable:
        L += ["## Undecidable", "",
              "Read failed, so these were NOT checked. Not the same as passing.", ""]
        for n in rep.undecidable:
            L.append(f"- {n}")
        L.append("")

    L += ["---", "",
          "Generated by `scripts/photo_select.py`. This tool flags and never "
          "drops: a BLOCK is a question for a human, not a deletion."]
    return "\n".join(L) + "\n"


def append_run(rep: Report, label: str) -> None:
    RUN_LOG.parent.mkdir(parents=True, exist_ok=True)
    row = {
        "label": label,
        "pool": rep.pool_dirs,
        "pool_size": rep.pool_size,
        "selected": len(rep.selected),
        "held": len(rep.denied),
        "blocking": len(rep.blocking),
        "warnings": len([f for f in rep.findings if f["level"] == "WARN"]),
        "scenes_in_pool": len(rep.clusters),
        "scenes_unused": len(rep.unrepresented),
        "verdict": "PASS" if rep.ok else "BLOCKED",
    }
    with RUN_LOG.open("a", encoding="utf-8") as fh:
        fh.write(json.dumps(row) + "\n")


# --------------------------------------------------------------------------
# CLI
# --------------------------------------------------------------------------

def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
    sub = ap.add_subparsers(dest="cmd", required=True)

    a = sub.add_parser("audit", help="audit a selection against its pool")
    a.add_argument("--pool", required=True,
                   help="directory, or comma-separated directories; walked recursively")
    a.add_argument("--select", required=True, help="comma-separated filenames")
    a.add_argument("--deny", default="", help="comma-separated held filenames")
    a.add_argument("--label", default="unlabelled")
    a.add_argument("--out", default="", help="write the markdown manifest here")
    a.add_argument("--log", action="store_true", help="append to the run history")

    c = sub.add_parser("cluster", help="show the distinct scenes in a pool")
    c.add_argument("--pool", required=True)

    e = sub.add_parser("eval", help="score a proposed selection against what a human kept")
    e.add_argument("--proposed", required=True, help="comma-separated filenames the tool proposed")
    e.add_argument("--kept", required=True, help="comma-separated filenames the human kept")
    e.add_argument("--pool", default="", help="optional pool dir(s) - enables within-scene swap extraction")
    e.add_argument("--label", default="unlabelled")
    e.add_argument("--notes", default="")

    sub.add_parser("history", help="print the run history")

    ns = ap.parse_args(argv)

    if ns.cmd == "cluster":
        by_name, _ = collect_pool([x.strip() for x in ns.pool.split(",") if x.strip()])
        fps = [fingerprint(by_name[n]) for n in sorted(by_name)]
        groups = cluster(fps)
        print(f"{len(fps)} frames -> {len(groups)} distinct scenes")
        for i, g in enumerate(groups, 1):
            marker = "  (one frame)" if len(g) == 1 else f"  ({len(g)} frames)"
            print(f"{i:3}. {g[0]}{marker}")
            for extra in g[1:]:
                print(f"       + {extra}")
        return 0

    if ns.cmd == "eval":
        proposed = [x.strip() for x in ns.proposed.split(",") if x.strip()]
        kept = [x.strip() for x in ns.kept.split(",") if x.strip()]
        scored = agreement(proposed, kept)
        swaps = []
        if ns.pool:
            by_name, _ = collect_pool([x.strip() for x in ns.pool.split(",") if x.strip()])
            fps = [fingerprint(by_name[n]) for n in sorted(by_name)]
            swaps = swap_pairs(proposed, kept, cluster(fps))
        record_eval(ns.label, scored, swaps, ns.notes)
        ag = scored.get("agreement"); ms = scored.get("miss")
        print(f"agreement: {'n/a' if ag is None else f'{ag:.0%}'}   "
              f"miss: {'n/a' if ms is None else f'{ms:.0%}'}   "
              f"(proposed {scored['n_proposed']}, kept {scored['n_kept']})")
        for n in scored.get("never_offered", []):
            print(f"  never offered: {n}   <- the trust-killer column")
        for sw in swaps:
            print(f"  swap in scene {sw['scene']}: tool {sw['tool_offered']} "
                  f"-> human {sw['human_took']}")
        print(f"recorded -> {EVAL_LOG}")
        return 0

    if ns.cmd == "history":
        if not RUN_LOG.exists():
            print("no runs recorded yet")
            return 0
        for line in RUN_LOG.read_text(encoding="utf-8").splitlines():
            r = json.loads(line)
            print(f"{r['verdict']:8} {r['label']:32} "
                  f"{r['selected']:>3}/{r['pool_size']:<4} frames  "
                  f"{r['blocking']} blocking, {r['warnings']} warn")
        return 0

    rep = audit([Path(x.strip()) for x in ns.pool.split(",") if x.strip()],
                [s.strip() for s in ns.select.split(",") if s.strip()],
                [s.strip() for s in ns.deny.split(",") if s.strip()])

    md = manifest_markdown(rep, ns.label)
    if ns.out:
        Path(ns.out).write_text(md, encoding="utf-8")
        print(f"manifest written: {ns.out}")
    else:
        print(md)

    if ns.log:
        append_run(rep, ns.label)

    return 0 if rep.ok else 1


if __name__ == "__main__":
    sys.exit(main())


# --------------------------------------------------------------------------
# the eval harness - the only accuracy numbers that mean anything
# --------------------------------------------------------------------------

EVAL_LOG = REPO_ROOT / "state" / "photo-select-evals.jsonl"


def agreement(proposed: list[str], kept: list[str]) -> dict:
    """Score a proposed selection against what the human actually kept.

    Two numbers, and they are not symmetric:

    AGREEMENT is how much of the proposal survived. It is the headline, and
    it is the forgiving one.

    MISS is the share of the human's own picks the tool never offered. This
    is the number that ends the relationship. Somebody who finds their best
    frame was never shown to them stops trusting the tool that day, and a
    high agreement rate does not buy that back. Tune for miss, report both.
    """
    p, k = set(proposed), set(kept)
    if not p and not k:
        return {"agreement": None, "miss": None, "n_proposed": 0, "n_kept": 0,
                "why": "nothing proposed and nothing kept - undecidable"}

    agree = round(len(p & k) / len(p), 4) if p else None
    miss = round(len(k - p) / len(k), 4) if k else None
    return {
        "agreement": agree,
        "miss": miss,
        "n_proposed": len(p),
        "n_kept": len(k),
        "survived": sorted(p & k),
        "dropped_by_human": sorted(p - k),
        "never_offered": sorted(k - p),
    }


def swap_pairs(proposed: list[str], kept: list[str], clusters: list[list[str]]) -> list[dict]:
    """The training signal, extracted.

    A swap inside one scene is worth far more than a global accept or reject.
    Scene, light and subject are held constant, so the only variable left is
    the thing the person actually cared about. A global reject confounds "bad
    frame" with "we already have one of these" and teaches the wrong lesson.

    Returns one row per within-scene swap: what the tool offered, what the
    human took instead, and which scene they were arguing about.
    """
    p, k = set(proposed), set(kept)
    out = []
    for i, group in enumerate(clusters):
        g = set(group)
        lost = sorted((p - k) & g)      # tool offered it, human dropped it
        won = sorted((k - p) & g)       # human reached past it for this one
        if lost and won:
            out.append({"scene": i, "tool_offered": lost,
                        "human_took": won, "members": sorted(g)})
    return out


def record_eval(label: str, scored: dict, swaps: list[dict],
                notes: str = "") -> None:
    """Append one evaluation. This file is the test log: every run of the
    tool against a real human decision, in one place, diffable over time.
    Agreement drift across runs for one person is the whole thesis."""
    EVAL_LOG.parent.mkdir(parents=True, exist_ok=True)
    row = {"label": label, "agreement": scored["agreement"],
           "miss": scored["miss"], "n_proposed": scored["n_proposed"],
           "n_kept": scored["n_kept"], "swaps": len(swaps), "notes": notes}
    with EVAL_LOG.open("a", encoding="utf-8") as fh:
        fh.write(json.dumps(row) + "\n")
