From 6ec74f1c6457837d469dfc144f13696c21672539 Mon Sep 17 00:00:00 2001 From: Bishwajeet Kumar Rajak Date: Sat, 13 Jun 2026 14:58:21 +0530 Subject: [PATCH] first commit --- .claude/settings.local.json | 10 + .env.example | 23 ++ __pycache__/config.cpython-39.pyc | Bin 0 -> 4137 bytes __pycache__/logging_config.cpython-39.pyc | Bin 0 -> 1283 bytes __pycache__/neft_inward_parser.cpython-39.pyc | Bin 0 -> 10203 bytes __pycache__/scheduler.cpython-39.pyc | Bin 0 -> 4980 bytes config.py | 92 ++++++ db/__init__.py | 6 + db/models.py | 50 ++++ db/oracle_connector.py | 111 +++++++ db/repository.py | 200 +++++++++++++ logging_config.py | 51 ++++ processors/__init__.py | 6 + processors/data_mapper.py | 132 +++++++++ processors/file_processor.py | 166 +++++++++++ requirements.txt | 23 ++ scheduler.py | 171 +++++++++++ sftp/__init__.py | 6 + sftp/file_monitor.py | 95 ++++++ sftp/sftp_client.py | 157 ++++++++++ upi.py | 34 +++ upi_parser.py | 270 ++++++++++++++++++ 22 files changed, 1603 insertions(+) create mode 100644 .claude/settings.local.json create mode 100644 .env.example create mode 100644 __pycache__/config.cpython-39.pyc create mode 100644 __pycache__/logging_config.cpython-39.pyc create mode 100644 __pycache__/neft_inward_parser.cpython-39.pyc create mode 100644 __pycache__/scheduler.cpython-39.pyc create mode 100644 config.py create mode 100644 db/__init__.py create mode 100644 db/models.py create mode 100644 db/oracle_connector.py create mode 100644 db/repository.py create mode 100644 logging_config.py create mode 100644 processors/__init__.py create mode 100644 processors/data_mapper.py create mode 100644 processors/file_processor.py create mode 100644 requirements.txt create mode 100644 scheduler.py create mode 100644 sftp/__init__.py create mode 100644 sftp/file_monitor.py create mode 100644 sftp/sftp_client.py create mode 100644 upi.py create mode 100644 upi_parser.py diff --git a/.claude/settings.local.json b/.claude/settings.local.json new file mode 100644 index 0000000..54559c9 --- /dev/null +++ b/.claude/settings.local.json @@ -0,0 +1,10 @@ +{ + "permissions": { + "allow": [ + "Bash(python3 -m venv:*)", + "Bash(source venv/bin/activate)", + "Bash(python:*)", + "Bash(pip install:*)" + ] + } +} diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..ed5239f --- /dev/null +++ b/.env.example @@ -0,0 +1,23 @@ +# Database Configuration +DB_USER=pacs_db +DB_PASSWORD=pacs_db +DB_HOST=testipksdb.c7q7defafeea.ap-south-1.rds.amazonaws.com +DB_PORT=1521 +DB_SERVICE_NAME=IPKSDB +DB_POOL_MIN=2 +DB_POOL_MAX=10 + +# SFTP Configuration +SFTP_HOST=192.168.1.100 +SFTP_PORT=22 +SFTP_USERNAME=ipks +SFTP_PASSWORD=secure_password +SFTP_BASE_PATH=/home/ipks/IPKS_FILES/REPORTS + +# Processing Configuration +POLL_INTERVAL_MINUTES=30 +BATCH_SIZE=100 +BANK_CODES=HDFC,ICICI,SBI,AXIS,PNB + +# Logging Configuration +LOG_LEVEL=INFO diff --git a/__pycache__/config.cpython-39.pyc b/__pycache__/config.cpython-39.pyc new file mode 100644 index 0000000000000000000000000000000000000000..b8cc4c2db2a1f165b96efcbe1bfa13603f49f186 GIT binary patch literal 4137 zcmaJ^OK;m&7UoM5CChT+B(2*?TlR&4ibS@YIGLnrCVnW1QCBu1Cmj$N5VY=88#fm~|1rUUoge0Mnns<;-SgJv*=FJC~F#JI+YpxxDw>*E!#L(Vd*EDER&3 zpMUfIJ*g=FqR#lQgw6vz(QhD3u@$B=%~x#|t?p|B-PTnaGkjxEunVff3%t0m+NDPd zE3o1lg%$TT+hm$sW+i66QS1sYvhs6H{GC_!6?=kD98?b#tea#NKCx5#MR}|3tFjN? z&eSgotD?PtaU&m}ME_zwK8?19{w45nu3R^T{$-3`c&pP0(>hB1V$!cHhe6Nl?+Q2e z!oV81f!pT;9>i8J6xQPMW2@)++!~6o%cIB(`qt1Ja^DMhqp}vdEVA5yS$nSUF*oLs z#e+Rhgn`^)&lR4#<@2af`41GOwCieWY`F0@+Df0tjvw~>Ty%*y{?pKTfG64j;R@8M z*cwx9ooP&mvJ7Uh0$Qk&mCzQM$;xO;timSHnyktu(Uw__O`)x@3+y7=2{sL(FQvsy z=44@UIP|@4CVBUi9(p1iyhoO)4i2syNB-fDTPQ_ECbK})__ z4ZPTc))H?0|C1Z-x{*#hBMm!HutSifSD-$ZIZ(U1d=;ve7wBP zyU#7JXEh*xW|91r7sXK{F=#0H?t~G9WV^coYgkXE!Jcc97Ob1`p1bQ~6R21aifJ+1 zH7#bAnU)}V0HYUWaHWzmQl)MIWCl-UfGAp3O+Ffl{0>_02v3Wp@B%L}jDkH|;h6;SF_a+TcI8kx^& zD{R=Ml_lqCr}e}x4c%_!uq}H6BO8mI&X4O)RwVoJdS^4a38#;};q!=XHM+N7+$Np& zICmTF@Ol*P#@pAM4Z)&@J8+XQa9>7^Za9$3H`bqQ+D3C>zG+WkYn=6Lb-CrV7k_Bk z#np}PJ1a}snd|Fo&JU~YRGlB|FaF8anse=XCDpt)1V_^tlqrQn4e#05WQ(gz-^DmZO zJ=nSR=+54Y_K#mS?V4P;wAg9Eb~Ya;pJXA1*3S~g&coHUR%iA}iw^6w>!z574#Xvp zv@By70Vh5@=}YJ)C5w=jz-8TIylavxgW?d`=3MkBh5sH;Br`qGBjjpoGII(!xlvcf zvG=3)0Ncb$tR5gok=IJU1a5L<3^}BP!XXj^e3s^pFf6eWa59rmB~B+n`Yc*kHrChH zoYnRw9A!~@#M8}I$JTDl*_EZm&E>~VXVq@mdUI~hF0ZXWa@JbUT5GAX+J3lBiX|P6 zY4^%7^nC>H7#`qCli!6a+m$Uh?ru8~5))F%&=trmTW;_iDTDDS1^f8Dj3DXwe2@F* zYM7G2ES@|oOhdyJHTm?Eeov&$Q_?9dQ%I)RD&wSumlNOk356Cj-YBo7CJ)sWWuFpj zG5L0bu58gYp+H>OlBpa)XW(9sI0QnO!p=?Y4bo%#duMrlrPWC`=jP^`*Fffp+#qt3 z$O4gDL~awg1Ja}^O`6iADNUNvq$y3B()b1H)W-5gUWl~(%1o?_xmTGlyk25C=3+!8XP7>m=H=RB9lp?5zq68~( z{su3vl^q48D)QToiWYfrM?oI-4=zggpH+;Pr~*DPY5Wix zdYctXMd+iHSkLle-0BL{m;m=oBuTzp$!kW;;0(1s3em`-x(Ra1j{(k-t1@GZLhM=Y zOgkL+v9WbuXWAHjP>d=eRAWoWK6%$lv};zoUOqu1t>D_byd^{^1UZA4B65Mq=OC%! z1-&pec0w;W7b1+!wB zvNZ(pyLe9hjz27D5f#Y0im1ivzT%<>thMCpM?9uV7C4dv^!;$_7;ei8qSy_(9ObQ+ z7Bh(CL1UMbccmL5GAjb#+*B|;FiG2kXZ^FZ87c@b9Jl|-2jVRseB z0)0xZ{tt3*z2@ie+Ee~Qffkur(WW6k$g705?i+6A zsaw=DoiP-4D}XP=L5|_oe*!V2p&#gi3M$wYrNpp}b4@N;!!~Y1u0hhUg}3pq37#zi z;h^U$cIkj-2%2u=?lj#^uW`1RP1Lx^nLCTUO~0YSzh;*nSnw8i8&3ox6um3w(r^47 z+W!jeZ|y)<(fU&q!#b|3X+B#i8ADpLyqd-5xtYh48c0@Gx-MkQD-lm~VCKLTQt4r2 zvA?L99$5}80e)P_cd{7mOxbOGAmo&<3bT_BU%WbboqPopxf;blWM*o05t};Jz}TRy zYi+fEhbDybeXh;8K3o4ZIN&-nc_|YeAI5`&itkwK`PROaTJsqwuI~^2I2@FR0}&6N zj|MMCgVThy&a^abAh?mhtOLGW4k4YKkPZ2aolRI8mi+y*3nO)UXTr0y)$+JrRVKNo za1vUtJcH=lpethAn@RH<6jZ7<)WBk0+VG}+t(O&&tG0JitCAZ7TK!X_Brl(DE8Ok@ z=>Nlr3vqM8YTdd^7{u0H)OnSD^R_nJK;urJL!D9*w7p$BQVeR0(fJr1`07n>+@|-s z9*?`w(h!xg+36nA8~I;EeG7U1^Z009m-1+m>-jlfNNJ8LIW^>*lYV2VCn%vLJLp#y99?_MYXz*$N*bbHWqu!4OUrMU*VH&U{5hejk$#spFOt!!xv_^TFU?c literal 0 HcmV?d00001 diff --git a/__pycache__/neft_inward_parser.cpython-39.pyc b/__pycache__/neft_inward_parser.cpython-39.pyc new file mode 100644 index 0000000000000000000000000000000000000000..0843f1fa274fb15012a3b32d43cd0541faed0ffc GIT binary patch literal 10203 zcmb_iTWlOhcJ1oNOwVv;IHF{Vq9wU4%aXL|0eWtE;Q-J@?#O^z!+vhTs0BzpDNHvzqoV^w9sa@bC&Qf1YU?*UB1a zT(7e-Q+K_rtGiJ)a5w5^!zx>iR5`_{omo#e?6Td+lrxQNIolX052$jho@?aGd8UcH zD6F#bpfJ{oJhjfsJH&t(+M|gfp1!AZn`dsB<(*;(rPdyeXHho5b12)TdgO(1^|5um zJd75*c>%S9yolP7S2eza58cxE(5hbE!w-s#urca@7{w^N(C#rle9J07E^NO09^)f? z4|?sz($APZapRCH+$-XYeKaG2{K^{h0CoRSoGA{0Yq45zg9^>)Lxp`9S%p z@PcLwYqR-Jn0MR9GN>iKjTz|;HB)CGuv8N_5-Jlx?1y!Q*BA7 zipIWkqZTYWhoi%eFHq?PsPLNHsWmIDreE^|(F~j$ix}N+dlfNGLy2HnHht$vAibvV zRe~B?9#zA+4H0b>mk|S=SU46x=*Or8H@!V_OIr_m$U5z>a_OyFu za(>FG`A%I_1IP2nkgdvA!wD9J8r(V3Y&k)#AskGt;$dbUz1qQyQE*-0%UaFSc=kR? z<8(=n(`Q7b*6`|a@m%w|SFiC4ZGkVOR%UFTsa1m5d;^~tTl359x`^!yZL0L@B^GC| zh`_D4u3Ql^%)RTp7reJTF)=xD`B{w1{5U1mB2=&s8$ZgMxK88pk0Oa6(#x96(WXV( zn#K)ouIn8wU~4)zBaQMXH`X<5suh;A4t~=ui7D4Yfh`Dx#BBPO-w9d{?H_hoZ3mX! z4`i+Fh$i>PRsSCd(Z(Q6fhBYfA5IFVW9>h5%;?u^5150GPfO9RV>{!*r8)QXg*V=q zoSB<)XQyJrUtWkUHD-y)47z-w$sybxym)M})ey%PYX0I4@2U{Nu_pG(tu=3WlDi~J z?tkV)@ka0%z9sHMXvGScLqNE{|~3h1o!l*ebh(yr~7o%3BN?$+W;+v$zpy zclFP7(yeze-kdg}(vms-fE|?sculx>ve86B8kfJSl$B{W58 zAc<3n=)`HPP7@ne(&b}R&aSJ~Op={ck_}pJceKNJ_GuL~MvGEZsm`)koFbE>C=Z9< ze|K`MeEEmZbtWh?G4|cdKYYG(5X)v(TKo~KwnZq1F+`;KHfM8Mpu`YDh?Rl(>ERtE zKCeRGAzkOr&Yey^Rq4w~h#61;J59*Z@B-4)W;`Z%xfQQd||4z*kBxMWrQqGQioA5t|FGRy|JBTu3%8 z)e&Mz!n1uD^J-rJ3U00HS6)C1KZQiIOsu4Q0=Wu3_Ww2v)!(61XfRM38K<&w1u*+d zRyO%#!V)QwCQ|}4V8RSoJOwjg!#t!#76u|KvQ-0hX_$e;_zbAJZl9!YRqwXQ@&Oo% zTvCd@d10x#97fLb0_qCe>(C}eZ{maZOkU(W&}OjrZa##5JNYh@6}f>;9FB*kr_Rp1 z=cX@B&Yp1-kPYibbuTlxzrq>YhpcF7xfHsJI;MMAJ9tC8+B*3KFit`Hs$l`jOOX2(RKq4V7i8nPR zadD;a{8ls0UTwh=d3Cqhic^r#VA+pzKrpxAUvYz#wum!iqsRvJvSdHv0g@3nnOU4w z50Jg(*zmp?tomS2sIv z*Eie6Zq)+Uou^qVBxfmBz?EX*;}1*Wc&|P5_aSH}Sf@g@0{@0g_&EVOh?3r69UXZV z0irubpkFoCU^%Z^>nI6Q1nZpL1Bkonz$Vb=#vT2PcKHZ;njH%*Gm#Y;k;%#p{hNv)|AR~lEoYIMy>apj+x z)=4uhJP;JF?%?StjrNL)5}o@{yF`X1+ev?rZf2AF1R0quc5fhjTPY18=6SEyRP;zN~^xyX!@jd&>+Ff8QHo4t%7B{PeuZq2X^0+ z0WxW-YA!dzv_cK;gaZzWx_m6$Hzg%tgp50J$qhImfja5TV*p#+tSD`sE)B&aq^{hh z0bF96Zq=*T7re?+yb}sRtW&s)XaN9dcagKiGskq6`i-611$%C? zf?m)~HvFY&4Cw{7i-jYPP#ZPS79K>TjChGsmSJoiXW%>9RSjbQUmZRCJfJ6?>k*WJ zo7}o%{p=7_$GB=n22b(y9jHcNMdm8&SeMY-1S?uKnAR*F*Zi3Y#f^v!LEtUF8lsZ3X zvY)aQd5DG=+}AK2Hj{ds&h%1D{6Wh3?}YZrqY{5P5_ zF`Av>%NyD-FISw#ay_Vl=7TFIXzY)7X^L}%cKWE3>~LwL*L2Ke!tQ9jp;30PYn)P& z;>YQxxKXb)#RC>W>@z)ED#PqlqaCa$zh%qr5zv#9Fo#QVJp%U2Iea@{f0g}Qzh=%Q z zWmy4AZJ-&qg@__c6*2tyKQwO9S(1O22E9yp5@No_T0OcVzQv$@Zud}9C17674( zB^CcVIdOzKu}PmFXWt^fLqVH@2l+b9bc&K6P%?rf&dmZZK_OJLQbqwq*{*wj;6fr| z19OX6RW6`y*~TOk>d9&9f1VOr#qCS;3d%mfeTWU~qj)07 z7-k3c@Y%1i$Ji3}a%FN!_If`e$7x)&wf;>c@CSeeAe2CjTjs~wM_Lu}7;fEySyh^? z16TA_mY~ZjThioP0gN$ECFRiBbgvG1`!kqmrqwhck461}PQ|WKRtEW$c5NXd(wYf> zyrFUDy*7M#GF0$>xwz?!z2sE9W~*6)pgMljTM}*sM!{JUD+y60(|NJE)NI{o9)n*g zy{;Df&H}s%#0ob2PTt>Ng%>tQApc1jnun-K&wo5k>c#ubLce@LTW4Gf}(w;apzJED&7Vr5O%{KuL^{#u8aE0cqxT2 zMj(~y#)DID&rF@3pE~2dHZ^%>YBo`A(q(xT3EX$G^-4*~16>nOB2&)7#uKlAq{XZz zo775ST8nb(`zMk6JzQ!THLGY1L0L^U`lSi4v&gb~cC5N<4 zy&qY9(pjaiA3`c`YqA(1rFh3GYwECuW}x4Org42li%jU}E4QJ`R~X2zzFSM`Mzmep zZtY{ed0Z~uxLtFUN9x_Ne)$-j?n;` zsvTsBNqowDyKjbhMY0o8Zll#WT7jjY)p%d4!X}64zHe}yc*09GKISoa3L!b%iyQOV zq3YkK`n7>|4dZ>#H{MV;1`FyQS&Ct7()ZIU0vR|Uh0f`@w@50*K}<4u zV+}a8bp%7&Z6eW&LW*wegdpwym-s0&nZ%&XS&SDxqZA~8xoS<+x&J)};XioK7y)lf znSqGz2AVAacHg8~DV0{*d>xN}OGPwMNw#=e(pt%5NWv!(AE!V@ z5*e$vT1!5*wmH^|i-Cze1r|~fz@Y!7kt7o$$0&J{hJDQOmum1+;O#YPzG6QJ%#?Dm z(eQ4@Ccwu}D2Q@cEG`z7uf!H&8i-3@qfzMFWt);4lu)2cnV2p&k!n*maHX8au&(l6 zf}73Q@EXD9Hzs&*B%7%gM|D<<^D4&uOI#{Gp;_d>?8P-?>@{{7BW#y3#D?%ZOc$PQ zV-(NkR{-F{`%U&26Y>+>>*pzidvu+qm;X1CgeP~vMmV7hf*~D~7|P{e!`-(KSf@Zi zDlj7Ol2u)vXr57d6ZvYG14Bs>CSB65T}Ld~T3P7Ef|Jow@1T4Fo)9q0Ky3MX_bn-e z5K=D?moe2lfIRI_xCN9*feXOpkId^#{snO%niZWzMP`&f1ssmj2;SKi1rReY0KPZM z;4ww1-Mtn%_|OQa03k}&((8#gWk))Ew%lDavU&a2D^S9L`5jtod8f>dtSaFkZ+DQx##(UIB8gDd<*+f@Gp240dcu=mx z7l_}h9;#M;NR_9MY*AjaVPwnZaQR1&DCKp!!5*+oj@3d67~=X}lNke030WQAP|wF zgIYiep6I?=L(e=7V1gQ7{zb>SqJaicT=Fc0Jl!FhWJGCh!&%4xQZj)Bc=|8`b-HVI zGEr(R6{MpKfCRLLLMwbgQ5h%$IaSx*!@Dp$ieb{5K+SRhC%~Yb0pPGDOU|H09`nQu zBXdLpsVjV@W{KYadW~u~zM?z#!K8IU|J3{`LRk!dTj@OLjo*TwjQ}=I!j2Hj=?l0y zirHMRi5uV^jauE4#71z;o3!upnsd!ceVOCPnVj>4zJSzI>`k5O9(87?UI$gjdE}WX z01ghJy9KjTlXDlQ6K}0Fh`l7!c|Qq`D4e;UOZJ$s=spFL(WXRs^0Sa9qkEq=8u_Ov zzc<+12>yME7edx9k`-xfh0eexwcmn8zcGL16(r&+C;V1EIaGR1Hn^#-4W^AdcB*3T$%SSZ2g2>k>NBB_6DrJ?w zK%R~@>we;%S?WcyL+yM@=;P&Q)XKo=-?lY~haUcF152U|j%L|Ab^-HjR@4z=fzw8a zm__aYA}u!ZBP^xk(Dhdso#NTOW0FVip@NQSw4?}%QYYB0?p-Mz9bk|}zudKP#9`JD zA>byg8=@oFW!3JQgAu?_2t1&CcKF1M(l-jM)2V>^*9zwd6l;wBk)kyI;LYrd-|#QY zpFQ`+lyiP+eiHwXG5KKT8y^3~&^UEsLb_P42XB7UV`rzoFyuOhmLF5n)s#O(J~qIm z)fxPDIVmJ^(p%mMk`cla4HlFBifok0KgBcTAl*Gg_-ZkjLBmLsZ~UL4us>1to+to! z%|1ami2k6L5QG#2Asy~a1flzS6)_!`pW|I&Ztmjw^OLjhIJ8^xk5Hu6R{k+^)A9?X zz=}Vk;yu|H!3dLojuNu?4IEUtE*2y2<9w>E{z(S@ zXZW3m{y_x)GZeV$07D*|cz)trCnioj?;d~7?PC|?^KZ}Jr!@nvk-mp}XS%0KL~H9f zs~UHvD6B_ER!*4T0x#(EbLx6n{tP$iQSyBxu`WfN?*3Cn|378mNEkCsxCVi5LOs&d zr9l6%fl4stX0>)j+Eo1sCA+CQP1AuE%0Vh3luCSIa=+s2xmTwz%ub!2oSTySsEW)> z;)*$x`;-zw5+z1zAtgCe_LLTh~2x()o@7l z%u36~Pyr&J3>eVeUVwY_#fKn2!pKjUYk&a0<*ZY_>LIz*M)oNr*wfQpUENh(^;H$a zdfnCVS$p{R&NpW??O*gV`!VtI34X~44byy$>CA|9Usq4VH`LSgO+3xW>RP^~6O9#> zy0&k3%YIqCmm;TI@hhrtM{c+3SGzU8rryg@z1#2`-8p|wy*tr-cfnuKHE#38XS%=i znZ_#2{aR!0Gs9nIRaX02^H=yHt3Ncv_k87h<~_f6T?e``vY~+ldcSo_J|*vet|F6U8iPxj#b# z+GbPF9TuiM?R0t5$lYC@22pQ!my5if@bsXs-oP7#YO>vncRITuo%^E*+Y!Ini(}qS zdm?x5@qRDqpgSN=TN~SVZ$=#+r+IY)=l`M?(;WEJ>Gf_QO;5Dx9J3z-FQ4F-e29i? zxEqb`1$SaF6PmJ^#Y%XVn9a&~+RR}UJj={wRXiP5gYN2iWox1(`SurKCqB`dx6>2e zs}$R6>)3Php}vLRP5hEi(WKflEz<%$)Asd|aZJ~Qet?6awbIh4l$Q2w%$AQ0T^qkU zlXqs~Se9J513h%Ys3ZBwiCWu&cH3-LL_~g8)~-_9_{>~i4|<{;{9uTy)M7@ zpp)!93HLcquf=>P4Lb3YP_TgPEqD+noLxIHwf-P?0#J0)Ab@XaNezu=IJ&Fbx^2$u za&fBh)`YN49yj?=pTtuNy@*b#jo^8i{vGW7$UN4eP!p;&m<3gpIkY3BCW#;r~aTWH`Y zO;ar6;VX?Ebo0`V2)kSRW!1lAOGHo5q(uJ*I`R{; zrXhqM>BpLArN#)LLn=ruj2oFTDv^cVS5Sain<)k19m=_{#j&NO<-ucJ8=AT%*Azu5 zGr!Z04C3-P(0rvaduTp2<4=p{b(nMN41OH1xttpNaG_!ioXhzZR(rCF_L;1^jvlV0 zB0re1Uf5+wM&iL(2?IL4*u#Z+z@N^}z{6w&^x=n{ByGvpZoPsU=#+Z>FoB_M{|d!U z2zz83(wp)iD^xxds$C0$b#GH%x-A5pdpfUdV7&9X_XE(5AGTq{J3Y9XjeGFt=n!zH zi3s8>M#VX5$~3=Jh!Ygz1WKH|#AzFV+#pGL_x52Y%`3ML+q_T5XIf{kVRGks|8YC|injND^wRAu(DJ}jv{%2{b*rP!^uU%#dO<@=!xi*IB^RaT>O zIE9nhXe&q7(4jTVJu-%sOy8f+oUFpC-x@_S#3x8k2*mu!z`;S?v0}o$?)hW%q+4wH8xN4M@qIwKRx;;v!3h8=W6A`E9E84 zEv8GE^^BlwX{@{D%zz$@VLhvlRx)rKHWZKLv5r^9bhLUkYp^rtp!vdk72R2kojWyl z9%C0y#tI5{@gYtqUZ_=<=!DSa>am8g|ERGw8jD@@ZT4CrLA(wuIP=^X+yELNQITG% z0}Z?kdOtGQ>+JFr4NAkgr*pB()bqM_ede}fcXIDQ2;6-04f#4U5t_mea_!@fKlZlLP$-<9Np2L8CoK;I(%VqJPdUuko2W~xzdhiS@1Eax{6PT4d)TKA;m zz5O7JAFhK*Q@*+Nu+#4o?=TWPWP?dUg{U*#O&VBsX0tctGMjlxyVyal{KE82it3(p zA!fH9@szL2InUefrD0TLZ0nw1O6gTLJa6luO}QXKAcoy!`iAvmjbYp_U{Pfj^486; z$PrbX5Br@UJ&Xe+Yye2E_N3Kn$@jKT29R|o>ASNR zH$FKc@$^8^Kwkc}HN_F-g$Zm=6id2RhOq8!iwm%q=2et6y3Oawa4UGo>qjm8bXq)pzq^PQW+A3QX z9x*9*i*(tqAY~tOs-!&@4g$#FSEeWT8>%}!xnC8$+e^7Ru($$VNM#T{liZ<4jHD*7 zQ0i1jR2eVq3BQ!2VVa27vE)?RlP@FroG1r36BIWC!o)|T8Hhv%azMjT*`Rg9`KM(% zxLJJ6H};V#F>4ILl!v087+Got0bvTnDnlh6@?xfq zYy`eN@F=r#0i(=O$=D5*Sd9RsM7gFY5U3_x#>tp7)Sv1AQuQPtEzZH*WlH$8{R#m4 zOXC;}5~NI(RYW>!m0yShe8GYOFqm=vQ&Bh#!@-jj6PAV`*!ndZx7TsGyi9 zSxA!Rf;_9>dr~=hT`HWYtZd1^yf`hoQJr-nVEtiQ;25Fjg%g>kCCvoxn~fQ0 zBXsgtsH#V*T8JBrMa`}KUMCjsfI?6uqA*WTqNvIk@muNnvgEYSn{c?N;FDZMqp7rWRX-0`HB>TMxbj)E2A4PV8eIJy(?(KSGh}@#W(zz} z(^B0?z!Lfgu;v9|rjE0{^7+ zq3^VXs!>o8r}zpw5AugfGI5ks=gKW|oDrxJDl#Iv8tOj+{3r52RWL;iv?zoK6%#4w`R)|h6BU-k2eeSt uHl!JKmYG8<-I7Lg=cD3+eowkm6^K=GaODmzd0zd{8jjict5$bhNB=J+$2jBw literal 0 HcmV?d00001 diff --git a/config.py b/config.py new file mode 100644 index 0000000..de3f895 --- /dev/null +++ b/config.py @@ -0,0 +1,92 @@ +#!/usr/bin/env python3 +""" +Configuration management for ACH file processing pipeline. +Loads and validates environment variables. +""" + +import os +from pathlib import Path +from logging_config import get_logger + +logger = get_logger(__name__) + + +class Config: + """Application configuration from environment variables.""" + + def __init__(self): + """Initialize configuration from environment.""" + self._validate_env_file() + self._load_database_config() + self._load_sftp_config() + self._load_processing_config() + + def _validate_env_file(self): + """Check if .env file exists.""" + if not Path('.env').exists(): + logger.warning(".env file not found. Using environment variables or defaults.") + + def _load_database_config(self): + """Load database configuration.""" + self.db_user = os.getenv('DB_USER', 'pacs_db') + self.db_password = os.getenv('DB_PASSWORD', 'pacs_db') + self.db_host = os.getenv('DB_HOST', 'testipksdb.c7q7defafeea.ap-south-1.rds.amazonaws.com') + self.db_port = int(os.getenv('DB_PORT', '1521')) + self.db_service_name = os.getenv('DB_SERVICE_NAME', 'IPKSDB') + self.db_pool_min = int(os.getenv('DB_POOL_MIN', '2')) + self.db_pool_max = int(os.getenv('DB_POOL_MAX', '10')) + + def _load_sftp_config(self): + """Load SFTP configuration.""" + self.sftp_host = os.getenv('SFTP_HOST', '43.225.3.224') + self.sftp_port = int(os.getenv('SFTP_PORT', '4650')) + self.sftp_username = os.getenv('SFTP_USERNAME', 'ipkssftp') + self.sftp_password = os.getenv('SFTP_PASSWORD', 'CqBx@j6G9vqNW#1') + self.sftp_base_path = os.getenv('SFTP_BASE_PATH', '/home/ipks/IPKS_FILES/REPORTS') + + def _load_processing_config(self): + """Load processing configuration.""" + self.poll_interval_minutes = int(os.getenv('POLL_INTERVAL_MINUTES', '30')) + self.batch_size = int(os.getenv('BATCH_SIZE', '100')) + self.bank_codes = self._parse_bank_codes() + self.log_level = os.getenv('LOG_LEVEL', 'INFO') + + def _parse_bank_codes(self): + """Parse bank codes from comma-separated environment variable.""" + codes_str = os.getenv('BANK_CODES', '0001,0002,0003,0004,0005,0006,0007,0009,0012,0013,0014,0015,0016,0017,0018,0020,0021') + return [code.strip() for code in codes_str.split(',') if code.strip()] + + def get_db_connection_string(self): + """Generate Oracle connection string.""" + return f"{self.db_user}/{self.db_password}@{self.db_host}:{self.db_port}/{self.db_service_name}" + + def validate(self): + """Validate critical configuration.""" + if not self.db_user or not self.db_password: + raise ValueError("Database credentials not configured") + if not self.sftp_username: + logger.warning("SFTP username not configured") + if not self.bank_codes: + raise ValueError("No bank codes configured") + logger.info(f"Configuration validated. Bank codes: {', '.join(self.bank_codes)}") + + +# Global config instance +config = None + + +def get_config(): + """Get or create global config instance.""" + global config + if config is None: + config = Config() + return config + + +if __name__ == '__main__': + cfg = get_config() + cfg.validate() + print(f"Bank Codes: {cfg.bank_codes}") + print(f"SFTP Host: {cfg.sftp_host}:{cfg.sftp_port}") + print(f"Database: {cfg.db_host}:{cfg.db_port}/{cfg.db_service_name}") + print(f"Poll Interval: {cfg.poll_interval_minutes} minutes") diff --git a/db/__init__.py b/db/__init__.py new file mode 100644 index 0000000..701b306 --- /dev/null +++ b/db/__init__.py @@ -0,0 +1,6 @@ +"""Database module for ACH file processing.""" + +from .oracle_connector import OracleConnector +from .repository import Repository + +__all__ = ['OracleConnector', 'Repository'] diff --git a/db/models.py b/db/models.py new file mode 100644 index 0000000..475fe21 --- /dev/null +++ b/db/models.py @@ -0,0 +1,50 @@ +from dataclasses import dataclass +from datetime import datetime +from decimal import Decimal +from typing import Optional + + +@dataclass +class UPIInwardRecord: + rrn: str + txn_date: str + txn_amt: Optional[Decimal] + credit_account: str + status: str + note: str + bank_code: str + txn_type: str # will be CREDIT / DEBIT from parser + + def to_dict(self): + return { + "BANKCODE": self.bank_code, + "RRN": self.rrn, + "TXN_DATE": self.txn_date, + "TXN_AMT": self.txn_amt, + "TXN_TYPE": self.txn_type, # converted to CR before insert + "BENEFICIARY_ACCOUNT": self.credit_account, + "STATUS": self.status, + "NOTE": self.note, + } + + +@dataclass +class ProcessedFile: + filename: str + bankcode: str + file_path: str + transaction_count: int + status: str = 'SUCCESS' + error_message: Optional[str] = None + processed_at: Optional[datetime] = None + + def to_dict(self): + return { + 'filename': self.filename, + 'bankcode': self.bankcode, + 'file_path': self.file_path, + 'transaction_count': self.transaction_count, + 'status': self.status, + 'error_message': self.error_message, + 'processed_at': self.processed_at or datetime.now(), + } \ No newline at end of file diff --git a/db/oracle_connector.py b/db/oracle_connector.py new file mode 100644 index 0000000..15ab91b --- /dev/null +++ b/db/oracle_connector.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python3 +""" +Oracle database connection pool manager using oracledb. +Manages connections with pooling and health checks. + +oracledb is the modern, simpler replacement for cx_Oracle. +No Oracle Instant Client required - uses Thick or Thin mode. +""" + +import oracledb +from logging_config import get_logger +from config import get_config + +logger = get_logger(__name__) + + +class OracleConnector: + """Manages Oracle database connections with pooling.""" + + def __init__(self): + """Initialize connection pool.""" + self.pool = None + self.config = get_config() + + def initialize_pool(self): + """Create connection pool.""" + try: + # Build connection string for oracledb + # Format: user/password@host:port/service_name + connection_string = ( + f"{self.config.db_user}/{self.config.db_password}@" + f"{self.config.db_host}:{self.config.db_port}/{self.config.db_service_name}" + ) + + # Create connection pool using oracledb API + # Note: oracledb uses 'min' and 'max' for pool sizing + self.pool = oracledb.create_pool( + dsn=connection_string, + min=self.config.db_pool_min, + max=self.config.db_pool_max, + increment=1, + ) + + logger.debug(f"Oracle connection pool initialized: min={self.config.db_pool_min}, max={self.config.db_pool_max}") + return True + except oracledb.DatabaseError as e: + logger.error(f"Failed to initialize connection pool: {e}", exc_info=True) + return False + except Exception as e: + logger.error(f"Unexpected error initializing pool: {e}", exc_info=True) + return False + + def get_connection(self): + """Get connection from pool.""" + if not self.pool: + self.initialize_pool() + + try: + conn = self.pool.acquire() + logger.debug("Connection acquired from pool") + return conn + except oracledb.DatabaseError as e: + logger.error(f"Failed to acquire connection: {e}", exc_info=True) + raise + except Exception as e: + logger.error(f"Unexpected error acquiring connection: {e}", exc_info=True) + raise + + def close_pool(self): + """Close connection pool.""" + if self.pool: + try: + self.pool.close() + logger.info("Connection pool closed") + except Exception as e: + logger.error(f"Error closing pool: {e}") + + def test_connection(self): + """Test database connectivity.""" + try: + conn = self.get_connection() + cursor = conn.cursor() + cursor.execute("SELECT 1 FROM dual") + result = cursor.fetchone() + cursor.close() + conn.close() + logger.debug("Database connection test successful") + return True + except Exception as e: + logger.error(f"Database connection test failed: {e}") + return False + + def __enter__(self): + """Context manager entry.""" + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + """Context manager exit.""" + self.close_pool() + + +# Global connector instance +_connector = None + + +def get_connector(): + """Get or create global connector instance.""" + global _connector + if _connector is None: + _connector = OracleConnector() + return _connector diff --git a/db/repository.py b/db/repository.py new file mode 100644 index 0000000..cb0a48a --- /dev/null +++ b/db/repository.py @@ -0,0 +1,200 @@ +#!/usr/bin/env python3 +""" +Data access layer for UPI inward file processing. +Handles CRUD operations and transaction management. +""" + +from typing import List, Optional +from logging_config import get_logger +from .oracle_connector import get_connector +from .models import UPIInwardRecord, ProcessedFile + +logger = get_logger(__name__) + + +class Repository: + """Data access layer for UPI inward processing.""" + + def __init__(self): + self.connector = get_connector() + + # --------------------------------------------------------- + # Account validation (same as NEFT) + # --------------------------------------------------------- + def validate_account_exists(self, account_number: str) -> bool: + + if not account_number: + return False + + last12 = str(account_number)[-12:] + + conn = self.connector.get_connection() + try: + cursor = conn.cursor() + cursor.execute( + "SELECT COUNT(*) FROM dep_account WHERE link_accno = :accno", + {'accno': last12} + ) + count = cursor.fetchone()[0] + return count > 0 + + except Exception as e: + logger.warning(f"Error validating account {account_number}: {e}") + return False + + finally: + cursor.close() + conn.close() + + # --------------------------------------------------------- + # UPDATED: UPI BULK INSERT WITH VALIDATION + # --------------------------------------------------------- + def bulk_insert_transactions(self, transactions: List[UPIInwardRecord]) -> tuple: + + if not transactions: + logger.warning("No transactions to insert") + return 0, 0 + + valid_transactions = [] + skipped_count = 0 + + for txn in transactions: + acct = txn.credit_account + + # Condition 1: account exists + account_valid = self.validate_account_exists(acct) + + # Condition 2: only CREDIT transactions + is_credit = (txn.txn_type == "CREDIT") + + if account_valid and is_credit: + # Force DB value to CR + txn.txn_type = "CR" + valid_transactions.append(txn) + else: + skipped_count += 1 + + if not valid_transactions: + logger.info(f"All {skipped_count} transactions skipped") + return 0, skipped_count + + conn = self.connector.get_connection() + cursor = None + + try: + cursor = conn.cursor() + + batch_data = [txn.to_dict() for txn in valid_transactions] + logger.info(batch_data) + + insert_sql = """ + INSERT INTO pacs_db.upi_inward_api_log ( + BANKCODE, + RRN, + TXN_DATE, + TXN_AMT, + TXN_TYPE, + CREDIT_ACCOUNT, + STATUS, + NOTE + ) VALUES ( + :BANKCODE, + :RRN, + :TXN_DATE, + :TXN_AMT, + :TXN_TYPE, + :CREDIT_ACCOUNT, + :STATUS, + :NOTE + ) + """ + + cursor.executemany(insert_sql, batch_data) + logger.info(f"executemany rowcount: {cursor.rowcount}") + + conn.commit() + logger.info("UPI transactions committed successfully") + + inserted_count = len(valid_transactions) + + logger.info(f"Inserted {inserted_count} UPI transactions") + return inserted_count, skipped_count + + except Exception as e: + if conn: + conn.rollback() + logger.error(f"Error inserting UPI transactions: {e}", exc_info=True) + raise + + finally: + if cursor: + cursor.close() + conn.close() + + # --------------------------------------------------------- + # FILE TRACKING (UNCHANGED) + # --------------------------------------------------------- + def is_file_processed(self, filename: str, bankcode: str) -> bool: + + conn = self.connector.get_connection() + cursor = None + + try: + cursor = conn.cursor() + cursor.execute( + """ + SELECT COUNT(*) + FROM neft_processed_files + WHERE filename = :filename + AND bankcode = :bankcode + """, + {'filename': filename, 'bankcode': bankcode} + ) + count = cursor.fetchone()[0] + return count > 0 + + except Exception as e: + logger.error(f"Error checking processed file: {e}", exc_info=True) + return False + + finally: + if cursor: + cursor.close() + conn.close() + + def mark_file_processed(self, processed_file: ProcessedFile) -> bool: + + conn = self.connector.get_connection() + cursor = None + + try: + cursor = conn.cursor() + + file_data = processed_file.to_dict() + + insert_sql = """ + INSERT INTO neft_processed_files ( + filename, bankcode, file_path, transaction_count, + status, error_message, processed_at + ) VALUES ( + :filename, :bankcode, :file_path, :transaction_count, + :status, :error_message, :processed_at + ) + """ + + cursor.execute(insert_sql, file_data) + conn.commit() + + logger.info(f"Marked file as processed: {processed_file.filename}") + return True + + except Exception as e: + if conn: + conn.rollback() + logger.error(f"Error marking file as processed: {e}", exc_info=True) + return False + + finally: + if cursor: + cursor.close() + conn.close() diff --git a/logging_config.py b/logging_config.py new file mode 100644 index 0000000..1b22df3 --- /dev/null +++ b/logging_config.py @@ -0,0 +1,51 @@ +import logging +import logging.handlers +import os +from pathlib import Path + +def setup_logging(log_level=logging.INFO, log_dir="logs"): + """ + Configure logging with both console and file handlers. + + Args: + log_level: logging level (default: logging.INFO) + log_dir: directory to store log files + """ + # Create logs directory if it doesn't exist + Path(log_dir).mkdir(exist_ok=True) + + # Get root logger + logger = logging.getLogger() + logger.setLevel(log_level) + + # Clear existing handlers + logger.handlers.clear() + + # Create formatter + formatter = logging.Formatter( + fmt='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + datefmt='%Y-%m-%d %H:%M:%S' + ) + + # Console handler + console_handler = logging.StreamHandler() + console_handler.setLevel(log_level) + console_handler.setFormatter(formatter) + logger.addHandler(console_handler) + + # File handler (rotating) + log_file = os.path.join(log_dir, 'app.log') + file_handler = logging.handlers.RotatingFileHandler( + log_file, + maxBytes=10 * 1024 * 1024, # 10MB + backupCount=5 + ) + file_handler.setLevel(log_level) + file_handler.setFormatter(formatter) + logger.addHandler(file_handler) + + return logger + +def get_logger(name): + """Get a logger instance for a specific module.""" + return logging.getLogger(name) diff --git a/processors/__init__.py b/processors/__init__.py new file mode 100644 index 0000000..4d08ef9 --- /dev/null +++ b/processors/__init__.py @@ -0,0 +1,6 @@ +"""Processors module for ACH file processing.""" + +from .data_mapper import UPIDataMapper +from .file_processor import FileProcessor + +__all__ = ['UPIDataMapper', 'FileProcessor'] diff --git a/processors/data_mapper.py b/processors/data_mapper.py new file mode 100644 index 0000000..26b04c8 --- /dev/null +++ b/processors/data_mapper.py @@ -0,0 +1,132 @@ +#!/usr/bin/env python3 +""" +Data mapper for UPI Switch feed. +Maps parsed UPI transactions (dicts) to UPIInwardRecord for database insertion. +""" + +from datetime import datetime +from decimal import Decimal +from typing import Dict, Any, List + +from logging_config import get_logger +from db.models import UPIInwardRecord + +logger = get_logger(__name__) + + +class UPIDataMapper: + """Maps parsed UPI transactions to UPIInwardRecord objects.""" + + # ------------------------- + # Helpers + # ------------------------- + + @staticmethod + def convert_date(date_str: str) -> str: + """ + Convert date from YYYY-MM-DD → DDMMYYYY + """ + try: + if not date_str: + return datetime.now().strftime("%d%m%Y") + + dt = datetime.strptime(date_str.strip(), "%Y-%m-%d") + return dt.strftime("%d%m%Y") + + except Exception as e: + logger.error(f"Error converting date '{date_str}': {e}") + return datetime.now().strftime("%d%m%Y") + + @staticmethod + def convert_amount(amount_in: Any) -> Decimal: + """ + Convert amount safely to Decimal + """ + try: + if isinstance(amount_in, Decimal): + return amount_in + + txt = (str(amount_in) or "").replace(",", "").strip() + return Decimal(txt) if txt else Decimal("0") + + except Exception as e: + logger.error(f"Error converting amount '{amount_in}': {e}") + return Decimal("0") + + @staticmethod + def normalize_status(status: str) -> str: + """ + Normalize status values + """ + try: + if not status: + return "" + + s = status.strip() + + if s.upper() == "SUCCESS": + return "SUCCESS" + elif s.upper() == "FAILURE": + return "FAILED" + else: + return s + + except Exception as e: + logger.error(f"Error normalizing status '{status}': {e}") + return status + + # ------------------------- + # Mapping + # ------------------------- + + @classmethod + def map_transaction( + cls, + parsed_txn: Dict[str, Any], + bankcode: str + ) -> UPIInwardRecord: + """ + Map single parsed UPI txn → UPIInwardRecord + """ + try: + txn_amt = cls.convert_amount(parsed_txn.get("amount")) + txn_date_raw = parsed_txn.get("txn_date", "") + txn_date = cls.convert_date(txn_date_raw) + + record = UPIInwardRecord( + rrn=(parsed_txn.get("rrn") or "").strip(), + txn_date=txn_date, + txn_amt=txn_amt, + credit_account=(parsed_txn.get("credit_account") or "").strip(), + status=cls.normalize_status(parsed_txn.get("status")), + note=(parsed_txn.get("note") or "").strip(), + bank_code=bankcode, + txn_type=(parsed_txn.get("type") or "").strip() + ) + + return record + + except Exception as e: + logger.error(f"Error mapping UPI transaction: {e}", exc_info=True) + raise + + @classmethod + def map_transactions( + cls, + parsed_transactions: List[Dict[str, Any]], + bankcode: str + ) -> List[UPIInwardRecord]: + """ + Map list of parsed UPI transactions + """ + records: List[UPIInwardRecord] = [] + + for txn in parsed_transactions: + try: + rec = cls.map_transaction(txn, bankcode) + records.append(rec) + except Exception as e: + logger.warning(f"Skipping transaction due to error: {e}") + + logger.info(f"Mapped {len(records)} UPI transactions for bank {bankcode}") + return records \ No newline at end of file diff --git a/processors/file_processor.py b/processors/file_processor.py new file mode 100644 index 0000000..65bde88 --- /dev/null +++ b/processors/file_processor.py @@ -0,0 +1,166 @@ +#!/usr/bin/env python3 +""" +Main file processor for end-to-end ACH file processing. +Orchestrates download, parsing, mapping, and database insertion. +""" + +import os +import tempfile +import gzip +import shutil +from pathlib import Path +from logging_config import get_logger +from upi_parser import UPIParser +from db.repository import Repository +from db.models import ProcessedFile +from sftp.sftp_client import SFTPClient +from .data_mapper import UPIDataMapper + +logger = get_logger(__name__) + + +class FileProcessor: + + + def __init__(self, repository: Repository = None, sftp_client: SFTPClient = None): + self.repository = repository or Repository() + self.sftp_client = sftp_client or SFTPClient() + self.temp_dir = tempfile.gettempdir() + + def process_file( + self, + filename: str, + bankcode: str, + remote_path: str + ) -> bool: + + local_path = os.path.join(self.temp_dir, filename) + + try: + logger.info(f"Starting processing: {filename} (bank: {bankcode})") + + # Step 1: Check if already processed + if self.repository.is_file_processed(filename, bankcode): + logger.info(f"File already processed for {bankcode}: {filename}") + return True + + # Step 2: Download file + if not self.sftp_client.download_file(remote_path, local_path): + raise Exception(f"Failed to download file: {remote_path}") + + # NEW: Handle .gz file + file_to_parse = local_path + + if filename.lower().endswith(".gz"): + logger.info(f"Extracting GZ file: {filename}") + + extracted_path = local_path[:-3] # remove .gz + + with gzip.open(local_path, 'rb') as f_in: + with open(extracted_path, 'wb') as f_out: + shutil.copyfileobj(f_in, f_out) + + file_to_parse = extracted_path + + # Step 3: Parse file + parser = UPIParser(file_to_parse) + transactions, metadata, summary = parser.parse() + + if not transactions: + logger.warning(f"No transactions found in {filename}") + + processed_file = ProcessedFile( + filename=filename, # ORIGINAL .gz NAME + bankcode=bankcode, + file_path=remote_path, + transaction_count=0, + status='SUCCESS' + ) + self.repository.mark_file_processed(processed_file) + return True + + # Step 4: Map transactions + mapped_records = UPIDataMapper.map_transactions(transactions, bankcode) + + # Step 5: Insert to database + inserted_count, skipped_count = self.repository.bulk_insert_transactions(mapped_records) + + # Step 6: Mark processed + processed_file = ProcessedFile( + filename=filename, # ORIGINAL .gz NAME + bankcode=bankcode, + file_path=remote_path, + transaction_count=inserted_count, + status='SUCCESS' + ) + self.repository.mark_file_processed(processed_file) + + logger.info( + f"Successfully processed {filename}: " + f"{inserted_count} inserted, {skipped_count} skipped (non-ipks accounts)" + ) + return True + + except Exception as e: + logger.error(f"Error processing {filename}: {e}", exc_info=True) + + try: + processed_file = ProcessedFile( + filename=filename, + bankcode=bankcode, + file_path=remote_path, + transaction_count=0, + status='FAILED', + error_message=str(e)[:2000] + ) + self.repository.mark_file_processed(processed_file) + except Exception as mark_error: + logger.error(f"Failed to mark file as failed: {mark_error}") + + return False + + finally: + # UPDATED CLEANUP (handles both .gz + extracted file) + try: + if os.path.exists(local_path): + os.remove(local_path) + + if 'file_to_parse' in locals() and file_to_parse != local_path: + if os.path.exists(file_to_parse): + os.remove(file_to_parse) + + logger.debug(f"Cleanup done for {filename}") + + except Exception as e: + logger.warning(f"Error cleaning up local file {local_path}: {e}") + + def process_files(self, files_to_process: list) -> dict: + + stats = { + 'total': len(files_to_process), + 'successful': 0, + 'failed': 0, + 'files': [] + } + + for filename, bankcode, remote_path in files_to_process: + success = self.process_file(filename, bankcode, remote_path) + stats['successful'] += 1 if success else 0 + stats['failed'] += 0 if success else 1 + stats['files'].append({ + 'filename': filename, + 'bankcode': bankcode, + 'success': success + }) + + logger.info(f"Processing complete: {stats['successful']}/{stats['total']} successful") + return stats + + def __enter__(self): + if self.sftp_client and not self.sftp_client.sftp: + self.sftp_client.connect() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + if self.sftp_client: + self.sftp_client.disconnect() \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..f55780d --- /dev/null +++ b/requirements.txt @@ -0,0 +1,23 @@ +# Core dependencies +python-dotenv==1.0.0 + +# Database (modern Oracle driver - simpler than cx_Oracle) +oracledb==2.0.0 + +# SFTP +paramiko==3.4.0 +cryptography==41.0.7 + +# Scheduling +schedule==1.2.0 + +# Configuration +python-decouple==3.8 + +# Timezone support +pytz==2023.3 + +# Development dependencies +pytest==7.4.0 +black==23.7.0 +flake8==6.0.0 diff --git a/scheduler.py b/scheduler.py new file mode 100644 index 0000000..5544d72 --- /dev/null +++ b/scheduler.py @@ -0,0 +1,171 @@ +#!/usr/bin/env python3 +""" +NEFT file processing scheduler. +Runs polling loop every 30 minutes to process new files. +""" + +import signal +import time +import sys +from datetime import datetime +from logging_config import get_logger, setup_logging +from config import get_config +from db import OracleConnector, Repository +from sftp import SFTPClient, FileMonitor +from processors import FileProcessor + +logger = get_logger(__name__) + + +class Scheduler: + """Main scheduler for NEFT file processing.""" + + def __init__(self): + """Initialize scheduler.""" + self.config = get_config() + self.config.validate() + self.running = True + self.cycle_count = 0 + + # Setup signal handlers for graceful shutdown + signal.signal(signal.SIGTERM, self._signal_handler) + signal.signal(signal.SIGINT, self._signal_handler) + + def _signal_handler(self, signum, frame): + """Handle shutdown signals gracefully.""" + logger.info(f"Received signal {signum}, shutting down gracefully...") + self.running = False + + def initialize_database(self): + """Initialize database connection and verify tables exist.""" + try: + connector = OracleConnector() + if connector.test_connection(): + logger.info("Database connection test passed") + repository = Repository() + # repository.verify_tables_exist() + return True + else: + logger.error("Database connection test failed") + return False + except SystemExit as e: + logger.error(f"Database initialization failed: {e}") + raise + except Exception as e: + logger.error(f"Error initializing database: {e}", exc_info=True) + return False + + def run_processing_cycle(self): + """Run single file processing cycle.""" + self.cycle_count += 1 + logger.info(f"=== Starting processing cycle {self.cycle_count} ===") + + sftp_client = SFTPClient() + repository = Repository() + + try: + # Connect to SFTP + if not sftp_client.connect(): + logger.error("Failed to connect to SFTP server") + return + + # Scan for new files across all banks + monitor = FileMonitor(sftp_client) + new_files = [] + today_str = datetime.now().strftime("%d%m%Y") + logger.info(f'listing file for {today_str}') + + for bank_code in self.config.bank_codes: + # Get list of files already processed for this specific bank + bank_processed = repository.get_processed_files(bank_code) + remote_path = f"{self.config.sftp_base_path}/{bank_code}/UPI" + + + pattern = f"*_{today_str}_*_UPI.txt.gz" + files = sftp_client.list_files(remote_path, pattern=pattern) + + + for filename in files: + if filename not in bank_processed: + full_path = f"{remote_path}/{filename}" + new_files.append((filename, bank_code, full_path)) + logger.info(f"Found new file: {filename} (bank: {bank_code})") + else: + logger.debug(f"Skipping already processed file for {bank_code}: {filename}") + + if not new_files: + logger.info("No new files to process") + return + + logger.info(f"Found {len(new_files)} new files to process") + + # Process files + processor = FileProcessor(repository, sftp_client) + stats = processor.process_files(new_files) + + # Log summary + logger.info(f"Cycle {self.cycle_count} complete:") + logger.info(f" Total files: {stats['total']}") + logger.info(f" Successful: {stats['successful']}") + logger.info(f" Failed: {stats['failed']}") + + # Call neft_api_txn_post procedure once per cycle to process all inserted transactions + # if stats['successful'] > 0: + # logger.info("Calling neft_api_txn_post procedure for all inserted transactions...") + # if repository.call_neft_api_txn_post(): + # logger.info("Transaction post-processing completed successfully") + # else: + # logger.error("Transaction post-processing failed") + + except Exception as e: + logger.error(f"Error in processing cycle: {e}", exc_info=True) + + finally: + sftp_client.disconnect() + + def run(self): + """Run scheduler main loop.""" + logger.info("="*80) + logger.info("NEFT_INWARD File Processing Scheduler Started") + logger.info(f"Poll Interval: {self.config.poll_interval_minutes} minutes") + logger.info(f"Bank Codes: {', '.join(self.config.bank_codes)}") + logger.info("="*80) + + # Initialize database + try: + if not self.initialize_database(): + logger.error("Failed to initialize database. Exiting.") + return + except SystemExit as e: + logger.error(f"Fatal error: {e}") + raise + + # Run processing loop + poll_interval_seconds = self.config.poll_interval_minutes * 60 + + while self.running: + try: + self.run_processing_cycle() + except Exception as e: + logger.error(f"Unexpected error in processing cycle: {e}", exc_info=True) + + # Wait for next cycle + if self.running: + logger.info(f"Waiting {self.config.poll_interval_minutes} minutes until next cycle...") + time.sleep(poll_interval_seconds) + + logger.info("Scheduler shutdown complete") + + +def main(): + """Main entry point.""" + # Setup logging + setup_logging() + + # Create and run scheduler + scheduler = Scheduler() + scheduler.run() + + +if __name__ == '__main__': + main() diff --git a/sftp/__init__.py b/sftp/__init__.py new file mode 100644 index 0000000..588ec13 --- /dev/null +++ b/sftp/__init__.py @@ -0,0 +1,6 @@ +"""SFTP module for ACH file processing.""" + +from .sftp_client import SFTPClient +from .file_monitor import FileMonitor + +__all__ = ['SFTPClient', 'FileMonitor'] diff --git a/sftp/file_monitor.py b/sftp/file_monitor.py new file mode 100644 index 0000000..163eeda --- /dev/null +++ b/sftp/file_monitor.py @@ -0,0 +1,95 @@ +#!/usr/bin/env python3 +""" +File monitoring and discovery for UPI files. +Scans SFTP directories for new files across multiple banks. + +Filename convention handled: + _DDMMYYYY__UPI.txt.gz + +Example: + TUMLUK_31052026_8C_UPI.txt.gz +""" + +import re +from typing import List, Tuple, Dict +from logging_config import get_logger +from config import get_config +from .sftp_client import SFTPClient + +logger = get_logger(__name__) + + +class FileMonitor: + """Monitors SFTP for new UPI files.""" + + def __init__(self, sftp_client: SFTPClient = None): + self.config = get_config() + self.sftp_client = sftp_client or SFTPClient() + + def scan_for_new_files(self, processed_filenames: List[str]) -> List[Tuple[str, str, str]]: + """ + Scan all bank directories for new UPI files. + + Returns: + List of (filename, bankcode, full_remote_path) + """ + new_files: List[Tuple[str, str, str]] = [] + + for bank_code in self.config.bank_codes: + remote_path = f"{self.config.sftp_base_path}/{bank_code}/UPI" + + # ✅ Pattern for UPI files + files = self.sftp_client.list_files( + remote_path, + pattern='*_UPI.txt.gz' + ) + + for filename in files: + if filename not in processed_filenames: + full_path = f"{remote_path}/{filename}" + new_files.append((filename, bank_code, full_path)) + logger.info(f"Found new UPI file: {filename} (bank: {bank_code})") + else: + logger.debug(f"Skipping already processed UPI file: {filename}") + + logger.info(f"UPI scan complete: Found {len(new_files)} new files") + return new_files + + @staticmethod + def parse_filename(filename: str) -> Dict[str, str]: + """ + Parse UPI filename. + + Expected format: + _DDMMYYYY__UPI.txt.gz + + Example: + TUMLUK_31052026_8C_UPI.txt.gz + """ + + pattern = r'^([A-Z]+)_(\d{2})(\d{2})(\d{4})_([A-Z0-9]+)_UPI\.txt\.gz$' + match = re.match(pattern, filename, flags=re.IGNORECASE) + + if not match: + logger.warning(f"Could not parse UPI filename: {filename}") + return {} + + location, day, month, year, code = match.groups() + + return { + 'filename': filename, + 'location': location, + 'day': day, + 'month': month, + 'year': year, + 'code': code, + 'timestamp': f"{day}/{month}/{year}" + } + + def __enter__(self): + if not self.sftp_client.sftp: + self.sftp_client.connect() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.sftp_client.disconnect() \ No newline at end of file diff --git a/sftp/sftp_client.py b/sftp/sftp_client.py new file mode 100644 index 0000000..4399915 --- /dev/null +++ b/sftp/sftp_client.py @@ -0,0 +1,157 @@ +#!/usr/bin/env python3 +""" +SFTP client for file operations. +Handles connection, file discovery, and download operations. +""" + +import paramiko +import os +from pathlib import Path +from logging_config import get_logger +from config import get_config + +logger = get_logger(__name__) + + +class SFTPClient: + """SFTP operations for ACH file processing.""" + + def __init__(self): + """Initialize SFTP client.""" + self.config = get_config() + self.sftp = None + self.ssh = None + + def connect(self) -> bool: + """ + Establish SFTP connection. + + Returns: + True if successful, False otherwise + """ + try: + # Create SSH client + self.ssh = paramiko.SSHClient() + self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + + # Connect + self.ssh.connect( + self.config.sftp_host, + port=self.config.sftp_port, + username=self.config.sftp_username, + password=self.config.sftp_password, + timeout=10 + ) + + # Get SFTP channel + self.sftp = self.ssh.open_sftp() + logger.info(f"Connected to SFTP server: {self.config.sftp_host}:{self.config.sftp_port}") + return True + + except Exception as e: + logger.error(f"Failed to connect to SFTP server: {e}", exc_info=True) + return False + + def disconnect(self): + """Close SFTP connection.""" + try: + if self.sftp: + self.sftp.close() + if self.ssh: + self.ssh.close() + logger.info("SFTP connection closed") + except Exception as e: + logger.error(f"Error closing SFTP connection: {e}") + + def list_files(self, remote_path: str, pattern: str) -> list: + """ + List files matching pattern in remote directory. + + Args: + remote_path: Path on SFTP server + pattern: File pattern to match (e.g., 'ACH_*.txt') + + Returns: + List of matching filenames + """ + if not self.sftp: + logger.error("SFTP not connected") + return [] + + try: + files = [] + try: + items = self.sftp.listdir_attr(remote_path) + except FileNotFoundError: + logger.warning(f"Directory not found: {remote_path}") + return [] + + import fnmatch + for item in items: + if fnmatch.fnmatch(item.filename, pattern): + files.append(item.filename) + + logger.debug(f"Found {len(files)} files matching {pattern} in {remote_path}") + return sorted(files) + + except Exception as e: + logger.error(f"Error listing files in {remote_path}: {e}", exc_info=True) + return [] + + def download_file(self, remote_path: str, local_path: str) -> bool: + """ + Download file from SFTP server. + + Args: + remote_path: Full path on SFTP server + local_path: Local destination path + + Returns: + True if successful, False otherwise + """ + if not self.sftp: + logger.error("SFTP not connected") + return False + + try: + # Create local directory if needed + Path(local_path).parent.mkdir(parents=True, exist_ok=True) + + # Download file + self.sftp.get(remote_path, local_path) + logger.info(f"Downloaded file: {remote_path} -> {local_path}") + return True + + except Exception as e: + logger.error(f"Error downloading file {remote_path}: {e}", exc_info=True) + return False + + def get_file_size(self, remote_path: str) -> int: + """ + Get size of remote file. + + Args: + remote_path: Full path on SFTP server + + Returns: + File size in bytes, or -1 if error + """ + if not self.sftp: + logger.error("SFTP not connected") + return -1 + + try: + stat = self.sftp.stat(remote_path) + return stat.st_size + except Exception as e: + logger.error(f"Error getting file size {remote_path}: {e}") + return -1 + + def __enter__(self): + """Context manager entry.""" + self.connect() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + """Context manager exit.""" + self.disconnect() diff --git a/upi.py b/upi.py new file mode 100644 index 0000000..c282ee3 --- /dev/null +++ b/upi.py @@ -0,0 +1,34 @@ +#!/usr/bin/env python3 +""" +Main application entry point. +Runs ACH file processing scheduler. +""" + +import logging +from logging_config import setup_logging, get_logger +from scheduler import Scheduler + +# Initialize logging +logging.getLogger("paramiko").setLevel(logging.WARNING) +logger = setup_logging(log_level=logging.INFO) +app_logger = get_logger(__name__) + + +def main(): + """Main application function.""" + app_logger.info("Application started") + + try: + # Run the scheduler + scheduler = Scheduler() + scheduler.run() + app_logger.info("Application completed successfully") + except KeyboardInterrupt: + app_logger.info("Application interrupted by user") + except Exception as e: + app_logger.error(f"An error occurred: {e}", exc_info=True) + raise + + +if __name__ == "__main__": + main() diff --git a/upi_parser.py b/upi_parser.py new file mode 100644 index 0000000..baeb3a8 --- /dev/null +++ b/upi_parser.py @@ -0,0 +1,270 @@ +import os +import logging +from typing import List, Dict, Tuple, Optional +from decimal import Decimal, InvalidOperation + +# ------------------------- +# Logger +# ------------------------- +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + + +# ------------------------- +# Helpers +# ------------------------- +def normalize_text(val: str) -> str: + return str(val or "").strip() + + +def to_decimal(val: str) -> Optional[Decimal]: + try: + return Decimal(val.strip()) + except Exception: + return None + + +# ------------------------- +# Parser Class +# ------------------------- +class UPIParser: + """ + Parser for UPI Bank Switch Report (NPCI Cycle) + """ + + EXPECTED_HEADER = [ + "rrn", + "type", + "txn_datetime", + "credit_account", + "status", + "amount", + "note", + ] + + def __init__(self, file_path: str): + self.file_path = file_path + self.transactions: List[Dict] = [] + self.file_metadata: Dict = {} + self.summary_data: Dict = {} + + # ------------------------- + # MAIN PARSE + # ------------------------- + def parse(self) -> Tuple[List[Dict], Dict, Dict]: + try: + rows = self._read_rows() + + self.file_metadata = { + "source_file": os.path.basename(self.file_path), + "row_count": len(rows), + "columns": self.EXPECTED_HEADER, + } + + for idx, row in enumerate(rows, start=1): + txn = self._row_to_transaction(row, idx) + if txn: + self.transactions.append(txn) + + self.summary_data = self._build_summary(self.transactions) + + logger.info(f"Parsed {len(self.transactions)} rows from {self.file_path}") + return self.transactions, self.file_metadata, self.summary_data + + except Exception as e: + logger.error(f"Error parsing file: {e}", exc_info=True) + raise + + # ------------------------- + # READ FILE + # ------------------------- + def _read_rows(self) -> List[List[str]]: + rows = [] + + with open(self.file_path, 'r', encoding='utf-8', errors='replace') as f: + for line in f: + line = line.strip() + + # Skip unwanted lines + if ( + not line + or "UPI BANK SWITCH REPORT" in line + or line.startswith("===") + or line.startswith("---") + or line.startswith("RRN") + ): + continue + + parts = line.split('|') + + if len(parts) < 7: + logger.debug(f"Skipping malformed row: {line}") + continue + + rows.append(parts[:7]) + + return rows + + # ------------------------- + # ROW -> TRANSACTION + # ------------------------- + def _row_to_transaction(self, row: List[str], row_num: int) -> Optional[Dict]: + + txn = { + "rrn": normalize_text(row[0]), + "type": normalize_text(row[1]), + "txn_datetime": normalize_text(row[2]), + "credit_account": normalize_text(row[3]), + "status": normalize_text(row[4]), + "amount": "", + "note": normalize_text(row[6]), + } + + # Amount conversion + amt = to_decimal(row[5]) + txn["amount"] = amt if amt is not None else "" + + # Split datetime + if txn["txn_datetime"]: + try: + date, time = txn["txn_datetime"].split(" ") + txn["txn_date"] = date + txn["txn_time"] = time + except Exception: + txn["txn_date"] = txn["txn_datetime"] + txn["txn_time"] = "" + + if not txn["rrn"]: + logger.debug(f"Skipping row {row_num}: Missing RRN") + return None + + return txn + + # ------------------------- + # SUMMARY + # ------------------------- + def _build_summary(self, txns: List[Dict]) -> Dict: + total_count = len(txns) + total_amount = Decimal("0") + + by_status: Dict[str, Dict] = {} + by_type: Dict[str, Dict] = {} + + for t in txns: + amt = t.get("amount") or Decimal("0") + + if isinstance(amt, str): + try: + amt = Decimal(amt) + except: + amt = Decimal("0") + + total_amount += amt + + status = t.get("status", "").upper() + txn_type = t.get("type", "").upper() + + # Status grouping + if status not in by_status: + by_status[status] = {"count": 0, "amount": Decimal("0")} + by_status[status]["count"] += 1 + by_status[status]["amount"] += amt + + # Type grouping + if txn_type not in by_type: + by_type[txn_type] = {"count": 0, "amount": Decimal("0")} + by_type[txn_type]["count"] += 1 + by_type[txn_type]["amount"] += amt + + # Convert Decimal → string + by_status_final = { + k: {"count": v["count"], "amount": f"{v['amount']:.2f}"} + for k, v in by_status.items() + } + + by_type_final = { + k: {"count": v["count"], "amount": f"{v['amount']:.2f}"} + for k, v in by_type.items() + } + + return { + "total_count": total_count, + "total_amount": f"{total_amount:.2f}", + "by_status": by_status_final, + "by_type": by_type_final, + } + + +# ------------------------- +# PRINT FUNCTIONS +# ------------------------- +def print_transactions(transactions: List[Dict], limit: Optional[int] = 50): + + headers = [ + ("RRN", 15), + ("TYPE", 8), + ("DATE", 10), + ("TIME", 10), + ("ACCOUNT", 20), + ("STATUS", 10), + ("AMOUNT", 12), + ("NOTE", 25), + ] + + header_line = " ".join([f"{h:<{w}}" for h, w in headers]) + print("\n" + "=" * len(header_line)) + print(header_line) + print("=" * len(header_line)) + + for i, txn in enumerate(transactions): + row = [ + txn.get("rrn", ""), + txn.get("type", ""), + txn.get("txn_date", ""), + txn.get("txn_time", ""), + txn.get("credit_account", ""), + txn.get("status", ""), + f"{txn.get('amount')}" if txn.get("amount") else "", + txn.get("note", ""), + ] + + print(" ".join(f"{str(val)[:w]:<{w}}" for val, (h, w) in zip(row, headers))) + + if limit and i + 1 >= limit: + print(f"... ({len(transactions) - limit} more rows)") + break + + print("=" * len(header_line)) + print(f"Total: {len(transactions)} records\n") + + +def print_metadata(metadata: Dict): + print("\n===== FILE METADATA =====") + for k, v in metadata.items(): + print(f"{k.upper():20}: {v}") + print("=========================\n") + + +def print_summary(summary: Dict): + print("\n===== SUMMARY =====") + for k, v in summary.items(): + print(f"{k.upper()}: {v}") + print("===================\n") + + +# ------------------------- +# MAIN RUNNER +# ------------------------- +if __name__ == "__main__": + + file_path = r"C:\Users\2780475\Desktop\Test\TUMLUK_31052026_8C_UPI.txt.gz" + + parser = UPISwitchParser(file_path) + + transactions, metadata, summary = parser.parse() + + print_metadata(metadata) + print_transactions(transactions, limit=50) + print_summary(summary) + + logger.info(f"Parsing complete. Extracted {len(transactions)} transactions") \ No newline at end of file