"""Storage backends. Two implementations, one interface:

  LocalStorage   - a folder on disk that mirrors the Drive job tree (dev / tests)
  DriveStorage   - the real shared Google Drive, via a service account

Both expose the same six calls. main.py never touches Drive directly.
"""
from __future__ import annotations

import csv
import io
import os
import re
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path

DOCS = "Docs"
SIGNED = "Signed"
REGISTER = "Register.csv"
CREW = "Crew.txt"
REGISTER_HEADER = ["timestamp", "job", "doc", "rev", "name", "signed_file", "user_agent"]

# Take-5: per-task, per-person hazard stop-and-think. Lad scans the foreman's QR, does it on his own
# phone. The record doubles as a task-start; the foreman taps Done (+ qty) later so the desk can get rates.
TAKE5 = "Take5"
TAKE5_REGISTER = "Take5-Register.csv"
HAZARDS = "Hazards.txt"          # one per line: "Hazard | standard control"  (seed from the prestart)
TAKE5_HEADER = ["id", "start", "job", "name", "task", "hazards", "controls", "other", "signed_file",
                "user_agent", "finish", "qty", "unit", "finished_by"]
DEFAULT_HAZARDS = [
    ("Manual handling — bar, mats, boards", "Two-man lifts on long or heavy gear. Cut to length before carrying."),
    ("Cutting — cut-off saw / grinder", "Guard on, glasses and face shield, gloves, hearing. Clear the cut area."),
    ("Exposed reo ends", "Caps on every exposed bar end. No stepping over uncapped bar."),
    ("Formwork — boards, props, step-ups", "Braced and pegged before you let go. Nails punched. Fingers clear."),
    ("Open excavation / trench edge", "Edge protection stays in place. Approach from the access side."),
    ("Mobile plant nearby", "Out of the swing radius. Eye contact with the operator. Spotter if entering."),
    ("Wet / slippery surface", "Three points of contact. No running. Battery tools only on wet ground."),
    ("Working at height / edge", "Handrail or harness. No leaning out. Tools tethered."),
    ("Other trades / public", "Stay inside the fence. Watch gear on the ground."),
    ("Heat / cold / hydration", "Water on site. Speak up if you're not right."),
]


# Hazard / near-miss / incident report. Sits beside Sign-On in "05 - Safety & Compliance/Incident Reports/"
# (the IMS folder every job already has). Feeds MSF22-3 at the desk and next Wednesday's toolbox.
INCIDENTS = "Incident Reports"
INCIDENT_REGISTER = "Incident-Register.csv"
INCIDENT_HEADER = ["id", "timestamp", "job", "kind", "name", "what", "where", "who", "action", "photo_file",
                   "pdf_file", "user_agent", "followup", "toolbox_date"]
INCIDENT_KINDS = {"hazard": "Hazard — I saw something unsafe", "nearmiss": "Near miss — nearly went wrong",
                  "incident": "Incident — someone hurt or something damaged"}


# Sign on / sign off. Append-only presence record, one row per tap. This is the job's muster: who
# was on site and when. It is never edited — a corrected time is a NEW row, attributed, not an
# overwrite. Hours are derived from it (app/timesheet.py), they are not stored here.
#
# event=adjust is the foreman correcting a time. It uses day/field/old/new/reason/by and leaves the
# original tap exactly where it was. The timesheet applies the latest adjust per field per day.
ATTENDANCE_REGISTER = "Attendance-Register.csv"
ATTENDANCE_HEADER = ["timestamp", "job", "name", "event", "lunch", "note", "source",
                     "day", "field", "old", "new", "reason", "by"]

# T&M / day-works docket. The money one. Lives with the job's site documents rather than in the
# safety folder, because it is a commercial record — it gets claimed, not audited.
TM_PARTS = ("07 - Site Documents", "TM Dockets")
TM_REGISTER = "TM-Register.csv"
TM_HEADER = ["id", "timestamp", "job", "date", "seq", "description", "men", "hours", "hours_total",
             "plant", "raised_by", "signed_by", "company", "signed_at", "pdf_file", "status"]

# "We're short of..." — one box and a photo. Sits with the sign-on register because it is the crew
# talking to the foreman, same as every other tap they make.
MATERIALS_REGISTER = "Materials-Register.csv"
MATERIALS_DIR = "Materials"
MATERIALS_HEADER = ["id", "timestamp", "job", "name", "item", "photo", "status", "by", "note"]
MATERIAL_STATUS = {"got": "got it on site", "ordered": "ordered", "not-needed": "not needed"}

# Photo drop — straight into the job's site diary, one folder per day, named time + who.
DIARY_PARTS = ("07 - Site Documents", "Site Diary")

# The board. Append-only, same rule as attendance: a drag is a row, nothing is ever edited, and the
# state of the board is what the rows add up to. That is what makes it a production record rather
# than a whiteboard — a card marked done with men and times on it prices itself.
#
# action:  created  a zone was opened            detail = the deck card it came off
#          on       a man went onto that zone    detail = his task, if we know it
#          off      he came off it               detail = why (lunch, knocked off)
#          home     sent home                    detail = the reason
#          moved    moved to another job         detail = the job
#          stage    the zone advanced            detail = formed|steel|inspected|poured|stripped|cleaned
#          estimate how long it is going to take detail = 1h|3h|today|tomorrow
#          done     the zone is finished         detail = ""
#          dismissed a Take-5 card was waved off detail = t5:<id>
BOARD_REGISTER = "Board-Register.csv"
BOARD_HEADER = ["timestamp", "job", "zone", "person", "action", "detail", "by"]

