Subversion Repositories SmartDukaan

package com.smartdukaan.cron.offercircular;

import java.io.File;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import com.spice.profitmandi.dao.repository.offers.CircularIngestRepository;

/**
 * Turns one uploaded circular PDF into rows in the {@code offers} schema.
 *
 * Everything for a document happens in a single transaction: the document's previous
 * derived rows are deleted and the new ones inserted, so a failure part-way leaves the
 * previous state intact rather than a half-populated set of offers. Config tables
 * (oem_division, scope_rule, product_alias, bank) are read but never written.
 */
@Service
public class CircularIngestService {

    private static final Logger LOGGER = LogManager.getLogger(CircularIngestService.class);

    private static final Pattern SELECTED_MODELS =
            Pattern.compile("selected\\s*model|selective\\s*model", Pattern.CASE_INSENSITIVE);
    private static final Pattern DATE_DMY = Pattern.compile("^(\\d{2})-(\\d{2})-(\\d{4})$");

    /** Store restrictions are prose inside the products / credit-cards cells. */
    private static final Object[][] CHANNEL_RULES = {
        {Pattern.compile("all stores except\\s+([A-Za-z ]+)", Pattern.CASE_INSENSITIVE), "EXCLUDE"},
        {Pattern.compile("only for\\s+([A-Za-z ]+)", Pattern.CASE_INSENSITIVE), "INCLUDE"},
        {Pattern.compile("only\\s+([A-Za-z ]+?)\\s*(?:mobile|stores)", Pattern.CASE_INSENSITIVE), "INCLUDE"},
    };

    private final CircularExtractor extractor;
    private final CircularIngestRepository repository;

    /** Constructor injection so the service can be exercised without a Spring context. */
    @Autowired
    public CircularIngestService(CircularExtractor extractor,
                                CircularIngestRepository repository) {
        this.extractor = extractor;
        this.repository = repository;
    }

    /** Counts and dropped-row reasons, for the notification email. */
    public static class Summary {
        private final Map<String, Integer> counts = new LinkedHashMap<>();
        private final Map<String, Integer> dropped = new TreeMap<>();
        private final Map<String, Integer> productStatuses = new TreeMap<>();
        private final List<String> warnings = new ArrayList<>();
        public Map<String, Integer> getCounts() { return counts; }
        public Map<String, Integer> getDropped() { return dropped; }
        public Map<String, Integer> getProductStatuses() { return productStatuses; }
        public List<String> getWarnings() { return warnings; }
        void bump(String key) { counts.merge(key, 1, Integer::sum); }
        void drop(String reason) { dropped.merge(reason, 1, Integer::sum); }
        void status(String s) { productStatuses.merge(s, 1, Integer::sum); }

        @Override public String toString() {
            StringBuilder sb = new StringBuilder(counts.toString());
            if (!productStatuses.isEmpty()) { sb.append(" products=").append(productStatuses); }
            if (!dropped.isEmpty()) { sb.append(" dropped=").append(dropped); }
            if (!warnings.isEmpty()) { sb.append(" warnings=").append(warnings.size()); }
            return sb.toString();
        }
    }

