# Source: DGFiP / data.gouv.fr, October 2026 release.
# Aggregate calculations: Palank, version 1.0.0, Open Licence 2.0.
import argparse, csv, gzip, hashlib, io, json, pathlib, urllib.request, datetime, collections

SOURCE = 'https://static.data.gouv.fr/resources/demandes-de-valeurs-foncieres-geolocalisees/20261020-114721/full-2025.csv.gz'
parser = argparse.ArgumentParser(description='Reproduce the Palank Rennes DVF 2025 aggregate audit.')
parser.add_argument('--output', default='.')
OUT = pathlib.Path(parser.parse_args().output)
OUT.mkdir(parents=True, exist_ok=True)


class DigestReader:
    def __init__(self, stream):
        self.stream, self.sha, self.bytes = stream, hashlib.sha256(), 0

    def read(self, size=-1):
        value = self.stream.read(size)
        self.sha.update(value)
        self.bytes += len(value)
        return value

    def seekable(self): return False
    def readable(self): return True


counts, seen, previous, group, split_ids = collections.Counter(), set(), None, [], 0
fields = ['id_mutation', 'date_mutation', 'numero_disposition', 'nature_mutation',
          'valeur_fonciere', 'code_commune', 'id_parcelle', 'code_type_local', 'surface_reelle_bati']


def finish(rows):
    if not rows: return
    local = [row for row in rows if row['code_commune'] == '35238']
    if not local: return
    counts['rennes_mutations_all_natures'] += 1
    if not all(row['nature_mutation'] == 'Vente' for row in rows): return
    counts['rennes_sale_mutations'] += 1
    if not any(row['code_type_local'] == '4' for row in local): return
    counts['professional_sale_mutations'] += 1
    professional = [row for row in rows if row['code_type_local'] == '4']
    counts['professional_rows'] += len(professional)
    flags = {
        'multiple_rows': len(rows) > 1,
        'with_housing': any(row['code_type_local'] in ['1', '2'] for row in rows),
        'multiple_parcels': len({row['id_parcelle'] for row in rows if row['id_parcelle']}) > 1,
        'multiple_dispositions': len({row['numero_disposition'] for row in rows}) > 1,
        'multiple_communes': len({row['code_commune'] for row in rows}) > 1,
        'missing_professional_surface': any(
            not row['surface_reelle_bati'] or float(row['surface_reelle_bati']) <= 0
            for row in professional
        ),
        'single_row_positive_surface': len(rows) == 1 and bool(professional[0]['surface_reelle_bati'])
            and float(professional[0]['surface_reelle_bati']) > 0
    }
    for key, value in flags.items(): counts[key] += int(value)


with urllib.request.urlopen(urllib.request.Request(SOURCE, headers={'User-Agent': 'Palank research data audit'}), timeout=120) as response:
    headers = {key: response.headers.get(key) for key in ['Content-Length', 'Last-Modified', 'ETag']}
    stream = DigestReader(response)
    with io.TextIOWrapper(gzip.GzipFile(fileobj=stream), encoding='utf-8', newline='') as text:
        for row in csv.DictReader(text):
            counts['source_rows'] += 1
            key = row['id_mutation']
            if key != previous:
                finish(group)
                if key in seen: split_ids += 1
                seen.add(key)
                previous, group = key, []
            group.append({field: row.get(field, '') for field in fields})
        finish(group)
    result = {
        'source': {
            'url': SOURCE, 'publisher': 'DGFiP / data.gouv.fr', 'vintage': '2026-10-20',
            'period': ['2025-01-01', '2025-12-31'],
            'retrievedAt': datetime.datetime.now(datetime.timezone.utc).isoformat(),
            'sha256': stream.sha.hexdigest(), 'compressedBytes': stream.bytes, 'headers': headers
        },
        'commune': {'code': '35238', 'name': 'Rennes'}, 'counts': dict(counts),
        'nonContiguousMutationGroups': split_ids, 'methodVersion': '1.0.0'
    }
    assert stream.sha.hexdigest() == '85854e21f21151e94748e6bf11657f801951854179f381b7f6abfeb9d80040a9', 'The source file differs from the published snapshot.'
    assert split_ids == 0, 'Mutation rows are not contiguous; another grouping pass is required.'
    (OUT / 'rennes-dvf-2025-audit.json').write_text(json.dumps(result, ensure_ascii=False, indent=2))
    print(json.dumps(result, ensure_ascii=False, indent=2), flush=True)
    assert split_ids == 0, 'Mutation rows are not contiguous; another grouping pass is required.'