# The foreman's voice notes against the debrief sections — WEATHER, ISSUES, NEXT DAY. The bits the
# registers cannot know because nobody taps them, so he says them instead.
DIARY_NOTES = "Diary-Notes.csv"
DIARY_NOTES_HEADER = ["timestamp", "section", "text", "by"]

# Who is where tomorrow. Company-level beside devices.csv, because "which job" is a cross-job answer
# and a per-job file could never be read by /today without knowing the job first — which is the one
# thing /today does not know.
ROSTER = "Roster.csv"
ROSTER_HEADER = ["date", "person", "job", "by", "timestamp", "source"]

# Vehicle prestart. Company plant, not job plant: a ute is on three jobs in a week, so the record
# is company-level beside devices.csv and people.csv.
#
# Written twice on purpose, same reasoning as the attendance register and the per-worker index:
# Fleet/<REGO>/Plant-Register.csv is the vehicle's own history, Fleet/Defects.csv is the open-defect
# index the foreman screen reads. Without the index every foreman page load would have to open
# eight files to find out whether anything is wrong.
FLEET_DIR = "Fleet"
PLANT_REGISTER = "Plant-Register.csv"
DEFECTS_REGISTER = "Defects.csv"
PLANT_HEADER = ["id", "timestamp", "rego", "vehicle", "name", "result", "fails", "note", "photo",
                "status", "resolved_by", "resolved_at"]

# Company-level, beside the jobs rather than inside one — a bloke's phone is his on every job.
# See app/people.py for what lives in here.
COMPANY = "_signon"
FOREMAN_FILE = "Foreman.txt"


# Pre-numbering jobs have no 4-digit prefix. JOB_ALIASES in .env maps a short URL key to a folder-name substring:
#   JOB_ALIASES=powerhouse:Ford Civil - Parramatta;doyle:Doyle Ground
def parse_aliases(text: str) -> dict[str, str]:
    out = {}
    for part in (text or "").split(";"):
        k, _, v = part.partition(":")
        if k.strip() and v.strip():
            out[k.strip().lower()] = v.strip()
    return out


def parse_foreman(text: str) -> tuple[str, str]:
    """Foreman.txt — one line, 'Matt Ryan | 0412 345 678'. Phone optional."""
    for line in (text or "").splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        n, _, p = line.partition("|")
        return n.strip(), p.strip()
    return "", ""


def parse_hazards(text: str) -> list[tuple[str, str]]:
    out = []
    for line in text.splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        h, _, c = line.partition("|")
        out.append((h.strip(), c.strip()))
    return out

_JOB_RE = re.compile(r"^(\d{4})\b")
_REV_RE = re.compile(r"\brev\s*([A-Za-z0-9.]+)", re.I)


@dataclass
class Doc:
    slot: str        # stable id used in the QR (filename with rev + ext stripped, slugged)
    filename: str    # actual file name in Docs/
    rev: str         # parsed from filename, "" if none


def slot_of(filename: str) -> str:
    stem = Path(filename).stem
    stem = _REV_RE.sub("", stem)
    return re.sub(r"[^a-z0-9]+", "-", stem.lower()).strip("-")


def rev_of(filename: str) -> str:
    m = _REV_RE.search(Path(filename).stem)
    return m.group(1) if m else ""


def signed_filename(doc: Doc, name: str, when: datetime) -> str:
    safe = re.sub(r"[^\w .-]+", "", name).strip()
    return f"{when:%Y-%m-%d_%H%M}_{Path(doc.filename).stem}_{safe}.pdf"


def short_label(folder_name: str) -> str:
    """'2643 - Ford Civil - Wentworth Point CC4 Works' -> 'WENTWORTH POINT CC4 WORKS'.

    What goes in the roster message and on the board column. A bloke reads the place, not the
    client — nobody has ever asked "am I on Ford Civil tomorrow".
    """
    name = _JOB_RE.sub("", folder_name or "").strip()
    parts = [s.strip() for s in re.split(r"[-–—]", name) if s.strip()]
    tail = parts[-1] if parts else (folder_name or "")
    return tail.upper()[:26]


def mime_for(filename: str) -> str:
    n = filename.lower()
    if n.endswith(".pdf"):
        return "application/pdf"
    if n.endswith(".csv"):
        return "text/csv"
    if n.endswith(".png"):
        return "image/png"
    return "image/jpeg"


