core/msproject.py

"""
msproject.py — Lectura/escritura de cronogramas MS Project (MSPDI XML).

Cambio estructural respecto al original: auditar_xml() ya NO escribe en
st.session_state. Ahora devuelve un AuditResult (dataclass) con TODO lo que las
capas superiores necesitan (df de tareas, bytes crudos, prefijo de namespace,
horas/día, nombre del proyecto, calendario detectado). Esto vuelve el parseo
reutilizable desde cualquier interfaz o test. La lógica de parseo/reescritura es
idéntica a app_tesis_final_V3.py (incluido [AUD-11], el modo Manual).
"""
import re
from dataclasses import dataclass, field
from datetime import datetime, timedelta, time as dtime
import xml.etree.ElementTree as ET

import pandas as pd


@dataclass
class AuditResult:
    """Resultado del parseo de un XML MSPDI, autocontenido (sin session_state)."""
    df: pd.DataFrame
    raw_bytes: bytes
    prefix: str
    hours_per_day: float
    project_name: str
    cal_dias: list | None = None
    cal_feriados: dict = field(default_factory=dict)

    @property
    def errores(self) -> pd.DataFrame:
        return self.df[self.df['Errores'] != 'OK']


def _to_bytes(file_or_bytes) -> bytes:
    """Acepta bytes o un objeto file-like y devuelve bytes."""
    if isinstance(file_or_bytes, (bytes, bytearray)):
        return bytes(file_or_bytes)
    file_or_bytes.seek(0)
    return file_or_bytes.read()


def extraer_calendario_xml(root, prefix):
    """Lee el calendario base del proyecto MSPDI: días laborables (weekday Python) y feriados."""
    map_daytype = {1: 6, 2: 0, 3: 1, 4: 2, 5: 3, 6: 4, 7: 5}  # DayType MSPDI (1=Dom..7=Sab) -> weekday() Py
    cal_uid_node = root.find(prefix + "CalendarUID")
    cal_default = cal_uid_node.text if cal_uid_node is not None else None
    dias_idx = set(); feriados = {}
    cals = root.find(prefix + "Calendars")
    if cals is None:
        return [0, 1, 2, 3, 4], {}, None
    elegido = None
    for cal in cals.findall(prefix + "Calendar"):
        uid = cal.findtext(prefix + "UID")
        if cal_default and uid == cal_default:
            elegido = cal; break
        if elegido is None and cal.findtext(prefix + "IsBaseCalendar") == "1":
            elegido = cal
    if elegido is None:
        elegido = cals.find(prefix + "Calendar")
    if elegido is None:
        return [0, 1, 2, 3, 4], {}, None
    wds = elegido.find(prefix + "WeekDays")
    if wds is not None:
        for wd in wds.findall(prefix + "WeekDay"):
            dt_node = wd.findtext(prefix + "DayType"); working = wd.findtext(prefix + "DayWorking")
            if dt_node:
                if working == "1":
                    py = map_daytype.get(int(dt_node))
                    if py is not None:
                        dias_idx.add(py)
            else:  # excepción embebida (forma antigua)
                tp = wd.find(prefix + "TimePeriod")
                if tp is not None and working == "0":
                    fd = tp.findtext(prefix + "FromDate"); td = tp.findtext(prefix + "ToDate")
                    if fd:
                        d0 = datetime.fromisoformat(fd).date(); d1 = datetime.fromisoformat(td).date() if td else d0
                        cur = d0
                        while cur <= d1:
                            feriados[cur.strftime('%Y-%m-%d')] = True; cur += timedelta(days=1)
    exc = elegido.find(prefix + "Exceptions")
    if exc is not None:  # excepciones forma nueva
        for e in exc.findall(prefix + "Exception"):
            if e.findtext(prefix + "DayWorking") == "0":
                tp = e.find(prefix + "TimePeriod")
                if tp is not None:
                    fd = tp.findtext(prefix + "FromDate"); td = tp.findtext(prefix + "ToDate")
                    if fd:
                        d0 = datetime.fromisoformat(fd).date(); d1 = datetime.fromisoformat(td).date() if td else d0
                        cur = d0
                        while cur <= d1:
                            feriados[cur.strftime('%Y-%m-%d')] = True; cur += timedelta(days=1)
    if not dias_idx:
        dias_idx = {0, 1, 2, 3, 4}
    return sorted(dias_idx), feriados, cal_default


