import { z } from 'zod';
import { BaseConnector, html as H, type ConnectorMeta, type CrawlContext, type RawRecordInput, type RawRecordLike } from '@rareindex/connectors';
import type { NormalizedRecord, NormalizedSale } from '@rareindex/shared';
import { makeSale } from '../_carlib/index.js';
import { clean, isNumisBundle, numisAttributes, numisCategory, parseAuctionDate, parseCoinGrade, realized } from '../_g5-numismatics-lib/index.js';

/**
 * NumisBids — aggregator of numismatic auction catalogues (≈ 370 houses) with prices realized.
 * Public pages: /results (closed sales with a "Prices realized" link → sale id), /sale/<sid> (sale
 * header + category list), /sale/<sid>/category/<cid>?pg=N (100 lots per page with estimate, image,
 * truncated description and "Price realized: 160 USD"). Prices are hammer prices ("Buyer's premium is
 * not included" on the /pr/<sid> page). Engine: Firecrawl (Cloudflare refuses plain HTTPS).
 */

const SITE = 'https://www.numisbids.com';
const PARSER_VERSION = '1.0.0';

export const SaleSchema = z.object({ sid: z.string(), firm: z.string(), title: z.string(), dateText: z.string().nullable() });
export const CategorySchema = z.object({ cid: z.string(), name: z.string(), count: z.number().int().nullable() });
export const LotSchema = z.object({
  lotNumber: z.string(),
  url: z.string(),
  title: z.string(),
  estimateText: z.string().nullable(),
  realizedText: z.string().nullable(),
  image: z.string().nullable(),
});
export const PayloadSchema = z.object({
  kind: z.literal('category_page'),
  sale: SaleSchema,
  category: CategorySchema,
  url: z.string(),
  page: z.number().int(),
  totalPages: z.number().int(),
  lots: z.array(LotSchema),
});
export type Payload = z.infer<typeof PayloadSchema>;

export interface ResultsEntry {
  sid: string;
  eventId: string | null;
  firm: string;
  title: string;
  subtitle: string | null;
  dayText: string | null;
}

/** /results → closed sales that publish prices realized on NumisBids (newest first as listed). */
export function parseResultsPage(htmlText: string): ResultsEntry[] {
  const $ = H.load(htmlText);
  const out: ResultsEntry[] = [];
  $('tr').each((_, tr) => {
    const $tr = $(tr);
    const pr = $tr.find('a[href*="/pr/"]').first().attr('href');
    const sid = pr?.match(/\/pr\/(\d+)/)?.[1];
    if (!sid || out.some((e) => e.sid === sid)) return;
    const firm = $tr.find('td.firmcell img').attr('alt') ?? clean($tr.find('a.descr').first().text()).replace(/^-\s*/, '');
    const titleA = $tr.find('a[href*="/event/"] b').first();
    const title = clean(titleA.text());
    const eventId = $tr.find('a[href*="/event/"]').first().attr('href')?.match(/\/event\/(\d+)/)?.[1] ?? null;
    const subtitle = clean($tr.find('a.descr[href*="/event/"]').first().text()) || null;
    const dayText = clean($tr.find('.datetext').first().text()) || null;
    if (!title) return;
    out.push({ sid, eventId, firm: clean(firm), title, subtitle, dayText });
  });
  return out;
}

export interface SalePage {
  sale: z.infer<typeof SaleSchema>;
  hasPrices: boolean;
  categories: z.infer<typeof CategorySchema>[];
}

