MS Project Schedule Parsing

from __future__ import annotations

import logging
from xml.etree.ElementTree import Element, parse

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-8s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("mspdi.parse")

# The MSPDI default namespace. ElementTree qualifies every tag with this URI,
# so all lookups must go through the registered prefix, never the bare tag.
MSPDI_NS = {"p": "http://schemas.microsoft.com/project"}


def _text(node: Element, path: str, default: str | None = None) -> str | None:
    """Namespace-safe child text lookup; returns default when absent or empty."""
    child = node.find(path, MSPDI_NS)
    if child is None or child.text is None or not child.text.strip():
        return default
    return child.text.strip()


def iter_task_elements(xml_path: str) -> list[Element]:
    """Yield every real <Task> element, skipping the reserved UID 0 summary row."""
    root = parse(xml_path).getroot()
    tasks_block = root.find("p:Tasks", MSPDI_NS)
    if tasks_block is None:
        raise ValueError("MSPDI file has no <Tasks> block; not a schedule export")

    real_tasks: list[Element] = []
    for task in tasks_block.findall("p:Task", MSPDI_NS):
        uid = _text(task, "p:UID")
        if uid is None:
            logger.warning("Task with no UID skipped")
            continue
        if uid == "0":
            continue  # UID 0 is the project summary; it carries no scope
        real_tasks.append(task)
    logger.info("Parsed %d real tasks from %s", len(real_tasks), xml_path)
    return real_tasks

import re
from datetime import datetime
from decimal import Decimal
from zoneinfo import ZoneInfo

# ISO 8601 working-time duration as MS Project writes it: PT<h>H<m>M<s>S.
_DURATION_RE = re.compile(r"^PT(\d+)H(\d+)M(\d+)S$")


def duration_to_days(raw: str, hours_per_day: Decimal = Decimal("8")) -> Decimal:
    """Convert an MSPDI PT#H#M#S working-time duration into working days."""
    match = _DURATION_RE.match(raw.strip())
    if match is None:
        raise ValueError(f"Unrecognized MSPDI duration format: {raw!r}")
    hours, minutes, seconds = (Decimal(g) for g in match.groups())
    total_hours = hours + (minutes / Decimal("60")) + (seconds / Decimal("3600"))
    # Round to 3 dp of a working day; a task is priced/scheduled in days, not seconds.
    return (total_hours / hours_per_day).quantize(Decimal("0.001"))


def to_aware(raw: str, project_tz: ZoneInfo) -> datetime:
    """Parse a bare MSPDI local timestamp and stamp it with the project zone."""
    parsed = datetime.fromisoformat(raw)  # e.g. 2026-07-15T08:00:00 — no offset
    if parsed.tzinfo is not None:
        return parsed  # already carried an offset; trust it
    return parsed.replace(tzinfo=project_tz)

from typing import Literal

WBS_RE = re.compile(r"^PROJ-\d{3}-(ARCH|STR|MEP|CIV|ELEC|PLMB)-\d{2}$")

AUTO_ROUTE = Decimal("0.92")    # site-canonical: >= 0.92 binds automatically
REVIEW_FLOOR = Decimal("0.75")  # 0.75–0.92 holds for a scheduler; < 0.75 quarantines

RouteState = Literal["auto_route", "human_review", "quarantine"]


def classify_confidence(score: Decimal) -> RouteState:
    if score >= AUTO_ROUTE:
        return "auto_route"
    if score >= REVIEW_FLOOR:
        return "human_review"
    return "quarantine"


def bind_wbs(
    outline_number: str,
    outline_crosswalk: dict[str, str],
    division_crosswalk: dict[str, str],
) -> tuple[str | None, Decimal, RouteState]:
    """Resolve an MSP OutlineNumber to a canonical WBS node with a confidence score."""
    if outline_number in outline_crosswalk:
        node = outline_crosswalk[outline_number]
        if not WBS_RE.match(node):
            raise ValueError(f"Crosswalk target violates PROJ-NNN-DIV-NN: {node!r}")
        return node, Decimal("0.98"), "auto_route"

    # Fall back to the parent outline (e.g. "1.2.3" -> "1.2") for a division rollup.
    parent = outline_number.rsplit(".", 1)[0] if "." in outline_number else outline_number
    if parent in division_crosswalk:
        return division_crosswalk[parent], Decimal("0.83"), "human_review"

    return None, Decimal("0.00"), "quarantine"  # never bind to a guessed node

from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

_LINK_TYPES = {"0": "FF", "1": "FS", "2": "SF", "3": "SS"}
LinkType = Literal["FF", "FS", "SF", "SS"]


class PredecessorLink(BaseModel):
    predecessor_uid: int = Field(ge=1)
    link_type: LinkType = "FS"
    lag_days: Decimal = Field(default=Decimal("0"))


