3j@&pdZddlmZddlZddlZddlmZmZddlmZddl m Z ddl m Z m Z ddlmZdd lmZmZdd lmZmZmZdd lmZdd lmZe rdd lmZddZddZ d ddZ d ddZ d ddZ! d ddZ"ddZ# d dZ$y)!u Staging pipeline — Story 9.1. Orchestrates: upload → parse → normalize → dedup → stage → review → commit. Architecture constraints (AR-11): - Uses a SEPARATE SQLite engine for staging (instance/import_staging.db) - Main DB is touched only to: create ImportBatch, check existing hashes, commit transactions - Staging rows are soft-deleted (committed_at set) after successful commit - Staging DB is managed by this module — callers never open it directly ) annotationsN)datetimedate)Decimal) TYPE_CHECKING) create_enginetext)Session) StagingBaseStagedTransactionModel)StagedTransaction ParseErrorcompute_dedup_hash)merchant_normalizer)flag_duplicates ImportBatchcdtd|d}tjj||S)Nz sqlite:///F)echo)rr metadata create_all)staging_db_pathengines 9/var/www/html/financials/app/services/staging_pipeline.py_get_staging_enginer"s/ Z'89 FF##F+ Mc ftjjtjjtjjtjjtd}tj |dtjj|dS)NinstanceT)exist_okzimport_staging.db)ospathjoindirname__file__makedirs) instance_dirs r_default_staging_pathr'(sc77<<PX@Y0Z []ghLKK t, 77<< &9 ::rchddlm}ddlm}ddlm}|xs t }t|} |jjdjD cgc]} | j| jd} } |jj|jj|jj!djD cgc]} | d } } |Dcgc]}|j}}t#|| }d}t%d |D}t'| 5}t)||D]\}}t+j,|j.| }|r|n |j0}t3|j4j7|j.||j8|j:||j<|j|j>||rd nd }|jA||d z }|jCddd||tE|dScc} wcc} wcc}w#1swY'xYw)a- Stage parsed transactions for user review. - Normalizes merchant names - Checks for duplicates against existing transaction hashes - Writes to staging DB - Returns summary dict with counts Returns: {"staged": int, "duplicates_flagged": int, "parse_errors": int} rdb Transaction)MerchantMappingT)user_confirmed) raw_pattern normalizedNc3&K|] }|sd yw)N).0fs r z%stage_transactions..Ys4!!A4s duplicatepending r merchant_rawmerchant_normalizedamount is_creditissuerconfidence_score dedup_hashraw_textimport_batch_idstatusr2)stagedduplicates_flagged parse_errors)#app.extensionsr*app.models.transactionr,app.models.merchant_mappingr-r'rquery filter_byallr/r0sessionr@filterisnotrsumr ziprapply_user_mappingsr:r;r r isoformatr<r=r?rAaddcommitlen) staged_txnsrFrBr>rr*r,r-r!rm user_mappingsrowexisting_hashest new_hashesduplicate_flags staged_count dup_countrMtxnis_dup user_namer0models rstage_transactionsre0s $"2;  535D  &F !&&000EIIK  Q\\BM //077 8N8N8T8TUY8Z[__aAO )441!,,4J4%j/BOL444I G{O< KC+??@P@PR_`I&/S5L5LJ*XX'') --$.zz--!$!5!5>> /&,{) E KK  A L' ( +0'L) Q  5 s H H)H#)CH((H1c|xs t}t|}t|5}|jtj tj |k(tjjd}|s"|j tjdk7}|jtjjj}|j|cdddS#1swYyxYw)z9Load staged transactions for a batch from the staging DB.Nrejected)r'rr rJr rNrB committed_atis_rCorder_byrdescrL expunge_all)rBrinclude_duplicatesr!rrMqrowss rget_staged_transactionsrpys  535D  &F  G MM0 1 8 8 " 2 2o E " / / 3 3D 9 "/66*DEAzz055::<=AAC   s CC33C<c>|xs t}t|}t|5} | jt|} | t d|d||| _||| _||| _||| _ ||| _ | jdddy#1swYyxYw)z'Update a single staged transaction row.NzStaged transaction z not found) r'rr getr ValueErrorr;r< category_id account_idrCrU) staged_idr;r<rtrurCrr!rrMrZs rupdate_staged_transactionrws  535D  &F Gkk0)< ;29+ZHI I  *&9C #  CJ  ")CO  !'CN  CJs A%BBc ddlm}ddlm}ddlm}|xs t }t|}d}t|5} | jtjtj|k(tjjdtjj!ddgj#} t%j&} | D]} | j(xs|} || j*| j,| j.| j0| j2d| j4| j6| j8| | j:| }|j<j?|| | _ d | _|d z }| jAddd|jjC|}|r'd |_t%j&|_ ||_"|j<jA|S#1swYixYw) a: Promote accepted staged transactions to the main Transaction table. - Only commits rows with status='pending' or 'accepted' (not rejected/duplicate) - Sets committed_at on each staging row after commit - Updates ImportBatch.status to 'committed' - Returns count of committed transactions rr)r+rNr8acceptedF) rr;r:r<r= is_manualr>r@r?rurtrB committedr2)#rGr*rHr,app.models.import_batchrr'rr rJr rNrBrhrirCin_rLrutcnowrurr;r:r<r=r>r@r?rtrMrTrUrr row_count)rBdefault_account_idrr*TxnModelrr!rr{staging_sessionronowrZacct_idnew_txnbatchs r commit_stagedrs">3  535D  &FI !O$$%;<CC " 2 2o E " / / 3 3D 9 " ) ) - -y*.E F  #% oo Cnn:(:GXX$'$;$; --zz--zz>>!$!5!5"OO / G JJNN7 #"C $CJ NI' *  ;!>    ! !/ 2E " %__.#JJ M!!s EG::Hc t|jj|j|j|j |j |j|j|j|j|d S)zMConvert a StagedTransaction dataclass to a StagedTransactionModel ORM object.r8r9) r rrSr:r;r<r=r>r?r@rA)rarBs rstaged_txn_to_modelrsc ! XX   !%%33zz--zz-->>'  rc|D]L}|jddj|jk(s5|jdsG|dcS|j|S)a  Story 9.5: Resolve category_id for a merchant from user mappings. user_mappings: list of dicts with 'normalized', 'category_id' default_category_map: {merchant_normalized: category_id} for built-in rules Returns category_id or None if no match. r0rt)rrlower)r;rYdefault_category_maprXs rauto_categorizersd$ 55r " ( ( *.A.G.G.I IaeeTaNb]# #$ # #$7 88r)rstr)returnr)N) rWzlist[StagedTransaction]rFzlist[ParseError]rBintr>rr str | Nonerdict)NF)rBrrrrmboolrzlist[StagedTransactionModel])NNNNNN)rvrr;rr<zDecimal | Nonert int | NonerurrCrrrrNone)rBrrrrrrr)rar rBrrr )r;rrYz list[dict]rzdict[str, int]rr)%__doc__ __future__rjsonr rrdecimalrtypingr sqlalchemyrr sqlalchemy.ormr app.models.staged_transactionr r app.services.pdf_parsers.baser rr app.servicesrapp.services.duplicate_detectorrr|rrr'rerprwrrrr3rrrs # # *"M[[,;3  ;#' F(F"FF  F  F  FV#'$" .'+!"!"&#        @#';;; ; ;|"999)9 9r