/**
 * Generic "past sales → lot results" connector skeleton shared by the g8 auction-house connectors.
 * A house exposes (1) an index of past sales and (2) per-sale lot pages (HTML or embedded JSON) that list
 * lot number, title, realised price and estimate. Subclasses implement the three parsers; crawl
 * (resumable, backfill-aware) and normalize (sale / auction_lot, native currency, premium basis labelled)
 * are shared so every house behaves identically.
 */
import { z } from 'zod';
import { BaseConnector, type ConnectorMeta, type CrawlContext, type RawRecordInput, type RawRecordLike } from '@rareindex/connectors';
import { parseGradeFromTitle } from '@rareindex/taxonomy';
import { AssetAttributesSchema, NormalizedAuctionLotSchema, NormalizedSaleSchema, type CurrencyCode, type ExtractionResult, type NormalizedRecord } from '@rareindex/shared';
import { brandFromSlug, hintFromLabel, slugFromTitle, watchReference, type DeptHint } from '../_auction-lib/categories.js';
import { englishLabel, isBundleMultilingual, strongLotHint, yearFromTitle } from './index.js';

export const SaleRefSchema = z.object({
  id: z.string(),
  title: z.string(),
  url: z.string(),
  /** ISO date (source's own sale date), null when the index does not show it */
  date: z.string().nullable(),
  location: z.string().nullable(),
  extra: z.record(z.string(), z.unknown()).default({}),
});
export type SaleRef = z.infer<typeof SaleRefSchema>;

export const ParsedLotSchema = z.object({
  lotNo: z.string(),
  title: z.string(),
  subtitle: z.string().nullable().default(null),
  description: z.string().nullable().default(null),
  url: z.string(),
  image: z.string().nullable().default(null),
  /** realised price as published (hammer or premium-inclusive per `premiumIncluded`) */
  price: z.number().nullable().default(null),
  currency: z.string().nullable().default(null),
  premiumIncluded: z.boolean().nullable().default(null),
  estimateLow: z.number().nullable().default(null),
  estimateHigh: z.number().nullable().default(null),
  /** sale date override (timed sales where each lot closes on its own day) */
  date: z.string().nullable().default(null),
  sold: z.boolean(),
  extra: z.record(z.string(), z.unknown()).default({}),
});
export type ParsedLot = z.infer<typeof ParsedLotSchema>;

export const SaleResultsPayloadSchema = z.object({
  kind: z.literal('sale_results'),
  url: z.string(),
  sale: SaleRefSchema,
  page: z.number(),
  totalLots: z.number().nullable(),
  lots: z.array(ParsedLotSchema),
});
export type SaleResultsPayload = z.infer<typeof SaleResultsPayloadSchema>;

export interface ParsedSalePage {
  lots: ParsedLot[];
  hasMore: boolean;
  totalLots: number | null;
  /** sale-level facts discovered on the lot page (date, location, premium basis) */
  sale?: Partial<Pick<SaleRef, 'date' | 'location' | 'title'>> & { extra?: Record<string, unknown> };
}

export interface HouseConfig {
  houseName: string;
  defaultCurrency: CurrencyCode;
  /** default `location` for sale records when the source does not give one */
  location: string | null;
  /** identifiers key, e.g. "aguttes_lot" → "<saleId>/<lotNo>" */
  idKey: string;
  /** default buyer-premium basis when the page does not label it (null = unknown) */
  premiumIncluded: boolean | null;
  /** engines for page fetches (default ['api']) */
  engines?: Array<'api' | 'firecrawl' | 'scrapfly'>;
  responseType?: 'text' | 'json';
  /** taxonomy slug when nothing matches (null = drop the lot) */
  fallbackSlug: string | null;
  /** politeness between requests (ms) */
  minIntervalMs?: number;
  /** cap of sale pages fetched per incremental run */
  maxPagesPerSale?: number;
  /** drop lots whose currency could not be read from the source (multi-currency houses) instead of defaulting */
  requireCurrency?: boolean;
}

type Cursor = { done?: string[]; pending?: Record<string, string>; backfill?: { index: number; page: number; itemsProcessed: number }; finished?: boolean };

/** Keep the pending map bounded (most recent 200 entries). */
function trimPending(p: Record<string, string>): Record<string, string> {
  const entries = Object.entries(p).sort((a, b) => b[1].localeCompare(a[1])).slice(0, 200);
  return Object.fromEntries(entries);
}

