update
This commit is contained in:
zhouyr9
@@ -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')
|
||||
|
||||
@@ -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 <a:clrScheme> 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)
|
||||
@@ -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)
|
||||
|
||||
@@ -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.",
|
||||
]
|
||||
}
|
||||
]
|
||||
@@ -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')
|
||||
@@ -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"]
|
||||
@@ -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"<title>(.*?)</title>", 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"<h1[^>]*>(.*?)</h1>", 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: <div class="...page...">
|
||||
2. id-based: <div id="Page...">
|
||||
3. header-based: split by repeated <h1> (or <h2>) 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"<div[^>]+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"<head>(.*?)</head>", 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 = "<html>" + head_html + "<body>" + body_fragment + "</body></html>"
|
||||
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"<div[^>]+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"<head>(.*?)</head>", 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 = "<html>" + head_html + "<body>" + body_fragment + "</body></html>"
|
||||
parts.append(frag)
|
||||
return parts
|
||||
|
||||
# 3) header-based: split by repeated <h1> (or <h2>) which often denote page titles in OneNote exports
|
||||
h1_matches = list(re.finditer(r"<h1[^>]*>", html, flags=re.I))
|
||||
if len(h1_matches) > 1:
|
||||
parts = []
|
||||
head_match = re.search(r"<head>(.*?)</head>", 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 = "<html>" + head_html + "<body>" + body_fragment + "</body></html>"
|
||||
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"<title>(.*?)</title>", 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")
|
||||
Reference in New Issue
Block a user