diff --git a/data/KSP_data/log_book.xlsx b/data/KSP_data/log_book.xlsx index 4b75950..91659d3 100644 Binary files a/data/KSP_data/log_book.xlsx and b/data/KSP_data/log_book.xlsx differ diff --git a/src/KSP_tools/KSP_wiki/utils/__init__.py b/src/KSP_tools/KSP_wiki/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/KSP_tools/KSP_wiki/utils/cal_rss_timestamp.py b/src/KSP_tools/KSP_wiki/utils/cal_rss_timestamp.py new file mode 100644 index 0000000..6df5113 --- /dev/null +++ b/src/KSP_tools/KSP_wiki/utils/cal_rss_timestamp.py @@ -0,0 +1,25 @@ +import datetime + +t0 = datetime.datetime(year=1951, month=1, day=1, hour=0, minute=0, second=0, microsecond=0) +now_time = 290263084.07456899 + +def cal_rss_timestamp(offset_seconds) -> str: + """Calculate RSS timestamp given offset in seconds from 1951-01-01 00:00:00. + + Args: + offset_seconds (int): Number of seconds offset from the base time. + + Returns: + str: Formatted timestamp string in 'YYYY-MM-DD HH:MM:SS' format. + """ + target_time = t0 + datetime.timedelta(seconds=offset_seconds) + return target_time.strftime("%Y-%m-%d %H:%M:%S") + +def cal_launch_date_via_met(met): + met_time = now_time - met + return cal_rss_timestamp(met_time) + +if __name__ == "__main__": + print(cal_launch_date_via_met(23083334.171755522)) + print('done') + diff --git a/src/KSP_tools/KSP_wiki/utils/sync_to_overview.py b/src/KSP_tools/KSP_wiki/utils/sync_to_overview.py new file mode 100644 index 0000000..ef13a36 --- /dev/null +++ b/src/KSP_tools/KSP_wiki/utils/sync_to_overview.py @@ -0,0 +1,700 @@ +"""Sync a given sheet's log entries into the Overview sheet in local_content_dir/log_book.xlsx. + +Usage (from project root): + python -m src.utils.sync_to_overview --sheet XH-01 + +This script: +- Loads local_content_dir/log_book.xlsx +- Reads all rows from the specified sheet +- Applies half-month attribution rules to start/end dates +- Maps each entry onto the Overview sheet's timeline row for that sheet +- Merges cells from the start half-month to the end half-month +- Copies the fill color from the source log_name cell + +Assumptions about the workbook layout are documented in the docstrings below. +""" + +from dataclasses import dataclass +from datetime import date, datetime +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple +import xml.etree.ElementTree as ET +from openpyxl.workbook.workbook import Workbook as _Workbook + +from openpyxl import load_workbook +from openpyxl.styles import PatternFill, Alignment +from openpyxl.worksheet.worksheet import Worksheet +from openpyxl.cell.cell import Cell, MergedCell + +# Paths and constants +PROJECT_ROOT = Path(__file__).resolve().parents[2] +LOG_BOOK_PATH = PROJECT_ROOT / "local_content_dir" / "log_book.xlsx" + +OVERVIEW_SHEET_NAME = "Overview" +HEADER_ROW_INDEX = 1 +OVERVIEW_TIMELINE_START_COL = 3 # Column C by default (used for horizontal layout) + + +@dataclass +class LogEntry: + start: date + end: date + name: str + fill: Optional[PatternFill] + + +def _normalize_header_name(value: Optional[str]) -> Optional[str]: + if not isinstance(value, str): + return None + v = value.strip().lower().replace("-", "_").replace(" ", "_") + return v or None + + +def get_log_sheet_headers(ws: Worksheet, header_row: int = HEADER_ROW_INDEX) -> Dict[str, int]: + """Return a mapping from logical header names to 1-based column indices. + + Required logical names: start_date, end_date, log_name + Optional: color + """ + headers: Dict[str, int] = {} + for col in range(1, ws.max_column + 1): + raw = ws.cell(row=header_row, column=col).value + norm = _normalize_header_name(raw) + if not norm: + continue + if norm in ("start", "start_date", "start_time"): + headers["start_date"] = col + elif norm in ("end", "end_date", "end_time"): + headers["end_date"] = col + elif norm in ("log_name", "name", "title"): + headers["log_name"] = col + elif norm in ("color", "colour", "fill"): + headers["color"] = col + required = {"start_date", "end_date", "log_name"} + missing = required - headers.keys() + if missing: + raise ValueError(f"Missing required columns in sheet {ws.title!r}: {sorted(missing)}") + return headers + + +def _parse_excel_date(value) -> Optional[date]: + """Best-effort parse of an Excel cell value into a date. + + Supports datetime/date values and common string formats. + """ + if value is None or value == "": + return None + if isinstance(value, datetime): + return value.date() + if isinstance(value, date): + return value + if isinstance(value, (int, float)): + # We avoid depending on workbook epoch; numeric dates are less expected here. + # Fallback: treat as ordinal days from 1900-01-01 if needed in future. + return None + if isinstance(value, str): + text = value.strip() + if not text: + return None + for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d"): + try: + return datetime.strptime(text, fmt).date() + except ValueError: + continue + return None + + +def _apply_tint(argb_hex: str, tint: float) -> str: + """Apply Excel tint to an ARGB hex string and return ARGB hex. + + Tint formula (approximate): + - tint in (-1, 1) + - for each RGB channel (0-255): + if tint < 0: result = channel * (1.0 + tint) + else: result = channel + (255 - channel) * tint + + argb_hex may be 8-char (ARGB) or 6-char (RGB) -- we assume ARGB and preserve alpha. + Returns uppercase 8-char ARGB string. + """ + s = argb_hex.strip().lstrip("#") + if len(s) == 6: + alpha = 255 + rgb = s + elif len(s) == 8: + alpha = int(s[0:2], 16) + rgb = s[2:] + else: + return argb_hex.upper() + + try: + r = int(rgb[0:2], 16) + g = int(rgb[2:4], 16) + b = int(rgb[4:6], 16) + except Exception: + return argb_hex.upper() + + def mix(channel: int, t: float) -> int: + if t < 0: + v = int(round(channel * (1.0 + t))) + else: + v = int(round(channel + (255 - channel) * t)) + return max(0, min(255, v)) + + r2 = mix(r, tint) + g2 = mix(g, tint) + b2 = mix(b, tint) + + return f"{alpha:02X}{r2:02X}{g2:02X}{b2:02X}" + + +def _coerce_fill(value) -> Optional[PatternFill]: + """Convert various fill-like objects into a concrete solid PatternFill. + + Rules: + - Only convert when the source is a solid fill (or has an explicit fgColor rgb/index/theme) + - Do not convert transparent/empty colors (alpha == '00' or missing) to avoid black blocks + - Support Color.rgb (ARGB or RGB), Color.indexed via openpyxl.styles.colors.COLOR_INDEX + - If we can't find a clear solid color, return None so caller won't apply a fill + """ + if value is None: + return None + + # If already a concrete PatternFill with solid type and a usable fgColor, try to reuse + try: + if isinstance(value, PatternFill): + ptype = getattr(value, "patternType", None) or getattr(value, "fill_type", None) + fg = getattr(value, "fgColor", None) + if ptype and str(ptype).lower() in ("solid", "darkgray", "gray125", "gray0625") and fg is not None: + rgb = getattr(fg, "rgb", None) or getattr(fg, "value", None) + if isinstance(rgb, str) and rgb.strip(): + hex_str = rgb.strip().lstrip("#") + if len(hex_str) == 6: + hex_str = "FF" + hex_str + if len(hex_str) == 8 and not hex_str.startswith("00"): + return PatternFill(fill_type="solid", fgColor=hex_str) + # otherwise fallthrough + except Exception: + pass + + # If the object indicates no pattern (no fill), bail out to avoid assigning black + try: + ptype = getattr(value, "patternType", None) or getattr(value, "fill_type", None) + if ptype is None or str(ptype).strip().lower() in ("none", ""): + return None + except Exception: + pass + + # Generic proxy/object with color attributes + fg = getattr(value, "fgColor", None) or getattr(value, "start_color", None) + rgb = None + if fg is not None: + rgb = getattr(fg, "rgb", None) or getattr(fg, "value", None) + # try indexed + if rgb is None: + idx = getattr(fg, "indexed", None) + if isinstance(idx, int): + try: + from openpyxl.styles.colors import COLOR_INDEX + + hex_str = COLOR_INDEX[idx].lstrip("#") + if len(hex_str) == 6: + hex_str = "FF" + hex_str + # avoid default/black colors + if len(hex_str) == 8 and not hex_str.startswith("00") and hex_str.upper() != "FF000000": + return PatternFill(fill_type="solid", fgColor=hex_str) + except Exception: + pass + if isinstance(rgb, str) and rgb.strip(): + hex_str = rgb.strip().lstrip("#") + # normalize to ARGB + if len(hex_str) == 6: + hex_str = "FF" + hex_str + if len(hex_str) == 8 and not hex_str.startswith("00") and hex_str.upper() != "FF000000": + # apply tint if present + tint = getattr(fg, "tint", None) + if tint is not None: + try: + hex_str = _apply_tint(hex_str, float(tint)) + except Exception: + pass + return PatternFill(fill_type="solid", fgColor=hex_str) + # theme-based color (fg.theme) + theme_idx = getattr(fg, "theme", None) + if theme_idx is not None and _CURRENT_WB is not None: + try: + global _THEME_COLORS_CACHE + if _THEME_COLORS_CACHE is None: + _THEME_COLORS_CACHE = _load_theme_colors(_CURRENT_WB) + if isinstance(theme_idx, int) and _THEME_COLORS_CACHE and theme_idx < len(_THEME_COLORS_CACHE): + theme_hex = _THEME_COLORS_CACHE[theme_idx] + if theme_hex and theme_hex.upper() != "FF000000": + tint = getattr(fg, "tint", None) + if tint is not None: + try: + theme_hex = _apply_tint(theme_hex, float(tint)) + except Exception: + pass + return PatternFill(fill_type="solid", fgColor=theme_hex) + except Exception: + pass + + # As a last-ditch, the object itself might expose a string color + valstr = None + try: + if isinstance(value, str): + valstr = value + else: + valstr = str(value) + except Exception: + valstr = None + if isinstance(valstr, str): + s = valstr.strip().lstrip("#") + if len(s) == 6: + hex_str = "FF" + s + if hex_str.upper() != "FF000000": + return PatternFill(fill_type="solid", fgColor=hex_str) + if len(s) == 8 and not s.startswith("00") and s.upper() != "FF000000": + return PatternFill(fill_type="solid", fgColor=s) + + return None + + +def _load_theme_colors(wb: _Workbook) -> List[str]: + """Parse workbook.loaded_theme XML and return list of SRGB ARGB hex strings in scheme order. + + This is best-effort. We look for children and for each child prefer 'srgbClr' @val. + Returned hex strings are normalized to ARGB (8 chars, FF prefixed when needed). + """ + colors: List[str] = [] + try: + data = getattr(wb, "loaded_theme", None) + if data is None: + return colors + if isinstance(data, bytes): + data = data.decode("utf-8") + root = ET.fromstring(data) + # find any clrScheme element + clr = None + for node in root.findall('.//'): + tag = node.tag + if tag.endswith('clrScheme'): + clr = node + break + if clr is None: + return colors + for child in list(clr): + # find srgbClr descendant + srgb = child.find('.//') + val = None + # search for any element with 'srgbClr' in tag + for desc in child.iter(): + if desc.tag.endswith('srgbClr'): + val = desc.get('val') + break + if val: + s = val.strip().lstrip('#') + if len(s) == 6: + colors.append('FF' + s.upper()) + elif len(s) == 8: + colors.append(s.upper()) + else: + colors.append('') + else: + colors.append('') + except Exception: + return [] + return colors + + +def parse_log_row(ws: Worksheet, row_index: int, headers: Dict[str, int]) -> Optional[LogEntry]: + start_cell: Cell = ws.cell(row=row_index, column=headers["start_date"]) + end_cell: Cell = ws.cell(row=row_index, column=headers["end_date"]) + name_cell: Cell = ws.cell(row=row_index, column=headers["log_name"]) + + start = _parse_excel_date(start_cell.value) + end = _parse_excel_date(end_cell.value) + name_raw = name_cell.value + name = str(name_raw).strip() if name_raw is not None else "" + + if start is None and end is None and not name: + return None + + if start is None and end is not None: + start = end + if end is None and start is not None: + end = start + if start is None or end is None: + # Still not enough info + return None + + fill: Optional[PatternFill] = None + if "color" in headers: + color_cell = ws.cell(row=row_index, column=headers["color"]) + # Here we trust the user to put a proper hex like RRGGBB; minimal handling only. + if isinstance(color_cell.value, str) and color_cell.value.strip(): + hex_str = color_cell.value.strip().lstrip("#") + if len(hex_str) in (6, 8): + # openpyxl expects ARGB; if RGB given, prepend FF + if len(hex_str) == 6: + hex_str = "FF" + hex_str + fill = PatternFill(fill_type="solid", fgColor=hex_str) + if fill is None: + # annotate name_cell.fill access for static checkers + nf = name_cell.fill + fill = _coerce_fill(nf) + if nf is not None and fill is None: + # Debug info: collect fgColor attributes to help diagnose why coercion failed + try: + fg = getattr(nf, "fgColor", None) or getattr(nf, "start_color", None) + rgb = getattr(fg, "rgb", None) if fg is not None else None + theme = getattr(fg, "theme", None) if fg is not None else None + indexed = getattr(fg, "indexed", None) if fg is not None else None + tint = getattr(fg, "tint", None) if fg is not None else None + print(f"[sync_to_overview] sheet={ws.title!r} row={row_index} log_name={name!r} had fill proxy but coercion failed: fg.rgb={rgb!r} fg.theme={theme!r} fg.indexed={indexed!r} fg.tint={tint!r}") + except Exception: + print(f"[sync_to_overview] sheet={ws.title!r} row={row_index} log_name={name!r} had fill proxy but coercion failed (unable to inspect attributes)") + + if not name: + # If there is no name but valid dates, we still can render a colored block; use empty string. + name = "" + + return LogEntry(start=start, end=end, name=name, fill=fill) + + +def _shift_month(d: date, delta: int) -> date: + """Shift date by delta months, keeping day when possible. + + If the target month has fewer days, clamp to last day of month. + """ + month = d.month - 1 + delta + year = d.year + month // 12 + month = month % 12 + 1 + # Clamp day to last day of target month + from calendar import monthrange + + last_day = monthrange(year, month)[1] + day = min(d.day, last_day) + return date(year, month, day) + + +def attribute_start_date(d: date) -> date: + """Map start date to its month start. + + New rule: the start month is always the month of the original start date. + Returns a date representing the first day of that month (for timeline bucket comparisons). + """ + return date(d.year, d.month, 1) + + +def attribute_end_date(d: date) -> date: + """Map end date to its month start. + + New rule: the end month is the month containing the actual end date (no half-month cutoff). + Returns the first day of that month for timeline comparisons. + """ + return date(d.year, d.month, 1) + + +# --- New: support both horizontal (timeline across a header row) and vertical (timeline down column A) --- + +def _detect_overview_orientation(ws: Worksheet, header_row: int = HEADER_ROW_INDEX, start_col: int = OVERVIEW_TIMELINE_START_COL) -> str: + """Detect whether Overview worksheet uses a horizontal timeline (dates across header_row) + or a vertical timeline (dates down column A). Returns 'horizontal' or 'vertical'. + Simple heuristic: if column A (starting at row 2) contains many dates -> vertical. + Otherwise if header_row contains many dates across columns starting at start_col -> horizontal. + Defaults to 'horizontal'. + """ + # check column A for dates + col_a_date_count = 0 + for r in range(header_row + 1, min(ws.max_row, header_row + 200) + 1): + v = ws.cell(row=r, column=1).value + if _parse_excel_date(v) is not None: + col_a_date_count += 1 + if col_a_date_count >= 3: + return "vertical" + # check header row for dates across columns + row_date_count = 0 + for c in range(start_col, min(ws.max_column, start_col + 200) + 1): + v = ws.cell(row=header_row, column=c).value + if _parse_excel_date(v) is not None: + row_date_count += 1 + if row_date_count >= 3: + return "horizontal" + return "horizontal" + + +def get_overview_timeline( + ws: Worksheet, + header_row: int = HEADER_ROW_INDEX, + start_col: int = OVERVIEW_TIMELINE_START_COL, +) -> Tuple[str, List[Tuple[int, date]]]: + """Extract overview timeline. + + Returns (orientation, timeline) where orientation is 'horizontal' or 'vertical'. + - horizontal: timeline is list of (col_index, date) scanning header_row from start_col rightwards + - vertical: timeline is list of (row_index, date) scanning column A from header_row+1 downward + """ + orientation = _detect_overview_orientation(ws, header_row, start_col) + timeline: List[Tuple[int, date]] = [] + + if orientation == "horizontal": + empty_streak = 0 + max_empty_streak = 10 + for col in range(start_col, ws.max_column + 1): + cell = ws.cell(row=header_row, column=col) + d = _parse_excel_date(cell.value) + if d is None: + empty_streak += 1 + if empty_streak >= max_empty_streak: + break + continue + empty_streak = 0 + timeline.append((col, d)) + timeline.sort(key=lambda x: x[1]) + return orientation, timeline + + # vertical + empty_streak = 0 + max_empty_streak = 10 + for row in range(header_row + 1, ws.max_row + 1): + cell = ws.cell(row=row, column=1) + d = _parse_excel_date(cell.value) + if d is None: + empty_streak += 1 + if empty_streak >= max_empty_streak: + break + continue + empty_streak = 0 + timeline.append((row, d)) + timeline.sort(key=lambda x: x[1]) + return orientation, timeline + + +def find_bucket_span( + timeline: List[Tuple[int, date]], start_attr: date, end_attr: date +) -> Optional[Tuple[int, int]]: + if not timeline: + return None + # Match by year-month to be robust when timeline cells are mid-month + start_key = (start_attr.year, start_attr.month) + end_key = (end_attr.year, end_attr.month) + + ym_list: List[Tuple[int, Tuple[int, int]]] = [(idx, (d.year, d.month)) for idx, d in timeline] + + first_pos = None + for i, (_, ym) in enumerate(ym_list): + if ym >= start_key: + first_pos = i + break + if first_pos is None: + return None + + last_pos = None + for i in range(len(ym_list) - 1, -1, -1): + if ym_list[i][1] <= end_key: + last_pos = i + break + if last_pos is None or last_pos < first_pos: + return None + + return timeline[first_pos][0], timeline[last_pos][0] + + +def _has_merged_intersection(ws: Worksheet, min_row: int, max_row: int, min_col: int, max_col: int) -> bool: + """Return True if any existing merged cell range intersects the supplied rectangle.""" + for merged in ws.merged_cells.ranges: + if merged.min_row <= max_row and merged.max_row >= min_row and merged.min_col <= max_col and merged.max_col >= min_col: + return True + return False + + +def _unmerge_intersecting_ranges(ws: Worksheet, min_row: int, max_row: int, min_col: int, max_col: int) -> None: + to_unmerge = [] + for merged in list(ws.merged_cells.ranges): + if merged.min_row <= max_row and merged.max_row >= min_row and merged.min_col <= max_col and merged.max_col >= min_col: + to_unmerge.append(merged) + for merged in to_unmerge: + ws.unmerge_cells(str(merged.coord)) + + +def clear_overview_range(ws: Worksheet, sheet_pos: int, indices: List[int], orientation: str) -> None: + """Clear the destination cells for a given sheet in the Overview. + + - For horizontal orientation: sheet_pos is a row index, indices are column indices. + - For vertical orientation: sheet_pos is a column index, indices are row indices. + """ + if not indices: + return + min_idx = min(indices) + max_idx = max(indices) + + if orientation == "horizontal": + min_col = min_idx + max_col = max_idx + row_index = sheet_pos + # Unmerge any merged cells intersecting this row & column range + _unmerge_intersecting_ranges(ws, row_index, row_index, min_col, max_col) + for col in range(min_col, max_col + 1): + cell: Any = ws.cell(row=row_index, column=col) + cell.value = None + cell.fill = PatternFill() # clear fill + else: + # vertical orientation: we clear a column at sheet_pos between min_idx..max_idx rows + col_index = sheet_pos + min_row = min_idx + max_row = max_idx + _unmerge_intersecting_ranges(ws, min_row, max_row, col_index, col_index) + for row in range(min_row, max_row + 1): + cell: Any = ws.cell(row=row, column=col_index) + cell.value = None + cell.fill = PatternFill() # clear fill + + +def apply_log_to_overview( + ws: Worksheet, + sheet_pos: int, + start_idx: int, + end_idx: int, + log_name: str, + fill: Optional[PatternFill], + orientation: str, +) -> None: + """Apply a log entry to the Overview. + + - For horizontal: sheet_pos is row, start_idx/end_idx are columns -> merge horizontally (single row). + - For vertical: sheet_pos is column, start_idx/end_idx are rows -> merge vertically (single column). + """ + if start_idx > end_idx: + start_idx, end_idx = end_idx, start_idx + + # Conflict resolution: if any existing merged range intersects our intended target, shrink the start index forward + if orientation == "horizontal": + # target rectangle: (sheet_pos, start_idx..end_idx) + while start_idx <= end_idx and _has_merged_intersection(ws, sheet_pos, sheet_pos, start_idx, end_idx): + start_idx += 1 + if start_idx > end_idx: + return + if start_idx == end_idx: + cell: Any = ws.cell(row=sheet_pos, column=start_idx) + else: + ws.merge_cells(start_row=sheet_pos, start_column=start_idx, end_row=sheet_pos, end_column=end_idx) + cell: Any = ws.cell(row=sheet_pos, column=start_idx) + else: + # vertical: target rectangle: (start_idx..end_idx, sheet_pos) + while start_idx <= end_idx and _has_merged_intersection(ws, start_idx, end_idx, sheet_pos, sheet_pos): + start_idx += 1 + if start_idx > end_idx: + return + if start_idx == end_idx: + cell: Any = ws.cell(row=start_idx, column=sheet_pos) + else: + ws.merge_cells(start_row=start_idx, start_column=sheet_pos, end_row=end_idx, end_column=sheet_pos) + cell: Any = ws.cell(row=start_idx, column=sheet_pos) + + # Defensive: if we somehow got a MergedCell (read-only), find the actual top-left cell + try: + if isinstance(cell, MergedCell): + for merged in ws.merged_cells.ranges: + if merged.min_row <= cell.row <= merged.max_row and merged.min_col <= cell.column <= merged.max_col: + cell = ws.cell(row=merged.min_row, column=merged.min_col) + break + except Exception: + # ignore and continue + pass + + cell.value = log_name + cell.alignment = Alignment(horizontal="center", vertical="center") # type: ignore[attr-defined] + if fill is not None: + cell.fill = fill # type: ignore[attr-defined] + + +def _find_overview_position_for_sheet(ws: Worksheet, sheet_name: str, orientation: str) -> Optional[int]: + """Find where the sheet's overview cells live. + + - horizontal orientation: sheet names are expected in column A (like before) -> return the row index + - vertical orientation: sheet names are expected in header row (ROW 1) across columns -> return the column index + """ + if orientation == "horizontal": + for row in range(2, ws.max_row + 1): # assume header in row 1 + val = ws.cell(row=row, column=1).value + if isinstance(val, str) and val.strip() == sheet_name: + return row + return None + # vertical + for col in range(2, ws.max_column + 1): # assume column A is timeline, so sheet headers start at B + val = ws.cell(row=HEADER_ROW_INDEX, column=col).value + if isinstance(val, str) and val.strip() == sheet_name: + return col + return None + + +def sync_to_overview(source_sheet_name: str, workbook_path: Optional[Path] = None) -> None: + global _CURRENT_WB, _THEME_COLORS_CACHE + wb_path = Path(workbook_path) if workbook_path is not None else LOG_BOOK_PATH + if not wb_path.exists(): + raise FileNotFoundError(f"Workbook not found: {wb_path}") + + wb = load_workbook(wb_path, data_only=False) + _CURRENT_WB = wb + _THEME_COLORS_CACHE = None + + if source_sheet_name not in wb.sheetnames: + raise KeyError(f"Sheet not found: {source_sheet_name}") + if OVERVIEW_SHEET_NAME not in wb.sheetnames: + raise KeyError(f"Overview sheet not found: {OVERVIEW_SHEET_NAME}") + + ws_src = wb[source_sheet_name] + ws_ov = wb[OVERVIEW_SHEET_NAME] + + orientation, timeline = get_overview_timeline(ws_ov) + timeline_indices = [idx for idx, _ in timeline] + + overview_pos = _find_overview_position_for_sheet(ws_ov, source_sheet_name, orientation) + if overview_pos is None: + if orientation == "horizontal": + raise ValueError(f"Could not find row in Overview with name {source_sheet_name!r} in column A") + else: + raise ValueError(f"Could not find column in Overview with header {source_sheet_name!r} in row 1") + + headers = get_log_sheet_headers(ws_src) + + # Clear the destination range first + clear_overview_range(ws_ov, overview_pos, timeline_indices, orientation) + + for row in range(HEADER_ROW_INDEX + 1, ws_src.max_row + 1): + entry = parse_log_row(ws_src, row, headers) + if entry is None: + continue + start_attr = attribute_start_date(entry.start) + end_attr = attribute_end_date(entry.end) + if end_attr < start_attr: + end_attr = start_attr + span = find_bucket_span(timeline, start_attr, end_attr) + if span is None: + continue + idx_start, idx_end = span + apply_log_to_overview(ws_ov, overview_pos, idx_start, idx_end, entry.name, entry.fill, orientation) + + wb.save(wb_path) + # clear global + _CURRENT_WB = None + _THEME_COLORS_CACHE = None + + +def main(sheet: str, workbook: Optional[str] = None) -> None: + """Run sync_to_overview with explicit parameters. + + Args: + sheet: source sheet name (e.g. 'XH-01') + workbook: path to workbook file; if None, use default LOG_BOOK_PATH + """ + wb_path = Path(workbook) if workbook is not None else LOG_BOOK_PATH + sync_to_overview(sheet, wb_path) + + +if __name__ == "__main__": + # Configure these values here and call main directly. + # Edit these lines to change which sheet / workbook are processed. + sheet_name = "XH-01" + workbook_path = str(LOG_BOOK_PATH) + main(sheet_name, workbook_path) diff --git a/src/KSP_tools/KSP_wiki/utils/upload_log_book_to_db.py b/src/KSP_tools/KSP_wiki/utils/upload_log_book_to_db.py new file mode 100644 index 0000000..c53f5ab --- /dev/null +++ b/src/KSP_tools/KSP_wiki/utils/upload_log_book_to_db.py @@ -0,0 +1,151 @@ +"""Upload specified sheets from local_content_dir/log_book.xlsx into the database table public.log_book. + +Behavior: +- Reads each sheet in the provided list (currently set to ["XH-01"]). +- Uses the first row of each sheet as column names; expected to match the existing DB table columns. +- Appends a fixed column "vehicle" with the value equal to the sheet name. +- Skips fully-empty rows. + +Notes: +- The script uses src.pg_handler.db_handler.PgHandler and src.config.load_config.project_config for DB connection info. +- This script performs INSERTs; run with a backup of the workbook if needed. +""" +import os + +from pathlib import Path +from typing import List, Optional +from datetime import datetime, date +from openpyxl import load_workbook + +from src.PostgresSQL.db_handler import PgHandler +from src.config.load_config import project_config + +PROJECT_ROOT = Path(__file__).resolve().parents[4] +KSP_DATA_DIR = os.path.join(PROJECT_ROOT, "data", "KSP_data") + +def _parse_excel_date(value): + if value is None or value == "": + return None + if isinstance(value, datetime): + return value.date() + if isinstance(value, date): + return value + if isinstance(value, (int, float)): + return None + if isinstance(value, str): + text = value.strip() + if not text: + return None + for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d", "%Y-%m-%d %H:%M:%S"): + try: + return datetime.strptime(text, fmt).date() + except Exception: + continue + return None + + +def _read_sheet(ws): + """Read header row and data rows from worksheet. Returns (headers, rows). + + headers: list of header names (strings) + rows: list of lists of cell values (preserving original types where possible) + """ + headers: List[str] = [] + for col in range(1, ws.max_column + 1): + raw = ws.cell(row=1, column=col).value + if raw is None: + headers.append("") + else: + headers.append(str(raw).strip()) + + data_rows: List[List[Optional[object]]] = [] + for r in range(2, ws.max_row + 1): + row_vals: List[Optional[object]] = [] + non_empty = False + for c in range(1, ws.max_column + 1): + v = ws.cell(row=r, column=c).value + if isinstance(v, (datetime, date)): + v = _parse_excel_date(v) + elif isinstance(v, str): + v = v.strip() + if v not in (None, ""): + non_empty = True + row_vals.append(v) + if non_empty: + data_rows.append(row_vals) + return headers, data_rows + + +def upload_sheets(sheet_names: List[str], workbook_path) -> None: + wb_path = Path(workbook_path) + if not wb_path.exists(): + raise FileNotFoundError(f"Workbook not found: {wb_path}") + + wb = load_workbook(wb_path, data_only=True) + + # DB connection info from project_config + host = project_config.get('db_host') + port = project_config.get('db_port') + user = project_config.get('db_user') + password = project_config.get('db_password') + database = project_config.get('db_name') + + with PgHandler(host=host, port=port, user=user, password=password or None, database=database) as pg: + conn = pg._get_conn() + cur = conn.cursor() + + for sheet_name in sheet_names: + if sheet_name not in wb.sheetnames: + print(f"Sheet not found in workbook: {sheet_name}, skipping") + continue + ws = wb[sheet_name] + headers, rows = _read_sheet(ws) + if not headers or not any(h for h in headers): + print(f"No headers found in sheet {sheet_name}, skipping") + continue + + # Prepare columns: take non-empty header names only + cols = [h for h in headers if h] + if not cols: + print(f"Sheet {sheet_name} has no named columns, skipping") + continue + + # final columns include vehicle + db_columns = cols + ["vehicle"] + col_sql = ", ".join([f'"{c}"' for c in db_columns]) + placeholders = ", ".join(["%s"] * len(db_columns)) + insert_sql = f'INSERT INTO "public"."log_book" ({col_sql}) VALUES ({placeholders})' + + # Build values: map each row (which may contain more columns than headers length) to cols + values_to_insert = [] + for row_vals in rows: + # Truncate/extend row_vals to match number of defined headers + mapped = [] + for i, col_name in enumerate(cols): + if i < len(row_vals): + mapped.append(row_vals[i]) + else: + mapped.append(None) + mapped.append(sheet_name) # vehicle + values_to_insert.append(tuple(mapped)) + + if not values_to_insert: + print(f"No data rows for sheet {sheet_name}, skipping insert") + continue + + try: + cur.executemany(insert_sql, values_to_insert) + conn.commit() + print(f"Inserted {len(values_to_insert)} rows from sheet {sheet_name} into log_book") + except Exception as e: + conn.rollback() + print(f"Failed to insert rows from sheet {sheet_name}: {e}") + + cur.close() + + +if __name__ == "__main__": + WORKBOOK_PATH = os.path.join(KSP_DATA_DIR, 'log_book.xlsx') + SHEETS_TO_UPLOAD: List[str] = ["XH-01"] + upload_sheets(SHEETS_TO_UPLOAD, WORKBOOK_PATH) + diff --git a/src/KSP_tools/KSP_wiki/vessel_log/Stellaria_history.py b/src/KSP_tools/KSP_wiki/vessel_log/Stellaria_history.py new file mode 100644 index 0000000..4a92a30 --- /dev/null +++ b/src/KSP_tools/KSP_wiki/vessel_log/Stellaria_history.py @@ -0,0 +1,101 @@ +operating_history_raw = """ +2050-09-15 First module delivered to LEO via Vulture Shuttle +2053-02-02 Assembly Halted + +2053-05-11 Reconfiguring to ST-01 Stellaria + Only fitted with the Light Speed Engine. Other systems are still Xihe standard. Awaiting further upgrades. +2054-4-28 Reconfiguration Complete + +2054-05-12 Begin Trial Run in Earth Orbit, testing all systems +2054-08-22 Trial Run complete, all systems perform as expected. Including multiple run to GEO and Lunar Orbit before returning to LEO. + +2054-09-07 Begin Deep Space Test Mission + Uncrewed mission to Earth-Sun Lagrange Point 2 to test long-duration operations +2054-11-28 Test Mission Complete, all systems nominal. Returning to LEO. + +2055-12-01 Start maintenance +2055-02-28 Maintenance complete, all systems operational, ready for commissioning. + +2055-03-20 Picking up crew for Inner Solar System Exploration Mission 1 +2055-03-22 Earth Departure for Mecury + Use standard trajectory with 127km/s of departure burn +2055-04-17 Mercury Arrival + First crewed mission to Mercury + 2055-04-18 Orbital Survey of Mercury begins + 2055-06-14 First Crewed Surface Landing on Mercury + 2055-06-15 EVA operations commence + 2055-06-30 Surface operations complete, returning to orbit +2055-08-15 Departure from Mercury for Venus + Use standard trajectory with 85km/s of departure burn +2055-10-01 Venus Arrival + First crewed mission to Venus, no crewed landing attempted due to extreme surface conditions. Probe deployment only. + 2055-10-02 Orbital Survey of Venus begins + 2055-10-10 Landing Probe Deployment + 2055-11-10 Begin Upper Atmosphere Survey via Callisto MPV +2056-03-24 Departure from Venus for Earth + Use standard trajectory with 75km/s of departure burn +2056-04-12 Earth Arrival + +2056-04-15 Start maintenance +2056-06-01 Maintenance complete. + +2056-06-02 Picking up crew for Mars One Construction Mission 2 +2056-06-04 Picking up Mars One construction cargo +2056-06-06 Earth Departure for Mars + Use standard trajectory with 126km/s of departure burn +2056-07-01 Mars Arrival, First crewed mission to Mars using Stellaria + 2056-07-03 Crew descend to Martian Surface, Continuing the construction of Mars One Base + 2056-07-20 Heavy Cargo Delivery to Surface via Callisto MPV + 2056-09-17 Mars One Construction Mission 3 Crew Arrival via XH-01, joining Stellaria crew. First Crew Handover on Mars. + 2056-10-12 Mars One Construction Mission 2 Crew ascend to Stellaria +2056-10-15 Departure from Mars for Earth + Use Fast Return trajectory with 212km/s of departure burn +2056-11-03 Earth Arrival + +2056-11-04 Start maintenance +2056-12-10 Maintenance complete. + +2056-12-15 Start preparations for Europa Outpost Construction Mission 1 +2056-12-20 Europa Outpost Construction Package arrival via Qingtian Cargo Shuttle +2056-12-31 Picking up crew for Europa Outpost Construction Mission 1 +2057-01-02 Earth Departure for Europa + Use standard trajectory with 165km/s of departure burn + + + + +""" + + + + +operating_history = [ + { + "log_name": "Construction", + "start_time": "2050-09-15", + "end_time": "2053-02-02", + "mission_detail": "Construction", + "sub_log": + [ + "2050-09-15 First module delivered to LEO via Vulture Shuttle", + "2053-02-02 Assembly Halted", + ] + }, + { + "log_name": "Reconfiguring to ST-01 Stellaria", + "start_time": "2053-05-11", + "end_time": "2054-4-28", + "mission_detail": "Only fitted with the Light Speed Engine. Other systems are still Xihe standard. Awaiting further upgrades.", + "sub_log":[] + }, + { + 'log_name': "Trial Run", + 'start_time': "2054-05-12", + 'end_time': "2054-8-22", + 'mission_detail': "Trial Run before commissioning", + 'sub_log': [ + "2054-05-12 Begin Trial Run in Earth Orbit, testing all systems", + "2054-08-22 Trial Run complete, all systems perform as expected. Including multiple run to GEO and Lunar Orbit before returning to LEO.", + ] + } +] \ No newline at end of file diff --git a/src/KSP_tools/KSP_wiki/vessel_log/xihe_agent.py b/src/KSP_tools/KSP_wiki/vessel_log/xihe_agent.py new file mode 100644 index 0000000..50402cc --- /dev/null +++ b/src/KSP_tools/KSP_wiki/vessel_log/xihe_agent.py @@ -0,0 +1,323 @@ +import json +import os +import openai + +from pathlib import Path +from openai import OpenAI +from langchain_core.messages import HumanMessage, AIMessage + +from src.config.load_config import LLM, project_config + +PROJECT_ROOT = Path(__file__).parent.parent.parent.parent +DATA_PATH = os.path.join(PROJECT_ROOT, "data") + +xihe_main_wiki_path = os.path.join(PROJECT_ROOT,"local_content_dir","Exploration_Motherships","Xihe", "Xihe.md") +with open(xihe_main_wiki_path, "r", encoding="utf-8") as f: + XIHE_MAIN_WIKI_CONTEXT = f.read() + +with open(os.path.join(PROJECT_ROOT, "local_content_dir", "meta_info.md"), "r", encoding="utf-8") as f: + meta_info_content = f.read() + +user_input = """ +请根据给到你的材料,撰写羲和号(XH-01)的维基百科条目内容。 +你需要撰写多个章节,涵盖羲和号的各个方面。每个章节应当非常详细,内容应当丰富且专业。 +你应当根据提供的信息进行适度的扩展,但不可以脱离提供的信息范围。 +你应当保持中立的语气,避免使用主观评价性的语言。 +运行历史应当作为单独章节列出,时间线应当尽可能完整。 + +以下是羲和号的运行历史: +[ +{ + "log_name": "Construction", + "start_time": "2050-03-15", + "end_time": "2052-05-31", + "mission_detail": "Xihe XH-01 Construction", + "sub_log": + [ + "2050-03-15 First module delivered to LEO via Vulture Shuttle", + "2052-02-06 Assembly completed in LEO", + ] +}, +{ + "log_name": "Trial Run in Earth Orbit", + "start_time": "2052-06-01", + "end_time": "2052-11-30", + "mission_detail": "Trial Run before commissioning", + "sub_log": [] +}, +{ + "log_name": "Deep Space Test Mission", + "start_time": "2053-01-12", + "end_time": "2053-12-18", + "mission_detail": "Long term uncrewed test", + "sub_log": + [ + "2053-01-12 Deep Space Test Mission, uncrewed mission to Earth-Sun L2 point to test deep space operations", + "2053-12-18 Back to LEO" + ] +}, +{ + "log_name": "Maintenance and upgrades", + "start_time": "2054-01-07", + "end_time": "2054-06-28", + "mission_detail": "", + "sub_log": [] +}, +{ + "log_name": "Mars Mission Outbound", + "start_time": "2054-07-10", + "end_time": "2054-08-21", + "mission_detail": "Earth-Mars transfer using standard trajectory with 102km/s of departure burn", + "sub_log": [] +}, +{ + "log_name": "Mars Mission Demo at Mars", + "start_time": "2054-08-21", + "end_time": "2055-01-10", + "mission_detail": "Uncrewed, Deliver the first Mars-One base construction equipment package to Mars orbit.", + "sub_log": + [ + "2054-08-21 Mars Orbit Arrival, deploy equipment package to Mars surface via lander", + "2055-01-10 Mars Departure, return to Earth" + ] +}, +{ + "log_name": "Mars Mission Demo Return to Earth", + "start_time": "2055-01-10", + "end_time": "2055-02-28", + "mission_detail": "Mars-Earth transfer using standard trajectory with 97km/s of departure burn", + "sub_log": [] +}, +{ + "log_name": "Maintenance and upgrades", + "start_time": "2055-03-01", + "end_time": "2055-05-14", + "mission_detail": "", + "sub_log": [] +}, +{ + "log_name": "Mars One base construction mission 1 outbound", + "start_time": "2055-06-15", + "end_time": "2055-07-30", + "mission_detail": "Earth-Mars transfer using standard trajectory with 115km/s of departure burn", + "sub_log": [ + "2055-06-15 Picking up crew for Mars One base construction mission 1", + "2055-06-18 Earth Departure for Mars" + ] +}, +{ + "log_name": "Mars One base construction mission 1 at Mars", + "start_time": "2055-07-30", + "end_time": "2056-02-07", + "mission_detail": "First Mars One base construction mission, begin base construction", + "sub_log": [ + "2055-07-30 Mars Arrival, begin base construction", + "2056-02-07 Mars Departure, return to Earth" + ] +}, +{ + "log_name": "Mars One base construction mission 1 Return to Earth", + "start_time": "2056-02-07", + "end_time": "2056-03-18", + "mission_detail": "Mars-Earth transfer using standard trajectory with 122km/s of departure burn", + "sub_log": [ + "2056-02-07 Mars Departure, return to Earth. First to use quick return trajectory.", + "2056-03-18 Earth Arrival, crew return" + ] +}, +{ + "log_name": "Maintenance and upgrades", + "start_time": "2056-03-19", + "end_time": "2056-06-20", + "mission_detail": "", + "sub_log": [] +}, +{ + "log_name": "Mars One base construction mission 3 outbound", + "start_time": "2056-07-30", + "end_time": "2056-09-15", + "mission_detail": "Earth-Mars transfer using standard trajectory with 95km/s of departure burn", + "sub_log": [] +}, +{ + "log_name": "Mars One base construction mission 3 at Mars", + "start_time": "2056-09-15", + "end_time": "2057-02-12", + "mission_detail": "Third Mars One base construction. First crew hand-off on Mars(with Mars One base construction 2).", + "sub_log": [ + "2056-09-15 Mars Arrival, continue base construction", + "2057-02-12 Mars Departure, return to Earth." + ] +}, +{ + "log_name": "Mars One base construction mission 3 Return to Earth", + "start_time": "2057-02-12", + "end_time": "2057-03-25", + "mission_detail": "Mars-Earth transfer using standard trajectory with 122km/s of departure burn", + "sub_log": [ + "2057-02-12 Mars Departure, return to Earth.", + "2057-03-25 Earth Arrival, crew return." + ] +}, +{ + "log_name": "Maintenance", + "start_time": "2057-03-26", + "end_time": "2057-04-25", + "mission_detail": "", + "sub_log": [] +}, +{ + "log_name": "Mars Expedition 2 Outbound", + "start_time": "2057-05-05", + "end_time": "2057-06-22", + "mission_detail": "Earth-Mars transfer using standard trajectory with 110km/s of departure burn", + "sub_log": [ + "2057-05-05 Picking up crew for Mars Expedition 2", + "2057-05-09 Earth Departure for Mars" + ] +}, +{ + "log_name": "Mars Expedition 2 at Mars", + "start_time": "2057-06-22", + "end_time": "2058-04-18", + "mission_detail": "Second long term Mars Expedition, first mission to Mars One base after the construction is completed. First mission involving space tourism to Mars.", + "sub_log": [ + "2057-06-22 Mars Arrival", + "2057-06-25 Mars EDL and surface operations begin", + "2058-04-10 Begin Mars Departure preparations", + "2058-04-18 Mars Departure for Earth" + ] +}, +{ + "log_name": "Mars Expedition 2 Return to Earth", + "start_time": "2058-04-18", + "end_time": "2058-05-20", + "mission_detail": "Mars-Earth transfer using standard trajectory with 115km/s of departure burn", + "sub_log": [ + "2058-04-18 Mars Departure for Earth", + "2058-05-20 Earth Arrival, crew return" + ] +}, +{ + "log_name": "Maintenance", + "start_time": "2058-05-21", + "end_time": "2058-06-18", + "mission_detail": "", + "sub_log": [] +}, +{ + "log_name": "Europa Expedition 1 Outbound", + "start_time": "2058-07-10", + "end_time": "2058-09-25", + "mission_detail": "Earth-Jupiter transfer using standard trajectory with 160km/s of departure burn", + "sub_log": [ + "2058-07-10 Picking up crew for Europa Expedition 1", + "2058-07-12 Earth Departure for Jupiter system", + "2058-09-25 Europa Arrival" + ] +}, +{ + "log_name": "Europa Expedition 1 at Jupiter System", + "start_time": "2058-09-25", + "end_time": "2059-07-10", + "mission_detail": "First standard crew expedition to Europa Outpost to begin exploration of Jupiter system Long term surface operation at Europa Outpost, followed by exploration of IO and Ganymede", + "sub_log": [ + "2058-09-25 Europa Arrival", + "2058-09-27 Europa descend, begin surface operation at Europa Outpost", + "2058-12-15 Europa ascend, begin Jupiter system exploration", + "2058-12-20 Begin IO exploration", + "2059-01-20 Return to Europa Outpost", + "2059-06-20 Begin Europa Outpost departure preparations", + "2059-06-25 Europa Expedition 1 crew ascend from Europa Outpost", + "2059-07-05 begin Ganymede exploration", + "2059-07-10 Departure for Earth, upgrade planned." + ] +}, +{ + "log_name": "Europa Expedition 1 Return to Earth", + "start_time": "2059-07-10", + "end_time": "2059-11-15", + "mission_detail": "Uses standard trajectory with 112km/s of departure burn Parked to Star Port", + "sub_log": [ + "2059-07-10 Departure for Earth.", + "2059-11-15 Earth Arrival, Park to Star Port." + ] +}, +{ + "log_name": "Maintenance and upgrade", + "start_time": "2059-11-16", + "end_time": "2060-02-19", + "mission_detail": "Greenhouse upgrade to support long term deep space missions", + "sub_log": [] +}, +{ + "log_name": "Europa Expedition 3 Outbound", + "start_time": "2060-02-20", + "end_time": "2060-05-10", + "mission_detail": "Uses standard trajectory with 163km/s of departure burn", + "sub_log": [ + "2060-02-20 Picking up Europa Expedition 3 crew, On its way to explore Jupiter systems before sending crew to Europa Outpost", + ] +}, +] +""" + +prompt = f""" +# 角色 +你是一个航天领域的维基百科撰写专家,精通航天技术和科学。你熟悉维基百科的格式和风格指南,能够撰写符合其标准的内容。 + +# 任务 +你的任务是根据用户提供的信息,帮助用户以维基百科的风格撰写相关的航天器维基百科条目内容。 + +# 注意点 +- 交给你的信息是一个架空世界观下的航天器信息,请基于这些信息撰写内容,避免引入现实世界的航天器知识。 +- 你需要根据提供的信息撰写多个章节,涵盖航天器的各个方面。 +- 你应当根据提供的信息进行适度的扩展,但不可以脱离提供的信息范围。 +- 你应当保持中立的语气,避免使用主观评价性的语言。 +- 你应当以markdown格式输出内容,内容应当仿照维基百科的模式,分成多个章节。 +- 不要提及航天器的研发机构或国家背景,这是一个架空世界观下的航天器,因此不涉及现实世界的国家或组织,也不要编造一个组织或者机构。 +- 对于载具的名称,如果能够在提供的信息中找到对应的中文名称,应当尽量使用中文名称,否则应当遵循原材料中的名称。 +- 避免出现外部链接,所有的超链接都应当是指向维基百科内部的其他条目。 + +# 输出格式要求 +你应当以markdown格式输出内容,内容应当仿照维基百科的模式,分成多个章节。为了保证维基百科的风格,最开始一定要有一个简介段落。 +你的输出一定要保持中立的语气,避免使用主观评价性的语言。 +你应当根据提供的信息进行适度的扩展,但不可以脱离提供的信息范围。 + +你应当以中文来完成撰写 +如果涉及到其他的航天器,你应当以超链接的形式链接到对应的航天器维基百科条目。维基百科的根目录url为: /home。给到你的材料当中已经包含了其他航天器的一些url信息,你可以参考这些信息来创建超链接。如果没有相关信息,你可以自己指定一个。比如说,如果提到“奥丁一号”,你可以创建一个链接指向“/home/Odin-1”: [奥丁一号](/home/Odin-1) + +以下是提供给你的材料: +{XIHE_MAIN_WIKI_CONTEXT} + +以下是一些通用材料: +{meta_info_content} + +以下是你的任务需求: +{user_input} +""" + +message_list = [ + { + "role": "user", + "content": prompt + } +] + +# llm_resp = LLM.stream(prompt) +# full_content = '' +# for chunk in llm_resp: +# print(chunk.content, end="", flush=True) +# full_content += chunk.content + +llm_resp = LLM.invoke(prompt) +full_content = llm_resp.content + +message_list.append( + { + "role": "assistant", + "content": full_content + } +) + +print('done') diff --git a/src/OneNote_tool/__init__.py b/src/OneNote_tool/__init__.py new file mode 100644 index 0000000..e7b1c3b --- /dev/null +++ b/src/OneNote_tool/__init__.py @@ -0,0 +1,28 @@ +"""OneNote_tool package exports utilities to load OneNote documents. + +Currently exposes load_OneNote_document. +""" +import importlib + +try: + # dynamic import to get reference if available + _mod = importlib.import_module(".onenote", __package__) + _load_OneNote_document = getattr(_mod, "load_OneNote_document", None) +except Exception: + _load_OneNote_document = None + + +def load_OneNote_document(*args, **kwargs): + """Load OneNote document (wrapper). + + Uses an eagerly-imported function when available (helps some IDEs/statics); otherwise + imports the implementation lazily to avoid import-time side effects. + """ + if _load_OneNote_document is not None: + return _load_OneNote_document(*args, **kwargs) + mod = importlib.import_module(".onenote", __package__) + func = getattr(mod, "load_OneNote_document") + return func(*args, **kwargs) + + +__all__ = ["load_OneNote_document"] diff --git a/src/OneNote_tool/onenote.py b/src/OneNote_tool/onenote.py new file mode 100644 index 0000000..c30b729 --- /dev/null +++ b/src/OneNote_tool/onenote.py @@ -0,0 +1,323 @@ +"""OneNote loader utilities. + +Provides load_OneNote_document which supports: +- Loading from local exported OneNote HTML directory or single .html/.mht file. +- Loading via Microsoft Graph API when provided with "graph" source and an access token. + +This implementation keeps dependencies optional: only `requests` is required for Graph fetches. +""" +from typing import Optional, Dict, Any, List +import pathlib +import mimetypes +import logging +import re +import email +from email import policy + +logger = logging.getLogger(__name__) + +# Minimal contract: +# - input: path_or_source (str), options (dict) +# - output: Dict with keys: "type" ("local"|"graph"), "pages": list of {"title","content","id","lastModified"} + + +def _read_local_html_file(path: str) -> str: + with open(path, "rb") as f: + data = f.read() + # try to decode utf-8, fallback to latin1 + for enc in ("utf-8", "utf-8-sig", "latin-1"): + try: + return data.decode(enc) + except Exception: + continue + return data.decode("latin-1", errors="ignore") + + +def _decode_part_payload(part) -> str: + # part is an email.message.Message + raw = part.get_payload(decode=True) + if raw is None: + return "" + ch = part.get_content_charset() + if ch: + try: + return raw.decode(ch, errors="replace") + except Exception: + pass + for enc in ("utf-8", "utf-8-sig", "latin-1"): + try: + return raw.decode(enc) + except Exception: + continue + return raw.decode("latin-1", errors="replace") + + +def _extract_title_from_html(html: str) -> str: + m = re.search(r"(.*?)", html, flags=re.I | re.S) + if m: + return m.group(1).strip() + # try to find first h1/h2 as fallback + m2 = re.search(r"]*>(.*?)", html, flags=re.I | re.S) + if m2: + return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", "", m2.group(1))).strip() + return "(untitled)" + + +def _split_html_into_pages(html: str) -> List[str]: + """Heuristic split of a single HTML that may contain multiple OneNote pages. + + Tries multiple strategies in order and returns the first that yields more than one fragment: + 1. class-based:
+ 2. id-based:
+ 3. header-based: split by repeated

