import { z } from 'zod';
import { BaseConnector, type ConnectorMeta, type CrawlContext, type RawRecordInput, type RawRecordLike } from '@rareindex/connectors';
import type { NormalizedRecord, NormalizedSale } from '@rareindex/shared';
import { dateWords, makeSale, md, money, splitMarkdownItems, vehicleAttributes } from '../_carlib/index.js';

const BASE = 'https://rmsothebys.com';
const PARSER_VERSION = '1.0.0';

export const AuctionSchema = z.object({ code: z.string(), name: z.string().nullable(), dateText: z.string().nullable() });
export const LotSchema = z.object({ slug: z.string(), url: z.string(), title: z.string(), lotNumber: z.string().nullable(), priceText: z.string().nullable(), status: z.string().nullable(), image: z.string().nullable() });
export const PagePayloadSchema = z.object({ kind: z.literal('lots_page'), auction: AuctionSchema, lots: z.array(LotSchema) });
export type PagePayload = z.infer<typeof PagePayloadSchema>;

/** /results/ markdown: "Auction Name | 13 - 15 August 2026 | [View Results](…/auctions/mo26/lots/)" blocks. */
export function parseResults(markdown: string): z.infer<typeof AuctionSchema>[] {
  const out: z.infer<typeof AuctionSchema>[] = [];
  const re = /\[View Results\]\(https:\/\/rmsothebys\.com\/auctions\/([a-z0-9]+)\/lots\/[^)]*\)/g;
  let m: RegExpExecArray | null;
  while ((m = re.exec(markdown))) {
    const before = markdown.slice(Math.max(0, m.index - 600), m.index);
    const lines = before.split('\n').map((l) => l.trim()).filter((l) => l && !l.startsWith('!') && !/^[‹›]+$/.test(l));
    const dateText = [...lines].reverse().find((l) => /\d{1,2}\s*[-–]?\s*\d{0,2}\s*[A-Z][a-z]+\s+\d{4}|^[A-Z][a-z]+\s+\d{1,2},?\s+\d{4}|Bidding Closes/.test(l)) ?? null;
    const name = [...lines].reverse().find((l) => l !== dateText && !/View Results|Bidding Closes/.test(l) && l.length > 3) ?? null;
    if (!out.some((a) => a.code === m![1])) out.push({ code: m[1]!, name, dateText: dateText?.replace(/^Bidding Closes\s*/i, '') ?? null });
  }
  return out;
}

