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; }

}