/liu&`dZddlZddlmZddlZejeZGddZdS)z ImdbTransformer - DuckDB-based transformation (v2) Sostituisce database.py + processor.py. Legge TSV direttamente via DuckDB, produce imdb.parquet. N)Pathc eZdZdZ ddedededed ed ef d Zd Zd Z de defdZ defdZ defdZ defdZdefdZdefdZdZdefdZdS)ImdbTransformerz{ Trasformatore DuckDB per i dataset IMDb. Carica TSV filtrati in-memory, esegue join per step, scrive parquet. r4GBtemp_dirparquet_output max_directors max_actorsduckdb_threadsduckdb_memory_limitct||_t||_||_||_||_||_d|_dSN)rrr r r r r _con)selfrr r r r r s 'C:\PYTHON\ImdbUpdate\src\transformer.py__init__zImdbTransformer.__init__sKX ">22*$,#6  cptj|_|jd|jd|jd|jdt |jdd}|jd|d|S)NzSET threads TO 1zSET memory_limit=''z"SET preserve_insertion_order=false\/zSET temp_directory=')duckdbconnectrexecuter strrreplace)r temp_dir_strs r __enter__zImdbTransformer.__enter__sN$$  ,--- Jt/GJJJKKK >???4=))11$<<  @@@@AAA rcX|jr"|jd|_dSdSr)rclose)rargss r__exit__zImdbTransformer.__exit__*s3 9  IOO   DIII  rfilesreturnc`||d||d||d||d||d||S)zd Entry point. Carica tabelle, trasforma, scrive parquet. Ritorna row count. title_basics title_akas title_ratingstitle_principals name_basics)_load_title_basics_load_title_akas_load_title_ratings_load_title_principals_load_name_basics_build_new_imdb_write_parquet)rr$s r transformzImdbTransformer.transform/s n 5666 eL1222   !7888 ##E*<$=>>> u]3444 ""$$$rpathcntd|jt|dd}|jd|d|jdd}td|d d dS) NzCaricamento title_basics: rrz CREATE OR REPLACE TABLE title_basics AS SELECT tconst, titleType, SUBSTR(primaryTitle, 1, 100) AS primaryTitle, runtimeMinutes, startYear, genres FROM read_csv('z', delim=' ', header=true, nullstr='\N', ignore_errors=true) WHERE titleType IN ('movie', 'tvMiniSeries', 'tvMovie', 'tvSeries') z!SELECT COUNT(*) FROM title_basicsrz title_basics: ,z righe (filtrate per titleType)loggerinfonamerrrrfetchonerr4path_strcounts rr,z"ImdbTransformer._load_title_basics@s <<<===t99$$T3//  %        !!"EFFOOQQRST OuOOOOPPPPPrcntd|jt|dd}|jd|d|jdd}td|d d dS) NzCaricamento title_akas: rrzp CREATE OR REPLACE TABLE title_akas AS SELECT titleId, title FROM read_csv('z', delim=' ', header=true, nullstr='\N', ignore_errors=true) WHERE region = 'IT' zSELECT COUNT(*) FROM title_akasrz title_akas: r6z righe (solo region=IT)r7r<s rr-z ImdbTransformer._load_title_akasOs :ty::;;;t99$$T3//  %       !!"CDDMMOOPQR EUEEEEFFFFFrcntd|jt|dd}|jd|d|jdd}td|d d dS) NzCaricamento title_ratings: rrz CREATE OR REPLACE TABLE title_ratings AS SELECT tconst, averageRating, numVotes FROM read_csv('_', delim=' ', header=true, nullstr='\N', ignore_errors=true) z"SELECT COUNT(*) FROM title_ratingsrz title_ratings: r6 righer7r<s rr.z#ImdbTransformer._load_title_ratings\s =$)==>>>t99$$T3//  %       !!"FGGPPRRSTU 7777788888rcntd|jt|dd}|jd|d|jdd}td|d d dS) NzCaricamento title_principals: rrz CREATE OR REPLACE TABLE title_principals AS SELECT tconst, nconst, ordering, category FROM read_csv('z', delim=' ', header=true, nullstr='\N', ignore_errors=true) WHERE category IN ('director', 'actor') z%SELECT COUNT(*) FROM title_principalsrz title_principals: r6z righe (director+actor)r7r<s rr/z&ImdbTransformer._load_title_principalshs @TY@@AAAt99$$T3//  %       !!"IJJSSUUVWX K5KKKKLLLLLrcntd|jt|dd}|jd|d|jdd}td|d d dS) NzCaricamento name_basics: rrzv CREATE OR REPLACE TABLE name_basics AS SELECT nconst, primaryName FROM read_csv('rAz SELECT COUNT(*) FROM name_basicsrz name_basics: r6rBr7r<s rr0z!ImdbTransformer._load_name_basicsus ; ;;<<<t99$$T3//  %       !!"DEENNPPQRS 5e555566666rctdtd|jdtd|jd|jdtd|jd|jdtd |jd |jd d }td |dddS)z Costruisce new_imdb in step separati per limitare il picco di memoria. Ogni tabella intermedia viene materializzata prima del join finale. z'Costruzione new_imdb (step separati)...z Step 1/4: titoli italianiz CREATE OR REPLACE TABLE italian_titles AS SELECT titleId, ANY_VALUE(title) AS ti FROM title_akas GROUP BY titleId z Step 2/4: registiaH CREATE OR REPLACE TABLE directors AS SELECT p.tconst, STRING_AGG(n.primaryName, ', ' ORDER BY p.ordering) AS regista FROM ( SELECT tp.tconst, tp.nconst, tp.ordering, ROW_NUMBER() OVER (PARTITION BY tp.tconst ORDER BY tp.ordering) AS rn FROM title_principals tp WHERE tp.category = 'director' AND tp.tconst IN (SELECT tconst FROM title_basics) ) p JOIN name_basics n ON p.nconst = n.nconst WHERE p.rn <= z' GROUP BY p.tconst z Step 3/4: attoriaC CREATE OR REPLACE TABLE actors AS SELECT p.tconst, STRING_AGG(n.primaryName, ', ' ORDER BY p.ordering) AS cast_str FROM ( SELECT tp.tconst, tp.nconst, tp.ordering, ROW_NUMBER() OVER (PARTITION BY tp.tconst ORDER BY tp.ordering) AS rn FROM title_principals tp WHERE tp.category = 'actor' AND tp.tconst IN (SELECT tconst FROM title_basics) ) p JOIN name_basics n ON p.nconst = n.nconst WHERE p.rn <= z Step 4/4: join finalea1 CREATE OR REPLACE TABLE new_imdb AS SELECT b.tconst AS IMDB_CODICE, b.titleType AS TIPOL, b.primaryTitle AS "TO", it.ti AS TI, TRY_CAST(b.runtimeMinutes AS INTEGER) AS DUR, TRY_CAST(b.startYear AS INTEGER) AS ANNO, b.genres AS GENERE, r.averageRating AS VOTO, r.numVotes AS VOTANTI, d.regista AS REGISTA, a.cast_str AS "CAST" FROM title_basics b LEFT JOIN italian_titles it ON b.tconst = it.titleId LEFT JOIN title_ratings r ON b.tconst = r.tconst LEFT JOIN directors d ON b.tconst = d.tconst LEFT JOIN actors a ON b.tconst = a.tconst zSELECT COUNT(*) FROM new_imdbrz new_imdb: r6rBN)r8r9rrr r r;)rr>s rr1zImdbTransformer._build_new_imdbsd  =>>>  1222       )***   -      "  ()))   ?      "  -...     * !!"ABBKKMMaP 25222233333rc |jjddt|jdd}t d|j|jd|d|jd|d  d }|j j d z }t d |d d|dd|S)z3Scrive new_imdb su file parquet. Ritorna row count.T)parentsexist_okrrzScrittura parquet: z; COPY (SELECT * FROM new_imdb) TO 'zP' (FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 100000) zSELECT COUNT(*) FROM 'rriz Parquet scritto: r6z righe, z.1fz MB) r parentmkdirrrr8r9rrr;statst_size)r parquet_strr>size_mbs rr2zImdbTransformer._write_parquets "(((EEE$-..66tSAA  ?$*=??@@@         !!"I;"I"I"IJJSSUUVWX%**,,4 D K%KKK7KKKKLLL rN)rrrr)__name__ __module__ __qualname____doc__rintrrrr#dictr3r,r-r.r/r0r1r2rrrrsq <=EJ   t  # 58 !$ ?B        %t % % % % %" Qt Q Q Q Q GT G G G G 9 9 9 9 9 M4 M M M M 7d 7 7 7 7 M4M4M4^rr) rRloggingpathlibrr getLoggerrOr8rrUrrrYs   8 $ $UUUUUUUUUUr