(or

) elements which often mark page titles + + Returns list of HTML fragments (each a standalone-ish HTML string) or single-element list if heuristics fail. + """ + # 1) class-based (existing) + pattern_class = re.compile(r"]+class=[\"'][^\"']*(?:page|onenote-page|onenotepage|oneNotePage)[^\"']*[\"'][^>]*>", flags=re.I) + matches = list(pattern_class.finditer(html)) + if len(matches) > 1: + parts = [] + head_match = re.search(r"(.*?)", html, flags=re.I | re.S) + head_html = head_match.group(0) if head_match else "" + for i, m in enumerate(matches): + start = m.start() + end = matches[i + 1].start() if i + 1 < len(matches) else len(html) + body_fragment = html[start:end] + frag = "" + head_html + "" + body_fragment + "" + parts.append(frag) + return parts + + # 2) id-based: look for divs whose id begins with 'Page' or 'page' or 'id="P' common OneNote exports + pattern_id = re.compile(r"]+id=[\"']?(?:Page|page|P)_[^\s\"'>]*[\"']?[^>]*>", flags=re.I) + matches_id = list(pattern_id.finditer(html)) + if len(matches_id) > 1: + parts = [] + head_match = re.search(r"(.*?)", html, flags=re.I | re.S) + head_html = head_match.group(0) if head_match else "" + for i, m in enumerate(matches_id): + start = m.start() + end = matches_id[i + 1].start() if i + 1 < len(matches_id) else len(html) + body_fragment = html[start:end] + frag = "" + head_html + "" + body_fragment + "" + parts.append(frag) + return parts + + # 3) header-based: split by repeated

(or

) which often denote page titles in OneNote exports + h1_matches = list(re.finditer(r"]*>", html, flags=re.I)) + if len(h1_matches) > 1: + parts = [] + head_match = re.search(r"(.*?)", html, flags=re.I | re.S) + head_html = head_match.group(0) if head_match else "" + for i, m in enumerate(h1_matches): + start = m.start() + end = h1_matches[i + 1].start() if i + 1 < len(h1_matches) else len(html) + body_fragment = html[start:end] + frag = "" + head_html + "" + body_fragment + "" + parts.append(frag) + return parts + + # fallback: no multi-page split detected + return [html] + + +def _parse_mht_bytes(data: bytes) -> List[Dict[str, Any]]: + """Parse MHT/MHTML bytes and return a list of pages (dicts with id/title/content). + + Strategy: + - Use email.parser.BytesParser to parse the multipart MHT. + - For each part with content-type text/html, treat as one page. + - If there's only a single html part but it seems to contain multiple pages, try heuristic splitting. + """ + msg = email.parser.BytesParser(policy=policy.default).parsebytes(data) + html_parts: List[str] = [] + ids: List[str] = [] + # walk parts + if msg.is_multipart(): + for part in msg.walk(): + ctype = part.get_content_type() + if ctype == "text/html": + try: + html = _decode_part_payload(part) + except Exception: + html = _decode_part_payload(part) + html_parts.append(html) + # try to get a useful id from Content-Location or Content-ID + loc = part.get("Content-Location") or part.get("Content-ID") or "" + ids.append(loc) + else: + # not multipart: try parse as text/html + ctype = msg.get_content_type() + if ctype == "text/html": + html_parts.append(_decode_part_payload(msg)) + ids.append("") + pages: List[Dict[str, Any]] = [] + if len(html_parts) == 0: + return pages + if len(html_parts) == 1: + # maybe this single html contains multiple OneNote pages; try splitting heuristically + subpages = _split_html_into_pages(html_parts[0]) + if len(subpages) > 1: + for idx, sp in enumerate(subpages): + title = _extract_title_from_html(sp) or f"page_{idx}" + pages.append({"id": ids[0] or f"mht_part_{idx}", "title": title, "content": sp}) + return pages + # otherwise single page + title = _extract_title_from_html(html_parts[0]) + pages.append({"id": ids[0] or "mht_root", "title": title, "content": html_parts[0]}) + return pages + # multiple html parts -> each is a page + for idx, html in enumerate(html_parts): + title = _extract_title_from_html(html) + pages.append({"id": ids[idx] if idx < len(ids) else f"mht_part_{idx}", "title": title, "content": html}) + return pages + + +def load_OneNote_document(path_or_source: str, options: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + """Load OneNote content. + + If `path_or_source` is a local filesystem path to a directory or file, load local HTML/MHT exports. + If `path_or_source` == "graph", options must contain `access_token` and either `notebook_id` or `site_id` + `notebook_id`. + + Returns a dict with parsed pages. + """ + options = options or {} + + p = pathlib.Path(path_or_source) + if path_or_source == "graph": + # fetch via Microsoft Graph + token = options.get("access_token") + if not token: + raise ValueError("Graph source requires 'access_token' in options") + notebook_id = options.get("notebook_id") + # simple graph pagination for pages + import requests + + headers = {"Authorization": f"Bearer {token}"} + pages = [] + if not notebook_id: + # try current user's notebooks + url = "https://graph.microsoft.com/v1.0/me/onenote/notebooks" + else: + url = f"https://graph.microsoft.com/v1.0/me/onenote/notebooks/{notebook_id}/sections" + resp = requests.get(url, headers=headers) + resp.raise_for_status() + data = resp.json() + # This is a thin implementation: collect pages by iterating sections then pages + sections = data.get("value", []) + for sec in sections: + sec_id = sec.get("id") + sec_title = sec.get("displayName") + # list pages in section + pages_url = f"https://graph.microsoft.com/v1.0/me/onenote/sections/{sec_id}/pages" + r2 = requests.get(pages_url, headers=headers) + if r2.status_code != 200: + continue + for pjson in r2.json().get("value", []): + content_url = pjson.get("contentUrl") + page_id = pjson.get("id") + title = pjson.get("title") + last_modified = pjson.get("lastModifiedDateTime") + # fetch content (HTML) + try: + c_resp = requests.get(content_url, headers=headers) + c_resp.raise_for_status() + content = c_resp.text + except Exception as e: + logger.exception("Failed to fetch OneNote page content: %s", e) + content = "" + pages.append({"id": page_id, "title": title, "content": content, "lastModified": last_modified, "section": sec_title}) + return {"type": "graph", "pages": pages} + + # Local path handling + if not p.exists(): + raise FileNotFoundError(f"Path not found: {path_or_source}") + + pages = [] + if p.is_file(): + mtype, _ = mimetypes.guess_type(p.as_posix()) + if (p.suffix.lower() == ".mht") or (mtype == "message/rfc822") or (p.suffix.lower() == ".mhtml"): + # .mht parsing: try to parse parts and split into pages + with open(p.as_posix(), "rb") as f: + data = f.read() + try: + mht_pages = _parse_mht_bytes(data) + if mht_pages: + pages.extend(mht_pages) + else: + # fallback to raw read + content = _read_local_html_file(p.as_posix()) + pages.append({"id": p.name, "title": p.stem, "content": content}) + except Exception as e: + logger.exception("Failed to parse MHT file %s: %s", p, e) + content = _read_local_html_file(p.as_posix()) + pages.append({"id": p.name, "title": p.stem, "content": content}) + else: + content = _read_local_html_file(p.as_posix()) + pages.append({"id": p.name, "title": p.stem, "content": content}) + return {"type": "local", "pages": pages} + + # directory: search for .html/.htm/.mht/.mhtml + exts = (".html", ".htm", ".mht", ".mhtml") + for fp in sorted(p.rglob("*")): + if fp.suffix.lower() in exts and fp.is_file(): + try: + if fp.suffix.lower() in (".mht", ".mhtml"): + with open(fp.as_posix(), "rb") as f: + data = f.read() + mht_pages = _parse_mht_bytes(data) + if mht_pages: + for mp in mht_pages: + pages.append({"id": fp.as_posix() + "::" + mp.get("id", ""), "title": mp.get("title"), "content": mp.get("content")}) + continue + # else fall through to raw read + content = _read_local_html_file(fp.as_posix()) + except Exception as e: + logger.exception("Failed to read file %s: %s", fp, e) + continue + # try to extract title from HTML + m = re.search(r"(.*?)", content, flags=re.I | re.S) + title = m.group(1).strip() if m else fp.stem + pages.append({"id": fp.as_posix(), "title": title, "content": content}) + return {"type": "local", "pages": pages} + + +def _safe_filename(s: str) -> str: + # create a filesystem-safe filename + s = s or "page" + s = re.sub(r"[\\\/\:\*\?\"<>\|]", "_", s) + s = re.sub(r"\s+", "_", s) + return s[:180] + + +def _maybe_debug_save(pages: List[Dict[str, Any]], options: Optional[Dict[str, Any]] = None) -> None: + """If options contains 'debug_save_dir', save each page's HTML into that directory for inspection.""" + options = options or {} + outdir = options.get("debug_save_dir") + if not outdir: + return + outp = pathlib.Path(outdir) + try: + outp.mkdir(parents=True, exist_ok=True) + except Exception: + return + for idx, page in enumerate(pages): + try: + title = page.get("title") or f"page_{idx}" + fname = f"{idx:03d}_{_safe_filename(title)}.html" + fp = outp.joinpath(fname) + with open(fp, "w", encoding="utf-8") as f: + f.write(page.get("content") or "") + except Exception: + logger.exception("Failed to write debug page %s", page.get("id")) + + +# Small smoke-run when executed directly +if __name__ == "__main__": + import sys + if len(sys.argv) > 1: + path = sys.argv[1] + else: + path = r'C:\Users\zhouyr9\Documents\2025.mht' + out = load_OneNote_document(path) + print(f"Found {len(out['pages'])} pages")