/** Sale header (firm, auction title, closing date) + category list, from any /sale/<sid>… page. */
export function parseSaleHeader(htmlText: string, sid: string): SalePage | null {
  const $ = H.load(htmlText);
  const status = $('.salestatus .text').first();
  const firm = clean(status.find('.name').first().text());
  const title = clean(status.find('b').first().text());
  if (!firm || !title) return null;
  const statusHtml = status.html() ?? '';
  const dateText = clean(statusHtml.match(/<\/b>\s*(?:&nbsp;|\s)*([^<]+)<br/i)?.[1] ?? '') || null;
  const hasPrices = status.find(`a[href*="/pr/${sid}"]`).length > 0 || /View prices realized/i.test(status.text());
  const categories: z.infer<typeof CategorySchema>[] = [];
  $(`a[href*="/sale/${sid}/category/"]`).each((_, a) => {
    const href = $(a).attr('href') ?? '';
    const cid = href.match(/\/category\/(\d+)/)?.[1];
    if (!cid || href.includes('?') || categories.some((c) => c.cid === cid)) return;
    const label = clean($(a).text());
    const m = label.match(/^(.*?)\s*\((\d+)\)\s*$/);
    if (!m) return; // navigation links to the same category without a count (e.g. "Go back to browse lots")
    categories.push({ cid, name: m[1]!.trim(), count: Number(m[2]) });
  });
  return { sale: { sid, firm, title, dateText }, hasPrices, categories };
}

/** Category page → lots (100 per page) + "Page X of Y". */
export function parseCategoryPage(htmlText: string, sale: z.infer<typeof SaleSchema>, category: z.infer<typeof CategorySchema>, url: string): Payload {
  const $ = H.load(htmlText);
  const pg = $('.salenav .small').first().text().match(/Page\s+(\d+)\s+of\s+(\d+)/i);
  const lots: z.infer<typeof LotSchema>[] = [];
  $('div.browse').each((_, el) => {
    const e = $(el);
    const lotA = e.find('.lot a').first();
    const url = lotA.attr('href');
    const lotNumber = clean(lotA.text()).replace(/^Lot\s+/i, '');
    if (!url || !lotNumber) return;
    const title = clean(e.find('.summary a').first().text());
    if (!title) return;
    lots.push({
      lotNumber,
      url: url.startsWith('http') ? url : SITE + url,
      title,
      estimateText: clean(e.find('.estimate .rateclick').first().text()) || clean(e.find('.estimate').first().text().replace(/^Estimate:\s*/i, '')) || null,
      realizedText: clean(e.find('.realized .rateclick').first().text()) || clean(e.find('.realized').first().text().replace(/^Price realized:\s*/i, '')) || null,
      image: e.find('.browseimg img').attr('src') ?? null,
    });
  });
  return { kind: 'category_page', sale, category, url, page: pg ? Number(pg[1]) : 1, totalPages: pg ? Number(pg[2]) : 1, lots };
}

interface Cursor {
  doneSales?: string[];
  inProgress?: { sid: string; sale: z.infer<typeof SaleSchema>; categories: z.infer<typeof CategorySchema>[]; catIndex: number; page: number } | null;
  /** backfill: next (older) sale id to inspect, descending */
  backfillSid?: number | null;
  minSeenSid?: number | null;
  done?: boolean;
  updatedAt?: string;
}

export class NumisBidsConnector extends BaseConnector {
  readonly version = '1.0.0';
  readonly parserVersion = PARSER_VERSION;
  protected override minIntervalMs = 2500;
  override readonly urlPatterns = [/numisbids\.com\/sale\/\d+/i];

