diff --git a/src/app/api/portfolio/[portfolioId]/bulk-uploads/[uploadId]/address-corrections/route.ts b/src/app/api/portfolio/[portfolioId]/bulk-uploads/[uploadId]/address-corrections/route.ts new file mode 100644 index 00000000..6ae418fa --- /dev/null +++ b/src/app/api/portfolio/[portfolioId]/bulk-uploads/[uploadId]/address-corrections/route.ts @@ -0,0 +1,43 @@ +import { NextRequest, NextResponse } from "next/server"; +import { getServerSession } from "next-auth"; +import { z } from "zod"; + +import { AuthOptions } from "@/app/api/auth/[...nextauth]/authOptions"; +import { upsertUprnCorrection } from "@/lib/bulkUpload/addressMatches"; + +// Confirm one flagged combiner row (ADR-0057): either assign a UPRN chosen from +// OS Places, or assert the address has none. Keyed by `source_row_id` (there is +// no property.id yet — the finaliser creates identities). Upsert semantics, so +// re-confirming overwrites. The confirmation is overlaid onto the combiner CSV +// at dispatchFinaliser. +const AssignSchema = z.object({ + sourceRowId: z.string().min(1), + uprn: z.string().regex(/^\d+$/, "UPRN must be numeric"), + address: z.string().min(1), + postcode: z.string().min(1), +}); +const NoUprnSchema = z.object({ + sourceRowId: z.string().min(1), + markedNoUprn: z.literal(true), +}); +const BodySchema = z.union([NoUprnSchema, AssignSchema]); + +export async function POST( + request: NextRequest, + { params }: { params: Promise<{ portfolioId: string; uploadId: string }> }, +) { + const session = await getServerSession(AuthOptions); + if (!session) return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + + const { uploadId } = await params; + const parsed = BodySchema.safeParse(await request.json().catch(() => null)); + if (!parsed.success) { + return NextResponse.json( + { error: parsed.error.issues[0]?.message ?? "Invalid request" }, + { status: 400 }, + ); + } + + await upsertUprnCorrection(uploadId, parsed.data, session.user?.email ?? undefined); + return NextResponse.json({ ok: true }, { status: 200 }); +} diff --git a/src/app/api/portfolio/[portfolioId]/bulk-uploads/[uploadId]/address-matches/route.ts b/src/app/api/portfolio/[portfolioId]/bulk-uploads/[uploadId]/address-matches/route.ts new file mode 100644 index 00000000..a585aba7 --- /dev/null +++ b/src/app/api/portfolio/[portfolioId]/bulk-uploads/[uploadId]/address-matches/route.ts @@ -0,0 +1,22 @@ +import { NextRequest, NextResponse } from "next/server"; +import { getServerSession } from "next-auth"; + +import { AuthOptions } from "@/app/api/auth/[...nextauth]/authOptions"; +import { getFlaggedAddresses } from "@/lib/bulkUpload/addressMatches"; + +// Read-only: the combiner rows address2uprn could not confidently match +// (unmatched / ambiguous_duplicate / invalid_postcode), joined with any saved +// correction. Drives the pre-finalise "Addresses" review tab (ADR-0057); the +// Finalise gate blocks until every flagged row is resolved. +export async function GET( + _request: NextRequest, + { params }: { params: Promise<{ portfolioId: string; uploadId: string }> }, +) { + const session = await getServerSession(AuthOptions); + if (!session) return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + + const { uploadId } = await params; + const flagged = await getFlaggedAddresses(uploadId); + const unresolved = flagged.filter((f) => !f.resolved).length; + return NextResponse.json({ flagged, unresolved }, { status: 200 }); +} diff --git a/src/lib/bulkUpload/addressMatches.ts b/src/lib/bulkUpload/addressMatches.ts new file mode 100644 index 00000000..198a4c0b --- /dev/null +++ b/src/lib/bulkUpload/addressMatches.ts @@ -0,0 +1,177 @@ +import { eq } from "drizzle-orm"; +import * as XLSX from "xlsx"; + +import { db } from "@/app/db/db"; +import { bulkAddressUploads } from "@/app/db/schema/bulk_address_uploads"; +import { bulkUploadUprnCorrections } from "@/app/db/schema/bulk_upload_uprn_corrections"; +import { createRetrofitDataS3Client } from "@/app/utils/s3"; +import { SOURCE_ROW_ID_COLUMN } from "./s3Keys"; + +// Columns address2uprn writes onto each combiner row (ADR-0057). `status` is the +// per-row outcome; a withheld/ambiguous/no-match row has an empty `uprn`. +const UPRN_COL = "address2uprn_uprn"; +const STATUS_COL = "address2uprn_status"; +const LEXISCORE_COL = "address2uprn_lexiscore"; + +// A combiner row is confidently matched only when address2uprn kept a UPRN AND +// did not flag it. Everything else — unmatched, ambiguous_duplicate, +// invalid_postcode, error, or a legacy row with no status but no UPRN — needs +// the user to resolve it on the Addresses tab before finalise. +function isConfidentlyMatched(row: Record): boolean { + const uprn = (row[UPRN_COL] ?? "").trim(); + const status = (row[STATUS_COL] ?? "").trim(); + if (!uprn) return false; + return status === "" || status === "matched"; +} + +export interface FlaggedAddressRow { + sourceRowId: string; + address: string; + postcode: string; + status: string; + lexiscore: number | null; + // The candidate address2uprn found but did not trust (null when none). + suggestedUprn: string | null; + // The user's confirmation (from bulk_upload_uprn_corrections), if any. + resolvedUprn: string | null; + resolvedAddress: string | null; + markedNoUprn: boolean; + // Resolved = user picked a UPRN OR asserted there is none. + resolved: boolean; +} + +function parseS3Uri(uri: string): { bucket: string; key: string } { + const url = new URL(uri); // s3://bucket/key... + const key = url.pathname.startsWith("/") ? url.pathname.slice(1) : url.pathname; + if (!url.hostname || !key) throw new Error(`Malformed S3 URI: ${uri}`); + return { bucket: url.hostname, key }; +} + +async function readCombinerRows(s3Uri: string): Promise[]> { + const { bucket, key } = parseS3Uri(s3Uri); + const s3 = createRetrofitDataS3Client(); + const result = await s3.getObject({ Bucket: bucket, Key: key }).promise(); + const body = result.Body?.toString("utf-8"); + if (!body) return []; + const wb = XLSX.read(body, { type: "string" }); + const sheet = wb.Sheets[wb.SheetNames[0]]; + // Every cell as a string so a numeric UPRN never arrives as a float. + return XLSX.utils.sheet_to_json>(sheet, { defval: "", raw: false }); +} + +function buildAddress(row: Record): string { + return ["Address 1", "Address 2", "Address 3"] + .map((c) => (row[c] ?? "").trim()) + .filter(Boolean) + .join(", "); +} + +/** + * The combiner rows that need user confirmation, joined with any saved + * correction. Reads the combiner CSV from `combinedOutputS3Uri` (ADR-0057). + * Returns [] until the combiner has run. + */ +export async function getFlaggedAddresses( + uploadId: string, +): Promise { + const [row] = await db + .select({ combinedOutputS3Uri: bulkAddressUploads.combinedOutputS3Uri }) + .from(bulkAddressUploads) + .where(eq(bulkAddressUploads.id, uploadId)); + if (!row?.combinedOutputS3Uri) return []; + + const [rows, corrections] = await Promise.all([ + readCombinerRows(row.combinedOutputS3Uri), + db + .select() + .from(bulkUploadUprnCorrections) + .where(eq(bulkUploadUprnCorrections.uploadId, uploadId)), + ]); + + const correctionByRow = new Map(corrections.map((c) => [c.sourceRowId, c])); + + return rows + .filter((r) => !isConfidentlyMatched(r)) + .map((r) => { + const sourceRowId = (r[SOURCE_ROW_ID_COLUMN] ?? "").trim(); + const uprn = (r[UPRN_COL] ?? "").trim(); + const rawScore = (r[LEXISCORE_COL] ?? "").trim(); + const lexiscore = rawScore === "" ? null : Number(rawScore); + const c = correctionByRow.get(sourceRowId); + const resolvedUprn = c?.uprn ?? null; + const markedNoUprn = c?.markedNoUprn ?? false; + return { + sourceRowId, + address: buildAddress(r), + postcode: (r["postcode"] ?? "").trim(), + status: (r[STATUS_COL] ?? (uprn ? "matched" : "unmatched")).trim(), + lexiscore: lexiscore !== null && Number.isFinite(lexiscore) ? lexiscore : null, + suggestedUprn: uprn || null, + resolvedUprn, + resolvedAddress: c?.address ?? null, + markedNoUprn, + resolved: resolvedUprn !== null || markedNoUprn, + }; + }); +} + +/** Count of flagged rows the user has not yet resolved — gates Finalise. */ +export async function countUnresolvedFlaggedAddresses(uploadId: string): Promise { + const flagged = await getFlaggedAddresses(uploadId); + return flagged.filter((f) => !f.resolved).length; +} + +export type UprnCorrectionInput = + | { + sourceRowId: string; + uprn: string; + address: string; + postcode: string; + markedNoUprn?: false; + } + | { sourceRowId: string; markedNoUprn: true }; + +/** Upsert one combiner row's confirmation (assign a UPRN, or mark no-UPRN). */ +export async function upsertUprnCorrection( + uploadId: string, + input: UprnCorrectionInput, + userId?: string, +): Promise { + const values = input.markedNoUprn + ? { + uploadId, + sourceRowId: input.sourceRowId, + uprn: null, + address: null, + postcode: null, + markedNoUprn: true as const, + userId: userId ?? null, + } + : { + uploadId, + sourceRowId: input.sourceRowId, + uprn: input.uprn, + address: input.address, + postcode: input.postcode, + markedNoUprn: false as const, + userId: userId ?? null, + }; + + await db + .insert(bulkUploadUprnCorrections) + .values(values) + .onConflictDoUpdate({ + target: [ + bulkUploadUprnCorrections.uploadId, + bulkUploadUprnCorrections.sourceRowId, + ], + set: { + uprn: values.uprn, + address: values.address, + postcode: values.postcode, + markedNoUprn: values.markedNoUprn, + userId: values.userId, + updatedAt: new Date(), + }, + }); +}