export function parseLotsPage(markdown: string, auction: z.infer<typeof AuctionSchema>): PagePayload {
  const chunks = splitMarkdownItems(markdown, /^\[!\[[^\]]*\]\([^)]+\)\]\(https:\/\/rmsothebys\.com\/auctions\/[a-z0-9]+\/lots\//m);
  const lots: z.infer<typeof LotSchema>[] = [];
  for (const c of chunks) {
    const url = c.match(/\]\((https:\/\/rmsothebys\.com\/auctions\/[a-z0-9]+\/lots\/([a-z0-9-]+)\/)\)/);
    if (!url) continue;
    const body = c.match(/\[\*\*[^*]+\*\*\s*\\?\s*\n?([\s\S]*?)\]\(https:\/\/rmsothebys\.com\/auctions\/[a-z0-9]+\/lots\//)?.[1] ?? c;
    const lines = body.split('\n').map((l) => l.replace(/\\+$/, '').replace(/^\\+/, '').trim()).filter(Boolean);
    const title = lines[0] ?? null;
    if (!title) continue;
    const lotLine = lines.find((l) => /^Lot\s+\S+/.test(l)) ?? '';
    const lotNumber = lotLine.match(/^Lot\s+([A-Za-z0-9.]+)/)?.[1] ?? null;
    const priceText = lotLine.match(/\|\s*(.+)$/)?.[1]?.trim() ?? null;
    const status = lines.find((l) => /^(Sold|Not Sold|Withdrawn|Lot Sold|Lot Closed)$/i.test(l)) ?? null;
    lots.push({ slug: url[2]!, url: url[1]!, title: md.clean(title), lotNumber, priceText: priceText && /\d/.test(priceText) ? priceText : null, status, image: md.image(c) });
  }
  return { kind: 'lots_page', auction, lots };
}

const MEMORABILIA = /\b(sculpture|poster|sign|helmet|model|artwork|painting|trophy|pedal car|neon|literature|memorabilia|watch)\b/i;

export class RMSothebysConnector extends BaseConnector {
  readonly version = '1.0.0';
  readonly parserVersion = PARSER_VERSION;
  protected override minIntervalMs = 2000;

  async *crawl(ctx: CrawlContext): AsyncIterable<RawRecordInput> {
    const auctionsPerRun = Number(this.meta.config.auctionsPerRun ?? 3);
    const done = new Set<string>(Array.isArray(ctx.options.cursor?.doneAuctions) ? (ctx.options.cursor!.doneAuctions as string[]) : []);
    const year = ctx.options.mode === 'backfill' ? Number(ctx.options.cursor?.year ?? new Date().getUTCFullYear()) : new Date().getUTCFullYear();
    const listUrl = `${BASE}/results/${ctx.options.mode === 'backfill' ? `?year=${year}` : ''}`;
    await this.throttle();
    const list = await ctx.fetch(listUrl, { waitForMs: 5000, expect: ['title', 'date'], parse: (r) => (r.markdown ? { title: parseResults(r.markdown)[0]?.name ?? null, date: parseResults(r.markdown)[0]?.dateText ?? null } : null), minQuality: 0.3 });
    if (!list.success || !list.markdown) {
      ctx.anomaly('page_fetch_failed', `${listUrl}: ${list.error ?? list.httpStatus}`);
      return;
    }
    const now = Date.now();
    const skipSealed = this.meta.config.skipSealed !== false;
    const auctions = parseResults(list.markdown).filter((a) => !done.has(a.code)).filter((a) => !(skipSealed && /^sealed/i.test(a.name ?? ''))).filter((a) => {
      const d = dateWords(a.dateText);
      return !d || d.getTime() <= now;
    });
    let count = 0;
    let processed = 0;
    for (const auction of auctions) {
      if (processed >= auctionsPerRun || ctx.signal?.aborted || this.reached(ctx, count)) break;
      const url = `${BASE}/auctions/${auction.code}/lots/`;
      await this.throttle();
      const res = await ctx.fetch(url, {
        waitForMs: 7000,
        expect: ['title', 'price', 'status'],
        parse: (r) => {
          const p = r.markdown ? parseLotsPage(r.markdown, auction) : null;
          const f = p?.lots.find((l) => l.priceText && /sold/i.test(l.status ?? ''));
          return f ? { title: f.title, price: money(f.priceText)?.amount ?? null, status: f.status } : p?.lots.length ? { title: p.lots[0]!.title } : null;
        },
        minQuality: 0.3,
      });
      processed++;
      if (!res.success || !res.markdown) {
        ctx.anomaly('page_fetch_failed', `${url}: ${res.error ?? res.httpStatus}`);
        continue;
      }
      const payload = parseLotsPage(res.markdown, auction);
      if (payload.lots.length === 0) {
        ctx.anomaly('empty_page', url);
        continue;
      }
      count++;
      done.add(auction.code);
      yield { url, externalId: `auction:${auction.code}:first40`, kind: 'sale', engine: res.engine, httpStatus: res.httpStatus, payload, fetchedAt: res.fetchedAt };
      await ctx.setCursor({ doneAuctions: [...done].slice(-200), year: ctx.options.mode === 'backfill' && auctions.every((a) => done.has(a.code)) ? year - 1 : year, updatedAt: new Date().toISOString() });
    }
  }

  async normalize(raw: RawRecordLike): Promise<NormalizedRecord[]> {
    const p = PagePayloadSchema.parse(raw.payload);
    const saleDate = dateWords(p.auction.dateText);
    if (!saleDate) return [];
    const out: NormalizedSale[] = [];
    for (const lot of p.lots) {
      if (!lot.priceText || !/^(sold|lot sold)$/i.test(lot.status ?? '')) continue;
      const m = money(lot.priceText, 'USD');
      if (!m) continue;
      const memorabilia = MEMORABILIA.test(lot.title) && !/^\d{4}\s/.test(lot.title);
      const attributes = vehicleAttributes(lot.title, { identifiers: { rm_lot: `${p.auction.code}-${lot.slug}` }, metadata: { auction: p.auction.name, auction_code: p.auction.code, auction_dates: p.auction.dateText }, ...(memorabilia ? { categorySlug: 'automotive_memorabilia' as const } : {}) });
      out.push(makeSale({ meta: this.meta, sourceUrl: lot.url, externalId: `${p.auction.code}-${lot.slug}`, rawTitle: lot.title, attributes, price: m.amount, currency: m.currency, saleDate, buyerPremiumIncluded: true, auctionHouse: "RM Sotheby's", lotNumber: lot.lotNumber, imageUrls: lot.image ? [lot.image] : [], observedAt: raw.fetchedAt, parserVersion: PARSER_VERSION }));
    }
    return out;
  }
}

export default (meta: ConnectorMeta) => new RMSothebysConnector(meta);