    /**
     * Ingests one document. Requires a NEW transaction so each document commits or
     * rolls back on its own - one bad circular must not abort a whole batch.
     */
    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Throwable.class)
    public Summary ingest(int documentId) throws Exception {
        Map<String, Object> document = repository.selectDocument(documentId);
        if (document == null) {
            throw new IllegalArgumentException("no such circular: " + documentId);
        }
        // Defence in depth. The scheduler already refuses to claim a curated circular,
        // but this is the method that DELETES rows, so the guard belongs here too -
        // otherwise any future caller (a reprocess endpoint, a migration, a test)
        // silently erases decisions a human made that the PDF cannot reproduce.
        if (isCurated(document)) {
            throw new IllegalStateException("circular " + documentId
                    + " has been manually curated; re-ingest would erase those decisions. "
                    + "Delete and re-upload the PDF as a new document if it must be re-derived.");
        }

        String storedPath = (String) document.get("storedPath");
        String filename = (String) document.get("sourceFilename");
        if (storedPath == null || filename == null) {
            throw new IllegalStateException("circular " + documentId + " has no stored PDF");
        }
        File pdf = new File(storedPath, filename);
        if (!pdf.isFile()) {
            throw new IllegalStateException("stored PDF missing on disk: " + pdf.getAbsolutePath());
        }

        ScopeConfig scope = ScopeConfig.load(repository);
        Map<String, Integer> bankAliases = repository.selectBankAliases();
        CatalogIndex catalog = CatalogIndex.load(repository);
        ProductAliases aliases = ProductAliases.load(repository);
        Set<String> ownChannels = repository.selectOwnChannels();

        CircularExtractor.Result extracted = extractor.extract(pdf);
        Summary summary = new Summary();
        summary.getWarnings().addAll(extracted.getWarnings());
        summary.counts.put("rowsInPdf", extracted.getRows().size());

        repository.deleteDerivedRows(documentId);

        Map<Integer, Integer> rowNoByPage = new LinkedHashMap<>();
        for (CircularRow row : extracted.getRows()) {
            ScopeConfig.Decision decision = scope.evaluate(row.getOemLabel(), row.getEligibleProducts());
            if (!decision.isKeep()) {
                summary.drop(decision.getReason());
                continue;
            }
            // A row restricted INCLUSIVELY to a retailer that is not us belongs to
            // somebody else. Detected before anything is written, so it is dropped
            // rather than stored and later cleaned up.
            List<ChannelScope> channels = detectChannelScopes(row);
            String foreign = foreignExclusive(channels, ownChannels);
            if (foreign != null) {
                summary.drop("exclusive to another retailer: " + foreign);
                continue;
            }
            int rowNo = rowNoByPage.merge(row.getPageNo(), 1, Integer::sum);
            persistRow(documentId, row, rowNo, decision, channels,
                    bankAliases, catalog, aliases, summary);
        }
        // Footnotes last: they are page-scoped rules that apply to whatever rows
        // survived on that page.
        FootnoteParser.Result notes =
                FootnoteParser.parse(extracted.getFootnotes(), bankAliases);
        summary.getWarnings().addAll(notes.getWarnings());
        for (FootnoteParser.Condition condition : notes.getConditions()) {
            repository.insertCondition(documentId, condition.getPageNo(),
                    condition.getConditionType(), condition.getBankId(),
                    condition.getTenureMonths(), condition.getNoteText(),
                    condition.getRawText());
            summary.bump("conditions");
        }

        summary.counts.put("rowsLoaded", summary.counts.getOrDefault("offers", 0));
        LOGGER.info("circular {} ingested: {}", documentId, summary);
        return summary;
    }

    /** One detected store restriction, before it is known whether the row survives. */
    private static final class ChannelScope {
        final String inclusion;
        final String channelRef;
        final String rawText;
        ChannelScope(String inclusion, String channelRef, String rawText) {
            this.inclusion = inclusion;
            this.channelRef = channelRef;
            this.rawText = rawText;
        }
    }

    private List<ChannelScope> detectChannelScopes(CircularRow row) {
        // the prose appears in either the products or the credit-cards cell
        String haystack = nullToEmpty(row.getEligibleProducts()) + " "
                + nullToEmpty(row.getCreditCards());
        List<ChannelScope> found = new ArrayList<>();
        for (Object[] rule : CHANNEL_RULES) {
            Matcher m = ((Pattern) rule[0]).matcher(haystack);
            while (m.find()) {
                String ref = m.group(1).trim().replaceAll("[.)]+$", "").trim();
                if (!ref.isEmpty()) {
                    found.add(new ChannelScope((String) rule[1], ref, m.group()));
                }
            }
        }
        return found;
    }

    /** @return the foreign retailer this row is exclusive to, or null if it is ours. */
    private String foreignExclusive(List<ChannelScope> channels, Set<String> ownChannels) {
        for (ChannelScope scope : channels) {
            if ("INCLUDE".equals(scope.inclusion)
                    && !ownChannels.contains(scope.channelRef.toUpperCase())) {
                return scope.channelRef;
            }
        }
        return null;
    }

    private void persistRow(int documentId, CircularRow row, int rowNo,
                            ScopeConfig.Decision decision, List<ChannelScope> channels,
                            Map<String, Integer> bankAliases,
                            CatalogIndex catalog, ProductAliases aliases, Summary summary) {
        LocalDate start = parseDate(row.getStartDateRaw());
        LocalDate end = parseDate(row.getEndDateRaw());
        if (start == null || end == null) {
            summary.drop("unparseable dates");
            return;
        }

        BankTextParser.Result credit = BankTextParser.parse(row.getCreditCards(), bankAliases);
        BankTextParser.Result debit = BankTextParser.parse(row.getDebitCards(), bankAliases);

        int offerId = repository.insertOffer(documentId, row.getOemLabel(),
                benefitTiming(row.getAdditionalCashback()), row.getDescription(),
                credit.isAllBanks() || debit.isAllBanks(), start, end);
        summary.bump("offers");

        repository.insertRawRow(documentId, row.getPageNo(), rowNo, row.getOemLabel(),
                row.getAdditionalCashback(), row.getDescription(), row.getEmiTenure(),
                row.getDebitCards(), row.getCreditCards(), row.getEligibleProducts(),
                row.getStartDateRaw(), row.getEndDateRaw(), "PARSED", offerId);

        BenefitParser.Result benefits = BenefitParser.parse(row.getDescription());
        if (benefits.getWarning() != null) {
            summary.drop("description: " + benefits.getWarning());
        }
        for (Map.Entry<String, BenefitParser.Benefit> e : benefits.getBenefits().entrySet()) {
            BenefitParser.Benefit b = e.getValue();
            repository.insertBenefit(offerId, e.getKey(), b.getCalcType(),
                    b.getFlatAmount(), b.getPercent(), b.getMaxAmount());
            summary.bump("benefits");
        }

        TenureParser.Result tenures = TenureParser.parse(row.getEmiTenure());
        if (tenures.getWarning() != null) {
            summary.drop("tenure: " + tenures.getWarning());
        }
        for (TenureParser.Tenure t : tenures.getTenures()) {
            repository.insertTenure(offerId, t.getMonths(), t.getScheme());
            summary.bump("tenures");
        }

        for (int bankId : credit.getBankIds()) {
            repository.insertBank(offerId, bankId, "CREDIT", "INCLUDE");
            summary.bump("banks");
        }
        for (int bankId : debit.getBankIds()) {
            repository.insertBank(offerId, bankId, "DEBIT", "INCLUDE");
            summary.bump("banks");
        }
        if (credit.isUpi()) {
            Integer upiId = bankAliases.get("UPI");
            if (upiId != null) {
                repository.insertBank(offerId, upiId, "UPI", "INCLUDE");
                summary.bump("banks");
            }
        }
        for (String name : credit.getUnresolved()) {
            summary.getWarnings().add("unmapped bank token: " + name);
        }

        for (ChannelScope scope : channels) {
            repository.insertChannelScope(offerId, scope.inclusion, "RETAILER",
                    scope.channelRef, scope.rawText);
            summary.bump("channelScopes");
        }
        persistProducts(offerId, row, decision, catalog, aliases, summary);
    }


    private void persistProducts(int offerId, CircularRow row, ScopeConfig.Decision decision,
                                 CatalogIndex catalog, ProductAliases aliases, Summary summary) {
        List<String> entities = ProductNames.split(row.getEligibleProducts());
        // Apple is restricted to iPhone by scope rule, so an iPad/Mac/AirPods entity
        // riding along in an iPhone row must not be matched against Mobile Phone.
        entities = decision.filterEntities(entities);

        List<ProductMatcher.Candidate> candidates =
                catalog.candidates(decision.getBrand(), decision.getCategoryId());

        for (String raw : entities) {
            if (SELECTED_MODELS.matcher(raw).find()) {
                repository.insertProduct(offerId, raw, "CATEGORY_ALL", null, null, "REVIEW", null);
                summary.bump("products");
                summary.status("CATEGORY_ALL");
                continue;
            }
            ProductAliases.Alias alias = aliases.find(row.getOemLabel(), raw);
            if (alias != null) {
                if (alias.isIgnore()) {
                    repository.insertProduct(offerId, raw, "UNRESOLVED", null, null, "IGNORED", null);
                    summary.bump("products");
                    summary.status("IGNORED");
                    continue;
                }
                if (alias.isPin()) {
                    repository.insertProduct(offerId, raw,
                            alias.getCatalogId() != null ? "VARIANT" : "MODEL",
                            alias.getCatalogId(), alias.getSuperCatalogId(), "CONFIRMED",
                            java.math.BigDecimal.ONE);
                    summary.bump("products");
                    summary.status("CONFIRMED");
                    continue;
                }
                // REWRITE: match on the catalog's spelling but keep the circular's
                // variant spec, so 12+256GB still selects the right SKU
                for (ProductMatcher.Match m : ProductMatcher.matchAll(alias.rewrite(raw), candidates)) {
                    String status = ("AUTO_EXACT".equals(m.getMatchStatus())
                            || "AUTO_MODEL".equals(m.getMatchStatus()))
                            ? "CONFIRMED" : m.getMatchStatus();
                    repository.insertProduct(offerId, raw, m.getMatchLevel(), m.getCatalogId(),
                            m.getSuperCatalogId(), status, m.getMatchScore());
                    summary.bump("products");
                    summary.status(status);
                }
                continue;
            }
            // fan out: memory stated -> one variant; memory absent -> every variant
            for (ProductMatcher.Match m : ProductMatcher.matchAll(raw, candidates)) {
                repository.insertProduct(offerId, raw, m.getMatchLevel(), m.getCatalogId(),
                        m.getSuperCatalogId(), m.getMatchStatus(), m.getMatchScore());
                summary.bump("products");
                summary.status(m.getMatchStatus());
            }
        }
    }

    /** TINYINT(1) arrives as Boolean under Connector/J and as a Number under Hibernate. */
    private static boolean isCurated(Map<String, Object> document) {
        Object value = document.get("manuallyCurated");
        if (value instanceof Boolean) {
            return (Boolean) value;
        }
        if (value instanceof Number) {
            return ((Number) value).intValue() == 1;
        }
        return value != null && "1".equals(value.toString());
    }

    static String benefitTiming(String cell) {
        String key = cell == null ? "" : cell.replaceAll("\\s+", "").toLowerCase();
        if ("instant".equals(key)) { return "INSTANT"; }
        if ("deferred".equals(key)) { return "DEFERRED"; }
        if ("upi".equals(key)) { return "UPI"; }
        if (key.contains("instant") && key.contains("deferred")) { return "INSTANT_OR_DEFERRED"; }
        return "NONE";
    }

    static LocalDate parseDate(String raw) {
        if (raw == null) {
            return null;
        }
        Matcher m = DATE_DMY.matcher(raw.trim());
        if (!m.matches()) {
            return null;
        }
        return LocalDate.of(Integer.parseInt(m.group(3)),
                Integer.parseInt(m.group(2)), Integer.parseInt(m.group(1)));
    }

    private static String nullToEmpty(String s) { return s == null ? "" : s; }
}