RjC *dZddlmZddlZddlZddlmZgdZgdgdgd gd gd d gd dgddgddgddggdd Z d/dZ ddddddddddd Z hdhdhd hd!d"Z d0d#Z d$Zd1d%Zd1d&Zd2d'Zd3d(Zd4d)Zd5d*Zd+Zd6d,Zd7d8d-Z d7 d9d.Zy):aEPG spreadsheet -> Crystal schedule frame. The dashboard's file-upload path expects a schedule with the columns `validate_schedule()` requires: event_name, country_name, channel_name, telecast_type and hour. A raw EPG export has none of those directly, so this module derives them. Nothing here writes to tvviewers. The only DB access is three read-only reference lookups (channel bridge, event vocabulary), which the caller supplies as DataFrames so this module stays testable offline. Every derived row carries a `confidence` column. Rows below "high" are meant to land in the existing analyst-review screen rather than be trusted silently -- a wrong telecast_type moves an estimate by an order of magnitude. ) annotationsN)classify) channel_namechannel_countries prog_title prog_st_time)rchannel epg_channel)r country_namecountry territory)r description programme program_titletitle)r country_start_time start_timeprog_start_time) prog_en_timecountry_end_timeend_time prog_end_time telecast_type sports_event event_name sports_teamsteams match_levelstagesports_category sub_genre)country_timezonetimezonetz) rrrr rrrrrr!r#cL|j}|jDcic]*}t|jj |,}}t j D]8\}}||jvr|D]}|j|}|||||<8:|Scc}w)zAdd canonical column names for whichever aliases this file happens to use. Returns a copy. Original columns are left in place, so nothing downstream that reads the file's own names breaks. )copycolumnsstrstriplowerCOLUMN_ALIASESitemsget)dfoutcr+canonaliasesasrcs 4/var/www/html/crystal/crystal2/engine/epg_adapter.py canon_columnsr74s '')C02 ;1SV\\^ ! ! #Q & ;E ;(..0w CKK   A))A,CWE    J icciplodir8testcricket>eplfifauefalaligasoccerfootball>nrlsixrugbynations>atpwtarolandtennis wimbledon)rHrNrQrVc4tfdtDS)zTrue if this looks like a raw EPG listings export rather than a schedule. Detected by signature columns, not filename, so either file type can be uploaded through the same box. c3LK|]}|tjvywN)r7r().0r1r/s r6 z is_epg_export..`s!I!qM"%---Is!$)allREQUIRED_EPG_COLS)r/s`r6 is_epg_exportr_Zs I7HI II)epg_channel_bridgeepg_event_vocabepg_country_aliascddlm}|jd}|r8||tfdtDrt fdtDSt |S)aFetch the three reference tables the adapter needs. Read-only. Prefers the local snapshot written by worker/build_cache.py. The event vocabulary alone is a GROUP BY over 5.9M rows on a remote server; querying it per run cost ~85s before a single listing had been read. rPath fixtures_dirc3JK|]}|dz jywz.parquetN)existsr[nr1s r6r\z%load_epg_reference..rs%AqcN"**,As #c3TK|]}tj|dz !ywri)pd read_parquetrks r6r\z%load_epg_reference..ss%Qs(^);<Qs%()pathlibrfr.r] EPG_FRAMEStuplefetch_epg_reference)cfgrfcacher1s @r6load_epg_referencervfsL GGN #E K AjA AQjQQ Q s ##r`cddlm}||}tjd|}tjd|}tjd|}|||fS)zHAlways go to the server. build_cache.py calls this to make the snapshot.r)reference_enginez SELECT gsiq_channel, real_channel, COUNT(*) AS n FROM gl_ratings WHERE real_channel IS NOT NULL AND TRIM(real_channel) <> '' GROUP BY gsiq_channel, real_channel ORDER BY n DESC a8 SELECT a.event_name AS canonical, a.event_alias AS alias, COALESCE(g.n, 0) AS weight FROM event_alias a LEFT JOIN (SELECT event_name, COUNT(*) n FROM global_sports WHERE event_name IS NOT NULL GROUP BY event_name) g ON g.event_name = a.event_name WHERE a.event_alias IS NOT NULL AND TRIM(a.event_alias) <> '' UNION SELECT event_name, event_name, COUNT(*) FROM global_sports WHERE event_name IS NOT NULL AND TRIM(event_name) <> '' GROUP BY event_name z SELECT country, alias_country FROM country_alias WHERE alias_country IS NOT NULL AND TRIM(alias_country) <> '' ) referencerxrnread_sql)rtrxengbridgevocabaliass r6rsrsxsj+ 3 C [[ F KK   E" KK  E 5% r`cddl}ddlm}|j||j }t |\}}}t ||||S)zCConvenience wrapper: load reference tables from config, then adapt.rNre)jsonrprfloads read_textrvadapt)epg config_pathrrfrtr|r}r~s r6 adapt_epgrsF **T+&002 3C-c2FE5 feU ++r`cztjddt|jj S)z;Loose key for event matching: lowercase, alphanumeric only.z [^a-z0-9]+ )resubr)r+r*)ss r6_normrs( 66-c!flln 5 ; ; ==r`ct}t|jD]>}tj ||}|j d|jD@|S)Nc3&K|] }|s| ywrZ)r[xs r6r\z_tokens..s-11-s)setrsplitSYNONYMSr.update)rr0ws r6_tokensrsR %C 1X^^ . LLA  -aggi--. Jr`cNtjD]\}}||zs |cSyrZ) SPORT_MARKERSr-)toksfammarkss r6_familyrs/#))+ U %<J r`g{Gz?c g}|jD]q\}\}}t|}t|dkr#|j||t |xsdt t|t d|Dt|fs|S)a Tokenise every alias once. resolve_event used to call _tokens() on all 5,378 aliases for every row it was given. On a 5,400-row file that is ~29 million tokenisations to answer 33 distinct questions, and it dominated the whole pipeline at 71 seconds. rc3<K|]}|jywrZ)isdigit)r[ts r6r\z!_prepare_vocab..s4 4s)r-rlenappendintranyr)r}r0r~r2weighta_tokss r6_prepare_vocabrs C"'++-Ev v;?  FE3v{#3WWU^5L4V44eElD E E Jr`czt|}|syt|}g}|jt|}|r|j d|d|dft|}|xs t |D]h\}} } } } } ||ks| r0t jdt j| zdz|sB|r| r| |k7rL|j t|| | fj|sytd|D}|Dcgc]}|dk(s|dt|zk\s|}}t|xs|\}} }|dk(r| dfS| |d k\rdfSd fScc}w) aIMap a free-text EPG title onto a canonical global_sports event_name. vocab maps normalised alias -> (canonical event name, evidence row count). Returns (event_name, confidence). Two rules decide it. The alias with the most tokens wins, so a specific event beats a generic one. But a candidate is only admissible if it carries a meaningful share of the evidence held by the best-evidenced candidate -- otherwise a 4-row misspelling of an event outranks the 25,934-row real one purely for being longer, and every row attached to it becomes unestimable. )Nlowcrrz\bc3(K|] \}}}| ywrZr)r[_rs r6r\z resolve_event..s(wq!Q(srhighmedium) rrr.rrrrsearchescapermaxEVIDENCE_FLOOR)rr}preparedt_tokst_famcandsexactt_normrr2ra_fam has_digita_normbest_wr1viablen_tokrs r6 resolve_eventrsbU^F  FOE)+E IIeEl #E  b%(E!H-. 5\F<||jXd?<|j[d@AScc}wcc}} wcc}wcc}}} wcc}} wcc}wcc}wcc}}}wcc}}w)Ba Convert a raw EPG frame into a Crystal schedule frame. channel_bridge: columns gsiq_channel, real_channel (from gl_ratings) event_vocab: columns canonical, alias, weight (event_alias + global_sports) country_alias: columns country, alias_country (from country_alias) z&EPG file is missing required columns: z, z. Accepted names for each: z; c3VK|]!}|ddjt|#yw)z = /N)joinr,)r[r1s r6r\zadapt..s*N3sxxq(9:;<Ns'))rrr )subset real_channel gsiq_channelrleft)left_onright_onhowrr alias_countryr crjt|jj|SrZ)r.r)r*r+)vamaps r6zadapt...s%dhhs1v||~335q9r`r coerce)errors prog_datehourstart_dtrend_dtrr9)index)nanNone-rc|dSNrrr_clss r6rzadapt..Csar`c|dSNrrrs r6rzadapt..D$q'!*r`)_tt_ttcrrr tt_confidencerr~ canonicalrrc|dSrrr_ress r6rzadapt..Us471:r`c|dSrrrs r6rzadapt..Vsd1gajr`)r ev_confidence)axisrrc|dSrrr_alts r6rzadapt..]rr`c|dSrrrs r6rzadapt..^sT!WQZr`r)rrr!r#r)rrrc|SrZr)rs r6rzadapt..isPQr`)key confidenceepg_confidence)r(broadcast_date)rr rrrrrrrrrrrrr!r#rrn_rawn_duplicates_removedT)drop).r7r^r( ValueErrorrrdrop_duplicatesdropnamergezip isinstancer)r*r+maprn to_datetimedtdaterNaTastypeSeriesrreplacefillnauniquer DataFramewhererrrrconcatisnarlocrrrminr-renameattrs reset_index)#rchannel_bridge event_vocab country_aliasr/r1missingrn_dedupbrr4tsstated_titlesrderivedrr}revneedsecolrankr_ttr_evrckrinvr0rrrrs# @@@@r6rrsn s B+CQq /BqCGC 4tyy7I I ) * IINgN N OP P GE  #Q  RBc"goG   '7  8(0@A "nn56(>v  OB /0B~ S%7 o > i@XY5Aqa%!'')A $$&)5 /33 9;> >*8 Y>AZ+Z &ZZ>Z-Z 90Z% Z,)r/ pd.DataFramereturnr&)r/r&r'bool)rtdictr'z/tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame])rr&r'r&)rr)r'r))rr)r'set[str])rr*r'z str | None)r}dict[str, tuple[str, int]]r'listrZ)rr)r}r+r'ztuple[str | None, str]) rr&rr&rr&rzpd.DataFrame | Noner'r&)__doc__ __future__rrpandasrn epg_rulesrr^r,r7rrr_rqrvrsrrrrrrrrrr`r6r1s# %DV]`Z**(,7('2'1+[9?$0 5? 12R b ?G /=  JL $$( V,> "/5j04w&&w&#w&-w&9Ew&r`