"""Deck discovery — map the inbox onto (company, period) grading units. The inbox is organized by company: /data/inbox//. The subfolder name is the company slug; the reporting period is parsed from each filename (2026-Q2, Q2 2026, 2026-H1, 2026-05, FY2026 ...). Files whose period cannot be parsed form a period-less unit that the extractor's own deck.period can later fill in. Files at the inbox root are skipped (we would not know the company) and reported so the UI can nag the operator. """ from __future__ import annotations import os import re SUPPORTED_EXTS = {".pdf", ".pptx", ".docx", ".txt", ".md", ".markdown", ".text"} # Canonical period forms: "2026-Q2", "2026-H1", "2026-05", "FY2026". _PERIOD_PATTERNS = [ # 2026-Q2 / 2026Q2 / 2026_Q2 / 2026 Q2 (re.compile(r"(? str: """Lowercase [a-z0-9-] slug for a company folder name.""" s = re.sub(r"[^a-z0-9]+", "-", (name or "").strip().lower()).strip("-") return s or "company" def parse_period_from_name(filename: str) -> str | None: """Canonical period parsed from a filename, or None.""" base = os.path.basename(filename) for pat, canon in _PERIOD_PATTERNS: m = pat.search(base) if m: return canon(m) return None def canonicalize_period(text: str | None) -> str | None: """Normalize a free-form period string ("2026-q2", "FY 2026", "2026/05") to the canonical form, or None. Extractor-supplied periods and target periods must pass through here so ledger lookups match filename periods.""" if not text: return None s = str(text).strip().replace("/", "-") if period_sort_key(s) != _UNKNOWN_KEY: return s # already canonical for pat, canon in _PERIOD_PATTERNS: m = pat.search(s) if m: return canon(m) return None def period_sort_key(period: str | None) -> tuple: """(year, start_month, granularity_rank); unknown/None sorts last.""" if not period: return _UNKNOWN_KEY m = re.fullmatch(r"((?:19|20)\d{2})-Q([1-4])", period) if m: return (int(m.group(1)), (int(m.group(2)) - 1) * 3 + 1, _GRAN_Q) m = re.fullmatch(r"((?:19|20)\d{2})-H([12])", period) if m: return (int(m.group(1)), (int(m.group(2)) - 1) * 6 + 1, _GRAN_H) m = re.fullmatch(r"((?:19|20)\d{2})-(0[1-9]|1[0-2])", period) if m: return (int(m.group(1)), int(m.group(2)), _GRAN_M) m = re.fullmatch(r"FY((?:19|20)\d{2})", period) if m: return (int(m.group(1)), 1, _GRAN_FY) return _UNKNOWN_KEY def discover(inbox_dir: str) -> dict: """Scan the inbox into grading units. Returns {"units": [{company_slug, period, period_source, files, ignored}], "skipped": [root-level file names]}. Units are grouped by (company, parsed period), sorted by company then oldest period first (period-less last).""" units: dict[tuple, dict] = {} skipped: list[str] = [] if not os.path.isdir(inbox_dir): return {"units": [], "skipped": []} for entry in sorted(os.listdir(inbox_dir)): if entry.startswith("."): continue path = os.path.join(inbox_dir, entry) if os.path.isfile(path): skipped.append(entry) continue if not os.path.isdir(path): continue company = slugify(entry) for fn in sorted(os.listdir(path)): if fn.startswith("."): continue fpath = os.path.join(path, fn) if not os.path.isfile(fpath): continue period = parse_period_from_name(fn) key = (company, period_sort_key(period), period) unit = units.setdefault(key, { "company_slug": company, "period": period, "period_source": "filename" if period else "unknown", "files": [], "ignored": [], }) ext = os.path.splitext(fn)[1].lower() if ext in SUPPORTED_EXTS: unit["files"].append(os.path.abspath(fpath)) else: unit["ignored"].append(fn) out = [u for _, u in sorted(units.items(), key=lambda kv: (kv[0][0], kv[0][1])) if u["files"]] for u in out: u["files"].sort() u["ignored"].sort() return {"units": out, "skipped": skipped}