export abstract class SaleResultsConnector extends BaseConnector {
  readonly parserVersion = '1.0.0';
  abstract readonly house: HouseConfig;

  /** Fetch + parse the index of past sales (most recent first is not required; we sort by date). */
  abstract listSales(ctx: CrawlContext): Promise<SaleRef[]>;
  /** URL of page `page` (1-based) of a sale's lot list. */
  abstract salePageUrl(sale: SaleRef, page: number): string;
  /** Parse one lot page (HTML text or JSON). Return null when the document is not a lot page. */
  abstract parseSalePage(res: ExtractionResult, sale: SaleRef, page: number): ParsedSalePage | null;
  /**
   * Taxonomy slug for a lot: strong multilingual lot cues (a Rolex in a "Moderne Kunst" sale is a watch) win,
   * then the department hint derived from the sale title / department (any supported language), then a keyword
   * sweep over title + description, then the house fallback (null = drop).
   */
  categoryFor(sale: SaleRef, lot: ParsedLot): string | null {
    return resolveCategory(`${sale.title} ${String(sale.extra.department ?? '')} ${String(sale.extra.title_fr ?? '')}`, lot, this.house.fallbackSlug);
  }

  protected get salesPerRun(): number {
    return Number(this.meta.config.salesPerRun ?? 3);
  }

  protected async fetchSalePage(ctx: CrawlContext, sale: SaleRef, page: number): Promise<{ url: string; res: ExtractionResult; parsed: ParsedSalePage | null }> {
    const url = this.salePageUrl(sale, page);
    await this.throttle(url);
    const res = await ctx.fetch(url, {
      engines: this.house.engines ?? ['api'],
      responseType: this.house.responseType ?? 'text',
      timeoutMs: 60_000,
      expect: ['title', 'price'],
      minQuality: 0.3,
      parse: (r) => {
        const p = this.parseSalePage(r, sale, page);
        const sold = p?.lots.find((l) => l.price);
        return p?.lots.length ? { title: p.lots[0]!.title, price: sold?.price ?? null, currency: sold?.currency ?? null } : null;
      },
    });
    const parsed = res.success ? this.parseSalePage(res, sale, page) : null;
    return { url, res, parsed };
  }

  /** Yield every page of one sale (stops at hasMore=false, empty page or the per-sale cap). */
  protected async *crawlSale(ctx: CrawlContext, sale: SaleRef, startPage = 1, onPage?: (page: number, lots: number) => Promise<void>): AsyncIterable<RawRecordInput> {
    const cap = this.house.maxPagesPerSale ?? 40;
    let page = startPage;
    for (; page < startPage + cap; page++) {
      if (ctx.signal?.aborted) return;
      const { url, res, parsed } = await this.fetchSalePage(ctx, sale, page);
      if (!parsed) {
        ctx.anomaly(page === 1 ? 'sale_parse_failed' : 'pagination_failure', `${sale.id} p${page}: ${res.error ?? res.httpStatus}`);
        return;
      }
      if (parsed.sale) Object.assign(sale, { date: parsed.sale.date ?? sale.date, location: parsed.sale.location ?? sale.location, title: parsed.sale.title ?? sale.title, extra: { ...sale.extra, ...(parsed.sale.extra ?? {}) } });
      if (parsed.lots.length === 0) {
        if (page === 1) ctx.anomaly('selector_missing', `${sale.id}: no lots parsed`);
        return;
      }
      const payload: SaleResultsPayload = { kind: 'sale_results', url, sale: { ...sale }, page, totalLots: parsed.totalLots, lots: parsed.lots };
      yield { url, externalId: `${sale.id}:p${page}`, kind: 'sale', engine: res.engine, httpStatus: res.httpStatus, payload, fetchedAt: res.fetchedAt };
      if (onPage) await onPage(page, parsed.lots.length);
      if (!parsed.hasMore) return;
    }
    ctx.anomaly('pagination_failure', `${sale.id}: page cap ${cap} reached`);
  }

