YzTwpryx

1require 'tiny_tds'
2require 'pg'
3require 'dotenv/load'
4require 'pry'
5
6def types_map(mssql_type)
7  map = {
8    bigint: 'bigint',
9    binary: 'bytea',
10    bit: 'boolean',
11    char: 'char',
12    character: 'character',
13    date: 'date',
14    datetime: 'timestamp',
15    datetime2: 'timestamp',
16    datetimeoffset: 'timestamp with time zone',
17    decimal: 'decimal',
18    dec: 'dec',
19    double: 'double',
20    float: 'double precision',
21    int: 'integer',
22    money: 'money',
23    nchar: 'char',
24    ntext: 'text',
25    numeric: 'numeric',
26    nvarchar: 'varchar',
27    real: 'real',
28    rowversion: 'bytea',
29    smalldatetime: 'timestamp',
30    smallint: 'smallint',
31    smallmoney: 'money',
32    text: 'text',
33    time: 'time',
34    timestamp: 'bytea',
35    tinyint: 'smallint',
36    uniqueidentifier: 'char(16)',
37    varbinary: 'bytea',
38    varchar: 'text',
39    xml: 'xml'
40  }
41  map[mssql_type.to_sym]
42end
43
44def mssql_client_params(city)
45  {
46    username: ENV['MS_SQL_USERNAME'],
47    password: ENV['MS_SQL_PASSWORD'],
48    host: ENV['MS_SQL_HOST'],
49    timeout: 300,
50    database: "taxi_#{city}"
51  }
52end
53
54def pg_client_params(dbname)
55  {
56    dbname: dbname,
57    host: ENV['POSTGRES_DB_HOST'] || 'localhost',
58    port: ENV['POSTGRES_DB_PORT'] || '5432',
59    user: ENV['POSTGRES_DB_USERNAME'] || 'postgres',
60    password: ENV['POSTGRES_DB_PASSWORD'] || 'ferraru'
61  }
62end
63
64pg_master_client = PG::Connection.new(pg_client_params('postgres'))
65
66cities = ["voronezh", "tula", "kozelsk", "lyudinovo", "kondrovo", "aleksin", "abakan", "almaty", "angarsk", "astana", "astrakhan", "barnaul", "berezniki", "birobidzhan", "blagoveschensk", "bryansk", "cha", "cheboksary", "chelyabinsk", "cherepovets", "chita", "dimitrovgrad", "dzerzhinsk", "eburg", "gatchina", "habarovsk", "irkutsk", "ivanovo", "izhevsk", "kaliningrad", "kaluga", "karaganda", "kazan", "kemerovo", "kirov", "kostroma", "krasnodar", "krasnoyarsk", "kungur", "kursk", "kyzyl", "lipetsk", "lysva", "magnitogorsk", "maykop", "miass", "moscow", "murmansk", "nchelny", "ncherkassk", "nevi", "nmoskovsk", "nn", "novok", "novokuznetsk", "novorossiysk", "novosibirsk", "ntagil", "obninsk", "okt", "omsk", "orenburg", "orsk", "penza", "perm", "petrozavodsk", "praga", "pskov", "puralsk", "rnd", "ryazan", "salavat", "samara", "saransk", "saratov", "serpukhov", "sgorsk", "shakhty", "smolensk", "sochi", "spb", "stavropol", "sterl", "syk", "syzran", "taganrog", "tambov", "tolyatti", "tomsk", "tver", "tyumen", "ufa", "ulan-ude", "ulyanovsk", "vladimir", "vladivostok", "vnovgorod", "volgograd", "vologda", "votkinsk", "yaroslavl", "yola", "zlatoust"]
67
68threads = []
69cities.each do |city|
70  threads << Thread.new(city) do |city|
71    pg_master_client.exec("CREATE DATABASE #{city}")
72    puts "DB #{city} created"
73  end
74end
75threads.each { |thr| thr.join }
76
77pg_master_client.finish
78
79threads = []
80cities.each do |city|
81  threads << Thread.new(city) do |city|
82    puts "Thread was created"
83
84    pg_client = PG::Connection.new(pg_client_params(city))
85
86    mssql_client = TinyTds::Client.new(mssql_client_params(city))
87
88    begin
89      tables = mssql_client.execute("SELECT name FROM sys.Tables").each(symbolize_keys: true)
90    rescue Exception => e
91      puts e
92      puts "Error on #{city}"
93      mssql_client = TinyTds::Client.new(mssql_client_params(city))
94      tables = mssql_client.execute("SELECT name FROM sys.Tables").each(symbolize_keys: true)
95    end
96
97    tables.each do |table|
98      get_columns_query = "
99        SELECT c.name name, t.name type
100        FROM sys.Columns c
101        JOIN sys.Types t
102        ON c.system_type_id = t.system_type_id
103        WHERE object_id=OBJECT_ID('#{table[:name]}')
104      "
105
106      begin
107        columns = mssql_client.execute(get_columns_query).each(symbolize_keys: true)
108      rescue Exception => e
109        puts e
110        puts "Error on #{city} #{table[:name]}"
111        mssql_client = TinyTds::Client.new(mssql_client_params(city))
112        columns = mssql_client.execute(get_columns_query).each(symbolize_keys: true)
113      end
114
115      pg_table_attributes = ''
116      columns.each do |column|
117        pg_table_attributes += "#{column[:name]} #{types_map(column[:type])}," if column[:name].downcase != 'id' && column[:type] != 'sysname'
118      end
119
120      pg_client.exec(
121        %Q(
122          CREATE TABLE IF NOT EXISTS #{table[:name]} (
123            id bigserial PRIMARY KEY,
124            #{pg_table_attributes.chop}
125         );
126        )
127      )
128
129      begin
130        rows = mssql_client.execute("SELECT * from #{table[:name]}").each(symbolize_keys: true)
131        mssql_client.close
132      rescue Exception => e
133        puts e
134        puts "Error on #{city} #{table[:name]}"
135        mssql_client = TinyTds::Client.new(mssql_client_params(city))
136        rows = mssql_client.execute("SELECT * from #{table[:name]}").each(symbolize_keys: true)
137        mssql_client.close
138      end
139
140      rows.each do |row|
141        row.delete(:ID)
142        row.delete(:id)
143        row.delete(:Id)
144
145        values = row.values.map do |val|
146          if val.nil?
147            'null'
148          elsif val.is_a?(Integer)
149            val
150          elsif val.is_a?(Float)
151            val
152          elsif val.is_a?(BigDecimal)
153            val.to_f
154          else
155            "'#{val}'"
156          end
157        end
158
159        values = values.join(',')
160        keys = row.keys.join(',')
161
162        data = pg_client.exec(%Q(INSERT INTO #{table[:name]} (#{keys}) VALUES (#{values}) returning *;))[0]
163        puts "Data #{data} was saved to #{table[:name]}"
164      end
165    end
166
167    pg_client.finish
168  end
169end
170
171threads.each { |thr| thr.join }