o }i:@sTddlmZddlZddlZddlZddlZeeZddl m Z m Z m Z m Z mZddlZddlmZddlmZddlmZddlmZddlmZmZmZmZmZd Zd Zd Zd Z d Z!dZ"dZ#dAddZ$dBddZ%dCddZ&dBdd Z'dedfdDd(d)Z(dEdFd,d-Z) dGd0d1Z*dHd4d5Z+dId6d7Z,efdJd9d:Z-efdJd;d<Z.dKdLd?d@Z/dS)M) annotationsN)IterableListOptionalDictAny) HTTPAdapter)Retry)datetime)SQLAlchemyError)db SubscriberContent SentMessage SMSOutboxZ7996z3https://fin.finsightngr.online:30701/callbacks/pushd)g?)g?g333333?items List[dict]sizeintreturnIterable[List[dict]]ccs.tdt||D] }||||Vq dS)Nr)rangelen)rrir:/home/www/bk.finsightngr.online/finsigh_v2/app/sms_push.py_chunksr itemDict[str, Any]NonecCsN|d}t|tr!|dr#t|dkr%d|dd|d<dSdSdSdS)Nmsisdn0 234)get isinstancestr startswithr)r!r$rrr_normalize_msisdn_inline"s   r-cCs"zt|WStytdw)Nz%pcode must be an integer (e.g., 2273))r Exception ValueError)valrrr_to_int's   r1cCsx|d}|d}d|vo|ddv}d|vo|ddv}|s$td|dur,tdt||d<||As:td dS) z Provider schema requires: - msisdn (E.164 / 234XXXXXXXXXX) - pcode (INTEGER) - exactly one of: message OR link (provider examples show message) r$pcodemessage)NlinkzEach item must include msisdn.Nz'Each item must include pcode (integer).z4Each item must have exactly one of: message OR link.)r)r/r1)r!r$r2Zhas_msgZhas_lnkrrr_validate_item-s   r6List[Dict[str, Any]]batch_id Optional[str] endpoint_urlr+ extra_headersOptional[Dict[str, str]]c Cs|D] }t|t|q|sdtjdd}ddd}|r'||t|}d}}g} t} t |t D]} || d} d} t dt dD]}zb| j || |td }|j}z|jd d drk|}nd |ji}Wntyd |ji}Ynw| ||d d|krdkrnn |t| 7}Wqtd|} td|t ||Wntjy}z|} td|t |WYd}~nd}~ww|t krt|tdd}t |qJ|t| 7}t!dt| | t tjt"q:||||| d}t#d||S)z Send items to /callbacks/push in controlled chunks with manual retry/backoff. Returns: { "batch_id": "...", "total": N, "sent": x, "failed": y, "responses": [ { "status": 200, "body": {...} }, ... ] } zFIN-Nzapplication/json)z Content-TypeAcceptr)r8rr()jsonheaderstimeoutz content-typer4raw)statusbodyi,zHTTP z,Push attempt %s/%s failed (HTTP %s). Body=%szPush attempt %s/%s raised %rg?z-Giving up on chunk (%s items). Last error: %r)r8totalsentfailed responseszPush summary: %s)$r6r-uuiduuid4hexupdaterrequestsSessionr BATCH_LIMITrRETRIESpost CONNECT_READ status_coder@r)lowerr,r?textr.append RuntimeErrorloggingwarningRequestException BASE_BACKOFFrandomuniformtimesleeperror CHUNK_PAUSE_Sinfo)rr8r:r;itr@rFrGrHrIsessionchunkpayloadlast_excattemptresprCrDeZsleep_ssummaryrrrpush_sms_batchCs          rmr=urlc Csz%tjdd|i|d}|jr|jr |jdr#|jWSW|SW|SW|StjyD}ztd|d|WYd}~|Sd}~ww)z? Try TinyURL; fall back to original link on any error. z"https://tinyurl.com/api-create.phprn)paramsrAhttpz[Shorten] failed for z: N) rNr)okrVr,stripr[rYrZ)rnrArrkrrr shorten_links  rtr$ str | NonecCs:|sdS|}|drt|dkrd|ddS|S)z@Ensure 234XXXXXXXXXX style; returns None if input is None/empty.Nr%r&r'r()rrr,r)r$rrr_normalize_msisdns rvr5 Optional[int]cCsR|sdSz|d}d|vr|d}t||dWSWdSty(YdSw)z7Expect .../content///.N/contentr()splitindexrr.)r5partsidxrrr_extract_content_id_from_links   r~cCsz9tjtjdtjd}|}|jtjdiddtj t d|d|t ddWStyQtj t dd d d YSw) zQ Set has_received_message=False for all active subscribers (idempotent). TF)synchronize_sessionz.[MorningReset] Reset has_received_message for z active subscribers.Z)resetatz[MorningReset] DB errorrZdb_error)rra)r queryfiltersubscribeStatusis_has_received_messagecountrMr recommitloggerrcr utcnow isoformatr rollback exception)qrrrr&reset_has_received_message_for_activess    rsenderc Cstjtjdtjdtjd}|s&t ddddSg}g}g}|D]=}t |j }|s8q.t |j}|jpAd}d|} |||| |d t|j} |t| |j||||d dd ||q.|sudt|dSd tjdd } t|| d} | dd} z,|D] }d|_tj|q|D]}| rdnd|_t| |_tj|qtjWnt ytj!t "dYnwt d| d| dt|| t|| dS)ae Send 1 SMS per eligible subscriber immediately (no queue). Eligibility: active, has news_link, has_received_message=False. Message: "Your news link today is " pcode = servicePack; msisdn = phone. Scales to tens of thousands (push helper chunks and retries), but for very large lists prefer the queue version below. TNFz8[MorningPush] Nothing to send (no eligible subscribers).r)rGskippedr4Your news link today is r$r2r3rZqueued content_id recipient_idr2r$ short_linkrprovider_statusprovider_responsez FIN-PUSH-r=r8rGrqrHz,[MorningPush] DB error while marking/loggingz[MorningPush] batch_id= sent=z total=)r8rFrG)#r rrrr news_linkisnotrallrrcrvphonert servicePackrWr~ridrrJrKrLrmr)r readdrr+rrr rr)rtargetsrZto_markZlogssubr$shortr2r3rr8rlrGlogrrrpush_news_links_to_activessr               rc Cs&d}d}tjtjdtjdtjdtj }d} |tj |k | }|s5q|D]2}t |j}|sD|j }q7t|j}|jpMd} d|} tjt|j || || dd |d 7}|j }q7ztjWntytjtd Ynwq%td |d d|iS)z Enqueue one outgoing SMS per eligible subscriber (fast DB inserts). A separate worker drains the queue in batches (process_sms_outbox_once). irTNFr4rpending) subscriber_idr$r2rr3rCr(z*[SMSQueue] enqueue commit failed mid-batchz[SMSQueue] Enqueued z items.Zenqueued)r rrrrrrrorder_byrasclimitrrvrrtrr rerrrr rrrrc) rZPAGEaddedrlast_idpagerr$rr2msgrrrenqueue_sms_for_activesTsP        rrc Csxtjjddtj|}|sdddSdd|D}d|djd|d j}t||d }| d ddk}d}zS|D]I}|j d 7_ |r|d |_ t j |j }|rw|jswd |_tjtt|j|j|j|jd|jdt|d|d 7}nd|_ t||_tj|qAtjWntytjtdYnwtd|d|dt |t ||dS)z Drain up to `limit` pending items from SMSOutbox: - send via push_sms_batch (which chunks & retries) - mark Subscriber.has_received_message = True on success - log into SentMessage r)rCr)ZpickedrGcSs"g|] }|j|j|j|jdqS)rr).0rsrrr s"z+process_sms_outbox_once..z FIN-PUSHQ--rrGr(TNrqrrHz [SMSQueue] process commit failedz[SMSQueue] batch_id=rz picked=)!rr filter_byrrrrrrmr)attemptsrCr rrr rerrr~rr2r$rr+ last_errorrr rrrrcr) rrrr8rlrqrGrsrrrrprocess_sms_outbox_oncesV         r)rrrrrr)r!r"rr#)rr) rr7r8r9r:r+r;r<rr")r=)rnr+rr+)r$rurru)r5rurrw)rr")rr+rr")r)rrrr")0 __future__rrYr_rJr] getLogger__name__rtypingrrrrrrNZrequests.adaptersrZurllib3.util.retryr r Zsqlalchemy.excr app.modelsr r rrrSENDER_DEFAULTZPUSH_URLrPrQrSr\rbr r-r1r6rmrtrvr~rrrrrrrrsB            h  V3