278 lines
8.2 KiB
Python
278 lines
8.2 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from openpyxl import load_workbook
|
|
from sqlalchemy import delete, select
|
|
|
|
from app.extensions import db
|
|
from app.models import Asset, AssetLogEntry
|
|
|
|
|
|
EVENT_LINE_PATTERN = re.compile(r"^(\d{4}-\d{2}-\d{2})\s+(.+)$")
|
|
LOG_BOOK_IGNORED_SHEETS = {"Overview", "Model"}
|
|
|
|
BODY_LOCATION_KEYWORDS = [
|
|
("earth-sun l2", "Earth-Sun L2"),
|
|
("earth-sun lagrange point 2", "Earth-Sun L2"),
|
|
("star port", "Star Port"),
|
|
("iapetus", "Iapetus"),
|
|
("callisto", "Callisto"),
|
|
("ganymede", "Ganymede"),
|
|
("europa", "Europa"),
|
|
("jupiter", "Jupiter"),
|
|
("saturn", "Saturn"),
|
|
("neptune", "Neptune"),
|
|
("mars", "Mars"),
|
|
("venus", "Venus"),
|
|
("mercury", "Mercury"),
|
|
("lunar orbit", "Lunar Orbit"),
|
|
("leo", "LEO"),
|
|
("earth orbit", "Earth"),
|
|
("earth", "Earth"),
|
|
("sun", "Sun"),
|
|
]
|
|
|
|
LOG_BOOK_ASSET_SPECS = {
|
|
"ST-01": {
|
|
"name": "万星源号 ST-01",
|
|
"asset_type": "Exploration Mothership",
|
|
"program": "Stellaria",
|
|
"home_region": "LEO",
|
|
"note": "First Class of Stellaria",
|
|
},
|
|
"XH-01": {
|
|
"name": "羲和号 XH-01",
|
|
"asset_type": "Exploration Mothership",
|
|
"program": "羲和计划",
|
|
"home_region": "LEO",
|
|
"note": "羲和计划首舰",
|
|
},
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class LogBookSheetSummary:
|
|
sheet_name: str
|
|
asset_name: str
|
|
state_entries: int = 0
|
|
event_entries: int = 0
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"sheet_name": self.sheet_name,
|
|
"asset_name": self.asset_name,
|
|
"state_entries": self.state_entries,
|
|
"event_entries": self.event_entries,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class LogBookImportSummary:
|
|
workbook: str
|
|
imported_assets: list[LogBookSheetSummary] = field(default_factory=list)
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"workbook": self.workbook,
|
|
"imported_assets": [item.to_dict() for item in self.imported_assets],
|
|
}
|
|
|
|
|
|
def import_log_book_data(workbook_path: str | Path) -> LogBookImportSummary:
|
|
path = Path(workbook_path)
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"Workbook not found: {path}")
|
|
|
|
workbook = load_workbook(path, data_only=True, read_only=True)
|
|
summary = LogBookImportSummary(workbook=path.name)
|
|
|
|
try:
|
|
for sheet_name in _iter_importable_sheet_names(workbook.sheetnames):
|
|
asset_spec = _resolve_asset_spec(sheet_name)
|
|
sheet_summary = _import_sheet(workbook[sheet_name], sheet_name, asset_spec)
|
|
summary.imported_assets.append(sheet_summary)
|
|
db.session.commit()
|
|
except Exception:
|
|
db.session.rollback()
|
|
raise
|
|
|
|
return summary
|
|
|
|
|
|
def _iter_importable_sheet_names(sheet_names: list[str]) -> list[str]:
|
|
return [sheet_name for sheet_name in sheet_names if sheet_name not in LOG_BOOK_IGNORED_SHEETS]
|
|
|
|
|
|
def _resolve_asset_spec(sheet_name: str) -> dict[str, str | None]:
|
|
asset_spec: dict[str, str | None] = {
|
|
"name": sheet_name,
|
|
"asset_type": "Vehicle",
|
|
"program": None,
|
|
"home_region": None,
|
|
"note": f"Imported from log_book.xlsx sheet {sheet_name}.",
|
|
}
|
|
asset_spec.update(LOG_BOOK_ASSET_SPECS.get(sheet_name, {}))
|
|
return asset_spec
|
|
|
|
|
|
def _import_sheet(worksheet: object, sheet_name: str, asset_spec: dict[str, str]) -> LogBookSheetSummary:
|
|
asset = db.session.execute(select(Asset).where(Asset.name == asset_spec["name"])).scalar_one_or_none()
|
|
if asset is None:
|
|
asset = Asset(
|
|
name=asset_spec["name"],
|
|
asset_type=asset_spec["asset_type"],
|
|
program=asset_spec["program"],
|
|
home_region=asset_spec["home_region"],
|
|
note=asset_spec["note"],
|
|
)
|
|
db.session.add(asset)
|
|
db.session.flush()
|
|
|
|
asset.asset_type = asset_spec["asset_type"]
|
|
asset.program = asset_spec["program"]
|
|
asset.home_region = asset_spec["home_region"]
|
|
asset.note = asset_spec["note"]
|
|
|
|
db.session.execute(delete(AssetLogEntry).where(AssetLogEntry.asset_id == asset.id))
|
|
db.session.flush()
|
|
|
|
summary = LogBookSheetSummary(sheet_name=sheet_name, asset_name=asset.name)
|
|
for row in worksheet.iter_rows(min_row=2, values_only=True):
|
|
if not any(value not in (None, "") for value in row):
|
|
continue
|
|
|
|
log_name = _clean_text(row[0])
|
|
start_at = _coerce_datetime(row[1])
|
|
end_at = _coerce_datetime(row[2])
|
|
mission_detail = _clean_text(row[3])
|
|
sub_log = _clean_text(row[4])
|
|
explicit_location = _clean_text(row[7])
|
|
inferred_location = _infer_location(log_name, mission_detail, sub_log, explicit_location)
|
|
|
|
if log_name is None or start_at is None:
|
|
continue
|
|
|
|
db.session.add(
|
|
AssetLogEntry(
|
|
asset_id=asset.id,
|
|
entry_kind="state",
|
|
title=log_name,
|
|
state_label=log_name,
|
|
mission_label=log_name,
|
|
location=inferred_location,
|
|
start_at=start_at,
|
|
end_at=end_at,
|
|
summary=mission_detail,
|
|
note=sub_log,
|
|
)
|
|
)
|
|
summary.state_entries += 1
|
|
|
|
return summary
|
|
|
|
|
|
def _clean_text(value: object) -> str | None:
|
|
if value in (None, ""):
|
|
return None
|
|
cleaned = str(value).strip()
|
|
return cleaned or None
|
|
|
|
|
|
def _coerce_datetime(value: object) -> datetime | None:
|
|
if value in (None, ""):
|
|
return None
|
|
if isinstance(value, datetime):
|
|
normalized = value
|
|
else:
|
|
normalized = datetime.fromisoformat(str(value))
|
|
if normalized.tzinfo is None:
|
|
return normalized.replace(tzinfo=timezone.utc)
|
|
return normalized.astimezone(timezone.utc)
|
|
|
|
|
|
def _infer_location(
|
|
log_name: str | None,
|
|
mission_detail: str | None,
|
|
sub_log: str | None,
|
|
explicit_location: str | None,
|
|
) -> str | None:
|
|
if explicit_location:
|
|
return explicit_location
|
|
|
|
haystacks = [
|
|
(log_name or "").casefold(),
|
|
(mission_detail or "").casefold(),
|
|
(sub_log or "").casefold(),
|
|
]
|
|
for keyword, location in BODY_LOCATION_KEYWORDS:
|
|
if any(keyword in haystack for haystack in haystacks):
|
|
return location
|
|
|
|
if mission_detail and "transfer" in mission_detail.casefold():
|
|
return "Sun"
|
|
if log_name and "maintenance" in log_name.casefold():
|
|
return "LEO"
|
|
if log_name and "construction" in log_name.casefold():
|
|
return "LEO"
|
|
return None
|
|
|
|
|
|
def _infer_event_location(description: str, default_location: str | None) -> str | None:
|
|
lowered = description.casefold()
|
|
for keyword, location in BODY_LOCATION_KEYWORDS:
|
|
if keyword in lowered:
|
|
return location
|
|
return default_location
|
|
|
|
|
|
def _parse_sub_log_events(
|
|
asset_id: object,
|
|
parent_title: str,
|
|
parent_start_at: datetime,
|
|
parent_end_at: datetime | None,
|
|
default_location: str | None,
|
|
sub_log: str | None,
|
|
) -> list[AssetLogEntry]:
|
|
if sub_log is None:
|
|
return []
|
|
|
|
events = []
|
|
for raw_line in sub_log.splitlines():
|
|
line = raw_line.strip()
|
|
if not line:
|
|
continue
|
|
match = EVENT_LINE_PATTERN.match(line)
|
|
if match is None:
|
|
continue
|
|
|
|
event_at = _coerce_datetime(match.group(1))
|
|
if event_at is None:
|
|
continue
|
|
|
|
# Skip obvious workbook typos that fall far outside the parent interval.
|
|
if event_at < parent_start_at.replace(year=parent_start_at.year - 1):
|
|
continue
|
|
if parent_end_at is not None and event_at > parent_end_at.replace(year=parent_end_at.year + 1):
|
|
continue
|
|
|
|
description = match.group(2).strip()
|
|
events.append(
|
|
AssetLogEntry(
|
|
asset_id=asset_id,
|
|
entry_kind="event",
|
|
title=description,
|
|
state_label=None,
|
|
mission_label=parent_title,
|
|
location=_infer_event_location(description, default_location),
|
|
start_at=event_at,
|
|
end_at=None,
|
|
summary=parent_title,
|
|
note=line,
|
|
)
|
|
)
|
|
|
|
return events |