  async *crawl(ctx: CrawlContext): AsyncIterable<RawRecordInput> {
    const salesPerRun = Number(this.meta.config.salesPerRun ?? 2);
    const pagesPerRun = Number(this.meta.config.pagesPerRun ?? 30);
    const skipSections = ((this.meta.config.skipSections as string[] | undefined) ?? []).map((s) => s.toLowerCase());
    const cursor = { ...((ctx.options.cursor ?? {}) as Cursor) };
    const done = new Set(cursor.doneSales ?? []);
    let pages = 0;
    let yielded = 0;
    let salesFinished = 0;

    // Quality hints per page type so the router does not escalate to Scrapfly on pages Firecrawl rendered fine.
    const fetchPage = (url: string) => {
      const isResults = /\/results(?:$|[?#])/.test(url);
      const isCategory = /\/category\//.test(url);
      return ctx.fetch(url, {
        engines: ['firecrawl', 'scrapfly'],
        expect: isCategory ? ['title', 'price'] : ['title'],
        parse: (r) => {
          if (!r.html) return null;
          if (isResults) {
            const entries = parseResultsPage(r.html);
            return entries.length ? { title: `${entries.length} sales with prices realized` } : null;
          }
          const sid = url.match(/\/sale\/(\d+)/)?.[1] ?? '0';
          const sale = parseSaleHeader(r.html, sid);
          if (!isCategory) return sale ? { title: sale.sale.title } : null;
          const lots = parseCategoryPage(r.html, sale?.sale ?? { sid, firm: '', title: '', dateText: null }, { cid: '0', name: '', count: null }, url).lots;
          const priced = lots.find((l) => l.realizedText) ?? lots[0];
          return priced ? { title: priced.title, price: priced.realizedText ?? priced.estimateText } : null;
        },
      });
    };

    // Pick the sales to process: an unfinished one first, then new closed sales (incremental) or older ids (backfill).
    const queue: string[] = [];
    if (cursor.inProgress) queue.push(cursor.inProgress.sid);
    if (ctx.options.mode === 'backfill') {
      let sid = cursor.backfillSid ?? cursor.minSeenSid ?? null;
      if (sid === null) {
        await this.throttle();
        const res = await fetchPage(`${SITE}/results`);
        const entries = res.success && res.html ? parseResultsPage(res.html) : [];
        sid = entries.length ? Math.min(...entries.map((e) => Number(e.sid))) - 1 : null;
      }
      if (sid === null || sid <= 0) {
        await ctx.setCursor({ ...cursor, done: true, updatedAt: new Date().toISOString() });
        return;
      }
      for (let s = sid; s > 0 && queue.length < salesPerRun * 4; s--) if (!done.has(String(s))) queue.push(String(s));
    } else {
      await this.throttle();
      const res = await fetchPage(`${SITE}/results`);
      if (!res.success || !res.html) {
        ctx.anomaly('page_fetch_failed', `results: ${res.error ?? res.httpStatus}`);
      } else {
        const entries = parseResultsPage(res.html);
        if (!entries.length) ctx.anomaly('selector_missing', 'results page: no "Prices realized" rows');
        const sids = entries.map((e) => Number(e.sid)).filter((n) => Number.isFinite(n));
        if (sids.length) cursor.minSeenSid = Math.min(cursor.minSeenSid ?? Infinity, ...sids);
        for (const e of entries) if (!done.has(e.sid) && !queue.includes(e.sid)) queue.push(e.sid);
      }
    }

    for (const sid of queue) {
      if (ctx.signal?.aborted || salesFinished >= salesPerRun || pages >= pagesPerRun || this.reached(ctx, yielded)) break;
      let state = cursor.inProgress?.sid === sid ? cursor.inProgress : null;
      if (!state) {
        await this.throttle();
        const res = await fetchPage(`${SITE}/sale/${sid}`);
        pages++;
        const header = res.success && res.html ? parseSaleHeader(res.html, sid) : null;
        if (!header) {
          ctx.anomaly(res.success ? 'parse_failure_page' : 'page_fetch_failed', `sale ${sid}: ${res.error ?? res.httpStatus ?? 'no header'}`);
          if (ctx.options.mode === 'backfill' && res.success) await this.skipBackfillSale(ctx, cursor, done, sid); // removed / unpublished sale id
          continue;
        }
        if (!header.hasPrices || !header.categories.length) {
          // still open, or prices not published on NumisBids (house hosts them elsewhere)
          if (ctx.options.mode === 'backfill') await this.skipBackfillSale(ctx, cursor, done, sid);
          continue;
        }
        state = { sid, sale: header.sale, categories: header.categories.filter((c) => !skipSections.includes(c.name.toLowerCase())), catIndex: 0, page: 1 };
      }
      while (state.catIndex < state.categories.length) {
        if (ctx.signal?.aborted || pages >= pagesPerRun || this.reached(ctx, yielded)) break;
        const cat = state.categories[state.catIndex]!;
        const url = `${SITE}/sale/${sid}/category/${cat.cid}${state.page > 1 ? `?pg=${state.page}` : ''}`;
        await this.throttle();
        const res = await fetchPage(url);
        pages++;
        if (!res.success || !res.html) {
          ctx.anomaly('page_fetch_failed', `${url}: ${res.error ?? res.httpStatus}`);
          break;
        }
        const payload = parseCategoryPage(res.html, state.sale, cat, url);
        if (!payload.lots.length) ctx.anomaly('parse_failure_page', `${url}: no lots parsed`);
        if (payload.lots.some((l) => l.realizedText)) {
          yielded++;
          yield { url, externalId: `sale:${sid}:cat:${cat.cid}:p${payload.page}`, kind: 'sale', engine: res.engine, httpStatus: res.httpStatus, payload, fetchedAt: res.fetchedAt };
        }
        if (payload.page < payload.totalPages) state.page++;
        else {
          state.catIndex++;
          state.page = 1;
        }
        cursor.inProgress = state;
        await ctx.setCursor({ ...cursor, doneSales: [...done].slice(-300), updatedAt: new Date().toISOString() });
      }
      if (state.catIndex >= state.categories.length) {
        done.add(sid);
        salesFinished++;
        cursor.inProgress = null;
        if (ctx.options.mode === 'backfill') {
          cursor.backfillSid = Number(sid) - 1;
          await ctx.progress({ page: Number(sid), totalPages: null, itemsProcessed: yielded, reachedDate: parseAuctionDate(state.sale.dateText) });
        }
        await ctx.setCursor({ ...cursor, doneSales: [...done].slice(-300), updatedAt: new Date().toISOString() });
      }
    }
    if (ctx.options.mode === 'backfill' && (cursor.backfillSid ?? 1) <= 0) {
      // The campaign walks sale ids downwards; id 1 is the oldest sale NumisBids hosts.
      await ctx.setCursor({ ...cursor, done: true, doneSales: [...done].slice(-300), updatedAt: new Date().toISOString() });
    }
  }

  /** Backfill bookkeeping for a sale id that yields nothing (open sale, prices hosted elsewhere, removed id). */
  private async skipBackfillSale(ctx: CrawlContext, cursor: Cursor, done: Set<string>, sid: string): Promise<void> {
    done.add(sid);
    cursor.backfillSid = Number(sid) - 1;
    await ctx.setCursor({ ...cursor, doneSales: [...done].slice(-300), updatedAt: new Date().toISOString() });
  }

  async normalize(raw: RawRecordLike): Promise<NormalizedRecord[]> {
    const p = PayloadSchema.parse(raw.payload);
    const saleDate = parseAuctionDate(p.sale.dateText);
    if (!saleDate) return [];
    const out: NormalizedSale[] = [];
    const hint = `${p.category.name} ${p.sale.title}`;
    for (const lot of p.lots) {
      const price = realized(lot.realizedText);
      if (!price) continue; // unsold / withdrawn
      const categorySlug = numisCategory(lot.title, hint, 'coins');
      const g = parseCoinGrade(lot.title);
      const estimate = realized(lot.estimateText);
      const attributes = numisAttributes({
        categorySlug,
        title: lot.title,
        section: p.category.name,
        identifiers: { numisbids_lot: `${p.sale.sid}/${lot.lotNumber}` },
        metadata: { firm: p.sale.firm, auction: p.sale.title, numisbids_sale_id: p.sale.sid, numisbids_category_id: p.category.cid, estimate: estimate?.amount ?? null, estimate_currency: estimate?.currency ?? null, hammer_price: price.amount, buyer_premium: 'excluded (NumisBids prices realized are hammer prices)', title_truncated: /\.\.\.$/.test(lot.title) },
      });
      const sale = makeSale({
        meta: this.meta,
        sourceUrl: lot.url,
        externalId: `${p.sale.sid}-${lot.lotNumber}`,
        rawTitle: lot.title,
        attributes,
        price: price.amount,
        currency: price.currency,
        saleDate,
        buyerPremiumIncluded: false,
        auctionHouse: p.sale.firm,
        lotNumber: lot.lotNumber,
        imageUrls: lot.image ? [lot.image] : [],
        observedAt: raw.fetchedAt,
        parserVersion: PARSER_VERSION,
        confidence: g.grader ? 0.85 : 0.75,
        isBundle: isNumisBundle(lot.title, p.category.name),
        conditionRaw: g.conditionRaw,
      });
      sale.grade = { grader: g.grader, grade: g.grade, qualifier: g.qualifier, certificationNumber: g.certificationNumber };
      out.push(sale);
    }
    return out;
  }
}

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