def auditar_xml(file_or_bytes) -> AuditResult:
    """Parsea el XML MSPDI y devuelve un AuditResult autocontenido."""
    raw = _to_bytes(file_or_bytes)
    root = ET.fromstring(raw)
    prefix = root.tag.split("}")[0] + "}" if "}" in root.tag else ""
    title = root.find(prefix + "Title")
    project_name = title.text if (title is not None and title.text) else "Proyecto_Exportado"
    hours_per_day = 8.0
    h_pd_node = root.find(prefix + "MinutesPerDay")
    if h_pd_node is not None and h_pd_node.text:
        try:
            hours_per_day = float(h_pd_node.text) / 60.0
        except Exception:
            pass

    def parse_duration_days(dur_str):
        if not dur_str:
            return 0.0
        match = re.search(r'PT(\d+)H', dur_str)
        if match:
            return float(match.group(1)) / hours_per_day
        return 0.0

    def find_val(el, tag):
        x = el.find(prefix + tag)
        return x.text if x is not None else None

    tareas, uid_to_id, valid_ids = [], {}, []

    for task in root.iter(prefix + 'Task'):
        uid = find_val(task, 'UID')
        row_id = find_val(task, 'ID')
        active = find_val(task, 'Active')
        summary = find_val(task, 'Summary')
        if uid and row_id:
            uid_to_id[uid] = row_id
        if active != '0' and summary == '0' and row_id:
            try:
                valid_ids.append(int(row_id))
            except Exception:
                pass
    valid_ids.sort()

    for task in root.iter(prefix + 'Task'):
        active = find_val(task, 'Active')
        if active != '0':
            tid = int(find_val(task, 'ID') or 0)
            is_summary = (find_val(task, 'Summary') == '1')
            is_milestone = (find_val(task, 'Milestone') == '1')
            preds = []
            pred_links = []
            type_map = {'0': 'FF', '1': 'FS', '2': 'SF', '3': 'SS'}
            for link in task.findall(prefix + 'PredecessorLink'):
                p_uid = find_val(link, 'PredecessorUID')
                if p_uid:
                    pid = uid_to_id.get(p_uid, p_uid)
                    preds.append(pid)
                    ltype = type_map.get(find_val(link, 'Type') or '1', 'FS')
                    try:
                        pred_links.append((int(pid), ltype))
                    except Exception:
                        pass
            orig_preds = ", ".join(preds)
            errores = []
            if not is_summary and not is_milestone:
                constraint = int(find_val(task, 'ConstraintType') or '0')
                if not preds and tid > 1 and constraint <= 1:
                    prev = [x for x in valid_ids if x < tid]
                    sug = prev[-1] if prev else "N/A"
                    errores.append(f"Falta Predecesora (Sugerido ID {sug})")

            tareas.append({
                'ID': tid, 'Name': find_val(task, 'Name'), 'WBS': find_val(task, 'WBS'),
                'Start_XML': find_val(task, 'Start'), 'Finish_XML': find_val(task, 'Finish'),
                'Duration_Days': parse_duration_days(find_val(task, 'Duration')),
                'IsSummary': is_summary, 'IsMilestone': is_milestone,
                'OrigPreds': orig_preds, 'PredLinks': pred_links,
                'Errores': " | ".join(errores) if errores else "OK"
            })

    cal_dias, cal_feriados = None, {}
    try:
        cd, cf, _ = extraer_calendario_xml(root, prefix)
        cal_dias = cd
        cal_feriados = cf
    except Exception:
        cal_dias = None; cal_feriados = {}

    df = pd.DataFrame(tareas).sort_values('ID')
    return AuditResult(
        df=df, raw_bytes=raw, prefix=prefix, hours_per_day=hours_per_day,
        project_name=project_name, cal_dias=cal_dias, cal_feriados=cal_feriados,
    )


def generar_xml_ajustado(raw_bytes, prefix, df_final, hours_per_day):
    """Reescribe cada tarea del XML MSPDI con las fechas de CHRONOFLUX, preservando
    calendarios, dependencias y todo lo demás. [AUD-11] Modo Manual: las tareas hoja
    se marcan <Manual>1</Manual> para que Project respete las fechas sin recálculo.
    """
    NS = "http://schemas.microsoft.com/project"
    root = ET.fromstring(raw_bytes)
    by_id = {int(r['ID']): r for _, r in df_final.iterrows()}

    def fmt(d, hh, mm=0):
        return datetime.combine(d, dtime(hh, mm)).strftime('%Y-%m-%dT%H:%M:%S')

    def as_date(v):
        if v is None or isinstance(v, float):
            return None
        return v if hasattr(v, 'year') else datetime.fromisoformat(str(v)).date()

    for task in root.iter(prefix + 'Task'):
        idtxt = task.findtext(prefix + 'ID')
        if idtxt is None:
            continue
        try:
            tid = int(idtxt)
        except (TypeError, ValueError):
            continue
        if tid not in by_id:
            continue
        r = by_id[tid]
        is_sum = (task.findtext(prefix + 'Summary') == '1')
        is_mile = (task.findtext(prefix + 'Milestone') == '1')

        def setext(tag, val):
            el = task.find(prefix + tag)
            if el is None:
                el = ET.SubElement(task, prefix + tag)
            el.text = val

        d_ini = as_date(r.get('Inicio Nuevo'))
        d_fin = as_date(r.get('Fin Nuevo'))

        if is_sum:
            continue

        setext('Manual', '1')

        start_iso = fmt(d_ini, 8, 0) if d_ini else None
        finish_iso = fmt(d_fin, 17, 0) if d_fin else None

        if is_mile:
            ref = finish_iso or start_iso
            if ref:
                setext('Start', ref); setext('Finish', ref)
                setext('ManualStart', ref); setext('ManualFinish', ref)
                setext('EarlyStart', ref); setext('EarlyFinish', ref)
                setext('LateStart', ref); setext('LateFinish', ref)
                setext('Duration', 'PT0H0M0S')
                setext('ManualDuration', 'PT0H0M0S')
            continue

        if start_iso:
            setext('Start', start_iso)
            setext('ManualStart', start_iso)
            setext('EarlyStart', start_iso)
            setext('LateStart', start_iso)
        if finish_iso:
            setext('Finish', finish_iso)
            setext('ManualFinish', finish_iso)
            setext('EarlyFinish', finish_iso)
            setext('LateFinish', finish_iso)
        try:
            dnv = float(r.get('Duración Nueva'))
            horas = int(round(dnv * hours_per_day))
            dur_iso = f"PT{horas}H0M0S"
            setext('Duration', dur_iso)
            setext('ManualDuration', dur_iso)
        except (TypeError, ValueError):
            pass

    ET.register_namespace('', NS)

    try:
        fins = pd.to_datetime(df_final['Fin Nuevo'], errors='coerce').dropna()
        if len(fins):
            fd = root.find(prefix + 'FinishDate')
            if fd is not None:
                fd.text = pd.Timestamp(fins.max()).strftime('%Y-%m-%dT17:00:00')
    except Exception:
        pass
    return ET.tostring(root, encoding='UTF-8', xml_declaration=True)