  async *crawl(ctx: CrawlContext): AsyncIterable<RawRecordInput> {
    const cursor = (ctx.options.cursor ?? {}) as Cursor;
    // keep sales that have started (timed sales expose their end date in extra.end_date → wait for it)
    const sales = (await this.listSales(ctx)).filter((s) => {
      const when = typeof s.extra.end_date === 'string' ? s.extra.end_date : s.date;
      return !when || new Date(when).getTime() <= Date.now() + 86_400_000;
    });
    if (sales.length === 0) {
      ctx.anomaly('past_list_failed', 'no past sales found on the index');
      return;
    }
    if (ctx.options.mode === 'backfill') {
      yield* this.backfill(ctx, sales, cursor);
      return;
    }
    sales.sort((a, b) => (b.date ?? '').localeCompare(a.date ?? ''));
    const done = new Set<string>(cursor.done ?? []);
    // sales seen without any realised price yet (still running / results not published) → re-check after a few days
    const pending: Record<string, string> = { ...(cursor.pending ?? {}) };
    const recheckMs = Number(this.meta.config.pendingRecheckDays ?? 3) * 86_400_000;
    const maxUnfinished = Number(this.meta.config.maxUnfinishedPerRun ?? Math.max(this.salesPerRun * 3, 12));
    let processed = 0;
    let unfinished = 0;
    let count = 0;
    for (const sale of sales) {
      if (ctx.signal?.aborted || this.reached(ctx, count) || processed >= this.salesPerRun || unfinished >= maxUnfinished) break;
      if (done.has(sale.id)) continue;
      const seenAt = pending[sale.id] ? new Date(pending[sale.id]!).getTime() : 0;
      if (seenAt && Date.now() - seenAt < recheckMs && ctx.options.mode !== 'probe') continue;
      let pages = 0;
      let complete = true;
      for await (const raw of this.crawlSale(ctx, sale)) {
        pages++;
        count++;
        yield raw;
        // A first page without a single realised price = the sale is still running or results are not published yet:
        // keep its lots as auction_lot records but do not paginate further and do not mark the sale done.
        if (pages === 1 && !(raw.payload as SaleResultsPayload).lots.some((l) => l.sold)) {
          complete = false;
          unfinished++;
          break;
        }
        if (this.reached(ctx, count)) break;
      }
      if (!complete) {
        pending[sale.id] = new Date().toISOString();
        await ctx.setCursor({ done: [...done].slice(-500), pending: trimPending(pending) });
        continue;
      }
      processed++;
      if (pages > 0 && !this.reached(ctx, count)) {
        done.add(sale.id);
        delete pending[sale.id];
        await ctx.setCursor({ done: [...done].slice(-500), pending: trimPending(pending) });
      }
    }
  }

  /** Backfill: every past sale, oldest first, resumable at (sale index, page); ends with {finished:true}. */
  protected async *backfill(ctx: CrawlContext, sales: SaleRef[], cursor: Cursor): AsyncIterable<RawRecordInput> {
    if (cursor.finished) return;
    sales.sort((a, b) => (a.date ?? '').localeCompare(b.date ?? '') || a.id.localeCompare(b.id));
    let index = cursor.backfill?.index ?? 0;
    let itemsProcessed = cursor.backfill?.itemsProcessed ?? 0;
    let startPage = cursor.backfill?.page ?? 1;
    let fetched = 0;
    const maxPages = this.policy.backfillMaxPages;
    for (; index < sales.length; index++, startPage = 1) {
      const sale = sales[index]!;
      let stop = false;
      for await (const raw of this.crawlSale(ctx, sale, startPage, async (page, lots) => {
        itemsProcessed += lots;
        fetched++;
        await ctx.setCursor({ backfill: { index, page: page + 1, itemsProcessed } });
        await ctx.progress({ page: index + 1, totalPages: sales.length, itemsProcessed, reachedDate: sale.date ? new Date(sale.date) : null, cursor: { index, page: page + 1 } });
        if (fetched >= maxPages || ctx.signal?.aborted) stop = true;
      })) {
        yield raw;
        if (stop) return;
      }
      await ctx.setCursor({ backfill: { index: index + 1, page: 1, itemsProcessed } });
    }
    await ctx.setCursor({ finished: true, backfill: { index: sales.length, page: 1, itemsProcessed } });
    await ctx.progress({ page: sales.length, totalPages: sales.length, itemsProcessed, cursor: { finished: true } });
  }

