1#!/usr/bin/env python3 2# 3# Copyright (C) 2024 The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16 17import sqlite3 18 19class MetadataDb: 20 def __init__(self, db): 21 self.conn = sqlite3.connect(':memory:') 22 self.conn.row_factory = sqlite3.Row 23 with sqlite3.connect(db) as c: 24 c.backup(self.conn) 25 self.reorg() 26 27 def reorg(self): 28 # package_license table 29 self.conn.execute("create table package_license as " 30 "select name as package, pkg_default_applicable_licenses as license " 31 "from modules " 32 "where module_type = 'package' ") 33 cursor = self.conn.execute("select package,license from package_license where license like '% %'") 34 multi_licenses_packages = cursor.fetchall() 35 cursor.close() 36 rows = [] 37 for p in multi_licenses_packages: 38 licenses = p['license'].strip().split(' ') 39 for lic in licenses: 40 rows.append((p['package'], lic)) 41 self.conn.executemany('insert into package_license values (?, ?)', rows) 42 self.conn.commit() 43 44 self.conn.execute("delete from package_license where license like '% %'") 45 self.conn.commit() 46 47 # module_license table 48 self.conn.execute("create table module_license as " 49 "select distinct name as module, package, licenses as license " 50 "from modules " 51 "where licenses != '' ") 52 cursor = self.conn.execute("select module,package,license from module_license where license like '% %'") 53 multi_licenses_modules = cursor.fetchall() 54 cursor.close() 55 rows = [] 56 for m in multi_licenses_modules: 57 licenses = m['license'].strip().split(' ') 58 for lic in licenses: 59 rows.append((m['module'], m['package'],lic)) 60 self.conn.executemany('insert into module_license values (?, ?, ?)', rows) 61 self.conn.commit() 62 63 self.conn.execute("delete from module_license where license like '% %'") 64 self.conn.commit() 65 66 # module_installed_file table 67 self.conn.execute("create table module_installed_file as " 68 "select id as module_id, name as module_name, package, installed_files as installed_file " 69 "from modules " 70 "where installed_files != '' ") 71 cursor = self.conn.execute("select module_id, module_name, package, installed_file " 72 "from module_installed_file where installed_file like '% %'") 73 multi_installed_file_modules = cursor.fetchall() 74 cursor.close() 75 rows = [] 76 for m in multi_installed_file_modules: 77 installed_files = m['installed_file'].strip().split(' ') 78 for f in installed_files: 79 rows.append((m['module_id'], m['module_name'], m['package'], f)) 80 self.conn.executemany('insert into module_installed_file values (?, ?, ?, ?)', rows) 81 self.conn.commit() 82 83 self.conn.execute("delete from module_installed_file where installed_file like '% %'") 84 self.conn.commit() 85 86 # module_built_file table 87 self.conn.execute("create table module_built_file as " 88 "select id as module_id, name as module_name, package, built_files as built_file " 89 "from modules " 90 "where built_files != '' ") 91 cursor = self.conn.execute("select module_id, module_name, package, built_file " 92 "from module_built_file where built_file like '% %'") 93 multi_built_file_modules = cursor.fetchall() 94 cursor.close() 95 rows = [] 96 for m in multi_built_file_modules: 97 built_files = m['built_file'].strip().split(' ') 98 for f in built_files: 99 rows.append((m['module_id'], m['module_name'], m['package'], f)) 100 self.conn.executemany('insert into module_built_file values (?, ?, ?, ?)', rows) 101 self.conn.commit() 102 103 self.conn.execute("delete from module_built_file where built_file like '% %'") 104 self.conn.commit() 105 106 107 # Indexes 108 self.conn.execute('create index idx_modules_id on modules (id)') 109 self.conn.execute('create index idx_modules_name on modules (name)') 110 self.conn.execute('create index idx_package_licnese_package on package_license (package)') 111 self.conn.execute('create index idx_package_licnese_license on package_license (license)') 112 self.conn.execute('create index idx_module_licnese_module on module_license (module)') 113 self.conn.execute('create index idx_module_licnese_license on module_license (license)') 114 self.conn.execute('create index idx_module_installed_file_module_id on module_installed_file (module_id)') 115 self.conn.execute('create index idx_module_installed_file_installed_file on module_installed_file (installed_file)') 116 self.conn.execute('create index idx_module_built_file_module_id on module_built_file (module_id)') 117 self.conn.execute('create index idx_module_built_file_built_file on module_built_file (built_file)') 118 self.conn.commit() 119 120 def dump_debug_db(self, debug_db): 121 with sqlite3.connect(debug_db) as c: 122 self.conn.backup(c) 123 124 def get_installed_files(self): 125 # Get all records from table make_metadata, which contains all installed files and corresponding make modules' metadata 126 cursor = self.conn.execute('select installed_file, module_path, is_soong_module, is_prebuilt_make_module, product_copy_files, kernel_module_copy_files, is_platform_generated, license_text from make_metadata') 127 rows = cursor.fetchall() 128 cursor.close() 129 installed_files_metadata = [] 130 for row in rows: 131 metadata = dict(zip(row.keys(), row)) 132 installed_files_metadata.append(metadata) 133 return installed_files_metadata 134 135 def get_installed_file_in_dir(self, dir): 136 dir = dir.removesuffix('/') 137 cursor = self.conn.execute( 138 'select installed_file, module_path, is_soong_module, is_prebuilt_make_module, product_copy_files, ' 139 ' kernel_module_copy_files, is_platform_generated, license_text ' 140 'from make_metadata ' 141 'where installed_file like ?', (dir + '/%',)) 142 rows = cursor.fetchall() 143 cursor.close() 144 installed_files_metadata = [] 145 for row in rows: 146 metadata = dict(zip(row.keys(), row)) 147 installed_files_metadata.append(metadata) 148 return installed_files_metadata 149 150 def get_soong_modules(self): 151 # Get all records from table modules, which contains metadata of all soong modules 152 cursor = self.conn.execute('select name, package, package as module_path, module_type as soong_module_type, built_files, installed_files, static_dep_files, whole_static_dep_files from modules') 153 rows = cursor.fetchall() 154 cursor.close() 155 soong_modules = [] 156 for row in rows: 157 soong_module = dict(zip(row.keys(), row)) 158 soong_modules.append(soong_module) 159 return soong_modules 160 161 def get_package_licenses(self, package): 162 cursor = self.conn.execute('select m.name, m.package, m.lic_license_text as license_text ' 163 'from package_license pl join modules m on pl.license = m.name ' 164 'where pl.package = ?', 165 ('//' + package,)) 166 rows = cursor.fetchall() 167 licenses = {} 168 for r in rows: 169 licenses[r['name']] = r['license_text'] 170 return licenses 171 172 def get_module_licenses(self, module_name, package): 173 licenses = {} 174 # If property "licenses" is defined on module 175 cursor = self.conn.execute('select m.name, m.package, m.lic_license_text as license_text ' 176 'from module_license ml join modules m on ml.license = m.name ' 177 'where ml.module = ? and ml.package = ?', 178 (module_name, package)) 179 rows = cursor.fetchall() 180 for r in rows: 181 licenses[r['name']] = r['license_text'] 182 if len(licenses) > 0: 183 return licenses 184 185 # Use default package license 186 cursor = self.conn.execute('select m.name, m.package, m.lic_license_text as license_text ' 187 'from package_license pl join modules m on pl.license = m.name ' 188 'where pl.package = ?', 189 ('//' + package,)) 190 rows = cursor.fetchall() 191 for r in rows: 192 licenses[r['name']] = r['license_text'] 193 return licenses 194 195 def get_soong_module_of_installed_file(self, installed_file): 196 cursor = self.conn.execute('select name, m.package, m.package as module_path, module_type as soong_module_type, built_files, installed_files, static_dep_files, whole_static_dep_files ' 197 'from modules m join module_installed_file mif on m.id = mif.module_id ' 198 'where mif.installed_file = ?', 199 (installed_file,)) 200 rows = cursor.fetchall() 201 cursor.close() 202 if rows: 203 soong_module = dict(zip(rows[0].keys(), rows[0])) 204 return soong_module 205 206 return None 207 208 def get_soong_module_of_built_file(self, built_file): 209 cursor = self.conn.execute('select name, m.package, m.package as module_path, module_type as soong_module_type, built_files, installed_files, static_dep_files, whole_static_dep_files ' 210 'from modules m join module_built_file mbf on m.id = mbf.module_id ' 211 'where mbf.built_file = ?', 212 (built_file,)) 213 rows = cursor.fetchall() 214 cursor.close() 215 if rows: 216 soong_module = dict(zip(rows[0].keys(), rows[0])) 217 return soong_module 218 219 return None