Metro-Auto-Check-3

CREATE TABLE sandbox.yurbasov_metro_auto_check_3
AS
WITH task1 AS (
  SELECT t.v, s.phone_hash FROM sandbox.yurbasov_task_auto_check t
  LEFT JOIN secret.phones s ON t.v = s.phone
), chos1 AS (
  SELECT v, regexp_replace(phone_hash, '^[0]*', '') AS short_hash FROM task1 WHERE phone_hash IS NOT NULL
), base1 AS (
  SELECT v, hash_msisdn, fdt, startstid, stopstid,
  from_unixtime(unix_timestamp(fdt, 'dd.MM.yyyy HH:mm:ss'),'yyyy-MM-dd') AS d,
  DATEDIFF(fdt, from_unixtime(unix_timestamp(fdt, 'dd.MM.yyyy HH:mm:ss'),'yyyy-MM-dd')) tm,
  weekofyear(unix_timestamp(fdt, 'dd.MM.yyyy HH:mm:ss')) wy,
  PMOD(DATEDIFF(from_unixtime(unix_timestamp(fdt, 'dd.MM.yyyy HH:mm:ss'),'yyyy-MM-dd'), '2012-01-01'), 7) ww
  FROM chos1 t JOIN maxima_data.metro m ON t.short_hash = m.hash_msisdn
  WHERE DAY <= '2018-03-28' AND
   (
   from_unixtime(unix_timestamp(fdt, 'dd.MM.yyyy HH:mm:ss'),'yyyy-MM-dd') >= '2017-09-01'
   AND from_unixtime(unix_timestamp(fdt, 'dd.MM.yyyy HH:mm:ss'),'yyyy-MM-dd') <= '2018-02-28'
   )
), uniq1 AS (
  SELECT '0' AS r, 'A' AS g, v, hash_msisdn, startstid AS fromst, stopstid AS tost, 1 AS total, d, wy, MIN(ww) AS min_ww, MAX(tm) AS max_tm, MIN(tm) AS min_tm
  FROM base1 WHERE ww IN (0,1,2,3,4,5,6) GROUP BY v, hash_msisdn, startstid, stopstid, d, wy
), aggg1 AS (
  SELECT r, g, v, hash_msisdn, fromst, tost, SUM(total) AS cnt, MAX(d) AS max_d, MIN(d) AS min_d, MAX(max_tm) AS max_tm, MIN(min_tm) AS min_tm FROM uniq1 WHERE r = '0' GROUP BY v, hash_msisdn, fromst, tost, r, g HAVING SUM(total) >= 1
), res1 AS (
  SELECT a.cnt AS fromcnt, a.r, a.g, a.v, a.hash_msisdn, a.fromst, a.tost, a.min_d, a.max_d, a.max_tm, a.min_tm FROM aggg1 a
), anti1 AS (
  SELECT task1.v, res1.v AS z FROM task1 LEFT JOIN res1 ON task1.v = res1.v
), proto AS (
  SELECT res1.fromcnt AS tocnt, res1.min_d AS min_d, res1.max_d AS max_d,
  res1.v, res1.hash_msisdn, res1.r AS r, res1.g AS g, res1.fromst AS a, res1.tost AS b FROM res1
), raw AS (
  SELECT v, hash_msisdn, MAX(tocnt) AS tocnt, MIN(min_d) AS min_d, MAX(max_d) AS max_d,
  MAX(CONCAT(lpad(tocnt, 3, "0"), ' ', r, ' ', g, ' ', lpad(a, 3, "0"), ' ', lpad(b, 3, "0"))) AS info,
  substr(MAX(CONCAT(lpad(tocnt, 3, "0"), ' ', r, ' ', g, ' ', lpad(a, 3, "0"), ' ', lpad(b, 3, "0"))), 5, 1) AS r,
  substr(MAX(CONCAT(lpad(tocnt, 3, "0"), ' ', r, ' ', g, ' ', lpad(a, 3, "0"), ' ', lpad(b, 3, "0"))), 7, 1) AS g,
  CAST (substr(MAX(CONCAT(lpad(tocnt, 3, "0"), ' ', r, ' ', g, ' ', lpad(a, 3, "0"), ' ', lpad(b, 3, "0"))), 9, 3) AS INT) AS a,
  CAST (substr(MAX(CONCAT(lpad(tocnt, 3, "0"), ' ', r, ' ', g, ' ', lpad(a, 3, "0"), ' ', lpad(b, 3, "0"))), 13, 3) AS INT) AS b
  FROM proto GROUP BY v, hash_msisdn
), enrich AS (
  SELECT raw.tocnt, raw.min_d, raw.max_d, raw.v, raw.hash_msisdn,
 raw.r, raw.g, concat(m.station_name_rus, '-', raw.a) AS a, raw.b FROM raw
    LEFT JOIN sandbox.line_station_codes_new m
    ON raw.a = m.station_id
), done AS (
  SELECT enrich.tocnt, enrich.min_d, enrich.max_d, enrich.v, enrich.hash_msisdn,
 enrich.r, enrich.g, a, concat(m.station_name_rus, '-', enrich.b) AS b FROM enrich
    LEFT JOIN sandbox.line_station_codes_new m
    ON enrich.b = m.station_id
), res AS (
 SELECT s.phone, done.tocnt, done.min_d, done.max_d, done.r, done.g, done.a, done.b, done.v, done.hash_msisdn FROM done JOIN secret.phones s ON done.hash_msisdn = regexp_replace(s.phone_hash, '^[0]*', '')
 ), anti AS (
  SELECT a.v AS phone, '-' AS cnt, '-' AS r, '-' AS g, '-' AS a, '-'  AS b, '-' AS v, '-' AS msisdn, '-' AS min_d, '-' AS max_d FROM anti1 a WHERE a.z IS NULL
) SELECT concat('', phone) AS phone, CONCAT('', MAX(tocnt)) AS cnt, MAX(r) AS r, MAX(g) AS g, MAX(a) AS a, MAX(b) AS b, CONCAT('', MAX(v)) AS v, MAX(hash_msisdn) AS msisdn, MIN(min_d) AS min_d, MAX(max_d) AS max_d FROM res GROUP BY phone
UNION ALL
SELECT phone, cnt, r, g, a, b, v, msisdn, min_d, max_d FROM anti