  async normalize(raw: RawRecordLike): Promise<NormalizedRecord[]> {
    const p = SaleResultsPayloadSchema.parse(raw.payload);
    const out: NormalizedRecord[] = [];
    const saleDate = p.sale.date ? new Date(p.sale.date) : null;
    for (const lot of p.lots) {
      if (this.house.requireCurrency && !lot.currency) continue;
      const slug = this.categoryFor(p.sale, lot);
      if (!slug) continue;
      const text = `${lot.title} ${lot.subtitle ?? ''}`.trim();
      const g = parseGradeFromTitle(text);
      const isWatch = ['rolex', 'omega', 'patek_philippe', 'audemars_piguet', 'other_watches'].includes(slug);
      const attributes = AssetAttributesSchema.parse({
        categorySlug: slug,
        name: lot.title,
        model: lot.subtitle,
        brand: brandFromSlug(slug, text),
        // "réf. 5513" / "Ref 16233" — normalise the French/German abbreviation before the English reference parser
        reference: isWatch ? watchReference(text) ?? watchReference(text.replace(/\br[ée]f(?:[ée]rence|erenz)?\.?\s*/gi, 'Ref. ')) : null,
        year: yearFromTitle(text),
        identifiers: { [this.house.idKey]: `${p.sale.id}/${lot.lotNo}` },
        metadata: { sale_id: p.sale.id, sale_title: p.sale.title, sale_url: p.sale.url, estimate_low: lot.estimateLow, estimate_high: lot.estimateHigh, ...p.sale.extra, ...lot.extra },
      });
      const base = {
        connectorId: this.meta.id,
        sourceId: this.meta.sourceId,
        sourceUrl: lot.url,
        externalId: `${p.sale.id}:${lot.lotNo}`,
        rawTitle: lot.subtitle ? `${lot.title} — ${lot.subtitle}` : lot.title,
        description: lot.description,
        imageUrls: lot.image ? [lot.image] : [],
        attributes,
        grade: { grader: g.grader && g.grader !== 'raw' ? g.grader : null, grade: g.grader && g.grader !== 'raw' ? g.grade : null, qualifier: null, certificationNumber: null },
        condition: { condition: null, conditionRaw: null, completeness: null },
        observedAt: raw.fetchedAt,
        confidence: this.confidenceFor(slug, lot),
        parserVersion: this.parserVersion,
      };
      const currency = (lot.currency ?? this.house.defaultCurrency) as CurrencyCode;
      const lotDate = lot.date ? new Date(lot.date) : saleDate;
      const location = p.sale.location ?? this.house.location;
      if (lot.sold && lot.price && lotDate && !Number.isNaN(lotDate.getTime())) {
        out.push(NormalizedSaleSchema.parse({ kind: 'sale', ...base, saleType: 'auction', saleDate: lotDate, price: lot.price, currency, buyerPremiumIncluded: lot.premiumIncluded ?? this.house.premiumIncluded, quantity: 1, isBundle: isBundleMultilingual(text), location, auctionHouse: this.house.houseName, lotNumber: lot.lotNo }));
      } else {
        const status = lotDate && lotDate.getTime() < raw.fetchedAt.getTime() ? 'ended' : lotDate ? 'upcoming' : 'unknown';
        out.push(NormalizedAuctionLotSchema.parse({ kind: 'auction_lot', ...base, auctionHouse: this.house.houseName, auctionName: p.sale.title, lotNumber: lot.lotNo, startsAt: lotDate, endsAt: lotDate, estimateLow: lot.estimateLow, estimateHigh: lot.estimateHigh, currentBid: lot.sold ? lot.price : null, currency, status, location }));
      }
    }
    return out;
  }

  protected confidenceFor(_slug: string, _lot: ParsedLot): number {
    return 0.85;
  }
}

