182 lines
11 KiB
Python
182 lines
11 KiB
Python
"""Read-only XLSX ingestion into a private, indexed analysis snapshot. No publication.
|
||
|
||
Every source row remains addressable. Matching identifiers suggest the same entity;
|
||
they never imply a current trading status, contract, or an exhaustive census.
|
||
"""
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
from pathlib import Path
|
||
import sqlite3
|
||
import unicodedata
|
||
import xml.etree.ElementTree as ET
|
||
import zipfile
|
||
|
||
NS = '{http://schemas.openxmlformats.org/spreadsheetml/2006/main}'
|
||
SOURCES = {
|
||
'companies.xlsx': ('1ba4c4114b16b407e7746712d00d4f681246bbbc1b42351e96a31f7129acad30', 'EMPRESAS_LIST 2014-2022.xlsx', '2014–2022; data por registo quando disponível'),
|
||
'licensed.xlsx': ('1b5087b464e39e88c42def39655c2391aaaf684b4c832d6a77ff274b98889f00', 'Empresas Licenciadas.xlsx', 'Data por registo; cobertura temporal global não determinada'),
|
||
'tourism.xlsx': ('f02e086ed4973ef17234aa6cc54b98fbb7a6952ffb16ec4dc4f7d0c72af6400c', 'CADASTRO DE EMPREENDIMENTOS TURISTICOS 3, 4 E 5 ESTRELAS 30.10.24 18,00h.xlsx', '30/10/2024 (nome do exemplar); situação actual não confirmada'),
|
||
'cae.xlsx': ('83b4f1200988e99834c2ae3915346fbf237d128308554b67b5a080d4956bf506', 'CAE eBAU.xlsx', 'Versão não indicada'),
|
||
}
|
||
|
||
def norm(v):
|
||
return ''.join(c for c in unicodedata.normalize('NFKD', str(v)).casefold() if not unicodedata.combining(c)).strip()
|
||
|
||
def digest(v):
|
||
return hashlib.sha256(v.encode('utf-8')).hexdigest()
|
||
|
||
def rows(path):
|
||
# Shared strings are spooled to a temporary on-disk SQLite lookup, not a huge XML tree.
|
||
with zipfile.ZipFile(path) as z, sqlite3.connect('') as strings:
|
||
strings.execute('create table strings(id integer primary key,value text)')
|
||
if 'xl/sharedStrings.xml' in z.namelist():
|
||
with z.open('xl/sharedStrings.xml') as f:
|
||
i = 0
|
||
for _, e in ET.iterparse(f, events=('end',)):
|
||
if e.tag == NS+'si':
|
||
strings.execute('insert into strings values(?,?)', (i, ''.join(t.text or '' for t in e.iter(NS+'t'))))
|
||
i += 1
|
||
e.clear()
|
||
book = ET.fromstring(z.read('xl/workbook.xml'))
|
||
rels = {r.attrib['Id']: r.attrib['Target'] for r in ET.fromstring(z.read('xl/_rels/workbook.xml.rels'))}
|
||
for sheet in book.find(NS+'sheets'):
|
||
target = rels[sheet.attrib['{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id']]
|
||
target = target.lstrip('/') if target.startswith('/') else 'xl/'+target
|
||
with z.open(target) as f:
|
||
for _, e in ET.iterparse(f, events=('end',)):
|
||
if e.tag != NS+'row':
|
||
continue
|
||
result = {}
|
||
for c in e:
|
||
col = ''.join(ch for ch in c.attrib.get('r','') if ch.isalpha())
|
||
v = c.findtext(NS+'v', '')
|
||
if c.attrib.get('t') == 's':
|
||
hit = strings.execute('select value from strings where id=?', (int(v),)).fetchone()
|
||
v = hit[0] if hit else ''
|
||
elif c.attrib.get('t') == 'inlineStr':
|
||
v = ''.join(t.text or '' for t in c.iter(NS+'t'))
|
||
result[col] = ' '.join(v.split())
|
||
yield sheet.attrib['name'], int(e.attrib['r']), result
|
||
e.clear()
|
||
|
||
def main():
|
||
p = argparse.ArgumentParser()
|
||
p.add_argument('--sources', required=True)
|
||
p.add_argument('--output', required=True)
|
||
p.add_argument('--only-tourism', action='store_true', help='Rebuild only tourism in an existing private preview')
|
||
p.add_argument('--only-missing-names', action='store_true', help='Add nonempty company rows without a social denomination; never substitute the personal applicant name')
|
||
a = p.parse_args()
|
||
output = Path(a.output)
|
||
output.mkdir(parents=True, exist_ok=True)
|
||
manifest = {'version':1, 'access':'internal', 'sources':[], 'publication':False,
|
||
'limitations':'Registos administrativos históricos; cobertura incompleta, actividade actual, capacidade, procura e compromissos não confirmados. Ausência de registo significa informação insuficiente.'}
|
||
for alias, (expected, name, period) in SOURCES.items():
|
||
path = Path(a.sources)/alias
|
||
actual = hashlib.file_digest(path.open('rb'), 'sha256').hexdigest()
|
||
if actual != expected:
|
||
raise ValueError('Original changed: '+alias)
|
||
manifest['sources'].append({'alias':alias,'name':name,'sha256':actual,'bytes':path.stat().st_size,'period':period})
|
||
db = sqlite3.connect(output/'supply-preview.sqlite')
|
||
db.executescript('''create table if not exists records(id text primary key, source_hash text, sheet text, row_number integer, entity_key text, name text, province text, district text, cae text, activity text, sector text, source_date text, fingerprint text, duplicate_of text, identity_conflict integer default 0);
|
||
create index if not exists supply_place on records(province,district);
|
||
create index if not exists supply_entity on records(entity_key);
|
||
create index if not exists supply_fingerprint on records(fingerprint);''')
|
||
labels = {}
|
||
hierarchy = {}
|
||
for _, row, c in rows(Path(a.sources)/'cae.xlsx'):
|
||
if row > 1 and c.get('B') and c.get('F'):
|
||
labels[c['B']] = c['F']
|
||
hierarchy[c['A']] = c
|
||
# Supplied CAE hierarchy labels retained. Sector is the top-level CAE section,
|
||
# not an invented correspondence to the portal's twelve navigation sectors.
|
||
sector_by_code = {}
|
||
for node in hierarchy.values():
|
||
ancestor=node
|
||
visited=set()
|
||
while ancestor.get('D')!='1' and ancestor.get('C') in hierarchy and ancestor['A'] not in visited:
|
||
visited.add(ancestor['A'])
|
||
ancestor=hierarchy[ancestor['C']]
|
||
if ancestor.get('D')=='1':
|
||
sector_by_code[node['B']]=ancestor['F']
|
||
counts = {}
|
||
if a.only_tourism:
|
||
db.execute('delete from records where source_hash=?',(SOURCES['tourism.xlsx'][0],))
|
||
aliases=('companies.xlsx','licensed.xlsx') if a.only_missing_names else (('tourism.xlsx',) if a.only_tourism else ('companies.xlsx','licensed.xlsx','tourism.xlsx'))
|
||
for alias in aliases:
|
||
source_hash, _, period = SOURCES[alias]
|
||
count = 0
|
||
province = ''
|
||
previous_sheet=''
|
||
for sheet, row, c in rows(Path(a.sources)/alias):
|
||
if alias == 'tourism.xlsx':
|
||
if sheet!=previous_sheet:
|
||
province=''
|
||
previous_sheet=sheet
|
||
heading=norm(c.get('C','')).replace(' ','')
|
||
provinces={'cidadedemaputo':'MAPUTO CIDADE','maputocidade':'MAPUTO CIDADE','provinciademaputo':'MAPUTO PROVINCIA','maputoprovincia':'MAPUTO PROVINCIA'}
|
||
provinces.update({'provinciade'+norm(n).replace(' ',''):n for n in ['Gaza','Inhambane','Sofala','Manica','Tete','Zambézia','Nampula','Cabo Delgado','Niassa']})
|
||
if heading in provinces and not c.get('D'):
|
||
province=provinces[heading]
|
||
continue
|
||
# Explicit province section and ordinal identify an establishment.
|
||
# Missing licence/room/contact never excludes an authorised record.
|
||
name = c.get('C','')
|
||
if not name or not c.get('B','').isdigit():
|
||
continue
|
||
district = c.get('E','') if sheet=='Folha1' else ''
|
||
cae, activity, date, identifier = '', 'Alojamento (âmbito do exemplar; categoria por conferir)', period, ''
|
||
sector = labels.get('I','Alojamento e restauração — exemplar de alojamento')
|
||
else:
|
||
if row == 1:
|
||
continue
|
||
old = alias == 'companies.xlsx'
|
||
name = c.get('L','')
|
||
if a.only_missing_names and name:
|
||
continue
|
||
if not any(c.values()):
|
||
continue
|
||
# A missing social denomination is a data gap, not a reason to
|
||
# discard an authorised source row or publish an applicant's name.
|
||
name = name or 'Denominação não indicada'
|
||
province, district = c.get('Y' if old else 'V',''), c.get('X' if old else 'U','')
|
||
cae = c.get('S' if old else 'O','')
|
||
activity = labels.get(cae, '')
|
||
sector = sector_by_code.get(cae,'')
|
||
date = c.get('D','')
|
||
# Keep source serial verbatim; Excel origin must be demonstrated before conversion.
|
||
date = ('Serial Excel original: '+date) if date.replace('.','',1).isdigit() else (date or 'Não indicada')
|
||
nuit = c.get('K','')
|
||
identifier = ('nuit:'+nuit) if len(nuit)==9 and nuit.isdigit() and len(set(nuit))>1 else ''
|
||
rid = digest(source_hash+'|'+sheet+'|'+str(row))
|
||
entity = digest(identifier) if identifier else digest('unresolved:'+source_hash+'|'+sheet+'|'+str(row))
|
||
# Same projected administrative observation, not byte-identical rows:
|
||
# include licence and process identifiers; do not collapse distinct licences.
|
||
licence = (c.get('G','') if sheet!='Folha1' else '') if alias=='tourism.xlsx' else c.get('A','')
|
||
process = '' if alias=='tourism.xlsx' else c.get('B','')
|
||
fp = digest('|'.join(map(norm,[identifier,name,province,district,cae,date,licence,process])))
|
||
duplicate = db.execute('select id from records where fingerprint=? and id<>? limit 1',(fp,rid)).fetchone()
|
||
db.execute('insert or ignore into records values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)', (rid,source_hash,sheet,row,entity,name,province,district,cae,activity,sector,date,fp,duplicate[0] if duplicate else None,0))
|
||
count += 1
|
||
if count % 10000 == 0:
|
||
db.commit()
|
||
db.commit()
|
||
counts[alias] = count
|
||
print(json.dumps({'source':alias,'rows_processed':count}), flush=True)
|
||
db.execute('update records set identity_conflict=0')
|
||
db.execute("update records set identity_conflict=1 where entity_key in (select entity_key from records where name<>'Denominação não indicada' group by entity_key having count(distinct lower(name))>1)")
|
||
db.commit()
|
||
counts={alias:db.execute('select count(*) from records where source_hash=?',(SOURCES[alias][0],)).fetchone()[0] for alias in ('companies.xlsx','licensed.xlsx','tourism.xlsx')}
|
||
manifest.update(rows_processed=counts, records=db.execute('select count(*) from records').fetchone()[0],
|
||
duplicate_sightings=db.execute('select count(*) from records where duplicate_of is not null').fetchone()[0],
|
||
conflicting_identity_sightings=db.execute('select count(*) from records where identity_conflict=1').fetchone()[0],
|
||
unnamed_observations=db.execute("select count(*) from records where name='Denominação não indicada'").fetchone()[0],
|
||
unclassified_activity=db.execute("select count(*) from records where activity=''").fetchone()[0],
|
||
unresolved_territory=db.execute("select count(*) from records where district='' or province=''").fetchone()[0])
|
||
db.close()
|
||
(output/'preview.json').write_text(json.dumps(manifest,ensure_ascii=False,indent=2),encoding='utf-8')
|
||
print(json.dumps({k:v for k,v in manifest.items() if k not in ('sources',)},ensure_ascii=False))
|
||
|
||
if __name__ == '__main__':
|
||
main()
|