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['installed_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_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_soong_modules(self): 136 # Get all records from table modules, which contains metadata of all soong modules 137 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') 138 rows = cursor.fetchall() 139 cursor.close() 140 soong_modules = [] 141 for row in rows: 142 soong_module = dict(zip(row.keys(), row)) 143 soong_modules.append(soong_module) 144 return soong_modules 145 146 def get_package_licenses(self, package): 147 cursor = self.conn.execute('select m.name, m.package, m.lic_license_text as license_text ' 148 'from package_license pl join modules m on pl.license = m.name ' 149 'where pl.package = ?', 150 ('//' + package,)) 151 rows = cursor.fetchall() 152 licenses = {} 153 for r in rows: 154 licenses[r['name']] = r['license_text'] 155 return licenses 156 157 def get_module_licenses(self, module_name, package): 158 licenses = {} 159 # If property "licenses" is defined on module 160 cursor = self.conn.execute('select m.name, m.package, m.lic_license_text as license_text ' 161 'from module_license ml join modules m on ml.license = m.name ' 162 'where ml.module = ? and ml.package = ?', 163 (module_name, package)) 164 rows = cursor.fetchall() 165 for r in rows: 166 licenses[r['name']] = r['license_text'] 167 if len(licenses) > 0: 168 return licenses 169 170 # Use default package license 171 cursor = self.conn.execute('select m.name, m.package, m.lic_license_text as license_text ' 172 'from package_license pl join modules m on pl.license = m.name ' 173 'where pl.package = ?', 174 ('//' + package,)) 175 rows = cursor.fetchall() 176 for r in rows: 177 licenses[r['name']] = r['license_text'] 178 return licenses 179 180 def get_soong_module_of_installed_file(self, installed_file): 181 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 ' 182 'from modules m join module_installed_file mif on m.id = mif.module_id ' 183 'where mif.installed_file = ?', 184 (installed_file,)) 185 rows = cursor.fetchall() 186 cursor.close() 187 if rows: 188 soong_module = dict(zip(rows[0].keys(), rows[0])) 189 return soong_module 190 191 return None 192 193 def get_soong_module_of_built_file(self, built_file): 194 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 ' 195 'from modules m join module_built_file mbf on m.id = mbf.module_id ' 196 'where mbf.built_file = ?', 197 (built_file,)) 198 rows = cursor.fetchall() 199 cursor.close() 200 if rows: 201 soong_module = dict(zip(rows[0].keys(), rows[0])) 202 return soong_module 203 204 return None