"""rapport.py — intègre le « Rapport immobilier Québec » (article LaTeX, 26 sections, 67 figures PDF,
28 tableaux générés, valeurs clés, données HPI ACI + FRED) sous forme de mini-livre web."""
from __future__ import annotations
import hashlib
import json
import os
import re
import shutil
import subprocess
from datetime import datetime
from latex2html import Ctx, convert_chapter, strip_comments, read_group, plain_text
from bib import parse_bib
REPORT_DIR = "Rapport_Immobilier_Quebec"
def load_key_values(path: str) -> dict:
vals = {}
for m in re.finditer(r"\\newcommand\{\\(Vc[A-Za-z]+)\}\{([^}]*)\}", open(path, encoding="utf-8").read()):
vals[m.group(1)] = m.group(2)
return vals
def preprocess(src_dir: str) -> tuple[str, list[dict], dict]:
"""Retourne (texte des sections, liste [{'key','num','title','tex','appendix'}], valeurs clés)."""
main = open(os.path.join(src_dir, "rapport_immobilier_quebec.tex"), encoding="utf-8").read()
vals = load_key_values(os.path.join(src_dir, "tables", "valeurs_cles.tex"))
# corps : de la première \section à \bibliographystyle, puis annexes après \appendix
start = main.index("\\section{Introduction}")
bib_pos = main.index("\\bibliographystyle")
app_pos = main.index("\\appendix")
body = main[start:bib_pos] + "\n\\appendix\n" + main[app_pos + len("\\appendix"):main.index("\\end{document}")]
# inputs de tableaux
def inline_input(m):
p = os.path.join(src_dir, m.group(1) + (".tex" if not m.group(1).endswith(".tex") else ""))
return open(p, encoding="utf-8").read() if os.path.exists(p) else ""
body = re.sub(r"\\input\{([^}]+)\}", inline_input, body)
# valeurs clés
for k, v in sorted(vals.items(), key=lambda kv: -len(kv[0])):
body = re.sub(r"\\" + k + r"(\{\})?(?![A-Za-z])", lambda m, v=v: v, body)
body = strip_comments(body)
# découpage en sections → « chapitres »
parts = re.split(r"(?=^\\section\{)", body, flags=re.M)
sections = []
appendix = False
n = 0
for chunk in parts:
if not chunk.strip():
continue
if not chunk.startswith("\\section{"):
if "\\appendix" in chunk:
appendix = True
continue
if "\\appendix" in chunk:
# l'annexe commence après ce chunk
chunk_main, _, rest = chunk.partition("\\appendix")
chunk = chunk_main
pending_app = True
else:
pending_app = False
title, k = read_group(chunk, len("\\section"))
if appendix:
num = chr(ord("A") + sum(1 for s in sections if s["appendix"]))
else:
n += 1
num = str(n)
tex = "\\chapter{" + title + "}" + chunk[k:]
tex = tex.replace("\\subsubsection{", "\\subsection@@{").replace("\\subsection{", "\\section{").replace("\\subsection@@{", "\\subsection{")
tex = tex.replace("\\clearpage", "")
sections.append({"key": f"rapport-{num}", "num": num, "title": title, "tex": tex, "appendix": appendix})
if pending_app:
appendix = True
return body, sections, vals
def convert_figures(src_dir: str, images: list[str], cache_dir: str, out_dir: str) -> dict[int, str]:
"""Convertit les PDF de figures en SVG (cache) et les copie dans out_dir. Retourne {index: html}."""
os.makedirs(cache_dir, exist_ok=True)
os.makedirs(out_dir, exist_ok=True)
out = {}
for idx, rel in enumerate(images):
src = os.path.join(src_dir, rel)
if not os.path.exists(src):
out[idx] = ""
continue
if rel.endswith(".png") or rel.endswith(".jpg"):
name = os.path.basename(rel)
shutil.copy2(src, os.path.join(out_dir, name))
out[idx] = f'
'
continue
h = hashlib.sha1((rel + str(os.path.getmtime(src))).encode()).hexdigest()[:12]
name = os.path.splitext(os.path.basename(rel))[0] + "-" + h + ".svg"
cached = os.path.join(cache_dir, name)
if not os.path.exists(cached):
subprocess.run(["pdftocairo", "-svg", src, cached], check=True, timeout=120)
shutil.copy2(cached, os.path.join(out_dir, name))
head = open(cached, encoding="utf-8", errors="replace").read(600)
m = re.search(r'width="([\d.]+)pt" height="([\d.]+)pt"', head)
attrs = f' width="{int(float(m.group(1)) * 4 / 3)}" height="{int(float(m.group(2)) * 4 / 3)}"' if m else ""
out[idx] = f'
'
return out
def extract_hpi(src_dir: str) -> dict:
"""Extrait les séries HPI (mensuel non désaisonnalisé) des marchés d'intérêt."""
import openpyxl
path = os.path.join(src_dir, "data", "Not Seasonally Adjusted (M).xlsx")
wb = openpyxl.load_workbook(path, read_only=True, data_only=True)
wanted = {"AGGREGATE": "Canada", "QUEBEC": "Québec (province)", "MONTREAL_CMA": "RMR de Montréal", "QUEBEC_CMA": "RMR de Québec",
"ESTRIE": "Estrie", "MAURICIE": "Mauricie", "CENTRE_DU_QUEBEC": "Centre-du-Québec", "ONTARIO": "Ontario",
"GREATER_TORONTO": "Grand Toronto", "GREATER_VANCOUVER": "Grand Vancouver", "OTTAWA": "Ottawa", "CALGARY": "Calgary",
"BRITISH_COLUMBIA": "Colombie-Britannique", "ALBERTA": "Alberta", "HALIFAX_DARTMOUTH": "Halifax", "WINNIPEG": "Winnipeg"}
cols = ["Composite_HPI", "Single_Family_HPI", "One_Storey_HPI", "Two_Storey_HPI", "Townhouse_HPI", "Apartment_HPI", "Composite_Benchmark", "Single_Family_Benchmark", "Apartment_Benchmark"]
dates = None
series = {}
for sheet, label in wanted.items():
if sheet not in wb.sheetnames:
continue
ws = wb[sheet]
rows = [r for r in ws.iter_rows(values_only=True) if r and r[0] is not None]
header = [str(h) for h in rows[0]]
idx = {c: header.index(c) for c in cols if c in header}
data = [r for r in rows[1:] if isinstance(r[0], datetime)]
d = [r[0].strftime("%Y-%m") for r in data]
if dates is None or len(d) > len(dates):
dates = d
series[label] = {c: [(round(float(r[i]), 1) if isinstance(r[i], (int, float)) else None) for r in data] for c, i in idx.items()}
return {"dates": dates, "series": series, "source": "ACI/CREA — Indice des prix des propriétés MLS®, mensuel non désaisonnalisé, janv. 2005 = 100"}
def extract_fred(src_dir: str) -> dict:
meta = {"IR3TIB01CAM156N": ("Taux interbancaire 3 mois", "%"), "IRLTLT01CAM156N": ("Obligations 10 ans", "%"),
"CPALTT01CAM659N": ("Inflation (IPC, variation annuelle)", "%"), "LRUNTTTTCAM156S": ("Taux de chômage", "%"),
"DEXCAUS": ("Taux de change CAD/USD", "CAD par USD"), "NGDPRSAXDCCAQ": ("PIB réel (trimestriel)", "M$ 2017"),
"POPTOTCAA647NWDB": ("Population", "habitants"), "CANCPIALLMINMEI": ("IPC (indice)", "2015 = 100")}
out = {}
for fid, (label, unit) in meta.items():
p = os.path.join(src_dir, "data", "fred", fid + ".json")
if not os.path.exists(p):
continue
d = json.load(open(p, encoding="utf-8"))
obs = d.get("observations", []) if isinstance(d, dict) else d
pts = []
for o in obs:
try:
v = float(o["value"])
except (ValueError, KeyError, TypeError):
continue
pts.append([o["date"][:7], round(v, 3)])
if fid == "DEXCAUS": # quotidien → mensuel (moyenne)
agg = {}
for dte, v in pts:
agg.setdefault(dte, []).append(v)
pts = [[k, round(sum(v) / len(v), 4)] for k, v in sorted(agg.items())]
out[fid] = {"label": label, "unit": unit, "points": pts}
return out
def build_report(ctx_code: str, root: str, out_dir: str, cache_dir: str, finalize_cb=None):
"""Convertit le rapport. Retourne dict {sections: [ChapterResult+meta], vals, bib, images_html, hpi, fred}."""
src_dir = os.path.join(root, REPORT_DIR, "sources")
body, sections, vals = preprocess(src_dir)
bib = parse_bib(open(os.path.join(src_dir, "references.bib"), encoding="utf-8").read())
ctx = Ctx(ctx_code + "-RAPPORT", bib)
ctx.images = []
results = []
for s in sections:
url = f"/rapport/{s['num']}/"
r = convert_chapter(ctx, s["key"], s["num"], url, s["tex"])
r.appendix = s["appendix"]
results.append(r)
img_html = convert_figures(src_dir, ctx.images, os.path.join(cache_dir, "rapport-fig"), os.path.join(out_dir, "assets", "rapport"))
hpi = extract_hpi(src_dir)
fred = extract_fred(src_dir)
return {"sections": results, "vals": vals, "bib": bib, "ctx": ctx, "images": img_html, "hpi": hpi, "fred": fred, "src_dir": src_dir}