class ScheduleActivity(BaseModel):
    """Source-agnostic activity contract; P6 and MSP parsers both emit this shape."""

    model_config = ConfigDict(frozen=True)

    activity_id: str = Field(min_length=1, max_length=64)
    name: str = Field(min_length=1, max_length=240)
    wbs_node: str = Field(pattern=r"^PROJ-\d{3}-(ARCH|STR|MEP|CIV|ELEC|PLMB)-\d{2}$")
    start: datetime
    finish: datetime
    duration_days: Decimal = Field(ge=0)
    total_float_days: int
    is_critical: bool
    predecessors: list[PredecessorLink] = Field(default_factory=list)
    source: Literal["p6", "msp"]

    @field_validator("start", "finish")
    @classmethod
    def require_tz_aware(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("start/finish must be timezone-aware (see ISO 8601)")
        return v

    @model_validator(mode="after")
    def finish_after_start(self) -> "ScheduleActivity":
        if self.finish < self.start:
            raise ValueError("finish precedes start")
        # Total float below zero means the activity is behind and drives the path.
        if self.total_float_days <= 0 and not self.is_critical:
            raise ValueError("non-positive total float must be marked critical")
        return self


def parse_predecessors(task: Element) -> list[PredecessorLink]:
    """Decode every <PredecessorLink> under a task into typed links."""
    links: list[PredecessorLink] = []
    for link in task.findall("p:PredecessorLink", MSPDI_NS):
        uid = _text(link, "p:PredecessorUID")
        if uid is None:
            continue
        type_code = _text(link, "p:Type", default="1")  # default Finish-Start
        links.append(
            PredecessorLink(
                predecessor_uid=int(uid),
                link_type=_LINK_TYPES.get(type_code or "1", "FS"),
            )
        )
    return links


def build_activity(
    task: Element,
    wbs_node: str,
    project_tz: ZoneInfo,
    hours_per_day: Decimal = Decimal("8"),
) -> ScheduleActivity:
    """Assemble one ScheduleActivity from a normalized, bound MSPDI task element."""
    total_float = int(_text(task, "p:TotalSlack", default="0") or "0")
    return ScheduleActivity(
        activity_id=_text(task, "p:UID") or "",
        name=_text(task, "p:Name", default="(unnamed task)") or "(unnamed task)",
        wbs_node=wbs_node,
        start=to_aware(_text(task, "p:Start") or "", project_tz),
        finish=to_aware(_text(task, "p:Finish") or "", project_tz),
        duration_days=duration_to_days(_text(task, "p:Duration") or "PT0H0M0S", hours_per_day),
        total_float_days=total_float,
        is_critical=(_text(task, "p:Critical", default="0") == "1"),
        predecessors=parse_predecessors(task),
        source="msp",
    )

from datetime import datetime
from decimal import Decimal
from zoneinfo import ZoneInfo


def test_duration_decodes_to_working_days():
    assert duration_to_days("PT8H0M0S") == Decimal("1.000")
    assert duration_to_days("PT16H0M0S") == Decimal("2.000")
    assert duration_to_days("PT4H0M0S") == Decimal("0.500")


def test_bad_duration_raises():
    import pytest
    with pytest.raises(ValueError):
        duration_to_days("8 days")


def test_naive_timestamp_is_made_aware():
    tz = ZoneInfo("America/New_York")
    aware = to_aware("2026-07-15T08:00:00", tz)
    assert aware.tzinfo is not None
    assert aware.utcoffset() is not None


def test_exact_outline_binds_and_auto_routes():
    node, score, state = bind_wbs(
        "1.2.3",
        outline_crosswalk={"1.2.3": "PROJ-014-STR-02"},
        division_crosswalk={},
    )
    assert node == "PROJ-014-STR-02"
    assert state == "auto_route"
    assert score >= AUTO_ROUTE


def test_unmapped_outline_quarantines():
    node, score, state = bind_wbs("9.9.9", {}, {})
    assert node is None
    assert state == "quarantine"
    assert score < REVIEW_FLOOR


def test_activity_round_trips_and_matches_contract():
    tz = ZoneInfo("America/New_York")
    act = ScheduleActivity(
        activity_id="12",
        name="Erect structural steel — level 3",
        wbs_node="PROJ-014-STR-02",
        start=datetime(2026, 7, 15, 8, 0, tzinfo=tz),
        finish=datetime(2026, 7, 17, 17, 0, tzinfo=tz),
        duration_days=Decimal("3.000"),
        total_float_days=0,
        is_critical=True,
        source="msp",
    )
    restored = ScheduleActivity.model_validate_json(act.model_dump_json())
    assert restored.source == "msp"
    assert restored.wbs_node == act.wbs_node