def append_csv(path: Path, header: list[str], row: dict) -> None:
    """Append one row, widening the file first if we have gained a column since it was written.

    Registers written before adjustments existed have seven columns, not thirteen. Writing thirteen
    values under a seven-column header silently shears every row after it, so the file is rewritten
    with the wider header first. Every existing row survives; the new columns come out blank.
    """
    fields = header
    if path.exists():
        with path.open(newline="", encoding="utf-8") as f:
            existing = next(csv.reader(f), [])
        if existing:
            missing = [c for c in header if c not in existing]
            if not missing:
                fields = existing            # file already as wide as us (or wider) — keep its shape
            elif existing != header:
                fields = existing + missing
                with path.open(newline="", encoding="utf-8") as f:
                    old_rows = [dict(r) for r in csv.DictReader(f)]
                with path.open("w", newline="", encoding="utf-8") as f:
                    w = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
                    w.writeheader()
                    for r in old_rows:
                        w.writerow({k: (r.get(k) or "") for k in fields})
    new = not path.exists()
    with path.open("a", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
        if new:
            w.writeheader()
        w.writerow({k: row.get(k, "") for k in fields})


# --------------------------------------------------------------------------- local


class LocalStorage:
    def __init__(self, root: str, subpath: str, aliases: dict[str, str] | None = None):
        self.root = Path(root)
        self.subpath = subpath
        self.aliases = aliases or {}

    def _job_dir(self, job_no: str) -> Path | None:
        # root/<stage>/<job> or root/<job>. Two levels only — never rglob a Drive mount.
        # Numeric key = the 4-digit LFCS job number prefix. Non-numeric key (pre-numbering jobs, e.g. "powerhouse")
        # = case-insensitive substring of the folder name.
        if _JOB_RE.match(job_no):
            for pattern in (f"*/{job_no}*", f"{job_no}*"):
                for p in self.root.glob(pattern):
                    if p.is_dir() and _JOB_RE.match(p.name):
                        return p
            return None
        key = self.aliases.get(job_no.lower(), job_no).lower()
        for pattern in ("*/*", "*"):
            for p in self.root.glob(pattern):
                if p.is_dir() and key in p.name.lower():
                    return p
        return None

    def find_job(self, job_no: str) -> str | None:
        d = self._job_dir(job_no)
        return d.name if d else None

    def list_jobs(self) -> list[dict]:
        """Every job folder that has a Sign-On folder in it. That IS the live list.

        No config file to fall out of step — a job goes live the moment someone drops a prestart in
        it, and drops off the board when the folder is archived. Two levels only, same as _job_dir.
        """
        out, seen = [], set()
        for pattern in ("*/*", "*"):
            for d in sorted(self.root.glob(pattern)):
                if not d.is_dir() or d.name in seen or d.name.startswith("_"):
                    continue
                if not (d / self.subpath).is_dir():
                    continue
                seen.add(d.name)
                out.append({"key": self._key_for(d.name), "name": d.name, "label": short_label(d.name)})
        return [j for j in out if j["key"]]

    def _key_for(self, folder_name: str) -> str:
        """The bit that goes in a URL. Job number if it has one, else whichever alias points at it."""
        m = _JOB_RE.match(folder_name)
        if m:
            return m.group(1)
        low = folder_name.lower()
        for k, v in self.aliases.items():
            if v.lower() in low:
                return k
        return ""

    def _signon(self, job_no: str) -> Path:
        d = self._job_dir(job_no)
        if not d:
            raise FileNotFoundError(job_no)
        s = d / self.subpath
        (s / DOCS).mkdir(parents=True, exist_ok=True)
        (s / SIGNED).mkdir(parents=True, exist_ok=True)
        return s

    def list_docs(self, job_no: str) -> list[Doc]:
        docs = self._signon(job_no) / DOCS
        return sorted(
            (Doc(slot_of(f.name), f.name, rev_of(f.name)) for f in docs.glob("*.pdf")),
            key=lambda d: d.filename.lower(),
        )

    def get_doc(self, job_no: str, slot: str) -> tuple[Doc, bytes] | None:
        for d in self.list_docs(job_no):
            if d.slot == slot:
                return d, (self._signon(job_no) / DOCS / d.filename).read_bytes()
        return None

    def put_signed(self, job_no: str, filename: str, data: bytes) -> str:
        p = self._signon(job_no) / SIGNED / filename
        p.write_bytes(data)
        return str(p)

    def append_register(self, job_no: str, row: dict) -> None:
        p = self._signon(job_no) / REGISTER
        new = not p.exists()
        with p.open("a", newline="", encoding="utf-8") as f:
            w = csv.DictWriter(f, fieldnames=REGISTER_HEADER)
            if new:
                w.writeheader()
            w.writerow(row)

    def get_crew(self, job_no: str) -> list[str]:
        p = self._signon(job_no) / CREW
        if not p.exists():
            return []
        return [l.strip() for l in p.read_text(encoding="utf-8").splitlines() if l.strip()]

    def list_register(self, job_no: str) -> list[dict]:
        p = self._signon(job_no) / REGISTER
        if not p.exists():
            return []
        with p.open(newline="", encoding="utf-8") as f:
            return [dict(r) for r in csv.DictReader(f)]

    # -- take-5 ------------------------------------------------------------
    def _take5(self, job_no: str) -> Path:
        t = self._signon(job_no) / TAKE5
        t.mkdir(parents=True, exist_ok=True)
        return t

    def get_hazards(self, job_no: str) -> list[tuple[str, str]]:
        p = self._take5(job_no) / HAZARDS
        if p.exists():
            got = parse_hazards(p.read_text(encoding="utf-8"))
            if got:
                return got
        return DEFAULT_HAZARDS

    def put_take5(self, job_no: str, filename: str, data: bytes) -> str:
        p = self._take5(job_no) / filename
        p.write_bytes(data)
        return str(p)

    def _read_take5(self, job_no: str) -> list[dict]:
        p = self._take5(job_no) / TAKE5_REGISTER
        if not p.exists():
            return []
        with p.open(newline="", encoding="utf-8") as f:
            return [dict(r) for r in csv.DictReader(f)]

    def _write_take5(self, job_no: str, rows: list[dict]) -> None:
        p = self._take5(job_no) / TAKE5_REGISTER
        with p.open("w", newline="", encoding="utf-8") as f:
            w = csv.DictWriter(f, fieldnames=TAKE5_HEADER, extrasaction="ignore")
            w.writeheader()
            for r in rows:
                w.writerow({k: r.get(k, "") for k in TAKE5_HEADER})

    def append_take5(self, job_no: str, row: dict) -> None:
        rows = self._read_take5(job_no)
        rows.append(row)
        self._write_take5(job_no, rows)

    def list_take5(self, job_no: str) -> list[dict]:
        return self._read_take5(job_no)

    def finish_take5(self, job_no: str, take5_id: str, fields: dict) -> bool:
        rows = self._read_take5(job_no)
        hit = False
        for r in rows:
            if r.get("id") == take5_id:
                r.update(fields)
                hit = True
        if hit:
            self._write_take5(job_no, rows)
        return hit

    # -- incident / hazard / near-miss --------------------------------------
    def _incidents(self, job_no: str) -> Path:
        # sibling of Sign-On inside the same "05 - Safety & Compliance" folder
        p = self._signon(job_no).parent / INCIDENTS
        p.mkdir(parents=True, exist_ok=True)
        return p

    def put_incident_file(self, job_no: str, filename: str, data: bytes) -> str:
        p = self._incidents(job_no) / filename
        p.write_bytes(data)
        return str(p)

    def append_incident(self, job_no: str, row: dict) -> None:
        p = self._incidents(job_no) / INCIDENT_REGISTER
        new = not p.exists()
        with p.open("a", newline="", encoding="utf-8") as f:
            w = csv.DictWriter(f, fieldnames=INCIDENT_HEADER, extrasaction="ignore")
            if new:
                w.writeheader()
            w.writerow({k: row.get(k, "") for k in INCIDENT_HEADER})

    def list_incidents(self, job_no: str) -> list[dict]:
        p = self._incidents(job_no) / INCIDENT_REGISTER
        if not p.exists():
            return []
        with p.open(newline="", encoding="utf-8") as f:
            return [dict(r) for r in csv.DictReader(f)]

    # -- sign on / sign off --------------------------------------------------
    def append_attendance(self, job_no: str, row: dict) -> None:
        append_csv(self._signon(job_no) / ATTENDANCE_REGISTER, ATTENDANCE_HEADER, row)

    def list_attendance(self, job_no: str) -> list[dict]:
        p = self._signon(job_no) / ATTENDANCE_REGISTER
        if not p.exists():
            return []
        with p.open(newline="", encoding="utf-8") as f:
            return [dict(r) for r in csv.DictReader(f)]

    def get_foreman(self, job_no: str) -> tuple[str, str]:
        p = self._signon(job_no) / FOREMAN_FILE
        return parse_foreman(p.read_text(encoding="utf-8")) if p.exists() else ("", "")

    # -- anywhere else in the job folder ------------------------------------
    #
    # T&M dockets and site-diary photos do not live under Sign-On, so they get one generic pair of
    # calls rather than a bespoke pair each. `parts` is folder names under the job folder;
    # `signon_parts` hands back the Sign-On folder for the things that DO belong beside the register.

    @property
    def signon_parts(self) -> tuple:
        return tuple(x for x in self.subpath.replace("\\", "/").split("/") if x)

    def _under(self, job_no: str, parts, create: bool = True) -> Path:
        # create=False on the read paths. The foreman screen reads the T&M register on every load,
        # and 148 job folders do not want an empty "TM Dockets" in them because someone looked.
        d = self._job_dir(job_no)
        if not d:
            raise FileNotFoundError(job_no)
        p = d.joinpath(*parts)
        if create:
            p.mkdir(parents=True, exist_ok=True)
        return p

    def put_job_file(self, job_no: str, parts, filename: str, data: bytes) -> str:
        p = self._under(job_no, parts) / filename
        p.write_bytes(data)
        return str(p)

    def get_job_file(self, job_no: str, parts, filename: str) -> bytes | None:
        p = self._under(job_no, parts, create=False) / filename
        return p.read_bytes() if p.exists() else None

    def list_job_files(self, job_no: str, parts) -> list[str]:
        """Filenames only. The diary needs to know what is in today's photo folder, nothing more."""
        p = self._under(job_no, parts, create=False)
        return sorted(f.name for f in p.iterdir() if f.is_file()) if p.is_dir() else []

    def read_job_csv(self, job_no: str, parts, filename: str) -> list[dict]:
        p = self._under(job_no, parts, create=False) / filename
        if not p.exists():
            return []
        with p.open(newline="", encoding="utf-8") as f:
            return [dict(r) for r in csv.DictReader(f)]

    def append_job_csv(self, job_no: str, parts, filename: str, header: list[str], row: dict) -> None:
        append_csv(self._under(job_no, parts) / filename, header, row)

    def write_job_csv(self, job_no: str, parts, filename: str, header: list[str], rows: list[dict]) -> None:
        with (self._under(job_no, parts) / filename).open("w", newline="", encoding="utf-8") as f:
            w = csv.DictWriter(f, fieldnames=header, extrasaction="ignore")
            w.writeheader()
            for r in rows:
                w.writerow({k: (r.get(k) or "") for k in header})

    # -- company-wide (devices, people, fleet) --------------------------------
    def _company(self, *parts, create: bool = True) -> Path:
        p = self.root.joinpath(COMPANY, *parts)
        if create:
            p.mkdir(parents=True, exist_ok=True)
        return p

    def read_company_csv(self, filename: str, subdir=()) -> list[dict]:
        p = self._company(*subdir, create=False) / filename
        if not p.exists():
            return []
        with p.open(newline="", encoding="utf-8") as f:
            return [dict(r) for r in csv.DictReader(f)]

    def append_company_csv(self, filename: str, header: list[str], row: dict, subdir=()) -> None:
        append_csv(self._company(*subdir) / filename, header, row)

    def write_company_csv(self, filename: str, header: list[str], rows: list[dict], subdir=()) -> None:
        with (self._company(*subdir) / filename).open("w", newline="", encoding="utf-8") as f:
            w = csv.DictWriter(f, fieldnames=header, extrasaction="ignore")
            w.writeheader()
            for r in rows:
                w.writerow({k: (r.get(k) or "") for k in header})

    def put_company_file(self, filename: str, data: bytes, subdir=()) -> str:
        p = self._company(*subdir) / filename
        p.write_bytes(data)
        return str(p)

    def get_company_file(self, filename: str, subdir=()) -> bytes | None:
        p = self._company(*subdir, create=False) / filename
        return p.read_bytes() if p.exists() else None


# --------------------------------------------------------------------------- drive


class DriveStorage:
    """Google Drive via service account. Folder IDs cached per job for the process lifetime."""

    FOLDER = "application/vnd.google-apps.folder"

    def __init__(self, sa_json: str, jobs_root_id: str, subpath: str, aliases: dict[str, str] | None = None):
        self.aliases = aliases or {}
        from google.oauth2 import service_account
        from googleapiclient.discovery import build

        creds = service_account.Credentials.from_service_account_file(
            sa_json, scopes=["https://www.googleapis.com/auth/drive"]
        )
        self.svc = build("drive", "v3", credentials=creds, cache_discovery=False)
        self.root = jobs_root_id
        self.subpath = [s for s in subpath.replace("\\", "/").split("/") if s]
        self._cache: dict[str, dict[str, str]] = {}   # job_no -> {"job":id,"signon":id,"docs":id,"signed":id}

    # -- low level ---------------------------------------------------------
    def _q(self, q: str, fields="files(id,name,mimeType)"):
        out, tok = [], None
        while True:
            r = self.svc.files().list(
                q=q, fields=f"nextPageToken,{fields}", pageToken=tok,
                supportsAllDrives=True, includeItemsFromAllDrives=True, pageSize=200,
            ).execute()
            out += r.get("files", [])
            tok = r.get("nextPageToken")
            if not tok:
                return out

    def _children(self, parent: str, folders_only=False):
        q = f"'{parent}' in parents and trashed=false"
        if folders_only:
            q += f" and mimeType='{self.FOLDER}'"
        return self._q(q)

    def _child_named(self, parent: str, name: str, create=False) -> str | None:
        for f in self._children(parent):
            if f["name"] == name:
                return f["id"]
        if not create:
            return None
        r = self.svc.files().create(
            body={"name": name, "mimeType": self.FOLDER, "parents": [parent]},
            fields="id", supportsAllDrives=True,
        ).execute()
        return r["id"]

    def _find_job_folder(self, job_no: str) -> tuple[str, str] | None:
        # jobs root -> stage folders (Upcoming/Live/...) -> job folders. Two levels, then one.
        numeric = bool(_JOB_RE.match(job_no))
        key = self.aliases.get(job_no.lower(), job_no).lower()
        def hit(name: str) -> bool:
            return name.startswith(job_no) if numeric else key in name.lower()
        for stage in self._children(self.root, folders_only=True):
            for job in self._children(stage["id"], folders_only=True):
                if hit(job["name"]):
                    return job["id"], job["name"]
        for job in self._children(self.root, folders_only=True):
            if hit(job["name"]):
                return job["id"], job["name"]
        return None

    def _ids(self, job_no: str) -> dict[str, str]:
        if job_no in self._cache:
            return self._cache[job_no]
        found = self._find_job_folder(job_no)
        if not found:
            raise FileNotFoundError(job_no)
        cur, name = found
        for part in self.subpath:
            cur = self._child_named(cur, part, create=True)
        ids = {
            "job": found[0], "job_name": name, "signon": cur,
            "docs": self._child_named(cur, DOCS, create=True),
            "signed": self._child_named(cur, SIGNED, create=True),
        }
        self._cache[job_no] = ids
        return ids

    def _download(self, file_id: str) -> bytes:
        from googleapiclient.http import MediaIoBaseDownload
        buf = io.BytesIO()
        dl = MediaIoBaseDownload(buf, self.svc.files().get_media(fileId=file_id, supportsAllDrives=True))
        done = False
        while not done:
            _, done = dl.next_chunk()
        return buf.getvalue()

    def _upload(self, parent: str, name: str, data: bytes, mime: str, file_id: str | None = None) -> str:
        from googleapiclient.http import MediaIoBaseUpload
        media = MediaIoBaseUpload(io.BytesIO(data), mimetype=mime, resumable=False)
        if file_id:
            r = self.svc.files().update(fileId=file_id, media_body=media, fields="id", supportsAllDrives=True).execute()
        else:
            r = self.svc.files().create(
                body={"name": name, "parents": [parent]}, media_body=media, fields="id", supportsAllDrives=True
            ).execute()
        return r["id"]

    # -- interface ---------------------------------------------------------
    def find_job(self, job_no: str) -> str | None:
        try:
            return self._ids(job_no)["job_name"]
        except FileNotFoundError:
            return None

    def list_jobs(self) -> list[dict]:
        """Same question as the local backend, cached for the process — it is a folder walk.

        Only jobs holding the sign-on subpath count, so a stage folder full of finished work costs
        one listing and nothing else.
        """
        if "_jobs" in self._cache:
            return self._cache["_jobs"]["rows"]
        out, seen = [], set()
        groups = [self.root] + [s["id"] for s in self._children(self.root, folders_only=True)]
        for parent in groups:
            for job in self._children(parent, folders_only=True):
                if job["name"] in seen or job["name"].startswith("_"):
                    continue
                cur = job["id"]
                for part in self.subpath:
                    cur = self._child_named(cur, part, create=False)
                    if not cur:
                        break
                if not cur:
                    continue
                seen.add(job["name"])
                out.append({"key": self._key_for(job["name"]), "name": job["name"],
                            "label": short_label(job["name"])})
        rows = [j for j in out if j["key"]]
        self._cache["_jobs"] = {"rows": rows}
        return rows

    def _key_for(self, folder_name: str) -> str:
        m = _JOB_RE.match(folder_name)
        if m:
            return m.group(1)
        low = folder_name.lower()
        for k, v in self.aliases.items():
            if v.lower() in low:
                return k
        return ""

    def list_docs(self, job_no: str) -> list[Doc]:
        ids = self._ids(job_no)
        files = [f for f in self._children(ids["docs"]) if f["name"].lower().endswith(".pdf")]
        return sorted((Doc(slot_of(f["name"]), f["name"], rev_of(f["name"])) for f in files),
                      key=lambda d: d.filename.lower())

    def get_doc(self, job_no: str, slot: str) -> tuple[Doc, bytes] | None:
        ids = self._ids(job_no)
        for f in self._children(ids["docs"]):
            if f["name"].lower().endswith(".pdf") and slot_of(f["name"]) == slot:
                return Doc(slot, f["name"], rev_of(f["name"])), self._download(f["id"])
        return None

    def put_signed(self, job_no: str, filename: str, data: bytes) -> str:
        ids = self._ids(job_no)
        fid = self._upload(ids["signed"], filename, data, "application/pdf")
        return f"drive:{fid}"

    def append_register(self, job_no: str, row: dict) -> None:
        # Drive has no append; read-modify-write the CSV. Fine at sign-on volumes.
        ids = self._ids(job_no)
        existing = self._child_named(ids["signon"], REGISTER)
        buf = io.StringIO()
        w = csv.DictWriter(buf, fieldnames=REGISTER_HEADER)
        if existing:
            buf.write(self._download(existing).decode("utf-8"))
            if not buf.getvalue().endswith("\n"):
                buf.write("\n")
        else:
            w.writeheader()
        w.writerow(row)
        self._upload(ids["signon"], REGISTER, buf.getvalue().encode("utf-8"), "text/csv", file_id=existing)

    def get_crew(self, job_no: str) -> list[str]:
        ids = self._ids(job_no)
        fid = self._child_named(ids["signon"], CREW)
        if not fid:
            return []
        return [l.strip() for l in self._download(fid).decode("utf-8").splitlines() if l.strip()]

    def list_register(self, job_no: str) -> list[dict]:
        ids = self._ids(job_no)
        fid = self._child_named(ids["signon"], REGISTER)
        if not fid:
            return []
        return [dict(r) for r in csv.DictReader(io.StringIO(self._download(fid).decode("utf-8")))]

    # -- take-5 (same read-modify-write pattern as the register) -----------
    def _take5_id(self, job_no: str) -> str:
        ids = self._ids(job_no)
        if "take5" not in ids:
            ids["take5"] = self._child_named(ids["signon"], TAKE5, create=True)
        return ids["take5"]

    def get_hazards(self, job_no: str) -> list[tuple[str, str]]:
        fid = self._child_named(self._take5_id(job_no), HAZARDS)
        if fid:
            got = parse_hazards(self._download(fid).decode("utf-8"))
            if got:
                return got
        return DEFAULT_HAZARDS

    def put_take5(self, job_no: str, filename: str, data: bytes) -> str:
        return f"drive:{self._upload(self._take5_id(job_no), filename, data, 'application/pdf')}"

    def _read_take5(self, job_no: str) -> tuple[list[dict], str | None]:
        fid = self._child_named(self._take5_id(job_no), TAKE5_REGISTER)
        if not fid:
            return [], None
        return [dict(r) for r in csv.DictReader(io.StringIO(self._download(fid).decode("utf-8")))], fid

    def _write_take5(self, job_no: str, rows: list[dict], fid: str | None) -> None:
        buf = io.StringIO()
        w = csv.DictWriter(buf, fieldnames=TAKE5_HEADER, extrasaction="ignore")
        w.writeheader()
        for r in rows:
            w.writerow({k: r.get(k, "") for k in TAKE5_HEADER})
        self._upload(self._take5_id(job_no), TAKE5_REGISTER, buf.getvalue().encode("utf-8"), "text/csv", file_id=fid)

    def append_take5(self, job_no: str, row: dict) -> None:
        rows, fid = self._read_take5(job_no)
        rows.append(row)
        self._write_take5(job_no, rows, fid)

    def list_take5(self, job_no: str) -> list[dict]:
        return self._read_take5(job_no)[0]

    def finish_take5(self, job_no: str, take5_id: str, fields: dict) -> bool:
        rows, fid = self._read_take5(job_no)
        hit = False
        for r in rows:
            if r.get("id") == take5_id:
                r.update(fields)
                hit = True
        if hit:
            self._write_take5(job_no, rows, fid)
        return hit

    # -- incident / hazard / near-miss --------------------------------------
    def _incidents_id(self, job_no: str) -> str:
        ids = self._ids(job_no)
        if "incidents" not in ids:
            # parent of Sign-On = walk subpath minus its last part from the job folder
            cur = ids["job"]
            for part in self.subpath[:-1]:
                cur = self._child_named(cur, part, create=True)
            ids["incidents"] = self._child_named(cur, INCIDENTS, create=True)
        return ids["incidents"]

    def put_incident_file(self, job_no: str, filename: str, data: bytes) -> str:
        mime = "application/pdf" if filename.lower().endswith(".pdf") else "image/jpeg"
        return f"drive:{self._upload(self._incidents_id(job_no), filename, data, mime)}"

    def append_incident(self, job_no: str, row: dict) -> None:
        folder = self._incidents_id(job_no)
        existing = self._child_named(folder, INCIDENT_REGISTER)
        buf = io.StringIO()
        w = csv.DictWriter(buf, fieldnames=INCIDENT_HEADER, extrasaction="ignore")
        if existing:
            buf.write(self._download(existing).decode("utf-8"))
            if not buf.getvalue().endswith("\n"):
                buf.write("\n")
        else:
            w.writeheader()
        w.writerow({k: row.get(k, "") for k in INCIDENT_HEADER})
        self._upload(folder, INCIDENT_REGISTER, buf.getvalue().encode("utf-8"), "text/csv", file_id=existing)

    def list_incidents(self, job_no: str) -> list[dict]:
        fid = self._child_named(self._incidents_id(job_no), INCIDENT_REGISTER)
        if not fid:
            return []
        return [dict(r) for r in csv.DictReader(io.StringIO(self._download(fid).decode("utf-8")))]

    # -- sign on / sign off (read-modify-write, same as the register) --------
    def append_attendance(self, job_no: str, row: dict) -> None:
        # Read-rewrite rather than text-append: a register written before adjustments existed has
        # seven columns, and pasting a thirteen-column row under it shears the file. Same widening
        # rule as storage.append_csv on the local side.
        ids = self._ids(job_no)
        existing = self._child_named(ids["signon"], ATTENDANCE_REGISTER)
        rows: list[dict] = []
        fields = ATTENDANCE_HEADER
        if existing:
            rdr = csv.DictReader(io.StringIO(self._download(existing).decode("utf-8")))
            rows = [dict(r) for r in rdr]
            cur = list(rdr.fieldnames or [])
            missing = [c for c in ATTENDANCE_HEADER if c not in cur]
            fields = cur + missing if cur else ATTENDANCE_HEADER
        rows.append(row)
        buf = io.StringIO()
        w = csv.DictWriter(buf, fieldnames=fields, extrasaction="ignore")
        w.writeheader()
        for r in rows:
            w.writerow({k: (r.get(k) or "") for k in fields})
        self._upload(ids["signon"], ATTENDANCE_REGISTER, buf.getvalue().encode("utf-8"), "text/csv", file_id=existing)

    def list_attendance(self, job_no: str) -> list[dict]:
        ids = self._ids(job_no)
        fid = self._child_named(ids["signon"], ATTENDANCE_REGISTER)
        if not fid:
            return []
        return [dict(r) for r in csv.DictReader(io.StringIO(self._download(fid).decode("utf-8")))]

    def get_foreman(self, job_no: str) -> tuple[str, str]:
        ids = self._ids(job_no)
        fid = self._child_named(ids["signon"], FOREMAN_FILE)
        return parse_foreman(self._download(fid).decode("utf-8")) if fid else ("", "")

    # -- anywhere else in the job folder ------------------------------------
    @property
    def signon_parts(self) -> tuple:
        return tuple(self.subpath)

    def _under_id(self, job_no: str, parts, create: bool = True) -> str | None:
        # create=False on the read paths — a foreman opening the screen must not make folders.
        ids = self._ids(job_no)
        key = "|".join(parts)
        if key in ids:
            return ids[key]
        cur = ids["job"]
        for part in parts:
            cur = self._child_named(cur, part, create=create)
            if not cur:
                return None
        ids[key] = cur                          # only cached once it actually exists
        return cur

    def put_job_file(self, job_no: str, parts, filename: str, data: bytes) -> str:
        folder = self._under_id(job_no, parts)
        existing = self._child_named(folder, filename)
        return f"drive:{self._upload(folder, filename, data, mime_for(filename), file_id=existing)}"

    def get_job_file(self, job_no: str, parts, filename: str) -> bytes | None:
        folder = self._under_id(job_no, parts, create=False)
        fid = self._child_named(folder, filename) if folder else None
        return self._download(fid) if fid else None

    def list_job_files(self, job_no: str, parts) -> list[str]:
        folder = self._under_id(job_no, parts, create=False)
        return sorted(f["name"] for f in self._children(folder)) if folder else []

    def read_job_csv(self, job_no: str, parts, filename: str) -> list[dict]:
        folder = self._under_id(job_no, parts, create=False)
        fid = self._child_named(folder, filename) if folder else None
        if not fid:
            return []
        return [dict(r) for r in csv.DictReader(io.StringIO(self._download(fid).decode("utf-8")))]

    def append_job_csv(self, job_no: str, parts, filename: str, header: list[str], row: dict) -> None:
        rows = self.read_job_csv(job_no, parts, filename)
        rows.append(row)
        self.write_job_csv(job_no, parts, filename, header, rows)

    def write_job_csv(self, job_no: str, parts, filename: str, header: list[str], rows: list[dict]) -> None:
        folder = self._under_id(job_no, parts)
        existing = self._child_named(folder, filename)
        buf = io.StringIO()
        w = csv.DictWriter(buf, fieldnames=header, extrasaction="ignore")
        w.writeheader()
        for r in rows:
            w.writerow({k: (r.get(k) or "") for k in header})
        self._upload(folder, filename, buf.getvalue().encode("utf-8"), "text/csv", file_id=existing)

    # -- company-wide (devices, people, fleet) --------------------------------
    def _company_id(self, *parts, create: bool = True) -> str | None:
        key = "_company|" + "|".join(parts)
        if key in self._cache:
            return self._cache[key]["id"]
        cur = self._child_named(self.root, COMPANY, create=create)
        for part in parts:
            if not cur:
                return None
            cur = self._child_named(cur, part, create=create)
        if not cur:
            return None
        self._cache[key] = {"id": cur}
        return cur

    def read_company_csv(self, filename: str, subdir=()) -> list[dict]:
        folder = self._company_id(*subdir, create=False)
        fid = self._child_named(folder, filename) if folder else None
        if not fid:
            return []
        return [dict(r) for r in csv.DictReader(io.StringIO(self._download(fid).decode("utf-8")))]

    def append_company_csv(self, filename: str, header: list[str], row: dict, subdir=()) -> None:
        rows = self.read_company_csv(filename, subdir)
        rows.append(row)
        self.write_company_csv(filename, header, rows, subdir)

    def write_company_csv(self, filename: str, header: list[str], rows: list[dict], subdir=()) -> None:
        folder = self._company_id(*subdir)
        existing = self._child_named(folder, filename)
        buf = io.StringIO()
        w = csv.DictWriter(buf, fieldnames=header, extrasaction="ignore")
        w.writeheader()
        for r in rows:
            w.writerow({k: (r.get(k) or "") for k in header})
        self._upload(folder, filename, buf.getvalue().encode("utf-8"), "text/csv", file_id=existing)

    def put_company_file(self, filename: str, data: bytes, subdir=()) -> str:
        folder = self._company_id(*subdir)
        existing = self._child_named(folder, filename)
        return f"drive:{self._upload(folder, filename, data, mime_for(filename), file_id=existing)}"

    def get_company_file(self, filename: str, subdir=()) -> bytes | None:
        folder = self._company_id(*subdir, create=False)
        fid = self._child_named(folder, filename) if folder else None
        return self._download(fid) if fid else None


def from_env():
    from dotenv import load_dotenv
    load_dotenv()
    sub = os.getenv("SIGNON_SUBPATH", "05. Safety & Compliance/Sign-On")
    aliases = parse_aliases(os.getenv("JOB_ALIASES", ""))
    local = os.getenv("LOCAL_JOBS_ROOT")
    if local:
        return LocalStorage(local, sub, aliases)
    return DriveStorage(
        os.environ["GOOGLE_SERVICE_ACCOUNT_JSON"],
        os.environ["JOBS_ROOT_FOLDER_ID"],
        sub,
        aliases,
    )