/** Shared category resolution (see `SaleResultsConnector.categoryFor`). Exported for connectors with extra rules. */
export function resolveCategory(saleLabel: string, lot: ParsedLot, fallbackSlug: string | null): string | null {
  const text = `${lot.title} ${lot.subtitle ?? ''}`.trim();
  const full = `${text} ${lot.description ?? ''}`.slice(0, 500);
  const strong = strongLotHint(full);
  const saleHint: DeptHint = hintFromLabel(englishLabel(saleLabel));
  const hint: DeptHint = strong ?? saleHint;
  if (hint === 'wine') {
    if (/whisky|whiskey|威士忌|ウイスキー|bourbon|macallan|yamazaki|山崎|hibiki|響|yoichi|余市|karuizawa|輕井澤|軽井沢|springbank|bowmore|ardbeg|glenfiddich|dalmore|laphroaig|brora|port ellen/i.test(full)) return 'whisky';
    if (/cognac|armagnac|calvados|干邑|白蘭地|ブランデー/i.test(full)) return 'cognac';
    if (/\brum\b|\brhum\b|朗姆|ラム酒/i.test(full)) return 'rum';
    return 'wine';
  }
  if (hint === 'asian' || hint === 'antiquities') return slugFromTitle(text, hint) ?? 'antiques';
  if (hint === 'cars') {
    if (/\b(helmet|casque|helm|poster|affiche|plakat|trophy|trophée|suit|combinaison|gloves|gants|photograph|photographie|model|maquette|modell|miniature|book|livre|buch|sign|plaque|enamel|programme|program|watch|montre|mascot|mascotte|badge)\b/i.test(full) && !/\b(chassis|châssis|fahrgestell|telaio|\bvin\b|immatricul|registration|kilom|mileage)\b/i.test(full)) return 'automotive_memorabilia';
    return slugFromTitle(full, 'cars') ?? 'automobiles';
  }
  return slugFromTitle(text, hint) ?? slugFromTitle(full, hint) ?? (hint !== 'unknown' ? slugFromTitle('', hint) : null) ?? (saleHint !== 'unknown' ? slugFromTitle('', saleHint) : null) ?? fallbackSlug;
}

/** Unescape HTML entities commonly found in server-rendered auction pages. */
export function decodeEntities(s: string): string {
  return s
    .replace(/&nbsp;|&#160;/g, ' ')
    .replace(/&amp;/g, '&')
    .replace(/&lt;/g, '<')
    .replace(/&gt;/g, '>')
    .replace(/&quot;/g, '"')
    .replace(/&#39;|&apos;|&rsquo;|&#8217;/g, "'")
    .replace(/&euro;/g, '€')
    .replace(/&pound;/g, '£')
    .replace(/&eacute;/g, 'é')
    .replace(/&egrave;/g, 'è')
    .replace(/&agrave;/g, 'à')
    .replace(/&ccedil;/g, 'ç')
    .replace(/&ocirc;/g, 'ô')
    .replace(/&ecirc;/g, 'ê')
    .replace(/&uuml;/g, 'ü')
    .replace(/&ouml;/g, 'ö')
    .replace(/&auml;/g, 'ä')
    .replace(/&szlig;/g, 'ß')
    .replace(/&ordm;/g, 'º')
    .replace(/&#(\d+);/g, (_, n: string) => String.fromCodePoint(Number(n)))
    .replace(/&#x([0-9a-f]+);/gi, (_, h: string) => String.fromCodePoint(Number.parseInt(h, 16)));
}

/** Text content of an HTML fragment (tags stripped, entities decoded, whitespace collapsed). */
export function textOf(fragment: string | null | undefined): string {
  if (!fragment) return '';
  return decodeEntities(fragment.replace(/<br\s*\/?>/gi, ' ').replace(/<[^>]+>/g, ' ')).replace(/\s+/g, ' ').trim();
}

/** Split an HTML document into item chunks starting at each match of `startRe` (the chunk runs to the next match). */
export function chunksBetween(htmlText: string, startRe: RegExp, endBoundary?: RegExp): string[] {
  const re = new RegExp(startRe.source, startRe.flags.includes('g') ? startRe.flags : `${startRe.flags}g`);
  const idx: number[] = [];
  let m: RegExpExecArray | null;
  while ((m = re.exec(htmlText))) idx.push(m.index);
  if (idx.length === 0) return [];
  let end = htmlText.length;
  if (endBoundary) {
    const tail = htmlText.slice(idx[idx.length - 1]!);
    const e = tail.search(endBoundary);
    if (e > 0) end = idx[idx.length - 1]! + e;
  }
  return idx.map((s, k) => htmlText.slice(s, idx[k + 1] ?? end)).filter((c) => c.length > 0);
}

/** First capture group of `re` in `s`, entity-decoded and trimmed; null when absent. */
export function pick(s: string, re: RegExp): string | null {
  const m = s.match(re);
  return m ? textOf(m[1] ?? m[0]) || null : null;
}

export function absolute(base: string, href: string | null | undefined): string | null {
  if (!href) return null;
  try {
    return new URL(decodeEntities(href), base).toString();
  } catch {
    return null;
  }
}